LangGraph на Python: как создать AI-агента с инструментами

У учебного сервиса api статус HTTP 503. Нужно решить, что проверять дальше: сначала запросить статус, затем, если сервис отвечает ошибкой, открыть инструкцию по диагностике. В обычном Python-коде хватило бы нескольких if. Но если выбор следующего шага доверить языковой модели, появляется ещё одна задача: видеть, какую функцию она попросила вызвать, что функция вернула и почему выполнение закончилось.

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

Мы построим небольшой граф на Python с двумя инструментами: get_status и get_runbook. По ходу разберём, зачем нужны StateGraph, MessagesState, ToolNode и условное ребро, затем запустим пример и проверим ошибки.

Сначала роль модели сыграет обычная функция с предсказуемыми решениями. Вы увидите настоящий запуск LangGraph без API-ключа и платных запросов. Способность реальной LLM выбрать правильную проверку такой запуск не оценивает — об этом отдельно поговорим в конце.

Подготавливаем пример

Для этого руководства достаточно знать Python-функции и словари. LangGraph отвечает за узлы, состояние и переходы. Классы сообщений и декоратор @tool приходят из langchain-core. В официальном руководстве LangGraph эти части объединены в граф с инструментами.

Мы используем CPython 3.12 и langgraph==1.2.19 — версию от 30 октября 2026 года. Пример также закрепляет langchain-core==1.7.4 и pytest==9.2.0. Эти версии были установлены и проверены на CPython 3.12.12; зависимостям второго уровня установщик может выбрать более новые совместимые выпуски, поэтому повторная проверка после установки важна.

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

git clone https://github.com/xnova-tech/langgraph-tool-agent-example.git
cd langgraph-tool-agent-example
git checkout a9650baffd47bf10cea3772c7de4069c9441c9ed
python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements-dev.txt

Команды с .venv/bin/python относятся к macOS и Linux. В Windows исполняемый файл окружения находится в .venv\Scripts\python.exe; порядок создания и расположение каталогов описаны в документации Python. Основная проверка ниже выполнена на Python 3.12.12 в Linux-контейнере; это не обещание одинакового результата для всех систем.

После установки у вас должны быть agent_graph.py, requirements-dev.txt и каталог tests/ в текущей директории. Именно из корня публичного примера выполняются следующие команды. Если импорт не удаётся, прежде всего проверьте интерпретатор виртуального окружения и успешность установки зависимостей.

Какие функции разрешено вызывать

Представим два простых источника данных. get_status("api") возвращает учебную строку HTTP 503, а get_runbook("api") — подсказку «проверить журнал и upstream». Здесь upstream — сервис, от которого зависит наш API. Функции читают заранее заданные значения из памяти: ни к какому серверу они не подключаются.

Инструмент объявляется обычной Python-функцией с декоратором @tool. Декоратор позволяет передать модели имя, описание и схему аргументов функции. Затем ToolNode может выполнить такой вызов. В первом примере LangGraph показано, как модель получает доступ к объявленным инструментам.

Посмотрите на ограничение аргументов в get_status из agent_graph.py:

@tool
def get_status(service: str) -> str:
    """Прочитать учебный HTTP-статус сервиса по его имени."""
    if service not in {"api", "worker"}:
        raise ValueError("service not allowed")
    return {"api": "HTTP 503", "worker": "HTTP 200"}[service]

Функция разрешает только два имени. Если узел модели передаст другое значение, например путь к файлу, проверка завершится ошибкой. get_runbook устроен сходно, но принимает только api. Обе функции только читают заданные данные и не изменяют состояние.

Даже успешный вызов get_status пока сообщает только значение из учебного словаря. Он не объясняет причину 503 и не подтверждает состояние настоящего сервиса.

Как сообщения переходят между узлами

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

Нашему графу нужно хранить три вида сообщений:

• HumanMessage — вопрос: «Что проверить у api?».
• AIMessage — ответ узла model. Он может содержать обычный текст или список tool_calls: какой инструмент нужен, с какими аргументами и под каким идентификатором.
• ToolMessage — ответ исполненного инструмента. Его tool_call_id указывает на запрос, которому принадлежит результат.

Например, первое решение содержит вызов get_status с аргументом service="api" и идентификатором call-status. После выполнения в историю попадает ToolMessage со значением HTTP 503 и тем же tool_call_id. Благодаря этому можно сопоставить запрос и ответ даже тогда, когда инструментов несколько.

MessagesState хранит историю сообщений, а обновления поля messages объединяются специальным механизмом, а не заменяют всю историю. В разделе о MessagesState описан reducer add_messages: он добавляет новые сообщения и умеет обновлять существующие по идентификатору. В нашем маршруте так сохраняются вопрос, запрос инструмента и его результат.

Когда управление вернётся в model, узел увидит историю и сможет решить, нужна ли инструкция по диагностике. Здесь история сохраняется внутри одного запуска invoke. Для восстановления её между отдельными запусками понадобятся контрольные точки (checkpoints); в этом примере они не настроены.

Соединяем узлы в граф

Пора связать две функции и механизм выбора. В графе будет только два узла: model формирует очередное AIMessage, а tools выполняет запрошенные функции. ToolNode выполняет вызовы инструментов из сообщений модели и возвращает их результаты как сообщения инструментов. Описание ToolNode есть в справочнике LangGraph.

После model нам нужен выбор направления. Условный переход после ответа модели направляет выполнение к инструментам, если модель запросила вызов, и завершает цикл иначе. Готовая функция tools_condition реализует эту проверку; официальный quickstart показывает эквивалентный переход.

Для нашего значения HTTP 503 маршрут получается таким: START → model → tools → model → tools → model → END. Сначала model просит get_status, после ответа просит get_runbook, а после инструкции выдаёт заключительное сообщение без tool_calls. Именно отсутствие нового запроса позволяет закончить выполнение.

Полный agent_graph.py приведён ниже — его можно прочитать здесь или открыть в закреплённом репозитории. Обратите внимание на scripted_decision: она возвращает заранее выбранные AIMessage. Так можно отдельно проверить исполнение графа, не приписывая ему способности реальной языковой модели.

"""Минимальный граф с реальным ToolNode и заранее заданными решениями."""

from typing import Callable

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition


@tool
def get_status(service: str) -> str:
    """Прочитать учебный HTTP-статус сервиса по его имени."""
    if service not in {"api", "worker"}:
        raise ValueError("service not allowed")
    return {"api": "HTTP 503", "worker": "HTTP 200"}[service]


@tool
def get_runbook(service: str) -> str:
    """Прочитать учебную инструкцию по диагностике API."""
    if service != "api":
        raise ValueError("service not allowed")
    return "проверить журнал и upstream"


def scripted_decision(messages: list) -> AIMessage:
    """Тестовый выбор следующего шага; это не языковая модель."""
    results = [m for m in messages if isinstance(m, ToolMessage)]
    if not results:
        return AIMessage(
            content="",
            tool_calls=[{"name": "get_status", "args": {"service": "api"}, "id": "call-status"}],
        )
    last = results[-1]
    if last.name == "get_status" and last.content == "HTTP 503":
        return AIMessage(
            content="",
            tool_calls=[{"name": "get_runbook", "args": {"service": "api"}, "id": "call-guide"}],
        )
    if last.name == "get_status":
        return AIMessage(content="Сервис ответил; дополнительные проверки не запрошены.")
    return AIMessage(content=f"HTTP 503; следующий шаг: {last.content}.")


def build_graph(decide: Callable[[list], AIMessage] = scripted_decision):
    """Собрать граф; вместо decide можно передать адаптер реальной LLM."""

    def model_node(state: MessagesState) -> dict:
        return {"messages": [decide(state["messages"])]}

    builder = StateGraph(MessagesState)
    builder.add_node("model", model_node)
    builder.add_node("tools", ToolNode([get_status, get_runbook], handle_tool_errors=True))
    builder.add_edge(START, "model")
    builder.add_conditional_edges("model", tools_condition)
    builder.add_edge("tools", "model")
    return builder.compile()


def main() -> None:
    result = build_graph().invoke(
        {"messages": [HumanMessage(content="Что проверить у api?")]},
        config={"recursion_limit": 12},
    )
    for item in result["messages"]:
        if isinstance(item, ToolMessage):
            print(f"{item.name} -> {item.content}")
    print("completed: " + result["messages"][-1].content)


if __name__ == "__main__":
    main()

В build_graph узел model_node добавляет в состояние одно новое сообщение. ToolNode знает только о get_status и get_runbook. Строка builder.add_conditional_edges("model", tools_condition) выбирает следующий узел по последнему ответу, а builder.add_edge("tools", "model") возвращает управление после выполнения инструмента.

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

Запускаем граф и читаем результат

Перейдите в корень склонированного репозитория, где лежит agent_graph.py, и выполните:

.venv/bin/python -I agent_graph.py

Программа выводит:

get_status -> HTTP 503
get_runbook -> проверить журнал и upstream
completed: HTTP 503; следующий шаг: проверить журнал и upstream.

Здесь HTTP 503 взят из словаря в коде, а не из ответа реального HTTP-сервера. Узлы графа обрабатывают заранее подготовленное значение.

Смысл первых двух строк в том, что они выведены из ToolMessage, находящихся в возвращённой истории. Последняя строка — текст заключительного AIMessage. Если инструмент не запускался, его сообщения в истории не будет; одного красивого заключительного ответа для подтверждения выполнения недостаточно.

Запустите тесты из того же каталога:

.venv/bin/python -m pytest -p no:cacheprovider -p no:terminal tests

Параметр no:terminal отключает вывод pytest, поэтому при успешном завершении команда ничего не печатает. Шесть тестов проверяют маршрут с двумя инструментами, завершение без вызова, неизвестную функцию, запрещённый аргумент, ограничение числа шагов и списки допустимых значений.

Чтобы увидеть названия тестов и результат каждого из них, выполните .venv/bin/python -m pytest -v tests. В отличие от команды выше, этот запуск включает подробный вывод pytest.

Что изменится с настоящей языковой моделью

Сейчас model_node вызывает scripted_decision. Эта функция следует простому правилу: ещё нет ответа инструмента — запросить статус; получен HTTP 503 — запросить инструкцию; получена инструкция — закончить. Весь выбор заранее записан в Python.

При подключении LLM вместо этой функции нужен адаптер, который передаёт модели историю сообщений и получает от неё AIMessage. Обычно клиент модели связывают с разрешёнными инструментами через bind_tools; затем результат invoke возвращают в поле messages. Принцип такого узла показан в официальном quickstart.

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

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

Если граф работает не так, как ожидалось

Не импортируется langgraph. Убедитесь, что вы находитесь в склонированном репозитории и запуск идёт через .venv/bin/python. Если зависимости не установились, повторите установку из requirements-dev.txt и проверьте сообщения pip, прежде чем изменять код.

Модель запросила неподходящий инструмент или аргумент. Например, она может передать get_status(service="../../secret"). Проверка разрешённых имён в функции отклонит такой запрос. В проверенной связке ToolNode(..., handle_tool_errors=True) возвращает для неизвестного инструмента и отклонённого аргумента ToolMessage со статусом error. Это проверено отрицательными тестами; полный текст ошибки мы не закрепляем, поскольку он зависит от реализации. Назначение handle_tool_errors описано в справочнике ToolNode.

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

Запросы повторяются без конца. Ограничение числа итераций графа помогает остановить зацикливание, но само по себе не гарантирует корректность ответа. Параметр recursion_limit ограничивает количество шагов графа и может привести к GraphRecursionError — это описано в Graph API. Тест воспроизводит такой случай, когда модельный узел снова и снова запрашивает get_status. Проверьте условие завершения и историю tool_calls; просто повышать лимит не стоит.

Получился ответ, но причина 503 неизвестна. В примере функция возвращает заранее заданный статус, а инструкция — общую подсказку. Для настоящей диагностики нужны сетевые наблюдения и подтверждение результата после исправления. Ограничение прав инструментов и отдельная проверка опасных действий соответствуют рекомендациям OWASP об excessive agency.

Что стоит унести из примера

Мы начали с вопроса об ошибке API и дошли до проверяемой последовательности действий: get_status вернул учебный HTTP 503, затем get_runbook — рекомендацию, после чего model завершил работу. В истории можно связать каждый запрос инструмента с ответом по идентификатору вызова.

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

Если вам нужно всего два заранее известных вызова Python, достаточно обычной функции. LangGraph становится интереснее, когда порядок действий меняется, нужно явно описать развилки и контролировать цикл. При подключении настоящей LLM появится ещё одна самостоятельная задача — проверить качество её выбора, а не только исправность графа.

Рабочий пример, закреплённые зависимости и тесты доступны в репозитории XNova. Начните с офлайн-запуска и отрицательных тестов: они показывают фактический маршрут до подключения живой модели.

Источники

• Официальный quickstart: StateGraph, модельный узел, инструменты и переходы
• Graph API: состояние, reducers, узлы, переходы и лимит циклов
• Различие фиксированных процессов и агентов, использование инструментов
• ToolNode и обработки ошибок инструментов
• Условная маршрутизация инструментов
• Дата выпуска LangGraph 1.2.19 и требования к Python
• Безопасность и ограничение прав инструментов
• Сценарии pytest
• Виртуальные окружения Python 3.12
• Полный воспроизводимый пример, тесты и CI на закреплённом commit
• URL для клонирования публичного companion-репозитория