Agent Frameworks Reason, Temporal Survives: Устойчивая выполнение для агентов

Инженеры, разрабатывающие агентные системы в 2026 году, отвечали на вопросы о используемых ими фреймворках, и почти все упомянули LangGraph, CrewAI, AutoGen или какую-то их комбинацию. Ни один из этих инструментов не является плохим. Проблема в том, что вся дискуссия ведется на неверном уровне архитектуры.

В производственных агентных системах наблюдается повторяющийся паттерн. То, что будит инженера дежурной смены в 3 часа ночи, редко связано с фреймворком, определяющим логику работы агента. Почти всегда речь идет о одной группе проблем: процесс завершается на середине рабочего процесса, шаг выполняется дважды из-за отсутствия идемпотентности повторной попытки, извлечение пода прерывает длительно выполняемую задачу агента на 40-й минуте без возможности ее возобновления, или состояние рабочего процесса хранится в переменной Python, которая исчезает при пересоздании контейнера.

Ни LangGraph, ни CrewAI, ни AutoGen не обрабатывают всю эту цепочку процессов полностью, и это сделано намеренно, поскольку они предназначены для координации вызовов больших языковых моделей и использования инструментов. Системы производственных агентов действительно терпят неудачу из-за отсутствия надежной реализации — это гораздо более старая и менее интересная область, существующая примерно на десять лет раньше появления термина «агент». Temporal, система, разработавшая этот подход, практически не упоминается в дискуссиях о фреймворках агентов, однако тихо стала основой для всё большего числа серьезных реализаций агентов. Далее рассматривается, в чем сильные стороны трех популярных фреймворков, где проявляются ограничения каждой из абстракций и почему несовременный четвертый вариант заслуживает внимания со стороны бизнеса.

LangGraph

Вклад LangGraph заключается в преобразовании агента в явную графическую структуру, описывающую типизированные состояния, вместо косвенной последовательности запросов. Это имеет реальную ценность, поскольку поток управления становится поддающимся анализу и тестированию. Например, рабочий процесс сортировки может выглядеть следующим образом:

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class TriageState(TypedDict):
    ticket_id: str
    category: Literal["billing", "technical", "abuse", "unknown"]
    priority: int
    assigned_queue: str | None

def classify_node(state: TriageState) -> TriageState:
    category = llm_classify(state["ticket_id"])  # the only probabilistic step
    return {**state, "category": category}

def route_node(state: TriageState) -> str:
    # deterministic, testable, no LLM involved
    return {
        "billing": "billing_queue",
        "technical": "tech_queue",
        "abuse": "trust_and_safety_queue",
        "unknown": "human_review",
    }[state["category"]]

graph = StateGraph(TriageState)
graph.add_node("classify", classify_node)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_node)
app = graph.compile()

Это хорошая идея, текст написан четко. Проблемы возникают тогда, когда что-то должно существовать дольше одного вызова invoke() и соответствующего контрольного пункта. Подумайте о координации между сервисами, рабочих процессах, длительностью которых составляет несколько дней, или этапах выполнения, над которыми LangGraph не имеет контроля: задачах в Spark, очередях утверждений, проходящих через Slack и возвращающихся спустя несколько дней, или сервисах, которые другая команда развертывает по собственному графику. Механизм контрольных пунктов действительно работает, но он охватывает только собственную модель выполнения LangGraph. После передачи задачи инфраструктуре, не связанной с большими языковыми моделями, надежность становится чем-то, что необходимо реализовывать вручную.

CrewAI

Многопроцессную команду можно определить всего за несколько строк:

from crewai import Agent, Task, Crew

researcher = Agent(role="Research Analyst", goal="Find relevant market data", tools=[search_tool])
writer = Agent(role="Report Writer", goal="Turn research into a client-ready summary")

crew = Crew(agents=[researcher, writer], tasks=[
    Task(description="Research {company}'s Q2 earnings", agent=researcher),
    Task(description="Write a 3-paragraph summary for the client", agent=writer),
])
crew.kickoff(inputs={"company": "Acme Corp"})

Нет никакого инструмента, который мог бы превратить неопределенную идею о рабочем процессе с участием нескольких агентов в работающую демонстрацию за двадцать минут быстрее, чем CrewAI. Однако последствия этого ощущаются позже. Функция kickoff() не позволяет приостановить процесс на часы или дни, отсутствует механизм повторных попыток, кроме того, что предусмотрено для отдельных инструментов, а стабильное состояние сохраняется только тогда, когда это делает разработчик. CrewAI координирует работу больших языковых моделей, но не является движком рабочих процессов, и его никогда не рекламировали как таковой, хотя многие команды случайно начинают использовать его именно в этом качестве.

AutoGen

Минимальная конфигурация AutoGen включает помощника и прокси пользователя:

import autogen

assistant = autogen.AssistantAgent(name="assistant", llm_config={"config_list": config_list})
user_proxy = autogen.UserProxyAgent(name="user_proxy", human_input_mode="NEVER")
user_proxy.initiate_chat(assistant, message="Summarize this quarter's incident reports.")

Конверсационный, эмерджентный стиль AutoGen идеально подходит для исследовательской работы. Его версия 2026 года, оснащенная более структурированным рабочим инструментарием и шаблонами, основанными на событиях, даже склонна к разделению процессов оркестрации и самой конверсации. При этом основой по-прежнему остается транскрипт сообщений, а не надежный записанный отчет о выполнении операций. Когда процесс, в котором происходит конверсация, прерывается, остается лишь частичный лог чата, а не вычисления, которые можно было бы возобновить.

Общая проблема

Все три решения хорошо справляются с новыми задачами: заставлением больших языков моделей рассуждать, использовать инструменты и взаимодействовать с другими агентами, управляемыми большими языковыми моделями. Однако старая, более простая проблема — обеспечение корректного завершения многоэтапных процессов независимо от сбоев, перезагрузок, истечения времени или других препятствий — не решается ими. Это не их вина; ожидать этого от них — значит совершать категориальную ошибку.

Подумайте о реальном рабочем процессе, а не о демонстрации: сотрудник, который сверяет счета-фактуры поставщиков. Он извлекает элементы счета из PDF, сравнивает их с заказом на покупку в системе ERP, отмечает расхождения, превышающие установленный порог, для дальнейшего рассмотрения человеком, а затем может ждать ответа несколько дней перед тем, как внести утвержденные данные в бухгалтерию. Если какой-то шаг проваливается, его необходимо повторить без двойной записи. Перезапуск в процессе выполнения задачи не должен приводить к началу работы с нуля или к незаметному исчезновению данных.

┌────────────┐     ┌───────────────┐     ┌─────────────────┐     ┌───────────────┐
│  Extract   │ --> │ Match against │ --> │  Human approval │ --> │  Post to ERP  │
│  line items│     │ purchase order│     │  (can take days)│     │  (idempotent) │
└────────────┘     └───────────────┘     └─────────────────┘     └───────────────┘
      ▲                                          │
      └───────────── retry on failure ───────────┘

Никто не создавал LangGraph, CrewAI или AutoGen с целью гарантировать, что данная диаграмма останется актуальной после перезапуска контейнера на третий день, пока кто-то ещё рассматривает пункт 14. Речь здесь не идёт об оркестрации больших языковых моделей. Речь идёт о надёжной выполнимости, причём системы, которые справляются с этой задачей хорошо — в первую очередь Temporal — уже много лет обеспечивают выполнение неприметной работи на втором уровне: обработку платежей, выполнение заказов и микросервисы по шаблону saga, задолго до того, как термин «агент» стал использоваться для описания подобных программ.

Как работает Temporal

Модель Temporal кажется на первый взгляд удивительно простой. Рабочий процесс описывается в виде функции обычного кода, содержащего циклы, условные операторы и вызовы функций, причем Temporal обеспечивает сохранение истории выполнения этой функции. Если процесс сбоит в любой момент, Temporal перезапускает функцию на основе сохраненной истории событий до момента сбоя и продолжает работу детерминированно, без повторного выполнения побочных эффектов. Секрет заключается в том, что побочные эффекты, такие как вызов LLM, обращение к API или запись в базу данных, реализуются в отдельных функциях, называемых активностями. Temporal перепробует их независимо с настраиваемыми интервалами повторных попыток, в то время как функция рабочего процесса лишь координирует их.

Агент для согласования счетов ниже рассматривает вызов LLM как одну из нескольких активностей. Во-первых, идут активности — те части, которые взаимодействуют с внешним миром:

# activities.py — the parts that touch the outside world (LLMs, APIs, DBs)
from temporalio import activity
from dataclasses import dataclass

@dataclass
class Invoice:
    invoice_id: str
    vendor: str
    line_items: list[dict]

@activity.defn
async def extract_line_items(pdf_url: str) -> list[dict]:
    # calls a vision-capable LLM, or a dedicated OCR model — doesn't matter to the workflow
    text = await download_and_ocr(pdf_url)
    return await llm_extract_structured(text, schema=LINE_ITEM_SCHEMA)

@activity.defn
async def match_purchase_order(invoice: Invoice) -> dict:
    po = await erp_client.get_purchase_order(invoice.vendor)
    discrepancies = diff_line_items(invoice.line_items, po["line_items"])
    return {"po_id": po["id"], "discrepancies": discrepancies}
@activity.defn
async def post_to_erp(invoice_id: str, po_id: str) -> dict:
    # idempotency is enforced with a deterministic key derived from workflow inputs,
    # not a random UUID generated per attempt — this is the same lesson as always
    return await erp_client.post_journal_entry(
        idempotency_key=f"invoice-post-{invoice_id}",
        invoice_id=invoice_id,
        po_id=po_id,
    )

@activity.defn
async def notify_human_reviewer(invoice_id: str, discrepancies: list[dict]) -> None:
    await slack_client.post_message(
        channel="#ap-review",
        text=f"Invoice {invoice_id} has {len(discrepancies)} discrepancies needing review.",
    )

Далее идет рабочий процесс, который берет на себя роль оркестрации, которую обычно выполняли бы LangGraph или CrewAI, причем его надежность встроена и не зависит от использования чекпойнтов:

# workflows.py
from temporalio import workflow
from datetime import timedelta

@workflow.defn
class InvoiceReconciliationWorkflow:
    def __init__(self) -> None:
        self._approval_received = False
        self._approved = False

    @workflow.signal
    async def submit_human_decision(self, approved: bool) -> None:
        # A signal is how the outside world talks to a running, possibly-days-old workflow
        self._approval_received = True
        self._approved = approved

    @workflow.run
    async def run(self, pdf_url: str, invoice_id: str, vendor: str) -> dict:
        line_items = await workflow.execute_activity(
            extract_line_items, pdf_url,
            start_to_close_timeout=timedelta(minutes=2),
            retry_policy=workflow.RetryPolicy(maximum_attempts=3),
        )

        invoice = Invoice(invoice_id=invoice_id, vendor=vendor, line_items=line_items)
        match_result = await workflow.execute_activity(
            match_purchase_order, invoice,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=workflow.RetryPolicy(maximum_attempts=5),
        )

        if match_result["discrepancies"]:
            await workflow.execute_activity(
                notify_human_reviewer, invoice_id, match_result["discrepancies"],
                start_to_close_timeout=timedelta(seconds=10),
            )
            # This is the line that would be genuinely hard to write safely in the
            # other three frameworks: wait, durably, for up to 7 days, for a signal
            # that might arrive from a Slack button, a web form, anything —
            # while the workflow's entire state sits safely in Temporal's history,
            # costing zero compute while it waits.
            await workflow.wait_condition(
                lambda: self._approval_received,
                timeout=timedelta(days=7),
            )
            if not self._approved:
                return {"status": "rejected", "invoice_id": invoice_id}

        result = await workflow.execute_activity(
            post_to_erp, invoice_id, match_result["po_id"],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=workflow.RetryPolicy(maximum_attempts=5),
        )
        return {"status": "posted", "invoice_id": invoice_id, "erp_result": result}

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

Семидневное ожидание бесплатно и выдерживает любые ситуации. workflow.wait_condition не удерживает открытыми ни потоки, ни процессы, ни контейнеры в течение недели. Temporal хранит состояние рабочего процесса в собственном надежном хранилище, поэтому в это время рабочий процесс может быть перезапущен, сокращен до нуля или полностью завершить работу. Когда сигнал наконец поступает, Temporal воспроизводит историю событий и возобновляет работу именно на том месте, где было выполнено await. Попытка сделать то же самое с чекпоинтером LangGraph или группой CrewAI по сути означала бы пересоздание модели исходя из событий Temporal.

Попытки повторения выполняются автоматически, имеют узкую сферу применения и никогда не повторяют уже завершённую работу. Предположим, что функция post_to_erp срабатывает с ошибкой после успешного выполнения match_purchase_order. Temporal не будет повторно запускать match_purchase_order, поскольку её результат уже сохранён в постоянной истории. Сравните это с наивным подходом к той же идее:

# BEFORE — naive retry loop, the kind you write at 11pm under deadline pressure
async def reconcile_invoice_naive(pdf_url, invoice_id, vendor):
    for attempt in range(3):
        try:
            line_items = await extract_line_items(pdf_url)
            match_result = await match_purchase_order(Invoice(invoice_id, vendor, line_items))
            if match_result["discrepancies"]:
                await notify_human_reviewer(invoice_id, match_result["discrepancies"])
                approved = await wait_for_approval_somehow(invoice_id)  # good luck
                if not approved:
                    return {"status": "rejected"}
            return await post_to_erp(invoice_id, match_result["po_id"])
        except Exception:
            continue  # retries EVERYTHING from the top, including re-extracting the PDF

Эта наивная версия повторяет процесс извлечения данных с помощью LLM при каждой попытке, тратя токены и деньги на шаг, который уже сработал, причем функция wait_for_approval_somehow не предлагает надежного решения, если ей нужно существовать дольше времени перезапуска. Версия Temporal не решает эти проблемы с помощью дополнительного кода; благодаря специфике структуры выполнения такие проблемы вообще не возникают.

Простой Python означает возможность тестирования. Temporal включает фреймворк тестов, который позволяет пропустить ожидание в течение timedelta(days=7) и выполнять его за миллисекунды в рамках теста на единицу, а также позволяет имитировать действия без обращения к сети:

import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

@pytest.mark.asyncio
async def test_rejected_invoice_does_not_post_to_erp():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client, task_queue="test-queue",
            workflows=[InvoiceReconciliationWorkflow],
            activities=[extract_line_items, match_purchase_order, notify_human_reviewer, post_to_erp],
        ):
            handle = await env.client.start_workflow(
                InvoiceReconciliationWorkflow.run,
                args=["fake.pdf", "INV-001", "Acme"],
                id="test-invoice-001",
                task_queue="test-queue",
            )
            await handle.signal(InvoiceReconciliationWorkflow.submit_human_decision, False)
            result = await handle.result()
            assert result["status"] == "rejected"

Обратите внимание на отсутствие некоторых элементов: нет реального семидневного цикла сна, нет вмешательства в внутренние механизмы проверки состояния, нет ручного анализа процессов перезапуска. Надежность — это основа инфраструктуры, и не каждый тест должен её заново доказывать.

Использование обеих технологий

Сбалансированным решением не является отказ от LangGraph в пользу Temporal. Они работают на разных уровнях, и лучшие производственные решения 2026 года сочетают их: LangGraph или CrewAI для выполнения логических операций в рамках одного шага, Temporal — для управления рабочим процессом, объединяющим многодневные операции разных сервисов. Одно действие может включать один ограниченный шаг логического анализа вот так:

@activity.defn
async def run_agent_reasoning_step(ticket_text: str) -> dict:
    # LangGraph handles the bounded, single-turn reasoning —
    # classify, decide, draft — inside an activity that Temporal
    # retries and times out like any other side effect.
    result = triage_graph.invoke({"ticket_id": ticket_text, "audit_log": []})
    return {"category": result["category"], "priority": result["priority"]}

@workflow.defn
class SupportTicketWorkflow:
    @workflow.run
    async def run(self, ticket_id: str, ticket_text: str) -> dict:
        triage = await workflow.execute_activity(
            run_agent_reasoning_step, ticket_text,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=workflow.RetryPolicy(maximum_attempts=3),
        )
        if triage["priority"] >= 8:
            await workflow.execute_activity(page_oncall_engineer, ticket_id, ...)
            await workflow.wait_condition(lambda: self._acknowledged, timeout=timedelta(minutes=15))
        # ... routing, SLA timers, escalation chains — all durable, all testable

Это архитектура, которую рекомендуется использовать для команды, чей система агентов должна работать в производственных условиях. Фреймворк, оптимизированный для LLM, справляется с тем, для чего он предназначен — выполнением логических операций в рамках ограниченного числа шагов. Надежный движок выполнения выполняет свою основную функцию, обеспечивая то, чтобы весь процесс, независимо от того, длится ли он минуты или недели, и включает ли он вызовы LLM или утверждения людьми, завершался корректно или при выявлении ошибок происходило явное сообщение об этом. Ни тот, ни другой не должны выполнять функции друг друга.

Когда временные затраты слишком велики

Temporal имеет недостатки, из-за которых он плохо подходит в качестве стандартного решения для многих задач агентов. Это ещё одна распределённая система, которую необходимо запускать — будь то самостоятельно размещённый сервер или услуга Temporal Cloud, плюс рабочие процессы и очереди задач, то есть гораздо больше инфраструктуры, чем у процесса на Python с LangGraph. Код рабочего процесса также должен соблюдать строгие правила детерминизма. Прямое вызование datetime.now(), использование неконтролируемой случайности или выполнение операций ввода-вывода непосредственно в функции рабочего процесса строго запрещены. Командам требуется время, чтобы освоить это, а нарушение правил приводит к запутанным ошибкам недетерминизма при повторном запуске. Для краткой задачи с одним процессом, например для ответа на вопрос или краткого резюмирования документа с помощью одного вызова LLM, надёжная реализация представляет собой избыточность, поскольку не существует многоэтапного процесса, который мог бы пострадать от сбоя.

Правило принятия решения

Выбирайте надежную реализацию в качестве основы там, где рабочий процесс может длиться дольше одного этапа обработки, требует координации между различными сервисами или ожидания действий человека, либо когда речь идет о данных, которые нельзя обрабатывать дважды — например, о деньгах, запасах или электронных письмах, которые не должны отправляться дважды. Позвольте предпочтительной платформе LLM работать в качестве компонента внутри одного из ее этапов. Если всё начинается и заканчивается в рамках одного цикла запроса-ответа, пропустите ее использование, поскольку добавление такой компоненты решило бы проблему, которой пока нет.

Множество обсуждений на тему «Какую платформу агентов мне использовать» упускают из виду более важный вопрос: какой именно слой системы действительно требует усиления безопасности? LangGraph, CrewAI и AutoGen ведут полезные дискуссии о том, как структурировать процессы рассуждений больших языковых моделей и координацию между агентами. Однако вопрос о том, сможет ли агентская система выжить после неудачной развертки, сбоя сети или когда человеку потребуются три дня, чтобы нажать «одобрить», — это отдельный вопрос. Инструменты управления рабочими процессами практически тихо решили его, обеспечивая надежную работу ещё до того, как появилась концепция агентского ИИ. Популярные платформы настраивают то, что видно в демо-версиях. Но тот фактор, который игнорируется, определяет, останется ли что-либо функционировать через месяц.