15. Создание вашего первого графа с помощью LangGraph

В части IV мы определили инструменты, подключили их к LLM и реализовали цикл агента, который повторяет цикл «принять решение — выполнить». Управление циклом, запуск инструментов, когда LLM их запрашивал, понимание того, когда остановиться — мы вручную закодировали каждую часть этого процесса.

В этой главе мы построим тот же агент совершенно иным способом. Вместо того чтобы писать поток напрямую, мы зарегистрируем шаги (узлы) и правила соединения (рёбра) во фреймворке LangGraph и позволим ему заниматься исполнением. Поведение идентично главе 14, но способ построения меняется.

Эта глава охватывает четыре основных понятия — StateGraph, узлы, рёбра и State (состояние) — а затем перерабатывает цикл агента из главы 14 в граф LangGraph. В следующих главах глава 16 посвящена условной маршрутизации и готовым компонентам, а глава 17 — сохранению состояния, которое позволяет агенту возобновить работу с того места, где он был прерван.

15.1) Зачем нужны графы?

15.1.1) Ограничения существующего цикла агента

Давайте вернёмся к циклу агента из главы 14. Если убрать обработку ошибок и прочие детали, основная структура выглядела так:

# Цикл агента из главы 14 — основная структура (упрощённо)
messages = [
    SystemMessage(content="Вы полезный помощник."),
    HumanMessage(content=user_input),
]
 
for step in range(max_steps):
    # Просим LLM решить, каким будет следующее действие
    ai_message = llm_with_tools.invoke(messages)
    messages.append(ai_message)
 
    # Если вызов инструмента не запрошен, возвращаем финальный ответ
    if not ai_message.tool_calls:
        return ai_message.content
 
    # Выполняем запрошенные инструменты
    for tool_call in ai_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        messages.append(tool_message)

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

• Восстановление после сбоя — Если агент даёт сбой на шаге 7 из 10-шаговой исследовательской задачи, он должен иметь возможность возобновить работу с шага 7, а не начинать всё сначала.
• Запросы на одобрение — Прежде чем агент выполнит критическую операцию, он должен иметь возможность приостановиться и спросить у человека: «Можно продолжать?»
• Мониторинг в реальном времени — Пользователи должны иметь возможность видеть, что агент делает в данный момент и какие инструменты он вызывает.
• Визуализация и отладка — Должна быть доступна диаграмма, показывающая, как работает агент, чтобы при возникновении проблем можно было отследить, какой шаг пошёл не так.

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

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

Цикл агента из главы 14 обрабатывает всю логику напрямую, поэтому фреймворку не за что зацепиться. Чтобы воспользоваться тем, что предлагает LangGraph, нам нужно перестроить агента в структуре, которую LangGraph понимает — в граф. Именно об этом эта глава.

15.1.2) Что такое LangGraph?

LangGraph — это фреймворк оркестрации, который определяет и исполняет рабочие процессы агентов в виде графов. Граф здесь означает структуру, в которой каждый узел (шаг), выполняемый агентом, соединён рёбрами (правилами соединения).

В LangGraph вы разбиваете рабочий процесс на независимые узлы и соединяете их рёбрами. Затем LangGraph обходит граф, выполняя каждый узел по пути. Вот как выглядит цикл агента из главы 14, выраженный в виде графа:

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

В главе 14 весь рабочий процесс жил в циклах for, проверках if и другом написанном вручную коде. С LangGraph вы определяете, что делает каждый узел, и связываете узлы рёбрами. Короче говоря, вы переходите от кодирования рабочего процесса к его декларированию в виде структуры.

LangGraph не заменяет ничего из того, что вы изучили в главах 12–14. Определения инструментов, bind_tools(), tool_calls, ToolMessage — всё это по-прежнему используется внутри узлов, точно так же, как и раньше.

В следующем разделе поочерёдно рассматриваются основные компоненты LangGraph — StateGraph, State, узлы и рёбра.

15.2) Компоненты LangGraph: StateGraph, State, узлы, рёбра

В этом разделе поочерёдно рассматриваются четыре основных компонента LangGraph. Мы начнём с StateGraph — класса, который объединяет State, узлы и рёбра в граф, — а затем рассмотрим каждую из частей (State, узлы, рёбра), входящих в его состав.

15.2.1) StateGraph

StateGraph — это класс, используемый для построения графов в LangGraph. Вы указываете State, которым будет управлять граф, добавляете узлы, соединяете их рёбрами, а затем компилируете для получения исполняемого графа.

Давайте посмотрим, как это работает.

Создание экземпляра StateGraph

Вызовите конструктор StateGraph для создания экземпляра. Вам нужно передать схему State (сам класс) в качестве параметра. Здесь мы используем MessagesState — предопределённый State, который предоставляет LangGraph для управления списками сообщений. Подробности мы рассмотрим в разделе 15.2.2.

from langgraph.graph import StateGraph, MessagesState
 
builder = StateGraph(MessagesState)

Добавление узлов

Используйте add_node() для регистрации узла. Узел — это функция Python, которая принимает текущий State и возвращает те части, которые она хочет изменить. Функции узлов мы рассмотрим подробно в разделе 15.2.3.

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder.add_node(say_hello)    # имя узла становится "say_hello"

Соединение рёбер

Используйте add_edge(source, target) для соединения узлов. source — это то, откуда начинается ребро; target — куда оно ведёт. START и END — специальные маркеры точек входа и выхода графа. Рёбра мы рассмотрим в разделе 15.2.4.

from langgraph.graph import START, END
 
builder.add_edge(START, "say_hello")   # граф стартует → запуск say_hello
builder.add_edge("say_hello", END)     # say_hello завершается → конец графа

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

Вызов compile() проверяет структуру графа и создаёт исполняемый объект. Скомпилированный граф запускается с помощью invoke(), передавая начальные значения State.

graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)

Теперь давайте соберём всё вместе и построим простой граф:

from langgraph.graph import StateGraph, MessagesState, START, END
 
def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder = StateGraph(MessagesState)
 
builder.add_node(say_hello)
builder.add_edge(START, "say_hello")
builder.add_edge("say_hello", END)
 
graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)

Вывод:

hello world

Когда вы вызываете invoke(), граф выполняется в порядке START → say_hello → END. say_hello вернул словарь с ключом messages, и это значение было добавлено к списку messages в MessagesState. Мы разберём, как это работает, в разделе 15.2.2. Итог в том, что извлечение содержимого последнего сообщения даёт нам "hello world".

15.2.2) State: данные, которые протекают через граф

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

Определение State

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

from typing_extensions import TypedDict
 
class AgentState(TypedDict):
    messages: list       # список сообщений
    llm_calls: int       # счётчик вызовов LLM

Далее вы передаёте AgentState при создании StateGraph и используете его в качестве подсказки типа для ваших функций узлов.

Редьюсеры

Когда узел возвращает значение, соответствующее поле State обновляется. Поведение по умолчанию — перезапись: если узел возвращает {"llm_calls": 3}, llm_calls просто становится равным 3, независимо от того, каким оно было раньше.

Но некоторые поля нужно дополнять, а не перезаписывать. Что произойдёт, если messages будет перезаписан? Каждый раз, когда узел возвращает новое сообщение, вся история разговора исчезает. Для messages добавление — правильное поведение.

LangGraph позволяет задать отдельную стратегию обновления для каждого поля с помощью функции-редьюсера. Вы указываете редьюсер в качестве второго аргумента в Annotated:

from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # редьюсер: добавление
    llm_calls: int                            # без редьюсера: перезапись

add_messages — это редьюсер, предоставляемый LangGraph. Вместо замены списка он добавляет новые сообщения к тем, что уже есть. Поскольку этот редьюсер установлен для поля messages, любое значение, которое узел возвращает для messages, будет добавлено. У llm_calls нет редьюсера, поэтому возвращаемые значения просто перезаписывают то, что было там раньше.

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

MessagesState

LangGraph поставляется с предопределённым State под названием MessagesState. Вот как он выглядит под капотом:

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

Та же структура, которую мы только что рассмотрели — поле messages с уже подключённым редьюсером add_messages.

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

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int  # поведение перезаписи по умолчанию

15.2.3) Узлы: функции, обновляющие State

Узел — это функция Python, которая выполняет одну конкретную задачу внутри графа.

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

Две вещи, которые нужно знать при написании функций узлов:

Правило 1: она получает текущий State в качестве аргумента. LangGraph передаёт текущий объект State при запуске узла.

Правило 2: она возвращает только те части, которые хочет изменить, а не весь State. Узлы не изменяют State напрямую. Просто верните поля, которые хотите обновить, и LangGraph объединит их с существующим State согласно правилам редьюсера каждого поля.

Используйте add_node() для добавления узла в StateGraph:

builder.add_node(say_hello)          # имя функции "say_hello" становится именем узла
builder.add_node("my_node", my_func) # вы также можете указать имя явно

15.2.4) Рёбра: правила, соединяющие узлы

Ребро определяет, «после того как этот узел завершится, что запускается следующим?» Существует два типа.

Обычные рёбра

Обычное ребро связывает фиксированный переход между двумя узлами. Используйте add_edge(source, target) — source — это начальный узел, target — узел назначения.

builder.add_edge(START, "say_hello")       # когда граф стартует, запускаем say_hello
builder.add_edge("say_hello", "llm_call")  # после say_hello запускаем llm_call
builder.add_edge("llm_call", END)          # после llm_call завершаем граф

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

Условное ребро выбирает следующий узел во время выполнения на основе текущего State. Используйте add_conditional_edges(source, routing_function) — source — это начальный узел, а routing_function — функция, которая принимает текущий State и возвращает имя следующего узла:

from langgraph.graph import END
 
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"   # запрошен вызов инструмента → идём в tool_node
    return END               # вызов инструмента не запрошен → конец
 
builder.add_conditional_edges("llm_call", should_continue)

add_conditional_edges("llm_call", should_continue) сообщает LangGraph: «когда llm_call завершится, вызови should_continue, чтобы решить, что запускать следующим». should_continue направляет к "tool_node", если у последнего сообщения есть tool_calls, или к END, если нет. На практике это означает, что граф продолжает работу к узлу выполнения инструментов, когда LLM запрашивает вызов инструмента, и завершается, когда не запрашивает.

Теперь, когда мы рассмотрели все четыре компонента, в следующем разделе мы используем их для переработки цикла агента из главы 14 в граф LangGraph.

15.3) Переработка цикла агента в граф

Давайте перестроим цикл агента из главы 14 с помощью LangGraph. Поведение идентично главе 14 — LLM принимает решения, инструменты выполняются на основе запросов LLM, и цикл повторяется до завершения. Единственное, что меняется, — это то, как мы структурируем этот поток.

Вот как будет выглядеть готовый граф:

Граф циклически переключается между llm_call и tool_node, пока LLM не перестанет запрашивать вызовы инструментов, после чего он переходит к END. Давайте построим его шаг за шагом.

15.3.1) Определение State

Мы создаём подкласс MessagesState из раздела 15.2.2 для определения State агента. Поле messages наследуется от MessagesState, и мы добавляем поле llm_calls для отслеживания количества вызовов LLM.

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int    # счётчик вызовов LLM (перезапись)

Список messages будет накапливать пользовательские вводы (HumanMessage), ответы LLM (AIMessage) и результаты выполнения инструментов (ToolMessage) по порядку.

15.3.2) Построение узлов

Сначала давайте настроим инструменты и модель из главы 14:

from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """Получает текущую погоду для города."""
    fake_data = {"Tokyo": "18°C, облачно", "Cairo": "31°C, солнечно"}
    return fake_data.get(city, f"Нет данных о погоде для {city}.")
 
@tool
def calculate(expression: str) -> str:
    """Вычисляет простое арифметическое выражение. Пример: '3 * 21'."""
    return str(eval(expression))  # Внимание: eval() представляет угрозу безопасности. Не используйте в продакшене.
 
tools = [get_weather, calculate]
tool_map = {t.name: t for t in tools}
 
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)

Теперь давайте напишем две функции узлов.

Узел llm_call — Вызывает LLM и возвращает ответ:

def llm_call(state: AgentState):
    """Вызывает LLM и возвращает ответ."""
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

model_with_tools.invoke() вызывает LLM, и ответ упаковывается под ключом messages в возвращаемом словаре. Редьюсер добавляет его к существующим messages в AgentState. llm_calls возвращает текущий счётчик плюс 1, перезаписывая предыдущее значение.

Узел tool_node — Выполняет инструменты, запрошенные LLM, и возвращает результаты:

def tool_node(state: AgentState):
    """Выполняет инструменты, запрошенные LLM."""
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        results.append(tool_message)
    return {"messages": results}

Поскольку tool_node всегда выполняется сразу после llm_call, последним сообщением в messages гарантированно будет AIMessage, которое только что произвёл LLM. Поле tool_calls этого сообщения содержит вызовы инструментов, запрошенные LLM. Узел запускает каждый инструмент, собирает результаты в results и возвращает их под ключом messages — редьюсер заботится о добавлении их к существующему списку.

15.3.3) Условное ребро

После того как llm_call завершится, нам нужно условное ребро, чтобы решить, запускать ли tool_node или завершить граф. Это следует тому же шаблону из раздела 15.2.4:

from typing import Literal
from langgraph.graph import END
 
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    """Решает, выполнять ли инструменты или завершить граф."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

Если last_message.tool_calls присутствует, LLM запрашивает вызов инструмента, поэтому мы направляемся к tool_node. В противном случае мы направляемся к END, и граф завершается.

Подсказка возвращаемого типа Literal["tool_node", "__end__"] объявляет возможные пункты назначения, которые может вернуть эта функция. LangGraph нужна эта подсказка, чтобы правильно нарисовать пути условных рёбер при визуализации графа. Она не влияет на поведение во время выполнения.

"__end__" — это базовое строковое значение END. Поскольку Literal принимает только строковые литералы, мы пишем "__end__" вместо END.

15.3.4) Сборка и запуск графа

Пришло время связать всё воедино. Давайте соберём State, узлы и условное ребро в StateGraph и скомпилируем:

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(AgentState)
 
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
 
builder.add_edge(START, "llm_call")                         # старт → llm_call
builder.add_conditional_edges("llm_call", should_continue)  # llm_call → tool_node или END
builder.add_edge("tool_node", "llm_call")                   # tool_node → llm_call (цикл)
 
agent = builder.compile()

Ребро от tool_node обратно к llm_call создаёт цикл. Выполнение продолжает циклически повторяться, пока LLM не ответит финальным ответом вместо запроса очередного вызова инструмента, после чего цикл завершается.

Давайте запустим его:

from langchain_core.messages import HumanMessage
 
result = agent.invoke({
    "messages": [HumanMessage(content="Узнай температуру в Каире, затем умножь это число на 3.")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\nВсего вызовов LLM: {result['llm_calls']}")

Вывод:

Текущая температура в Каире: 31°C. Умноженная на 3 = 93.
 
Всего вызовов LLM: 3

Агент вызвал get_weather("Cairo"), увидел результат, вызвал calculate("31 * 3") и произвёл финальный ответ — тот же результат, который мы получили в главе 14.

Изучение полной истории сообщений показывает каждый шаг, записанный в messages, по порядку:

for message in result["messages"]:
    message.pretty_print()

Вывод:

================================ Human Message =================================
Узнай температуру в Каире, затем умножь это число на 3.
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_DiL9WF)
  Args:
    city: Cairo
================================= Tool Message =================================
Name: get_weather
31°C, солнечно
================================== Ai Message ==================================
Tool Calls:
  calculate (call_wa6RqWST)
  Args:
    expression: 31 * 3
================================= Tool Message =================================
Name: calculate
93
================================== Ai Message ==================================
Текущая температура в Каире: 31°C. Умноженная на 3 = 93.

15.3.5) Визуализация графа

В Jupyter-ноутбуке agent.get_graph().draw_mermaid_png() отрисовывает структуру графа в виде изображения прямо в выводе ячейки.

from IPython.display import Image, display
 
display(Image(agent.get_graph().draw_mermaid_png()))

В терминальной среде вместо этого сохраните его как файл PNG.

agent.get_graph().draw_mermaid_png(output_file_path="agent_graph.png")

Сгенерированное изображение:

Сплошные линии — это обычные рёбра, а пунктирные линии — условные рёбра. Эта диаграмма автоматически генерируется из кода.

15.3.6) Лимит рекурсии

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

Чтобы увидеть, как работает подсчёт, посмотрите на предыдущий запуск. Вызов и get_weather, и calculate посетил узлы в следующем порядке:

llm_call(1) → tool_node(2) → llm_call(3) → tool_node(4) → llm_call(5) → END

Всего 5 посещений узлов. Если вы установите recursion_limit равным 3, лимит сработает на 3-м посещении, и запуск будет прерван:

from langgraph.errors import GraphRecursionError
 
try:
    result = agent.invoke(
        {"messages": [HumanMessage(content="Узнай температуру в Каире, затем умножь это число на 3.")],
         "llm_calls": 0},
        config={"recursion_limit": 3},
    )
except GraphRecursionError:
    print("Агент достиг лимита рекурсии — остановка выполнения.")

Вывод:

Агент достиг лимита рекурсии — остановка выполнения.

Установите лимит, передав config={"recursion_limit": число} в invoke(). Правильное значение зависит от вашего сценария использования и сложности вашего графа. Начните с щедрого числа и настройте его через тестирование.