Практические советы: Ваша рамка для ИИ-агентов, скорее всего, не подходит. Вот как это исправить
Пошаговое руководство по практическим рекомендациям: ваша схема ИИ-агента, скорее всего, неподходящая. Вот как это исправить: контракты, проверки и готовые блоки кода для команд, использующих эту модель.
В этом руководстве пошагово показан путь от сырья до рабочей системы для следующего случая: ваша платформа ИИ-агентов, скорее всего, не подходит — вот как правильно её выбрать. Основное внимание уделяется практическим шагам, четким проверкам и коду, который можно просто добавить в репозиторий без необходимости угадывать намерения авторов. На этапе обзора необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не пытаясь угадать скрытое состояние системы. Рядом с функциональными результатами следует записывать время выполнения и стоимость токенов или запросов. Отображение затрат с самого начала помогает избежать неожиданных расходов при переходе от демо-версии к общедоступным средам.
Вопрос, который все задают наоборот
На этапе решения вопроса «Что спрашивают все», сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками окружения, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Создавайте контрольные точки после дорогостоящих операций. Функция возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг.
Ось 1: Насколько детерминированным должно быть ваше разветвление?
При работе над этапом «Насколько детерминировано?» оси 1 сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки со стороны человека и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Создавайте контрольные точки после дорогостоящих операций. Механизм возобновления работы не должен повторно оплачивать один и тот же вызов большой языковой модели, когда оператор пытается выполнить последующий этап.
# A branch where non-determinism is FINE — picking a tone for a summary email.
# If the agent occasionally phrases things slightly differently, nobody's paged.
def draft_summary_tone(context: dict) -> str:
return llm_call(
prompt=f"Summarize this incident in a {context['audience']}-appropriate tone.",
temperature=0.7, # variability here is a feature, not a bug
)
# A branch where non-determinism is NOT fine — deciding whether to page a human
# at 4am versus auto-remediating. This must be code, not a prompt.
def route_alert(alert: dict) -> str:
if alert["severity"] == "critical" and alert["service"] in PAGE_ALWAYS_SERVICES:
return "page_oncall"
if alert["auto_remediation_available"] and alert["confidence"] > 0.9:
return "auto_remediate"
if alert["severity"] == "critical":
return "page_oncall"
return "log_and_monitor"
Ось 2: Сколько времени существует одна единица работы?
При работе над этапом «Сколько времени» в Axis 2 сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые модули вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Вносите контрольные точки после дорогостоящих шагов. Механизм возобновления работы не должен повторно взимать плату за один и тот же вызов LLM при повторной попытке обработки последующего узла.
# Short-lived: starts and finishes inside one HTTP request.
# This is the "no framework needed" zone — a framework here is pure overhead.
async def handle_summarize_request(request: SummarizeRequest) -> SummarizeResponse:
text = await fetch_document(request.doc_id)
summary = await llm_summarize(text, max_tokens=300)
return SummarizeResponse(summary=summary)
# Long-lived: this alert might sit in "awaiting human ack" for six hours
# while the on-call engineer is asleep, then resume on a completely
# different process after a deploy rotated the pods underneath it.
class AlertTriageWorkflow:
async def run(self, alert: dict) -> dict:
decision = await self.classify_and_route(alert)
if decision == "page_oncall":
await self.page(alert)
await self.wait_for_ack(timeout_hours=1) # this line is the whole ballgame
...
При работе над этапом «Сколько времени» в Axis 2 сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Записывайте время выполнения, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам.
Ось 3: Что происходит, если шаг выполняется дважды?
Этап «Что происходит» на Оси 3 работает лучше всего, если рассматривать его как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма задачи. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф. Сохраняйте состояние графа простым и типизированным. Вложенные структуры данных скрывают информацию о том, какой узел записал какое поле, и мешают возобновлению работы после прерываний.
# BEFORE — looks fine in a demo, is a live incident waiting to happen
async def auto_remediate(alert: dict):
await restart_service(alert["service"]) # what if this activity gets retried?
# AFTER — idempotent by construction
async def auto_remediate(alert: dict, idempotency_key: str):
if await remediation_ledger.already_applied(idempotency_key):
logger.info("remediation already applied, skipping", key=idempotency_key)
return await remediation_ledger.get_result(idempotency_key)
result = await restart_service(alert["service"])
await remediation_ledger.record(idempotency_key, result)
return result
Ось 4: Кто нуждается в прочтении решения позже и в какой форме?
Axis 4, который требует настройки на этапе разработки, лучше всего работает, если рассматриваться как измеримая структура. Соберите один пример успешной работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно путь успешного выполнения задачи и путь восстановления после сбоя. Повторные попытки, проверки со стороны человека и обработка неработоспособных сообщений являются частью продукта, а не последующими доработками. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
# A framework-agnostic audit record — this is what actually matters
# in a postmortem, regardless of what orchestrated the steps.
@dataclass
class DecisionRecord:
alert_id: str
timestamp: float
step: str
reasoning: str # what the LLM said, verbatim
decision: str # the structured outcome, not prose
confidence: float | None
human_override: bool
async def log_decision(record: DecisionRecord):
await audit_store.insert(record)
# Also emit as a structured log line — cheap insurance for when
# the audit store itself is the thing that's down during an incident.
logger.info("agent_decision", **asdict(record))
Axis 5: В чём заключается реальное ограничение скорости работы вашей команды?
Этап Axis 5 What’s работает наилучшим образом, когда рассматривается как измеримая поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Предпочитайте небольшие, тестируемые единицы вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний. Этап Axis 5 What’s работает наилучшим образом, когда рассматривается как измеримая поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.
# Week-one prototype: prove the concept fast, accept the debt knowingly.
from crewai import Agent, Task, Crew
triage_agent = Agent(role="Alert Triage", goal="Decide how to handle infra alerts")
crew = Crew(agents=[triage_agent], tasks=[Task(description="Triage: {alert}", agent=triage_agent)])
crew.kickoff(inputs={"alert": alert_payload})
Axis 6: Каковы ваши бюджеты на задержку и затраты для каждого решения?
Для этапа Axis 6 What’s необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф. Включайте утверждение человека для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты охвата бизнес-логики.
# Expensive pattern: every routing decision is its own LLM call,
# multiplied across a multi-agent conversation with several turns.
# At alert volumes (hundreds/day, sometimes bursts of thousands during
# a real incident), this is a real line item, not a rounding error.
async def route_via_llm(alert: dict) -> str:
return await llm_call(f"How should we handle this alert? {alert}")
# Cheaper, faster, and more auditable: cheap deterministic pre-filtering
# in code, LLM reserved for genuinely ambiguous cases.
async def route_alert_efficiently(alert: dict) -> str:
if alert["service"] in KNOWN_NOISY_SERVICES and alert["severity"] == "low":
return "log_and_monitor" # zero LLM calls for the common case
if alert["signature"] in KNOWN_REMEDIATION_PLAYBOOK:
return "auto_remediate" # deterministic lookup, zero LLM calls
return await llm_call(f"Novel alert, needs judgment: {alert}") # LLM only when genuinely needed
Собираем всё вместе: путь принятия решений, а не дерево принятия решений
На этапе сборки необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо одновременно задокументировать успешный и аварийный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Внедрять утверждение человеком для операций, связанных с тратой денег или изменением производственных данных. Подключение компонентов во время компиляции не гарантирует полноты функционала продукта.
Is this unit of work stateless and finishes in seconds?
└─ YES → skip the framework entirely. Plain functions + retries. Ship it.
└─ NO, continue.
Does it need to survive process restarts / wait on humans for hours-to-days?
└─ YES → you need durable execution (Temporal or equivalent) as the backbone,
regardless of what else you pick for the reasoning layer.
└─ NO, continue.
Are the important branches safety- or compliance-critical
(money, infra changes, irreversible external actions)?
└─ YES → LangGraph-style explicit graphs, keep LLM scoped to narrow nodes.
└─ NO, mostly exploratory/creative → CrewAI or AutoGen are legitimate defaults.
Is this still a prototype whose findings might get thrown away?
└─ YES → optimize for speed of iteration over long-term correctness,
but write down when you'll revisit that tradeoff.
@activity.defn
async def classify_alert_activity(alert: dict) -> dict:
# LangGraph-style graph runs here — bounded reasoning, deterministic routing —
# inside an activity Temporal will retry and time-box like any other side effect.
result = alert_triage_graph.invoke({"alert": alert, "audit_log": []})
return {"decision": result["decision"], "confidence": result["confidence"]}
@workflow.defn
class AlertTriageWorkflow:
def __init__(self):
self._acked = False
@workflow.signal
async def acknowledge(self):
self._acked = True
@workflow.run
async def run(self, alert: dict) -> dict:
classification = await workflow.execute_activity(
classify_alert_activity, alert,
start_to_close_timeout=timedelta(seconds=20),
retry_policy=workflow.RetryPolicy(maximum_attempts=3),
)
if classification["decision"] == "page_oncall":
await workflow.execute_activity(page_oncall, alert, start_to_close_timeout=timedelta(seconds=10))
await workflow.wait_condition(lambda: self._acked, timeout=timedelta(hours=1))
if not self._acked:
await workflow.execute_activity(escalate_to_secondary, alert, start_to_close_timeout=timedelta(seconds=10))
elif classification["decision"] == "auto_remediate":
await workflow.execute_activity(
auto_remediate, alert, f"remediate-{alert['id']}",
start_to_close_timeout=timedelta(minutes=2),
retry_policy=workflow.RetryPolicy(maximum_attempts=2),
)
return {"alert_id": alert["id"], "decision": classification["decision"]}
Частые ошибки, с которыми вы постоянно сталкиваетесь
Чтобы избежать распространённых ошибок, необходимо заранее определить этапы, входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на сложную цепочку операций. Внедряйте человеческое утверждение для операций, связанных с тратой денег или изменением производственных данных. Настройка на этапе компиляции не гарантирует полноты бизнес-процесса. Чтобы избежать распространённых ошибок, необходимо заранее определить этапы, входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на ранних этапах предотвращает неожиданные счёты при переходе от демо-версии к общей среде.
Фактический ответ
На этапе определения фактического ответа сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Храните конфигурацию отдельно от кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять без необходимости просмотра всей структуры. Создавайте контрольные точки после дорогостоящих операций. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг.
Чек-лист операционной деятельности
Этап чек-листа операционной деятельности работает наилучшим образом, если рассматривать его как измеримую основу. Соберите один эталонный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия всем элементам, определите критерии успеха и не допускайте безответственного частичного выполнения задач.
Сохраняйте состояние графа в виде плоской структуры с явным типированием. Вложенные объекты скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
При наличии бюджета добавляйте тесты для проверки критического пути в процессе интеграционного тестирования с использованием фикстур, а не реальных платных API.
Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее помогает избежать неожиданных счетов при переходе с демо-среды в общедоступные среды.
Сохраняйте состояние графа в виде плоской структуры с явным типированием. Вложенные объекты скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
Перед внедрением новой стековой архитектуры заморозьте версии, сохраните эталонный вариант работы критического пути и уточните шаги для возврата к предыдущей версии. В общедоступных средах необходимы ограничения на количество запросов, проверки принадлежности пользователя и четко определенный ответственный за обновление секретов. Лучше выбирать надежность, чем креативные одноразовые демонстрации.
Примечание к пакету 72c003459fd6: не включайте ключи поставщиков в репозиторий, установите лимит токенов на сессию и храните транскрипции рядом с фиксами для оценки, чтобы последующие замены моделей оставались сопоставимыми.