Temporal и LangGraph: как плагин делает AI-агентов отказоустойчивыми на Python

Что изменилось с появлением плагина

Официальная интеграция Temporal с LangGraph даёт разработчикам готовый плагин, который превращает обычные AI-агентные графы в отказоустойчивые рабочие процессы. Вместо того чтобы писать инфраструктуру повторных попыток, таймаутов и восстановления после сбоев вручную, теперь достаточно описать граф привычным способом — через StateGraph или декораторы @entrypoint/@task — а плагин сам сохранит состояние выполнения в Temporal и гарантирует, что ни один шаг не будет потерян из-за падения worker’а, сетевого сбоя или перезапуска сервиса.

Ключевое изменение: LangGraph-граф теперь можно запускать не как обычную асинхронную функцию, а как полноценный Temporal Workflow с durable execution. Это означает, что промежуточные состояния, результаты нод и история выполнения сохраняются в Temporal, а не только в оперативной памяти Python-процесса. Для агентов, которые работают с LLM-вызовами, внешними API и цепочками шагов по 10–30 минут, это переход от «надеемся, что ничего не упадёт» к «платформа гарантирует восстановление».

Как устроена модель выполнения

Плагин поддерживает два стиля описания графов, и у каждого — своя семантика в Temporal.

Graph API (StateGraph) — вы описываете граф декларативно: ноды, рёбра, условия переходов. Внутри Workflow вызываете graph(...).compile().ainvoke(...), и каждая нода выполняется либо как Temporal Activity, либо прямо в Workflow. Выбор определяется метаданными, которые вы передаёте в add_node.

Functional API (@entrypoint / @task) — вы пишете код как обычные асинхронные функции, помечая их декораторами. Temporal сам управляет очередью задач и сериализацией состояния.

Принципиальное разделение: ноды, помеченные как "execute_in": "activity", получают настраиваемые таймауты (start_to_close_timeout, schedule_to_start_timeout) и политики повторных попыток (RetryPolicy). Ноды, выполняемые прямо в Workflow, работают в потоке выполнения и обязаны быть детерминированными — иначе Temporal не сможет воспроизвести историю и восстановить состояние после сбоя.

Практический рабочий метод: от установки до запуска worker’а

Ниже — минимальная последовательность действий, которая превращает обычный LangGraph-проект в отказоустойчивый агент на Temporal.

1. Установка зависимости. Пакет ставится как расширение Temporal с опцией [langgraph]. Требуется версия temporalio не ниже 1.27.0.

uv add "temporalio[langgraph]"
# или
pip install "temporalio[langgraph]"

2. Ограничение по версии Python. Для полной функциональности — Functional API, interrupt(), streaming из Workflow-нод — нужен Python 3.11 или новее. На более старых версиях плагин загружается, но выводит предупреждение, и перечисленные функции не работают, потому что зависят от распространения contextvars через asyncio.create_task().

3. Описание графа. Для Graph API каждая нода принимает метаданные:

g.add_node(
    "my_node",
    my_node,
    metadata={
        "execute_in": "activity",
        "start_to_close_timeout": timedelta(seconds=30),
        "retry_policy": RetryPolicy(maximum_attempts=3),
    },
)

4. Создание Workflow и компиляция графа:

@workflow.defn
class HelloWorldWorkflow:
    @workflow.run
    async def run(self, query: str) -> str:
        return await graph("hello-world").compile().ainvoke(query)

5. Конфигурация Worker’а. Плагин передаётся в конструктор Worker вместе с реестром графов:

plugin = LangGraphPlugin(graphs={"hello-world": build_graph()})
worker = Worker(
    client,
    task_queue="langgraph-hello-world",
    workflows=[HelloWorldWorkflow],
    plugins=[plugin],
)

6. Запуск локального сервера Temporal. Для тестирования используется Temporal development server, который должен быть запущен до старта worker’а.

Критические ограничения и конфликты конфигураций

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

Чеклист принятия решения о внедрении

Прежде чем добавлять Temporal в существующий LangGraph-проект, убедитесь в следующем:

• [ ] Версия Python в проекте ≥ 3.11, если нужен Functional API или стриминг из Workflow-нод.
• [ ] Установлена temporalio[langgraph] версии 1.27.0 или выше.
• [ ] Все ноды, выполняющие внешние вызовы (LLM, API, базы данных), помечены как "execute_in": "activity".
• [ ] Ни одна нода не использует встроенный retry_policy LangGraph — заменён на RetryPolicy из Temporal.
• [ ] Workflow-ноды не содержат недетерминированных операций.
• [ ] Temporal development server запущен локально для проверки воркфлоу.
• [ ] Определён task_queue и зарегистрирован LangGraphPlugin с актуальными графами.

Итоговая рекомендация

Если ваш LangGraph-агент выполняет цепочку шагов дольше нескольких секунд, обращается к внешним API или обрабатывает данные, которые нельзя потерять при сбое процесса, интеграция через плагин Temporal — это не архитектурное усложнение, а устранение целого класса инфраструктурных задач. Плагин берёт на себя персистентность, таймауты и повторные попытки, позволяя вам описывать только бизнес-логику графа. Единственное, что требуется — явно разделить ноды на Activities и Workflow-ноды и отказаться от LangGraph-политик повторных попыток в пользу Temporal-конфигурации. Дополнительным преимуществом становится возможность аудита и отладки: вся история выполнения графа, включая промежуточные состояния и результаты каждой ноды, доступна через Temporal Web UI, что значительно упрощает диагностику проблем в продакшене.

Источники

• LangGraph integration — Temporal Platform Documentation
• Temporal Python SDK on PyPI
• LangGraph Documentation