LangGraph в Python: состояния, параллельные узлы, точки сохранения и проверка человеком

Любой, кто создавал что-то большее, чем просто игрушку на основе LLM, знает о трех постоянных проблемах. Последовательные вызовы увеличивают задержку. Длинные рабочие процессы могут сорваться, теряя весь достигнутый прогресс. А в самый неподходящий момент модель может выдать уверенный, но совершенно неверный ответ.

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

В следующих разделах рассматривается процесс пошагово. Сначала рассматривается отдельный узел, затем параллельная обработка, условное маршрутизирование и постоянная память, после чего — прерывания с участием человека. В конце приводится полный рабочий процесс сортировки электронных писем, объединяющий все эти элементы. Каждый пример предназначен к запуску без изменений. SDK TypeScript предоставляет практически те же API, поэтому тот же подход к их использованию применим и там.

Настройка

Сначала требуются два пакета:

pip install langgraph langchain-openai

Затем переменные окружения для OpenAI и, по желанию, для отслеживания действий LangSmith:

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

Примеры были протестированы с langgraph 1.1.3 (март 2026 г.) и langchain-openai 1.1.10; для их работы требуется Python 3.10 или новее. В старых руководствах иногда используются LANGCHAIN_TRACING_V2 и LANGCHAIN_API_KEY; оба варианта по-прежнему работают, но теперь предпочтительными являются LANGSMITH_TRACING и LANGSMITH_API_KEY.

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

Определение состояния с помощью TypedDict

Состояние в LangGraph представляет собой обычный Python-объект TypedDict без каких-либо специальных базовых классов и декораторов, специфичных для фреймворка — это просто словари с указанием типов:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

Класс определяет структуру данных, проходящих через граф, и каждый узел читает из этой общей структуры и записывает в неё.

Почему использовать TypedDict, а не обычный словарь или dataclass? Он обеспечивает статическую проверку типов, поскольку его понимают как mypy, так и Pyright, наряду с типовыми хэштегами, и избавляет от нагрузки во время выполнения, свойственной dataclass или моделям Pydantic. LangGraph также может анализировать аннотации, чтобы определить схему, что важно для структурированного вывода и верификации.

Важной строкой является Annotated[list[str], operator.add], которая объявляет редьюсер. Это самая важная концепция в состоянии LangGraph, и почти у всех возникают трудности с ней при первом знакомстве.

Без редьюсера узел, возвращающий {"summary": "new value"}, просто заменяет сохранённое значение. Для скалярных полей, таких как summary, это именно то, что нужно. Проблемы возникают, когда два параллельных узла пишут в messages: сохраняется только тот, который закончит работу последним. Однако объявление operator.add указывает LangGraph на добавление данных, поэтому все вклады в messages накапливаются безопасно.

Для более сложной логики объединения требуется пользовательский редьюсер:

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

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

Этот редьюсер объединяет существующий список с новыми сообщениями, затем сокращает его до десяти самых свежих записей, что помогает при длинных разговорах, когда иначе количество токенов резко возрастало бы. Метод operator.add используется для простого добавления элементов. Для удаления дубликатов, ограничения размера списка или любого другого вида объединения помимо простого добавления требуется пользовательский редьюсер.

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

TypedDict против Pydantic

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

Pydantic используется для схем, с которыми взаимодействует большая языковая модель. Вызов .with_structured_output(SomeModel) предоставляет модели JSON-схему, к которой она должна соответствовать, и проверяет, что ответ соответствует этой схеме. Если ответ не соответствует, сразу возникает ошибка, и некорректные данные не могут попасть дальше по графу.

В последующем основном рабочем процессе строго соблюдается именно такое разделение: EmailState представляет собой TypedDict, служащий контейнером графа, а EmailClassification — это модель на Pydantic, описывающая результат работы большой языковой модели.

Создание первого узла

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

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Разбор кода:

• StateGraph(ChatState) создаёт граф, использующий ChatState в качестве схемы, а LangGraph читает аннотации для получения информации о редюсерах и типах полей.
• graph.add_node("chatbot", chatbot) регистрирует функцию под именем "chatbot", к которому обращаются определения рёбер.
• graph.add_edge(START, "chatbot") делает "chatbot" первым узлом, который будет выполнен. START — встроенный сентинел.
• graph.add_edge("chatbot", END) завершает выполнение после окончания работы chatbot. END — ещё один встроенный сентинел.
• graph.compile() проверяет структуру на наличие бесполезных рёбер и недоступных узлов, а затем возвращает готовый к выполнению объект CompiledGraph.

app.invoke({"messages": [...]}) выполняет граф синхронно и возвращает конечное состояние.

Узел оборачивает свою строку в список, поскольку messages использует operator.add в качестве редьюсера. Возврат {"messages": [response.content]} добавляет список из одного элемента к существующему, тогда как возврат простой строки приведёт к ошибке во время выполнения, поскольку operator.add не может добавить строку в список.

Правило заключается в соответствии того, что возвращает узел, с типом поля, включая редьюсер. Поле типа list[str] с operator.add принимает list[str], а простое поле типа str без редьюсера принимает str.

Это полностью рабочее приложение LangGraph, состоящее примерно из 20 строк. Следующим вопросом является то, что происходит внутри функции invoke().

Супершаги: как происходит выполнение

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

Это важно по двум причинам. Параллельное выполнение встроено в модель с самого начала и не добавляется позже. Кроме того, с помощью контрольного указателя LangGraph сохраняет полную копию состояния после завершения каждого супершага. Именно это позволяет восстанавливаться после сбоев: если процесс прерывается на полпути, выполнение возобновляется с последнего завершённого супершага.

После того как понятно, как передаётся состояние через один узел, следующий вопрос — как контролировать порядок выполнения или запускать узлы одновременно. Именно для этого и существуют рёбра.

Рёбра: последовательные, параллельные и условные

Ребра определяют способ перехода выполнения между узлами. LangGraph предлагает три типа ребер, которые в совокупности покрывают все возможные схемы маршрутизации. Самая простая форма уже была показана в примере с чат-ботом.

Последовательные ребра

В самом простом случае узел следует за другим узлом.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft выполняется в первую очередь. Как только он завершается, LangGraph объединяет поле draft с состоянием, после чего запускается review_draft, который может считать черновик из state["draft"]. Последовательность выполнения гарантируется, поскольку связь обеспечивает зависимость, и не требуется специального кода для управления порядком. Последовательные связи применимы в любом случае, когда узел B зависит от результата работы узла A: сначала генерация, затем проверка, сначала получение данных, затем их обработка, сначала классификация, затем направление.

Непреднамеренные циклы представляют собой распространённую проблему. По умолчанию LangGraph ожидает ориентированную ациклическую сеть, поэтому добавление связи от review обратно к generate приводит к ошибке на этапе компиляции, если только сеть не настроена явно как циклическая. Итеративные циклы обычно создаются с помощью условных связей, которые могут направлять поток обратно к более раннему узлу.

Параллельные связи

Когда две грани исходят из одного узла, их цели обрабатываются одновременно в том же супершаге. Здесь становятся необходимыми редьюсеры:

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Два ребра от START запланировали операции research и outline в один и тот же супершаг, и LangGraph выполняет их одновременно. У обеих операций также есть ребро, ведущее к combine, и LangGraph не запустит combine, пока все узлы, указывающие на него, не завершат свою работу, что служит автоматическим барьером синхронизации.

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

Расчёт задержки прост: если research занимает 2 секунды, а outline — 3, то в серийной версии потребуется 5 секунд, а в параллельной — 3 секунды, что соответствует длительности самой долгой операции. Для независимых вызовов LLM параллельная обработка позволяет добиться практически полного сокращения задержек.

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

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

Условные ребра

Когда следующий узел зависит от результата работы текущего, применяется функция add_conditional_edges:

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

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

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

Альтернатива — Command

Когда узлу необходимо обновить состояние и выполнить маршрутизацию на одном шаге, он может напрямую вернуть объект Command и пропустить вызов функции add_conditional_edges:

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

Одно значение, возвращаемое Command, содержит как обновление состояния (update=), так и выбор маршрута (goto=), поэтому узел записывает данные и выбирает следующее направление за один шаг.

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

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

Для этапа человеческого рассмотрения модуль email capstone опирается на функцию add_conditional_edges, поскольку маршрутизация здесь простая: он проверяет только приоритет и уровень уверенности, уже сохранённые в состоянии.

Память и механизмы сохранения состояния

После того как порядок выполнения операций и ветвление контролируются, возникает следующая проблема — сохранность данных: что происходит, когда процесс перезапускается посреди диалога?

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

MemorySaver для разработки

MemorySaver сохраняет состояние в памяти, что удобно для разработки и тестирования:

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

При использовании параметра checkpointer=memory скомпилированный граф сохраняет полную копию состояния после каждого «супершага». При повторном вызове функции invoke() с тем же значением thread_id LangGraph сначала загружает самый свежий чекпоинт для этой нити. thread_id функционирует как идентификатор сессии: его можно использовать повторно для продолжения работы или изменить, чтобы начать всё с нуля.

Второй вызов может не включать более ранние сообщения, поскольку они уже хранятся в чекпоинте. Если Алиса вызывает функцию с параметром {"messages": ["What should I learn first?"]}, LangGraph загружает чекпоинт "user-alice", находит сохранённый список и добавляет новую запись с помощью функции-редьюсера operator.add. После запуска чат-бота полная история сообщений хранится в элементе state["messages"].

Использование одного и того же thread_id для несвязанных разговоров — ошибка. Если "user-alice" снова используется для другой темы, чат-бот получает в контексте всю предыдущую переписку Алисы, поэтому каждая новая сессия должна иметь свой уникальный thread_id. Во время разработки функция app.get_state_history(config) отображает полную историю состояний для конкретного потока, и тот же механизм используется для отладки с возможностью «путешествия во времени» в LangSmith.

Надежные инструменты сохранения состояний для производства

MemorySaver хранится в памяти процесса, поэтому при его завершении состояние теряется. Для производственных целей требуется надежный бэкенд:

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() создает необходимые таблицы базы данных при первом запуске и далее не выполняет никаких действий, поэтому его достаточно вызвать один раз во время инициализации. API остается неизменным между разными бэкендами, поскольку механизмы контроля состояния работают через общую абстракцию. MemorySaver, PostgresSaver и RedisSaver следуют одному протоколу, поэтому код графа не знает, с кем именно он взаимодействует.

Выбор между ними:

• MemorySaver подходит для разработки, тестирования и демонстраций, где важность сохранения данных после перезагрузки невелика.
• PostgresSaver подходит для производственных рабочих процессов, требующих надежности и возможности просмотра истории контрольных точек.
• RedisSaver из пакета langgraph-redis подходит для сценариев с высокой пропускной способностью, где необходимы быстрые операции с контрольными точками, и при этом приемлемы компромиссы, связанные с сохранением данных в Redis.

Что сохраняется

Полная копия состояния хранится после каждого «супершага». Это позволяет осуществлять «путешествия во времени»: можно просмотреть состояние в любой момент истории, воспроизвести действия с выбранной точки контроля или отладить процесс, чтобы выяснить, где именно возникли ошибки. Тот же механизм используется для обработки прерываний. Когда узел вызывает функцию interrupt(), LangGraph нуждается в месте для хранения приостановленного состояния, чтобы позже возобновить его работу. Без такой точки контроля ничто не может сохранить состояние, и структура обработки прерываний разрушается.

Участие человека при прерываниях

Некоторые решения слишком важны, чтобы их автоматизировать. Прежде чем агент отправит письмо клиенту, одобрит возврат средств или развернет код, необходимо, чтобы человек их проверил и одобрил. LangGraph реализует это с помощью функции interrupt().

Функция interrupt()

Размещение функции interrupt() внутри узла прерывает её выполнение, записывает состояние в checkpointer и возвращает нагрузку вызывающему коду:

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

По порядку функция вызова выполняет четыре действия:

1. Она останавливает выполнение именно на той строке, где произошел вызов.
2. Она сохраняет полное состояние в контрольный указатель, поэтому он и требуется.
3. Она возвращает переданный пакет данных, а именно черновик текста, тому, кто вызвал invoke().
4. Она ждет, оставляя граф в приостановленном состоянии.

Переменная decision сначала ничего не получает, поскольку interrupt() возвращает значения только тогда, когда граф возобновляется с помощью Command(resume=value). В этот момент decision получает то, что было передано в параметр resume=.

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

Возобновление с помощью Command

Продолжение работы после прерывания означает повторное вызове графа с помощью Command(resume=value):

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

На первом этапе LangGraph запускает функцию generate, сохраняет черновик в состоянии и переходит к функции interrupt() в узле review. Затем он сохраняет состояние в чекпоинт и возвращает результат; результат первого вызова invoke() содержит данные о прерывании, а также черновик для рецензента. На втором этапе команда Command(resume="approve") заставляет LangGraph загрузить чекпоинт с именем "email-review-1", возобновить работу с момента вызова interrupt() и вернуть значение "approve" в качестве результата. Значение поля decision узла review становится "approve", устанавливается флаг approved=True, и выполнение переходит к шагу send.

Для обоих вызовов должен использоваться тот же config, поскольку LangGraph находит соответствующий чекпоинт по thread_id. Разный thread_id при втором вызове приведёт к указанию на несуществующий чекпоинт, и вызов завершится с ошибкой.

Чекпоинт является обязательным при прерываниях. Без него состояние паузы не имеет куда перейти, поэтому app.compile(checkpointer=memory) здесь не является опциональным.

Начиная с версии LangGraph 1.1.0, функция invoke() принимает параметр version="v2" и возвращает объект GraphOutput с атрибутами .value и .interrupts, что обеспечивает более надежную обработку прерываний с учетом типов данных; его стоит попробовать после того, как вы ознакомитесь с основами. Полезная нагрузка прерывания может быть любым сериализуемым значением — строкой, словарем или списком. В производственных средах обычно передают структурированный словарь, содержащий всю необходимую информацию для рецензента: черновик, классификацию, отправителя и URL рецензии, которую слой приложения может отобразить в интерфейсе или переслать через Slack или электронную почту.

Есть ещё одна подводная камень. Повторное вызов invoke() с полным начальным состоянием вместо Command(resume=value) является ошибкой: app.invoke({"messages": [...]}, config=config) на том же thread_id перезапускает граф для этой нити и удаляет состояние паузы. Прерванный граф следует продолжать исключительно с помощью Command(resume=value).

Итоговый проект: рабочий процесс сортировки электронных писем

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

Каждая из ранее рассмотренных концепций соответствует определённой части этого рабочего процесса:

• Состояние TypedDict с редюсерами представлено как EmailState, при этом используется оператор operator.add для обработки search_results.
• Последовательные связи соединяют read_email с classify_intent.
• Параллельные связи позволяют classify_intent разветвляться на search_docs и bug_tracking.
• Синхронизация через входящие связи означает, что оба параллельных узла поставляют данные в write_response.
• Условное маршрутизирование осуществляется с помощью функции route_by_priority, которая выбирает между ручной проверкой и автоматической отправкой.
• Структурированный вывод с использованием Pydantic имеет формат EmailClassification, который ограничивает ответ ЯИ.
• Интерупция выполняется через операцию human_review, которая приостанавливает работу в ожидании одобрения.
• Механизм контроля состояния — MemorySaver, который обеспечивает возможность интерупции и возобновления работы.

Состояние и схемы

Сначала идут состояние и схема Pydantic для структурированной классификации:

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification — это модель Pydantic, а не часть EmailState, что является примером разделения на TypedDict и Pydantic. Эта структура передаётся в ЯИИ в качестве цели в формате JSON schema, причём Pydantic проверяет полученный ответ и сразу же выдаёт ошибку, если он не соответствует. EmailState служит контейнером для графа и представляет собой TypedDict, поскольку LangGraph требует синтаксиса редьюсера и эффективной сериализации.

classification хранится в виде dict в EmailState, потому что после проверки ответа Pydantic функция .model_dump() преобразует его в обычный словарь перед сохранением в состояние. Это позволяет сохранять состояние в формате, сериализуемом любым checkpointerом, без необходимости использования Pydantic при восстановлении.

Самым важным полем является search_results: Annotated[list[str], operator.add]. Поскольку search_docs и bug_tracking выполняются параллельно и оба могут записывать результаты, редьюсер собирает их выходные данные вместе, тогда как без него одна из веток бы тихо перезаписала другую.

Создание узлов

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

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

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

classify_intent — это тот случай, когда Pydantic оказывается полезным. В строке structured_llm = llm.with_structured_output(EmailClassification) модель LLM оборачивается таким образом, что ответы должны соответствовать определенной схеме. Модель получает JSON-схему, полученную из EmailClassification, и обязана возвращать соответствующий валидный JSON; LangChain автоматически пытается выполнить запрос еще несколько раз перед тем, как выдать ошибку, если это не удается. Затем метод .model_dump() преобразует валидированный объект в обычный словарь для хранения в состоянии.

Два параллельных узла записывают данные в отдельные поля — search_results и bug_info — поэтому вызовы их LLM выполняются одновременно. Для поля search_results функция search_docs возвращает список ([f"[Docs] {response.content}"]), поскольку функция operator.add требует входных данных в виде списка. Функция bug_tracking возвращает обычную строку для поля bug_info; поскольку у этого поля нет функции сведения значений, достаточно простого присваивания.

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

human_review передаёт в функцию interrupt() структурированный словарь, а не простую строку, что является хорошей практикой в производственных средах. В этом словаре хранятся черновик, классификация и сообщение для рецензента, так что любая система, находящаяся с другой стороны — будь то веб-интерфейс, бот Slack или уведомление по электронной почте — получает всю необходимую информацию.

send_reply использует выражение state.get("approved", True) в качестве защитного значения по умолчанию; если значение approved по какой-то причине отсутствует в состоянии системы, функция всё равно отправляет сообщение. Это значение по умолчанию может быть изменено на False в случаях, когда профиль риска требует более осторожного подхода.

В реальных условиях search_docs и bug_tracking будут обращаться к реальным системам, таким как векторные базы данных, Jira или GitHub Issues, и декоратор LangChain @tool упрощает выполнение этих вызовов. Структура узлов остаётся прежней, причём вызов инструмента заменяет вызов большой языковой модели. Ещё одна проблема: write_response читает значение state["search_results"], поэтому если это всё ещё пустой список, доступ к нему может сорваться; этого можно избежать, инициализировав search_results: [] в начальном пакете данных вызова, чтобы редьюсер мог начать добавлять элементы именно оттуда.

Соединение узлов

Последовательные, параллельные и условные связи соединяют узлы между собой:

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority отправляет письма с высоким приоритетом на рассмотрение человека независимо от уровня уверенности. Классификации с уровнем уверенности ниже 0,8 также направляются на проверку, поскольку если модель не уверена в содержании письма, решение должен принять человек. Все остальные письма отправляются автоматически, что является разумным стандартным режимом в производственной среде, подлежащим настройке в зависимости от фактической частоты ошибок.

Этот механизм маршрутизации использует функцию add_conditional_edges, а не Command, поскольку route_by_priority представляет собой чистую функцию: она лишь анализирует текущее состояние и возвращает ключ, не создавая новых данных, что и предназначено для функции add_conditional_edges.

Вот все связи в графе:

• Последовательная обработка начинается с START, проходит через read_email и заканчивается в classify_intent.
• Расширение от classify_intent запускает одновременно как search_docs, так и bug_tracking.
• Возвратные связи передают результаты этих двух узлов в write_response.
• После write_response условная связь выбирает между human_review и send_reply.
• После завершения проверки human_review передаёт данные в send_reply, который заканчивается на этапе END.

Запуск рабочего процесса от начала до конца

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

config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

К этому моменту граф уже выполнил множество действий:

1. read_email проанализировал сообщение.
2. classify_intent пометил его как отчет о баге высокой приоритетности с использованием структурированного вывода Pydantic.
3. Поиск в документации и проверка в системе отслеживания багов выполнялись параллельно в рамках одного этапа обработки.
4. write_response подготовил ответ, используя всю эту информацию.
5. Узел human_review вызвал interrupt() и приостановил работу.

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

Затем рецензент утверждает запрос:

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph загрузил чекпоинт для "email-001", возобновил процесс human_review после вызова функции interrupt(), вернул значение "approve" в качестве результата, установил параметр approved=True в состоянии и перешел к функции send_reply, которая вернула статус подтверждения.

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

Для письма с низким приоритетом, но высокой степенью уверенности, функция route_by_priority возвращает значение "auto_send", и этап human_review пропускается. Все операции выполняются параллельно: считывание, классификация, поиск, составление проекта письма и его отправка; прерывание возникает только на том пути, где оно необходимо. В этом и заключается преимущество условного маршрутизирования.

Ещё одно важное замечание касается начального состояния. В отличие от Pydantic, структура TypedDict не устанавливает значения по умолчанию для отсутствующих полей. Если функция write_response пытается обратиться к элементу state["search_results"], а этот ключ отсутствует в начальном состоянии, возникает ошибка KeyError. Каждое поле должно присутствовать в словаре начального состояния, даже если изначально оно представляет собой пустую строку, пустой список или значение False.

Куда двигаться дальше

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

• Платформа LangGraph обеспечивает развертывание, управление чекпоинтами и горизонтальное масштабирование, что снимает нагрузку по эксплуатации инфраструктуры, когда цель — обработка реального трафика.
• LangSmith отслеживает и отлаживает каждую выполнение графа, показывая точное состояние на каждом этапе в случае некорректного поведения узла. Это становится важным, когда графы содержат пять и более узлов с условными путями.
• Подграфы объединяют более мелкие графы в более крупные многопроцессные системы. Каждый из них управляет собственным состоянием и взаимодействует через родительский граф, что является логичным следующим шагом после рабочих процессов с одним графом.
• Инструменты, вызываемые через LangChain с использованием @tool и ToolNode, позволяют агентам использовать внешние API, базы данных и сервисы прямо внутри графа; узлы search_docs и bug_tracking являются очевидными местами для их применения.
• Для сохранения данных в производственной среде необходимо установить langgraph-checkpoint-postgres, чтобы состояние сохранялось после перезагрузки, а замена API осуществляется всего одной строкой кода.

Краткий обзор

• Состояние представляет собой объект TypedDict. Поля, описанные как Annotated[list, operator.add], собирают значения при каждой записи, тогда как поля без функции обработки заменяются при каждой записи.
• Узел — это просто функция: она получает полное состояние и возвращает словарь с частичными обновлениями, после чего LangGraph выполняет их объединение.
• Рёбра определяют порядок выполнения. Используйте последовательные рёбра там, где важен порядок, параллельные рёбра для независимых операций, выполняющихся одновременно, и условные рёбра, когда данные должны выбрать нужную ветвь.
• Для хранения состояния графа используйте TypedDict, а для строгого контроля того, что возвращает большая языковая модель, применяйте Pydantic с методом .with_structured_output().
• Инструменты сохранения состояния обеспечивают его сохранность и возможность прерывания работы: MemorySaver во время разработки, PostgresSaver в производственной среде. Без таких инструментов прерывания невозможны.
• Прерывание — это всего лишь сохранённое, приостановленное состояние. Метод interrupt(payload) останавливает работу и сохраняет состояние, а метод Command(resume=value) возобновляет выполнение с точки, где оно было прервано.
• Параллелизм встроен в структуру: два рёбра, исходящих из одного узла, выполняют свои задачи одновременно, и каждое поле, записываемое этими узлами, требует функции сведения значений.

Вопросы для размышлений

1. MemorySaver подходит для разработки, а PostgresSaver — для производственной среды. Какие ещё фреймворки хранения заслуживают внимания для систем с высокой пропускной способностью, и какие компромиссы помогут сделать выбор?
2. route_by_priority отправляет письма с высоким приоритетом на проверку, а остальные — автоматически. Как можно настроить этот порог в реальной системе, и какие показатели покажут, что он установлен правильно?
3. Окончательная обработка маршрутизации происходит после write_response с использованием add_conditional_edges. Существует ли сценарий в рабочем процессе писем, где паттерн Command подошёл бы лучше, чем отдельная функция маршрутизации?

Связанные материалы

• Практические заметки: создание самокорректирующегося ИИ-агента с использованием Self-RAG и LangGraph — Подробное руководство по практическим заметкам: создание самокорректирующегося ИИ-агента с использованием Self-RAG и LangGraph, включающее шаблоны контрактов, проверок и готовые фрагменты кода для команд, внедряющих эту архитектуру.
• Создание агента для исследования финансовых рынков с использованием ReAct и LangGraph — Агент на базе LangGraph и ReAct берет тикер акции, последовательно запрашивает информацию по фундаментальным показателям, новостям, техническому анализу и инструментам аналитиков, а затем формирует краткий отчет с паузами для участия человека.

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

Практические заметки: Агент Text-to-SQL на Python: руководство по вызову инструментов LLM — Пошаговое руководство по использованию практических заметок: Агент Text-to-SQL на Python: руководство по вызову инструментов LLM, включающее шаблоны контрактов, проверок и готовые фрагменты кода для команд, внедряющих эту схему.

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

Самовосстанавливающаяся команда из нескольких агентов в LangGraph: архитектор, тестер, отладчик — Создание машинного состояния в LangGraph на Python, в котором архитектор, разработчик, среда тестирования pytest, отладчик и рецензент сотрудничают друг с другом до тех пор, пока тесты не пройдут.

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