LangGraph — runtime для долгоживущих stateful-агентов

Демо-агент, который отвечает на один вопрос, написать можно и без фреймворка. Агент, который живёт минутами, разговаривает с человеком между шагами, переживает перезапуск процесса и не теряет уже накопленный контекст, — это уже инженерная задача. Именно её решает LangGraph: низкоуровневый фреймворк оркестрации и runtime для построения, управления и деплоя долгоживущих stateful-агентов.

Проект написан на Python (есть версия для JS/TS — LangGraph.js), распространяется под лицензией MIT и набрал около 42 тысяч звёзд. Среди компаний, которые его используют, README называет Klarna, Replit и Elastic, а документация добавляет Uber и J.P. Morgan.

Ключевое отличие LangGraph от «ещё одного фреймворка для агентов» — он намеренно низкоуровневый. Он не прячет промпты и архитектуру агента за абстракцией: вы сами решаете, где в вашем коде стоит вызов модели, а где — обычная детерминированная логика. Документация прямо говорит, что перед началом работы стоит разобраться с моделями и инструментами, а тем, кто только начинает, предлагает более высокоуровневые абстракции LangChain.

Три сущности: state, nodes, edges

Весь граф собирается из трёх вещей:

• State — общая структура данных, текущий снимок состояния приложения. Обычно это TypedDict (поддерживаются также dataclass и Pydantic-модели, хотя Pydantic менее производителен).
• Nodes — функции, которые получают текущий state, делают что-то полезное и возвращают обновление.
• Edges — функции, которые решают, какой узел выполнять следующим. Это могут быть как фиксированные переходы, так и условные ветвления.

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

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

from typing import Annotated
from typing_extensions import TypedDict
import operator

class MessagesState(TypedDict):
    messages: Annotated[list, operator.add]
    llm_calls: int

Аннотация Annotated[list, operator.add] — это reducer, который дописывает новые сообщения в конец списка, а не заменяет его.

Сам граф создаётся через класс StateGraph, но компиляция обязательна — без неё использовать граф нельзя. Помимо базовых проверок структуры (например, «висячих» узлов без связей), compile позволяет передать runtime-аргументы: checkpointer, breakpoints.

from langgraph.graph import StateGraph, START, END

builder = StateGraph(MessagesState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)

builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
builder.add_edge("tool_node", "llm_call")

agent = builder.compile()

Детерминированные и агентные шаги в одном графе

Заявленная главная особенность LangGraph — возможность смешивать в одном графе обычный написанный руками код и шаги, где решение принимает модель. Там, где нужна предсказуемость и аудируемость, ставится детерминированный узел. Там, где нужна гибкость, ставится агентный шаг.

На уровне реализации это выглядит как алгоритм Pregel: граф выполняется дискретными «супер-шагами» (super-steps). Узлы, которые могут идти параллельно, попадают в один супер-шаг; последовательные — в разные. Узел активен, когда получает новое сообщение по одному из входящих рёбер, отрабатывает и возвращает обновления. Выполнение заканчивается, когда все узлы неактивны и в очереди нет сообщений. Именно на этом дизайне основаны checkpointing, time travel и durable execution.

Проект прямо называет источники вдохновения: Pregel и Apache Beam за алгоритм, а NetworkX — за форму публичного интерфейса.

Durable execution: чекпоинты вместо догадок

Durable execution — обещание, что агент переживает сбои. Платформа пишет состояние графа по шагам, и после падения работа продолжается ровно с того места, где остановилась, а не с начала.

Механизм persistence в LangGraph разделён на две независимые части:

Идентификатор thread_id — это, по сути, указатель на сохранённое состояние. Тот же thread_id продолжает тот же тред, новое значение начинает новый с пустым состоянием.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

graph = builder.compile(
    checkpointer=InMemorySaver(),
    store=InMemoryStore(),
)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi, my name is Bob."}]},
    {"configurable": {"thread_id": "thread-1"}},
)

Для продакшена InMemorySaver не подходит — он держит чекпоинты в RAM и теряет их при перезапуске процесса. Документация рекомендует PostgresSaver (PostgreSQL с async-поддержкой) или SqliteSaver для локальной разработки. Отдельно предупреждают про две частые проблемы: thread_id длиннее 255 символов не влезает в колонку Postgres (лечится UUID или хешем), и checkpoints могут расти бесконечно — нужен периодический prune или retention-политика.

Human-in-the-loop: interrupt вместо костылей

Классическая боль агентных систем — вставка «а точно?» посреди автоматического процесса. Обычно это решается самодельными флагами в state и перезапуском с нуля. В LangGraph есть встроенный механизм: функция interrupt().

Она ставит выполнение на паузу в конкретной точке кода, сохраняет state через checkpointer и ждёт ответа сколь угодно долго. Ключевое отличие от статических breakpoints — прерывания динамические: их можно поставить где угодно и сделать условными, прямо по вашей логике.

from langgraph.types import interrupt

def approval_node(state: State):
    approved = interrupt("Do you approve this action?")
    return {"approved": approved}

Дальше граф возобновляется повторным вызовом с Command(resume=...), и переданное значение становится возвращаемым значением interrupt() внутри узла — код продолжает исполняться с того места, где остановился. Важно: для возобновления нужен тот же самый thread_id, что был при прерывании.

Драйвить такой граф рекомендуется через event streaming: прерывания видны в stream.interrupts, факт паузы — в stream.interrupted, финальное состояние — в stream.output. Старый graph.invoke(...) тоже работает и отдаёт прерывания в result["__interrupt__"].

Стриминг: что происходит внутри

Стриминг в LangGraph — не украшение интерфейса, а основной способ наблюдать за агентом. У графа есть методы stream и astream, которым можно передать один или несколько режимов: updates, values, messages, custom, checkpoints, tasks, debug.

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node {node_name} updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Status: {chunk['data']['status']}")

Режим custom особенно полезен: через get_stream_writer() узел сам пишет произвольные статусы в поток — пользователь видит «сейчас придумаю шутку», а не пустой спиннер.

Для новых приложений документация рекомендует event streaming, появившийся в LangGraph 1.2: он даёт отдельные итераторы под каждую проекцию (messages, values, subgraphs, output), которые можно читать независимо, вместо ветвления по stream_mode. Унифицированный формат StreamPart с version="v2" требует LangGraph 1.1 или новее.

Два способа описать агента

В документации есть две равнозначные точки входа, и это часто удивляет:

• Graph API — явное описание узлов и рёбер через StateGraph. Подходит, когда структура сложная, её хочется визуализировать и переиспользовать как отдельные подграфы.
• Functional API — обычный код с циклами и условиями в одной функции. Отдельные шаги помечаются декоратором @task, а вся функция входа — @entrypoint(). Тот же runtime, та же persistence, но привычный control flow.

from langgraph.func import entrypoint, task

@task
def call_llm(messages: list[BaseMessage]):
    return model_with_tools.invoke(messages)

@entrypoint()
def agent(messages: list[BaseMessage]):
    model_response = call_llm(messages).result()

    while True:
        if not model_response.tool_calls:
            break
        results = [call_tool(tc).result() for tc in model_response.tool_calls]
        messages = add_messages(messages, [model_response, *results])
        model_response = call_llm(messages).result()

    return messages

Выбор между ними — вопрос вкуса и читаемости, а не возможностей.

CLI, Agent Server и деплой

Для локальной работы и продакшена есть отдельный инструмент — LangGraph CLI, ставящийся как langgraph-cli или через npx @langchain/langgraph-cli.

Приложение описывается конфигом langgraph.json — зависимости, список графов и файл с переменными окружения:

{
  "dependencies": ["langchain_openai", "./your_package"],
  "graphs": {
    "my_agent": "./your_package/your_file.py:agent"
  },
  "env": "./.env"
}

Наверх всё это выносится через Agent Server — API для создания и управления агентными приложениями. Он оперирует сущностями assistants, threads, runs и cron jobs, включает встроенную очередь задач и persistence. Три типа данных (core resources, checkpoints, store) по умолчанию лежат в PostgreSQL.

Отдельная деталь, которая снимает кучу кода: Agent Server сам подставляет checkpointer и store в граф во время выполнения. Конфигурировать их в коде графа не нужно и даже нежелательно — сервер управляет ими сам. Частоту записи чекпоинтов регулирует durability mode: async (по умолчанию) пишет после каждого шага, exit сохраняет только финальное состояние.

Ещё одна деталь про регистрацию графа: сервер умеет принимать как уже скомпилированный CompiledGraph (загружается один раз на старте контейнера — рекомендуемый вариант), так и фабрику (вызывается на каждый запуск, нужна только когда требуется per-run кастомизация).

Экосистема и место в стеке

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

Отдельно стоит отметить LangChain Skills — набор агентских скиллов, который ставится в кодинг-агента, чтобы тот лучше работал с задачами экосистемы LangChain, LangGraph и Deep Agents.

Когда LangGraph лишний

Честный список случаев, когда можно не связываться с фреймворком:

• Один запрос, один вызов модели, нет состояния между шагами. Тут хватит обычного SDK.
• Прототип, который нужно выкинуть через неделю. Граф, reducer и чекпоинты добавят многословности.
• Нужна готовая агентная архитектура «из коробки». Здесь LangGraph сознательно слишком низкоуровневый, и документация честно советует взять абстракции LangChain или Deep Agents.

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

Что почитать

• Документация на docs.langchain.com — концепции и гайды
• API reference — подробности по сигнатурам
• Quickstart — агент-калькулятор на Graph API или Functional API
• Graph API overview — state, nodes, edges, reducers
• Persistence — checkpoints, stores, подводные камни
• Interrupts — human-in-the-loop целиком
• Streaming — режимы и форматы потока
• LangGraph CLI и Agent Server — локальный запуск и деплой
• Кейсы компаний — как это выглядит в продакшене