Шесть примитивов LangGraph и способы скрытия моделей сбоев в каждом из них

Первый агент LangGraph обычно формируется быстро: он отвечает на вопрос, улучшает ответ и прекращает работу. Затем кто-то добавляет ветку для повторной попытки, и вдруг граф перестаёт завершаться, тратя кредиты API до момента остановки процесса. Иногда решением является просто отсутствующая рёбра, но настоящая проблема заключается в отсутствии понимания того, почему граф ведёт себя именно так.

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

Почему граф вместо цепи

Синтаксис трубопровода LangChain вида prompt | llm | parser удобен для однократной обработки данных через модель. Он перестаёт быть эффективным, как только агенту нужно принять решение: искать информацию или отвечать напрямую, попытаться снова или сдаться, обратиться к человеку или продолжить работу. Цепочка действий не имеет понятия «всё зависит от обстоятельств», поэтому разработчики окутывают вызовы цепочки операторами if, в результате чего со временем создаётся неофициальная, сложнее для отладки машина состояний.

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

1. Состояние: один общий объект и важные функции изменения состояния

Статус — это единственный объект, из которого читают и в который записывают все узлы. Без него контекст часто передается в качестве аргументов функций, и становится сложно понять, что именно знал тот или иной шаг. Приведённое ниже определение представляет собой TypedDict, содержащий вопрос, ответ и список сообщений, обновления которого объединяются с помощью редюсера add_messages.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add не является редюсером сообщений

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

add_messages разработан именно для этой цели. Он находит сообщения по идентификатору и заменяет существующее сообщение, если такой идентификатор уже присутствует, добавляя в список только действительно новые сообщения. Правило простое: используйте add_messages для полей, содержащих объекты HumanMessage и AIMessage, а operator.add — для обычных списков с накоплением данных, например, для ведения учета тех инструментов, которые использовались.

Сохраняйте минимальную структуру состояния

Вторая распространённая ошибка — проектирование состояния как схемы базы данных, при которой создаётся поле для каждой возможной потребности в будущем. Добавляйте поле только тогда, когда узел действительно его читает или записывает. Последствия игнорирования этого принципа очевидны: рассмотрим граф обработки документов, в котором в состоянии хранятся полные необработанные ответы больших языковых моделей вместе с метаданными об использовании токенов. Обработка 50 документов в цикле приводила к тому, что размер каждого этапа обработки достигал примерно 180 КБ, а время записи в Postgres превышало 400 мс, что слишком медленно для пользователей, ожидающих ответа. Решение было довольно простым: сократить объём состояния до трёх полей, которые действительно используются узлами на последующих этапах. Помните, что при использовании контрольных точек всё содержимое состояния сериализуется и сохраняется на каждом шаге.

2. Узлы: возвращайте только то, что изменилось

Узел — это обычная функция на Python. Она принимает состояние, выполняет свою работу и возвращает словарь, содержащий только те поля, которые были изменены. В этом и заключается весь контракт. В первом примере вызывается чат-модель OpenAI с вопросом, а ответ записывается в переменную answer.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

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

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Для этой версии необходимо, чтобы Ollama работал локально с загруженной моделью (ollama pull llama3.1) и установленным пакетом интеграции (pip install langchain-ollama). Установка параметра temperature=0 также делает результаты выполнения более предсказуемыми, что полезно при тестировании логики маршрутизации.

Возврат всего состояния мешает другим обновлениям

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

3. Прямые рёбра: всегда подключайте выход

Рёбра определяют, что будет выполняться далее. Прямое ребро является безусловным: когда узел A завершает работу, запускается узел B. Ниже приведена диаграмма, в которой заданы два узла, answer соединён с refine, refine — с END; также указана точка входа и произведена компиляция.

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

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

4. Условные ребра: где агент принимает решения

Условные рёбра делают граф агентом, а не фиксированной цепочкой операций. Функция маршрутизации проверяет состояние и возвращает метку; преобразователь переводит каждую метку в следующий узел. В этом примере краткий ответ (менее 50 символов) отправляется в refine, а все остальные — в END.

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

Обратите внимание, что это условное ребро заменяет прямое ребро answer → refine из предыдущего примера. Если зарегистрированы оба варианта, будут использоваться обе траектории, что редко бывает желательно.

Несоответствующие метки маршрутов вызывают явные, но неочевидные ошибки

Частой ошибкой здесь является функция маршрутизации, возвращающая строку, отсутствующую в словаре сопоставлений. В результате возникает довольно общая ошибка ключа, скрытая на нескольких уровнях глубоко в трейсе ошибки, и даже такая мелочь, как лишний пробель, может потребовать удивительно много времени. Надежная привычка: сначала написать словарь сопоставлений, а затем создать маршрутизатор, скопировав из него точные ключи. Еще лучше — определить метки один раз в качестве констант или аннотировать тип возвращаемого значения маршрутизатора с помощью Literal["refine", "done"], чтобы инструменты проверки типов и читатели сразу видели допустимые значения.

5. Пункты контроля: память, сохраняющаяся между вызовами

Чекпоинтер превращает вызов безсостоянийной функции в диалог с памятью. Без него каждый вызов app.invoke() начинается с нуля. С его помощью состояние сохраняется для каждой потока, и любой вызов, передающий в своей конфигурации тот же thread_id, продолжает работу с того места, где остановился предыдущий. В примере второй вызов на потоке user-session-42 помнит первый вопрос.

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver подходит только для локальной разработки. Он находится в памяти процесса, поэтому перезагрузка сервера удаляет всю информацию о диалогах. Для того чтобы реальные пользователи могли на чем-то полагаться, необходим постоянный бэкенд: SQLite для одного сервера или Postgres, когда несколько инстансов должны делить состояние.

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

Как показано в комментариях к установке, каждый бэкенд поставляется как отдельный пакет. В текущих версиях такие сохранители обычно создаются на основе строки подключения (например, с помощью from_conn_string), и для создания таблиц в Postgres требуется однократный вызов функции setup(); поэтому ознакомьтесь с документацией к checkpointer, чтобы узнать точные инструкции по инициализации для вашей версии.

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

6. Участие человека: статические точки останова против динамических перерывов

Схема, которую показывают большинство учебных материалов, — это interrupt_before, список имен узлов, на которых скомпилированная структура останавливается перед выполнением:

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

Этот подход работает и легко объясним, но он статичен. Точка остановки задается именем узла, её нельзя сделать условной, а также невозможно привязать к ней информацию о том, что должен проверить рецензент. Реальные требования быстро превосходят возможности этого подхода, поскольку «остановиться перед этим узлом» и «остановиться только тогда, когда сумма возврата превышает 500 долларов» — это разные правила, и только первое можно выразить таким образом.

Остановка изнутри узла с помощью interrupt()

Более гибкий вариант — это вызов interrupt() непосредственно изнутри узла. Приведённый ниже узел проверяет сумму возврата: если она превышает 500 долларов, он приостанавливается и передаёт информацию о проекте и сумме человеку для рассмотрения. Первый вызов invoke выполняется до момента этой паузы. Второй вызов передаёт параметр Command(resume="approve") в том же потоке, причём значение, переданное в resume, становится значением возврата функции interrupt(); таким образом узел либо продолжает работу, либо возвращает статус отмены. Для хранения состояния паузы требуется указатель на состояние, поскольку оно должно быть сохранено где-то во время ожидания.

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

В примере используются такие поля, как refund_amount, draft и task, которых нет в предыдущем типе состояния AgentState; в реальной схеме их следует объявить там.

Возобновление работы запускает весь узел заново

Поведение, которое удивляет людей: в режиме продолжения выполнения LangGraph не продолжает с строки interrupt(). Он перезапускает весь узел с самого начала, и на этот раз interrupt() возвращает значение для продолжения выполнения вместо того, чтобы приостановиться. Все коды, находящиеся до вызова, выполняются снова. Узел, увеличивающий счетчик перед прерыванием, будет увеличивать его дважды за каждое одобрение. Сделайте всё, что находится до interrupt(), идемпотентным или переместите побочные эффекты в более ранний узел. Тот же принцип применяется к вызовам API или записям в базу данных, размещаемым перед моментом приостановки.

Модель, которая сама себя одобряет, не является системой с участием человека

Какой бы механизм вы ни выбрали, задание модели вопроса «Стоит ли мне продолжать?» и доверие к её ответу не являются человеческим контролем, как бы это ни называлось. Это агент, самостоятельно утверждающий своё решение. Настоящий этап одобрения передаёт управление человеку, находящемуся вне структуры, и ожидает его ответа.

Шесть примитивов в одном взгляде

В приведённом ниже кратком описании каждая концепция сочетается с тем, что она делает, и типичной ошибкой, связанной с ней.

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

Разумный порядок сборки

Для первой настоящей графической структуры сначала обеспечьте полноценную работу цикла от начала до конца с использованием InMemorySaver без прерываний. Сохраняйте небольшой объем состояния, ограничивая его тем, что необходимо узлам, и убедитесь, что каждая условная ребро возвращает именно те метки, которые ожидается от соответствующей функции преобразования. Только после того, как всё будет работать корректно, можно внедрить постоянный указатель состояния и добавить прерывание в тот шаг, который действительно требует участия человека — обычно это операции с переводом денег, отправкой внешнего электронного письма или удалением данных.

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

Основные выводы

• Используйте add_messages для истории чата, а operator.add — только для обычных списков, и сокращайте объем хранимого состояния, поскольку оно сохраняется на каждом шаге.
• Возвращайте только измененные поля узлов; возврат полного состояния бесшумно перезапишет параллельные или предыдущие обновления.
• Укажите для каждого пути явный маршрут до END, а также ограничьте циклы повторных попыток, вместо того чтобы полагаться на лимит рекурсии.
• Определяйте метки маршрутизации на основе карты, чтобы они не отклонялись друг от друга.
• Рассматривайте InMemorySaver как инструмент только для разработки; замена контрольных точек стоит недорого, поэтому выполняйте ее до того, как пользователи начнут использовать граф.
• В качестве средства условного утверждения предпочитайте динамический вызов interrupt(), и сохраняйте код до того, как он станет идемпотентным, поскольку возобновление работы приведет к повторной обработке узла.

Справочная документация: документация Graph API для LangGraph, руководство по прерываниям и interrupt() — справка по API.