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

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

Что уже умеем

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

Что вы узнаете в рамках статьи

• что такое состояние графа и как его описать через TypedDict;
• как добавлять узлы и соединять их рёбрами START и END;
• как редьюсеры, включая add_messages, объединяют обновления состояния;
• как строить ветвления через add_conditional_edges;
• как собрать цикл агента вручную и ограничить его через recursion_limit.

Ключевые термины

• Граф — описание процесса в виде узлов и связей между ними.
• Состояние (state) — общий словарь данных, который читают и обновляют все узлы графа.
• Узел (node) — обычная функция Python: получает состояние и возвращает словарь с изменениями.
• Ребро (edge) — связь «после узла A выполнить узел B».
• Условное ребро — связь, у которой следующий узел выбирает функция-маршрутизатор.
• Редьюсер (reducer) — правило, по которому обновление узла объединяется с текущим значением поля.
• Шаг (step) — одно выполнение узлов графа; LangGraph считает шаги и прерывает граф при превышении лимита.

Состояние, узлы и рёбра

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

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

class State(TypedDict):
    text: str
    words: int

def count_words(state: State) -> dict:
    return {"words": len(state["text"].split())}

def shout(state: State) -> dict:
    return {"text": state["text"].upper()}

builder = StateGraph(State)
builder.add_node("count_words", count_words)
builder.add_node("shout", shout)
builder.add_edge(START, "count_words")
builder.add_edge("count_words", "shout")
builder.add_edge("shout", END)

graph = builder.compile()
print(graph.invoke({"text": "привет от графа"}))

Что увидим в консоли: {'text': 'ПРИВЕТ ОТ ГРАФА', 'words': 3}. START и END — служебные точки входа и выхода. Метод compile() проверяет, что граф собран корректно, и возвращает объект, который запускается так же, как агент: через invoke. Узел shout не вернул поле words, поэтому оно сохранилось после узла count_words. Частая ошибка — забыть compile() и вызвать invoke у builder: у строителя графа такого метода нет.

Редьюсеры: как объединяются обновления

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

import operator
from typing import Annotated, TypedDict

class State(TypedDict):
    log: Annotated[list[str], operator.add]  # списки складываются
    status: str                               # значение заменяется

def step_a(state: State) -> dict:
    return {"log": ["A выполнен"], "status": "A"}

def step_b(state: State) -> dict:
    return {"log": ["B выполнен"], "status": "B"}

Если соединить START → step_a → step_b → END и запустить граф с {"log": ["старт"], "status": ""}, в консоли увидим log из трёх записей ['старт', 'A выполнен', 'B выполнен'] и status равный 'B'.

Для сообщений есть специальный редьюсер add_messages из langgraph.graph.message. Он добавляет новые сообщения в конец, а сообщение с уже существующим id заменяет. Чтобы не писать это поле каждый раз, используйте готовый класс MessagesState из langgraph.graph: в нём уже есть поле messages с этим редьюсером. От него можно наследоваться и добавлять свои поля.

Условные переходы

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

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

model = init_chat_model("openai:gpt-6-luna", temperature=0)

class Ticket(TypedDict):
    text: str
    topic: str
    answer: str

def classify(state: Ticket) -> dict:
    reply = model.invoke(
        "Определи тему обращения. Ответь одним словом: billing или tech.\n"
        f"Обращение: {state['text']}"
    )
    return {"topic": reply.content.strip().lower()}

def billing(state: Ticket) -> dict:
    return {"answer": "Передано в финансовый отдел"}

def tech(state: Ticket) -> dict:
    return {"answer": "Передано инженерам поддержки"}

def route(state: Ticket) -> Literal["billing", "tech"]:
    return "billing" if state["topic"] == "billing" else "tech"

builder = StateGraph(Ticket)
builder.add_node("classify", classify)
builder.add_node("billing", billing)
builder.add_node("tech", tech)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, ["billing", "tech"])
builder.add_edge("billing", END)
builder.add_edge("tech", END)
graph = builder.compile()

result = graph.invoke({"text": "С карты дважды списали оплату за подписку"})
print(result["topic"], "->", result["answer"])

Что увидим: billing -> Передано в финансовый отдел. Обратите внимание на разделение ролей: модель только классифицирует текст, а решение о переходе принимает обычная функция route. Это делает поведение предсказуемым. Если модель ответит неожиданным словом, маршрутизатор всё равно отправит обращение в ветку tech, а не сломает граф. Третий аргумент — список возможных узлов — помогает LangGraph проверить граф и нарисовать его схему.

Цикл агента своими руками и recursion_limit

Ребро может вести назад, и тогда получается цикл. Именно так устроен агент: узел модели, узел инструментов и условное ребро «есть вызовы инструментов — идём в инструменты, нет — заканчиваем».

from langchain.messages import ToolMessage
from langchain.tools import tool
from langgraph.graph import MessagesState

@tool
def multiply(a: int, b: int) -> int:
    """Умножает два числа."""
    return a * b

tools_by_name = {"multiply": multiply}
model_with_tools = model.bind_tools([multiply])

def call_model(state: MessagesState) -> dict:
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def call_tools(state: MessagesState) -> dict:
    results = []
    for call in state["messages"][-1].tool_calls:
        output = tools_by_name[call["name"]].invoke(call["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=call["id"]))
    return {"messages": results}

def should_continue(state: MessagesState):
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", call_tools)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue, ["tools", END])
builder.add_edge("tools", "model")
agent = builder.compile()

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Сколько будет 17 * 23?"}]},
    config={"recursion_limit": 10},
)
print(result["messages"][-1].content)

Метод bind_tools передаёт модели описания инструментов, как вы делали вручную в уроке 1.6. Что увидим: 17 * 23 = 391. Граф прошёл три шага: модель попросила multiply, узел инструментов выполнил его, модель дала ответ.

Параметр recursion_limit в config ограничивает число шагов графа. Если лимит исчерпан, LangGraph выбрасывает исключение GraphRecursionError из langgraph.errors. В свежих версиях лимит по умолчанию очень большой, поэтому для графов с циклами задавайте его явно, исходя из разумного числа обращений к модели.

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

• Возвращать из узла всё состояние целиком. Это работает, но легко затереть чужие поля устаревшими значениями. Возвращайте только то, что узел действительно изменил.
• Забыть редьюсер у списка. Без operator.add или add_messages каждый узел перезаписывает список, и история пропадает. Если записи исчезают, первым делом проверьте Annotated.
• Доверять модели выбор перехода напрямую. Не возвращайте сырой ответ модели как имя узла. Пусть маршрутизатор сопоставляет ответ с известными ветками и имеет ветку по умолчанию.
• Цикл без ограничения. Для графа с обратным ребром всегда задавайте recursion_limit и обрабатывайте GraphRecursionError.

Наиболее частые вопросы

• Когда выбирать LangGraph, а когда create_agent? — Если модель свободно выбирает инструменты, хватит create_agent. Если в процессе есть обязательные шаги и ветки, которые вы хотите контролировать кодом, стройте граф.
• Можно ли вставить агента из create_agent узлом графа? — Да, скомпилированный агент сам является графом, его можно вызвать внутри функции-узла или добавить узлом напрямую.
• Работают ли в графе checkpointer и thread_id? — Да, checkpointer передаётся в compile(checkpointer=...), а thread_id — в config, как в уроке 2.4.
• Как увидеть схему графа? — Вызовите graph.get_graph().draw_mermaid(): получите текст схемы, который можно вставить в любой просмотрщик Mermaid.

Итоги

Отличная работа! Теперь вы умеете строить не только агентов «под ключ», но и собственные процессы: описывать состояние, добавлять узлы, соединять их рёбрами, ветвить путь условными переходами и собирать циклы. Вы своими руками собрали тот самый цикл «модель — инструменты», который раньше скрывался внутри create_agent, и знаете, как не дать ему работать бесконечно.

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