LangGraph: state machines for agents

LangGraph моделирует агентную систему как конечный автомат: узлы — это функции (обычно с вызовом LLM), рёбра — переходы, а всё общение между узлами идёт через единый объект состояния. Узел не вызывает другой узел напрямую — он возвращает частичное обновление состояния, и граф решает, куда идти дальше.

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]
    draft: str

def writer(state: State) -> dict:
    reply = llm.invoke(state["messages"])
    return {"messages": [reply], "draft": reply.content}

builder = StateGraph(State)
builder.add_node("writer", writer)
builder.add_edge(START, "writer")
builder.add_edge("writer", END)
graph = builder.compile()

Ключевая деталь — аннотация Annotated[list, add_messages]. Второй аргумент — редьюсер: функция, которая описывает, КАК новое значение сливается со старым. Для messages редьюсер add_messages дописывает новые сообщения в конец списка (и умеет заменять сообщение по id), а поле draft без редьюсера просто перезаписывается последним значением.

Редьюсеры — это не синтаксический сахар, а фундамент параллелизма: когда две ветки графа одновременно вернут обновление одного ключа, именно редьюсер определит результат слияния. Ключ без редьюсера при параллельной записи вызовет ошибку InvalidUpdateError — LangGraph честно отказывается молча терять данные.

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

from langgraph.prebuilt import ToolNode

def route(state: State) -> str:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return "done"

builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route, {"tools": "tools", "done": END})
builder.add_edge("tools", "agent")

Маршрутизатор — обычная детерминированная функция: её можно покрыть юнит-тестами без единого вызова LLM, и это одно из главных преимуществ явного графа перед «магическим» циклом внутри фреймворка. Решение модели (вызывать ли инструмент) остаётся вероятностным, но логика перехода — читаемая и проверяемая.

Против бесконечных циклов у LangGraph есть предохранитель recursion_limit (по умолчанию 25 супершагов): при превышении граф падает с GraphRecursionError вместо того, чтобы молча сжигать токены. Для продакшена лимит выставляют осознанно под задачу, а в состояние добавляют собственный счётчик итераций — он позволяет выйти из цикла мягко, с частичным результатом, а не с исключением.

Скомпилированный граф без персистентности живёт один вызов: упал процесс — пропал прогресс. Checkpointer сохраняет снимок состояния после каждого супершага, привязывая его к thread_id.

from langgraph.checkpoint.memory import MemorySaver

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "run-42"}}
graph.invoke({"messages": [("user", "Составь план миграции")]}, config)
# процесс можно убить: следующий invoke с тем же thread_id
# продолжит с последнего чекпойнта, а не с нуля

В проде вместо MemorySaver используют SqliteSaver или PostgresSaver из пакетов langgraph-checkpoint-sqlite и langgraph-checkpoint-postgres — интерфейс тот же, меняется одна строка. Это дёшево покупает три свойства сразу. Первое — устойчивость к сбоям: длинный мультиагентный прогон возобновляется с места падения, что критично при цене прогона в доллары. Второе — многосессионность: thread_id превращает граф в диалог с памятью, к которому можно вернуться завтра.

Третье свойство — time-travel. Метод graph.get_state_history(config) возвращает всю цепочку чекпойнтов, и можно перезапустить граф с любого из них, в том числе с изменённым состоянием: «а что бы ответил ревьюер, если бы спека была другой». Для отладки мультиагентных систем это заменяет часы гадания по логам — вы буквально проигрываете альтернативные ветки истории на реальном состоянии.

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

from langgraph.types import interrupt, Command

def approval_gate(state: State) -> dict:
    decision = interrupt({
        "question": "Влить ветку в dev?",
        "diff_summary": state["draft"],
    })
    return {"approved": decision == "yes"}

# ... пользователь посмотрел diff в UI и нажал кнопку:
graph.invoke(Command(resume="yes"), config)

Важно понимать механику: interrupt не «замораживает» Python-процесс. Граф завершает супершаг, пишет чекпойнт и отдаёт управление; пауза может длиться секунды или дни, процесс можно перезапустить — resume придёт в тот же thread_id и узел выполнится повторно уже с ответом человека. Отсюда требование: код узла до interrupt должен быть идемпотентным, потому что при возобновлении он выполнится ещё раз.

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