Практичні поради: ваша рамка для AI-агентів, ймовірно, не підходить. Ось як це виправити

У цьому посібнику описано процес створення системи від сировини до готового продукту для: вашої рамки AI Agent, яка, ймовірно, не підходить – ось як правильно її обрати. Основна увага приділяється конкретним крокам виконання, чітким перевіркам та коду, який можна просто додати до репозиторію без необхідності здогадуватися щодо його призначення. На етапі огляду необхідно визначити вхідні дані, виконавця кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись здогадатися про прихований стан системи. Реєструйте час виконання та витрати на токени чи запити разом із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ.

Питання, яке всі ставлять навпаки

Під час роботи на етапі «Запитання, яке ставить кожен», спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретних даних та флаги функцій мають знаходитися в одному місці, щоб оператори могли їх перевіряти, не читаючи весь код. Робіть контрольні пункти після дорогих операцій. Функція відновлення не повинна знову стягувати плату за той самий виклик LLM, коли оператор перезапускає пізнішу ланку.

Ось 1: Наскільки детермінованим має бути ваш розгалужений алгоритм?

Під час роботи над етапом «Яким чином визначальний процес?» (Axis 1 How deterministic) спочатку запишіть умови виконання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допоможе зберегти чесність подальших змін у коді. Одночасно задокументуйте шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Робіть контрольні позначки після дорогих кроків. Функція відновлення не повинна знову стягувати плату за той самий виклик LLM, коли оператор намагається знову виконати пізнішу операцію.

# 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"

Axis 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"]}

Поширені помилки, які ви постійно бачите

Щодо поширених помилок під час роботи на певній стадії, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Краще використовувати невеликі, тестовані одиниці коду замість об’ємних скриптів. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану структуру процесу. Встановлюйте людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення елементів під час компіляції не є гарантією повноти бізнес-функціоналу. Щодо поширених помилок під час роботи на певній стадії, необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Відображення витрат на ранньому етапі запобігає несподіваним рахункам під час переходу від демо-версії до спільного середовища.

Кроки.

Фактична відповідь

Під час роботи на етапі «Фактична відповідь» спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність подальших змін у коді. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Робіть контрольні позначки після дорогих операцій. Функція відновлення не повинна знову стягувати плату за той самий виклик LLM, коли оператор перезапускає пізнішу операцію.

Чек-лист для експлуатації

Етап «Чек-лист для експлуатації» працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок невдачі та примітку про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового завершення роботи.

Зберігайте стан графа у вигляді плоскої структури з визначеними типами даних. Вкладені блоки приховують інформацію про те, який вузол заповнив певне поле, і ускладнюють продовження роботи після перерв.

Додайте тест на базову функціональність, який перевіряє критичний шлях у процесі інтеграційного тестування за допомогою фікстур, а не реальних платних API, коли це дозволяють бюджетні обмеження.

Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Відомі заздалегідь витрати запобігають несподіваним рахункам під час переходу з демо-середовища у спільні середовища.

Зберігайте стан графа у вигляді плоскої структури з визначеними типами даних. Вкладені блоки приховують інформацію про те, який вузол заповнив певне поле, і ускладнюють продовження роботи після перерв.

Перш ніж піднімати стек на вищий рівень, заморозьте версії, створіть остаточний запис дій для критичного шляху та підтвердьте кроки для скасування змін. У спільних середовищах необхідні обмеження на кількість запитів, перевірки прав доступу та чіткий власник для зміни секретів. Віддавайте перевагу надійності перед креативними одноразовими демонстраціями.

Примітка до пакету 72c003459fd6: не включайте ключі постачальників у репозиторій, встановіть ліміт токенів на сеанс та зберігайте транскрипції поруч із фіксами для оцінки, щоб подальша заміна моделей залишалася порівнянною.