Основы LangGraph: состояние, узлы, рёбра

LangGraph моделирует рабочие процессы как графы. Узлы делают работу, рёбра определяют порядок.

#Когда графа не избежать

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

Граф собирается из трёх вещей: состояния (общая структура данных), узлов (функций, выполняющих работу) и рёбер (связей, задающих порядок). Всё остальное в LangGraph надстраивается над этой тройкой.

#Состояние

Состояние описывают через TypedDict, dataclass или Pydantic BaseModel. По умолчанию берут TypedDict: он быстрее остальных и не добавляет валидации там, где она не нужна.

from typing_extensions import TypedDict

class State(TypedDict):
    query: str
    documents: list[str]
    answer: str

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

Узел не обязан заполнять всё состояние. Он возвращает частичное обновление, а LangGraph объединяет его с текущим значением:

def retrieve(state: State):
    docs = search(state["query"])
    return {"documents": docs}   # остальные поля не трогаем

#Редукторы

Редуктор задаёт правило объединения. По умолчанию новое значение заменяет старое: вернули {"answer": "..."}, и прежний ответ пропал. Для накопления правило указывают через Annotated:

import operator
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    query: str
    documents: Annotated[list[str], operator.add]   # списки склеиваются

Редукторы решают конкретную проблему: параллельные ветви пишут в одно поле. Если два узла в одном шаге вернут {"documents": [...]} без редуктора, LangGraph не сможет выбрать победителя и сообщит об ошибке обновления. С operator.add результаты обеих ветвей просто складываются.

Для сообщений есть специальный редуктор add_messages:

from typing import Annotated
from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict

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

Он умнее простого сложения: новые сообщения добавляет в конец, а сообщение с уже существующим идентификатором заменяет. Именно это позволяет редактировать и удалять сообщения из истории, не пересобирая список вручную.

Готовый вариант такого состояния называется MessagesState. Его наследуют, когда нужны дополнительные поля:

from langgraph.graph import MessagesState

class State(MessagesState):
    user_id: str
    escalated: bool

#Узлы

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

from langgraph.runtime import Runtime

def classify(state: State) -> dict:
    return {"category": "billing"}

async def fetch(state: State, runtime: Runtime) -> dict:
    profile = await runtime.store.aget(("users",), state["user_id"])
    return {"profile": profile}

Имя узла задаётся первым аргументом add_node. Если его опустить, LangGraph возьмёт имя функции. Имена важнее, чем кажется: они попадают в трассировку, используются в условных рёбрах и в Command(goto=...).

Состояние в узле нельзя менять на месте. Запись state["documents"].append(doc) обходит редукторы, ломает восстановление из контрольной точки и приводит к трудноуловимым эффектам при параллельных ветвях. Всегда возвращайте новое значение.

#Рёбра

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

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

Если из узла выходят два обычных ребра, обе ветви выполняются параллельно. Это не ошибка проектирования, а штатный способ распараллелить работу, но поля, в которые пишут обе ветви, обязаны иметь редуктор.

#Компиляция и запуск

Перед использованием граф компилируют. На этом шаге проверяется связность и задаются параметры времени выполнения (чекпоинтер, кэш, точки останова):

graph = builder.compile()
result = graph.invoke({"query": "Как работает редуктор?", "documents": [], "answer": ""})

Входной словарь должен содержать поля, которые читают первые узлы. Ключ, объявленный в TypedDict, но не переданный на входе, останется отсутствующим, и обращение state["documents"] выбросит KeyError. Отсюда привычка: либо передавать полный вход, либо читать необязательные поля через state.get(...).

#Полный пример

Соберём маленький граф целиком: он классифицирует обращение и готовит ответ.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    text: str
    category: str
    reply: str

def classify(state: State):
    text = state["text"].lower()
    category = "billing" if "счёт" in text or "оплат" in text else "general"
    return {"category": category}

def reply(state: State):
    if state["category"] == "billing":
        return {"reply": "Передаю обращение в биллинг."}
    return {"reply": "Отвечаю на общий вопрос."}

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("reply", reply)
builder.add_edge(START, "classify")
builder.add_edge("classify", "reply")
builder.add_edge("reply", END)

graph = builder.compile()
result = graph.invoke({"text": "Не пришёл счёт за апрель", "category": "", "reply": ""})
print(result["reply"])

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

#Супершаг

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

Понимание супершагов объясняет два наблюдения. Во-первых, обновления параллельных узлов применяются одновременно, отсюда требование редукторов. Во-вторых, лимит рекурсии считает именно супершаги, а не вызовы модели: цикл из трёх узлов израсходует лимит втрое быстрее одиночного узла.

#Раздельные схемы входа и выхода

Иногда внутреннее состояние богаче того, что уместно принимать и отдавать наружу. Тогда объявляют отдельные схемы:

class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    answer: str

class OverallState(TypedDict):
    question: str
    scratchpad: list[str]
    answer: str

builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)

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

#Визуализация

Скомпилированный граф умеет рисовать себя. Схема в формате Mermaid помогает проверить, что рёбра расставлены так, как задумано:

print(graph.get_graph().draw_mermaid())

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

#Частые ошибки

Возврат полного состояния вместо частичного обновления. Формально это работает, но затирает изменения параллельных ветвей и делает узел зависимым от полей, которые его не касаются.

Изменение состояния на месте. state["items"].append(x) минует редуктор, поэтому при параллельном выполнении часть данных теряется.

Отсутствие редуктора у поля, в которое пишут несколько узлов одного супершага. Симптом: ошибка о конфликтующем обновлении.

Обращение к необязательному полю через квадратные скобки. Если поле не было передано на входе и ещё не установлено ни одним узлом, будет KeyError. Спасает state.get("field", default).

Устранение неисправностей