LangGraph на простом примере: состояние, узлы и условный переход

В поддержку приходят короткие письма: «Проверьте счёт за апрель» и «Нужен акт за апрель». Первое надо передать на проверку платежей, второе — на поиск документа. Третье письмо, «Есть вопрос», пока не даёт даже темы. У каждого обращения есть ID, и этот ID должен дойти до следующего шага, независимо от выбранного пути.

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

Какие данные проходят по графу

Начальное состояние содержит request_id, текст и customer_id. Узел classify добавляет kind, затем функция choose смотрит на полученное состояние. Если клиент неизвестен или тема не распознана, заявка идёт в clarify. В других случаях она попадает в billing либо documents. Конечный узел записывает next_step для вызывающего приложения. Таким образом, результат остаётся связан с исходным ID, а отсутствие клиента не превращается в запрос к чужому счёту.

В коде темы определяются по двум группам слов, чтобы пример запускался без ключа и внешнего сервиса. В рабочем процессе classify можно заменить вызовом модели или существующего классификатора, если письма приходят свободным языком. Остальная часть графа продолжит принимать ту же категорию и применять те же правила маршрута. Для неоднозначного письма функция возвращает unknown, а не угадывает первый подходящий отдел. Для запуска установите пакет командой pip install -U langgraph.

from typing import Literal, TypedDict
from langgraph.graph import START, END, StateGraph

class Ticket(TypedDict, total=False):
    request_id: str
    text: str
    customer_id: str | None
    kind: Literal["billing", "documents", "unknown"]
    next_step: str

def classify(state: Ticket) -> dict:
    text = state["text"].lower()
    billing_hit = "счёт" in text or "платёж" in text
    documents_hit = "акт" in text or "документ" in text
    if billing_hit and documents_hit:
        return {"kind": "unknown"}
    if billing_hit:
        return {"kind": "billing"}
    if documents_hit:
        return {"kind": "documents"}
    return {"kind": "unknown"}

def choose(state: Ticket) -> Literal["billing", "documents", "clarify"]:
    if not state.get("customer_id") or state["kind"] == "unknown":
        return "clarify"
    return state["kind"]

def billing(state: Ticket) -> dict:
    return {"next_step": "billing_lookup"}

def documents(state: Ticket) -> dict:
    return {"next_step": "document_search"}

def clarify(state: Ticket) -> dict:
    return {"next_step": "ask_customer_or_topic"}

builder = StateGraph(Ticket)
builder.add_node("classify", classify)
builder.add_node("billing", billing)
builder.add_node("documents", documents)
builder.add_node("clarify", clarify)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", choose, {
    "billing": "billing", "documents": "documents", "clarify": "clarify"
})
for branch in ("billing", "documents", "clarify"):
    builder.add_edge(branch, END)
graph = builder.compile()

cases = [
    {"request_id": "R-41", "text": "Проверьте счёт за апрель", "customer_id": "C-7"},
    {"request_id": "R-42", "text": "Нужен акт за апрель", "customer_id": "C-7"},
    {"request_id": "R-43", "text": "Есть вопрос", "customer_id": None},
    {"request_id": "R-44", "text": "Нужен акт и вопрос по счёту", "customer_id": "C-7"},
]
for case in cases:
    result = graph.invoke(case)
    print(result["request_id"], result["kind"], result["next_step"])

Первые три строки вывода будут такими: R-41 billing billing_lookup, R-42 documents document_search и R-43 unknown ask_customer_or_topic. Четвёртое письмо содержит обе темы и тоже попадёт в ask_customer_or_topic. Здесь billing_lookup и document_search — названия следующих действий приложения, а не выполненные обращения к платёжной системе или хранилищу. Именно это позволяет проверить маршрут без внешних побочных эффектов.

В приложении вызывающий код передаёт в graph.invoke() одну заявку и получает полный результат. По request_id он связывает действие с записью в своей системе, а по next_step выбирает обработчик. Так граф не теряет исходный контекст, когда ветки расходятся. Для регрессионного теста достаточно сохранять несколько входных заявок и ожидаемых next_step: изменение узла классификации сразу покажет, какие обращения ушли в другой отдел.

Где граф начинает окупаться

В StateGraph узел возвращает только обновлённые поля, а не собирает весь объект вручную после каждого шага. Условный переход выполняется после classify и читает уже обновлённое состояние. START задаёт вход, END закрывает каждую ветку, compile() создаёт исполняемый граф, а invoke() проводит одну заявку по нему. Если дальше понадобится сверить ID клиента, запросить акт, записать исход и обработать ошибку инструмента, новые узлы будут добавляться в понятные места, не смешивая классификацию, доступ и ответ в одном длинном вызове.

На трёх ветках ту же задачу можно решить обычной функцией Python. Причина выбрать граф появляется, когда процесс надо наблюдать и менять по шагам: например, после проверки платежа разрешён переход только к ответу либо к уточнению, а после поиска акта — к проверке его версии. Тогда схема маршрута становится частью проверяемого поведения. Сохранение состояния между запусками требует отдельного checkpointer; один compile() его не включает. Сначала зафиксируйте входы и ожидаемые пути, затем добавляйте память или повторные попытки там, где они нужны процессу.

Проверяйте и пограничные письма. Фраза «Нужен акт и вопрос по счёту» содержит оба учебных маркера; четвёртый кейс показывает, что классификатор отправляет её на уточнение. Если бизнес решит, что счёт всегда имеет приоритет, правило можно поменять, но ожидаемый результат теста тоже нужно изменить явно. Перед заменой простого классификатора моделью оставьте этот кейс в проверочном наборе: граф применит выданную категорию, поэтому качество определения темы остаётся отдельной проверяемой частью системы.

Если хотите пройти путь от такого маршрута до рабочего агентного сервиса, в платном курсе AI Agents PRO: LangGraph, AutoGen и LLMOps в продакшне собирают граф состояний, подключают инструменты и RAG, затем оформляют API, тесты качества и наблюдаемость. Открытый урок «FSM и DAG: оркестрация шагов» помогает сначала увидеть логику переходов. Откройте программу и пробный урок.

Рядом с этой темой полезны разбор повторного действия агента после таймаута, статья о разнице между планом и выполнением и другие материалы по инженерии агентов.