Друзья, привет! Возвращаюсь с продолжением.
В первой части мы разобрались, как поднять локальную LLM и пробросить к ней внешний доступ. Но до настоящей интеграции в продукт так и не добрались — модель работает, а что с ней делать дальше, непонятно. Сегодня исправляем это.
Изначально я хотел пойти по классическому пути: взять FastAPI, обернуть вокруг vLLM и получить привычный REST-сервис. Но чем глубже я погружался в тему, тем яснее становилось, что в рамках одной статьи невозможно нормально раскрыть все нюансы связки aiohttp и FastAPI. Слишком много инфраструктурных деталей, шаблонного кода и сопутствующей обвязки — в какой-то момент за этим теряется главное: сама логика работы с моделью.
Поэтому я решил сместить фокус в другую сторону — более интересную и, как мне кажется, более современную.
Сегодня поговорим про графовую инфраструктуру на базе локальных моделей — и не только локальных. Любых, поддерживающих OpenAI-совместимый протокол.
А теперь вопрос: что, если вам достаточно хорошо научиться писать граф — и вокруг него автоматически поднимется REST API, появится интерфейс для тестирования, трейсинг и мониторинг?
Экосистема LangGraph и откуда возьмется REST API
Я уже писал про LangChain, но до сегодняшнего момента за кадром оставалась «главная троица» инструментов, без которых архитектура будет неполной: LangGraph Server, LangSmith и SDK. Вот этот фундамент и разберем.
Но сначала — немного теории, без которой дальше будет сложно.
Почему ИИ так любит графы
Граф — это набор узлов и ребер, которые их соединяют. Звучит просто, но именно эта простота делает подход таким универсальным.
Технический граф — это удобный способ логически связывать между собой отдельные блоки обработки. Можно провести аналогию с классическим if/else, но с одним важным отличием: в графе вы можете вернуться в любую точку, перейти в любой узел, выстроить любой маршрут — и все это описывается декларативно, без лапши из условий.
Первая причина — роутинг
Рассмотрим простой пример. Есть узел, который принимает на вход данные. Внутри этого узла сидит модель, которая просто решает, куда двигаться дальше: в узел А или в узел Б. Это решение принимает не программист через if, а сама модель на основе контекста.
Этот простой механизм лежит в основе всей современной агентной архитектуры. Супервизоры, субагенты, мультиагентные системы — все это вариации идеи, что модель управляет потоком выполнения.
Вторая причина — состояние
В графе есть общая память, которая проходит через каждый узел. Называется она state. Каждый узел может читать из нее и писать в нее. Это позволяет строить сложные многошаговые сценарии, где каждый следующий шаг знает все о предыдущих.
Третья причина — чекпоинтер
Это механизм сохранения истории всех состояний графа. Благодаря нему можно буквально вернуться в любую точку и продолжить оттуда. Для отладки и продакшена это может быть очень важно.
Для понимания остального материала важно держать в голове четыре базовых понятия:
узел — логический блок. Внутри него выполняется какая-то работа: вызов модели, обращение к инструменту, обработка данных;
ребро — связь между узлами, определяет маршрут. Может быть статическим или условным — когда следующий узел выбирает модель;
состояние (state) — общая память графа, которая путешествует через все узлы;
чекпоинтер — история всех состояний с возможностью вернуться в любую точку.
Если вы хотите глубже погрузиться в эти концепции — загляните в мои предыдущие статьи по LangGraph, там все разобрано подробно.
Облачная инфраструктура для ваших проектов
Виртуальные машины в Москве, Санкт-Петербурге и Новосибирске с оплатой по потреблению.
Как из графа получить API
Каким бы удобным ни был граф — он остается просто скриптом. Голый граф никуда не прикрутить: ни к фронтенду, ни к мобильному приложению, ни к другому сервису. Нужен механизм, который превратит его в полноценный сервис.
Если вы бэкенд-разработчик, первое, что приходит в голову — написать обертку на FastAPI / Django и подключить граф как модуль. Идея рабочая, но она сразу тянет за собой целый хвост вопросов: как организовать сессии, как хранить историю диалога, как сделать стриминг, как не сломать состояние при параллельных запросах. И это еще до того, как вы написали хоть строчку бизнес-логики.
Я предлагаю другой подход.
Представьте систему, которая принимает на вход скомпилированный граф и сразу дает вам полноценный API со следующим из коробки:
вызов графа синхронно или стримом — токен за токеном;
автоматические сессии и управление тредами;
встроенный чекпоинтер — история каждого диалога сохраняется без вашего участия;
возможность вернуться в любую точку разговора и продолжить оттуда;
параллельные запросы без конфликтов состояния;
готовые эндпоинты для получения и обновления состояния графа программно.
Такая система и есть LangGraph Server. По сути, это рантайм вокруг вашего графа. Вы описываете логику, а сервер берет на себя всю инфраструктуру.
Но прежде чем лезть в продакшен, хочется все это потрогать руками и убедиться, что граф работает, как задумано. И здесь у LangGraph Server есть еще один козырь — LangGraph Studio.
Это визуальный интерфейс, который поднимается автоматически вместе с сервером в режиме разработки. Вы видите граф живьем: какой узел сейчас активен, что лежит в состоянии, как данные движутся между узлами. Можно отправить сообщение, посмотреть трейс выполнения, откатиться к предыдущему чекпоинту и попробовать снова. И все это прямо в браузере.
Если с логикой графа все устраивает, переходим к следующему шагу. Поднятый LangGraph Server — это полноценный REST-сервис, а значит его можно подключить куда угодно. В собственный бэкенд через Python SDK — об этом поговорим отдельно. Или в LangSmith — платформу от той же команды, которая дает мониторинг, трейсинг и аналитику по всем вызовам вашего графа в продакшене.
То есть путь выглядит так: написали граф → подняли сервер → потестировали в Studio → подключили LangSmith → прикрутили к своему проекту через SDK.
LangGraph SDK и зачем он нужен
Завершим теоретический блок разбором еще одного важного инструмента — LangGraph SDK.
Это библиотека той же экосистемы, которая позволяет интегрировать сгенерированный REST API LangGraph Server уже в ваши собственные продукты.
Возникает логичный вопрос: «зачем, если API уже и так есть?». И тут есть простой ответ — продакшен-уровень.
Вы можете написать граф и автоматически получить вокруг него API. Но напрямую открывать доступ к этому графу всем пользователям — не лучшая идея. Практически сразу возникает необходимость добавить авторизацию, собственные токены, очередь выполнения, биллинг, аудит и разграничение прав доступа. И вот здесь как раз становится полезен LangGraph SDK.
Технически это обертка вокруг сгенерированного REST API, такая же, как LangSmith, только для вашего бэкенда. Интегрируется буквально в несколько строк, а взамен дает:
возможность обращаться к удаленному графу так, будто он запущен локально;
получать токены, события и обновления состояния от SDK в удобном виде;
управлять тредами и состоянием через чистый Python-интерфейс;
синхронный (например, Django) и асинхронный (например, FastAPI) клиент.
По итогу схема выглядит так: LangGraph Server отвечает за граф и его инфраструктуру, LangSmith — за мониторинг, а SDK — за то, чтобы все это аккуратно жило внутри вашего продукта и не торчало наружу.
На этом с теорией заканчиваем — переходим к практике.
Чем сегодня займемся
Инструменты, которые мы сегодня рассматриваем, достаточно обширные и мощные. Глубоко и досконально вариться в каждом из них в рамках одной статьи не получится — формат не позволяет. Цель другая: дать четкое понимание принципов и общую картину того, как все это пишется и работает вместе.
Подготовка
Прежде чем идти дальше — разберемся с тем, что вам понадобится для повторения всего описанного ниже:
VPS-сервер — на нем будем запускать LangGraph CLI и FastAPI-проект с интегрированным SDK. Подойдет любой Linux-сервер с минимальным набором ресурсов.
Доступ к LLM — модель может быть локальной, поднятой через vLLM, llama.cpp или любой другой способ, либо облачной. Ключевое требование одно: поддержка OpenAI-совместимого протокола.
Базовое понимание графов и Python — если вы читали мои предыдущие статьи по LangGraph, этого более чем достаточно. Если нет — загляните туда перед тем, как продолжить.
Чтобы вы понимали, что нас ждет — вот полный маршрут:
Поднимаем LangGraph CLI на локальной машине.
Подключаем локальную модель — или облачную, принципиальной разницы нет.
Пишем несколько графов с ИИ.
Прикручиваем графы к CLI и смотрим, что получилось.
Тестируем и отлаживаем через LangGraph Studio.
Арендуем и настраиваем VPS.
Поднимаем CLI в продакшен-режиме на сервере.
Пишем FastAPI-приложение с интегрированным LangGraph SDK.
Поднимаем приложение рядом с сервером — получаем готовый стек.
Этих принципов вам будет достаточно, чтобы начать строить по-настоящему мощные ИИ-системы. Местами я буду проговаривать базовые вещи — это нужно, чтобы синхронизировать терминологию. Здесь читают люди с разным уровнем подготовки, и я стараюсь, чтобы никто не потерялся.
Что такое LangGraph CLI
LangGraph CLI — это утилита командной строки от команды LangChain, которая берет на себя всю инфраструктурную рутину вокруг вашего графа. Если коротко, то вы пишете граф, указываете его в конфиге, и CLI сам поднимает вокруг него полноценный сервер. Заниматься вручную FastAPI или самостоятельно писать роутеры не придется.
Под капотом CLI делает три вещи:
собирает образ — пакует ваш граф и зависимости в Docker-контейнер;
поднимает LangGraph Server — тот самый рантайм, который дает REST API, стриминг, треды и чекпоинтер из коробки;
открывает Studio — визуальный интерфейс для отладки графа прямо в браузере.
Работает в двух режимах: langgraph dev для локальной разработки — быстро, без Docker, с горячей перезагрузкой. И langgraph up для продакшена — поднимает полный стек через Docker Compose (сам CLI, база PostgreSQL и Redis).
Именно с CLI мы и начнем.
Поднимаем LangGraph CLI
Усложнять пока не будем. На этом этапе поднимаем все через стандартное виртуальное окружение Python в dev-режиме. До продакшена доберемся позже — когда будем деплоить на VPS.
Шаг 1. Подготовка окружения
Открываем любимую IDE, создаем пустую папку проекта и разворачиваем виртуальное окружение:
python3 -m venv venv source venv/bin/activate
Шаг 2. Устанавливаем LangGraph CLI
pip install "langgraph-cli[inmem]"
Флаг inmem подключает встроенный in-memory чекпоинтер — он нужен для работы в dev-режиме без внешней базы данных.
Шаг 3. Создаем шаблонный проект
В корне создаем папку app и в ней разворачиваем шаблон (команда выполняется в корневой папке, где стоит виртуальное окружение):
langgraph new app --template new-langgraph-project-python
Этот шаблон создаст минимальный чат-бот с памятью на Python — его мы и будем дорабатывать под свои нужды.
После выполнения данной команды, виртуальное окружение созданное на предыдущих этапах можно деактивировать и удалить папку venv. Тут дело в том, что внутри app будет использоваться собственное виртуальное окружение, которое управляется менеджером пакетов uv.
Шаг 4. Запускаем в dev-режиме
Переходим в папку app и запускаем сервер:
uv run langgraph dev
Эта команда подтянет недостающие пакеты и запустит CLI в DEV режиме.
При необходимости можно передать дополнительные параметры:
langgraph dev [OPTIONS] --host TEXT Хост (по умолчанию: 127.0.0.1) --port INTEGER Порт (по умолчанию: 2024) --no-reload Отключить автоперезагрузку --debug-port INTEGER Включить удаленную отладку --no-browser Не открывать браузер автоматически -c, --config FILE Путь к конфиг-файлу (по умолчанию: langgraph.json)
После запуска браузер автоматически откроется и перебросит вас в LangGraph Studio — визуальный интерфейс для отладки графа.
Для тестирования и отладки лучше использовать чистые версии Chrome или Firefox. В Yandex Browser могут быть проблемы с доменом localhost.
Шаг 5. Авторизация в LangSmith
При первом запуске вы увидите предупреждение:
It looks like your LangSmith API key is missing. Please make sure to add LANGSMITH_API_KEY to your .env file.
Это ожидаемо — без ключа Studio работает, но трейсинг недоступен. Если вы не авторизованы в LangSmith, вас перебросит на страницу входа. Заходим через Google или GitHub.
Шаг 6. Получаем API-ключ LangSmith
Для начала кликаем на шестеренку настроек и переходим на вкладку API Keys.
Нажимаем + API Key, далее сохраняем ключ — он показывается только один раз.
Шаг 7. Настраиваем переменные окружения
В папке app создаем файл .env:
LANGSMITH_PROJECT=habr-graph-cli LANGSMITH_API_KEY=lsv2_pt_ваш_ключ LLM_BASE_URL=https://your_domain/v1 LLM_KEY=ваш_ключ LLM_NAME=название_модели
Разберем каждую переменную:
LANGSMITH_PROJECT — имя проекта для группировки трейсов в дашборде LangSmith. Можно поставить любое;
LANGSMITH_API_KEY — ключ для трейсинга. С ним в дашборде будет видно каждый шаг графа: какие узлы сработали, входы и выходы каждого вызова модели, латентность, токены, ошибки. Без него граф работает нормально — просто без мониторинга;
LLM_BASE_URL — адрес вашей модели. Если поднимали LLM через vLLM или llama.cpp вместе с первой частью и пробросили ее наружу — указываем https://your_domain/v1. Если модель работает локально без проброса — указываем локальный адрес, например http://localhost:8000/v1;
LLM_KEY — ключ, который вы задавали при запуске vLLM или llama.cpp;
LLM_NAME — название модели, которое будет передаваться в запросах.
Небольшая ремарка по безопасности: в продакшене LLM не должна торчать наружу — она должна быть доступна только внутри сервера. LangGraph Server тоже не должен быть открыт напрямую. Но об этом подробнее поговорим в разделе про деплой.
После внесения изменений перезапускаем сервер. Если предупреждение о ключе исчезло, значит мы все делаем правильно.
Собираем простой чат на базе графа и нашей локальной модели
Основную подготовку мы выполнили — переходим к практике. Начнем с того, что разберем шаблонный граф и добавим в него настоящий интеллект. Нас интересует файл app/src/agent/graph.py.
Давайте сначала посмотрим на то, что там лежит по умолчанию, и разберем каждую часть:
#Шаблонный граф LangGraph с одним узлом. Возвращает заглушку. Заменяем логику под свои нужды.
from future import annotations from dataclasses import dataclass from typing import Any, Dict from langgraph.graph import StateGraph from langgraph.runtime import Runtime from typing_extensions import TypedDict class Context(TypedDict): #Параметры конфигурации графа. Передаются при создании ассистента или при вызове графа. Позволяют менять поведение графа без изменения кода. my_configurable_param: str @dataclass class State: #Состояние графа — общая память, которая проходит через все узлы. Каждый узел может читать из состояния и писать в него. changeme: str = "example" async def call_model(state: State, runtime: Runtime[Context]) -> Dict[str, Any]: """Основной узел графа — здесь происходит вся логика обработки. runtime.context дает доступ к параметрам конфигурации. """ return { "changeme": "output from call_model. " f"Configured with {(runtime.context or {}).get('my_configurable_param')}" } # Собираем граф: указываем схему состояния, добавляем узел, прописываем ребро от старта к узлу и компилируем. graph = ( StateGraph(State, context_schema=Context) .add_node(call_model) .add_edge("__start__", "call_model") .compile(name="New Graph") )
Шаблон рабочий, но модели здесь нет — узел просто возвращает заглушку. Исправим это.
Подключаем локальную модель и пишем чат
Нам нужно добавить в проект коннектор. Внутри langraph-cli проекта используется uv поэтому входим в папку app и внутри выполняем:
uv add langchain-openai
Полностью заменяем содержимое файла на следующее:
"""Простой чат-граф на базе локальной LLM через OpenAI-совместимый протокол.""" from future import annotations import os from dataclasses import dataclass, field from typing import Any, Dict, List from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph from langgraph.runtime import Runtime from typing_extensions import TypedDict from dotenv import load_dotenv load_dotenv() # Инициализируем клиент для нашей локальной модели. # ChatOpenAI умеет работать с любой моделью, поддерживающей OpenAI протокол — # vLLM, llama.cpp и другие отдают данные в том же формате. llm = ChatOpenAI( base_url=os.getenv("LLM_BASE_URL"), # адрес нашей локальной модели api_key=os.getenv("LLM_KEY", "not-needed"), # ключ, заданный при запуске модели model=os.getenv("LLM_NAME", "local-model"), # название модели temperature=0.7, extra_body={"chat_template_kwargs": {"enable_thinking": False}}, отключаем размышления, если не нужны. Параметр не обязательный. ) class Context(TypedDict): """Параметры конфигурации графа. Можно передать при создании ассистента или при вызове графа. """ system_prompt: str # системный промпт — задаем роль и поведение модели @dataclass class State: """Состояние графа — здесь живет история диалога. messages накапливает все сообщения: пользователя и модели. Каждый новый вызов графа получает актуальную историю. """ messages: List[BaseMessage] = field(default_factory=list) async def call_model(state: State, runtime: Runtime[Context]) -> Dict[str, Any]: """Основной узел — передаем историю сообщений в модель и получаем ответ. Если задан системный промпт — добавляем его первым сообщением. """ messages = state.messages # Получаем системный промпт из конфига если он есть system_prompt = (runtime.context or {}).get( "system_prompt", "Ты полезный ИИ-ассистент. Отвечай четко и по делу." ) # Собираем итоговый список сообщений для модели full_messages = [SystemMessage(content=system_prompt)] + messages # Вызываем модель response = await llm.ainvoke(full_messages) # Возвращаем обновленное состояние — добавляем ответ модели в историю return {"messages": messages + [response]} # Собираем граф graph = ( StateGraph(State, context_schema=Context) .add_node(call_model) .add_edge("__start__", "call_model") .compile(name="Chat Graph") )
Что изменилось по сравнению с шаблоном:
State теперь хранит историю сообщений, а не просто строку. Каждый новый вопрос пользователя добавляется в список — модель видит весь контекст диалога;
Context получил system_prompt — можно задать роль модели без изменения кода
call_model теперь реально вызывает LLM через ainvoke и возвращает ее ответ в состояние;
streaming=True — ответ будет стремиться токен за токеном, Studio это покажет в реальном времени.
Запускаем сервер, если еще не запущен:
langgraph dev
И переходим в Studio — там уже можно отправить первое сообщение и увидеть, как граф его обрабатывает.
Тестируем граф через LangGraph Studio
На этом этапе у нас уже есть работающий граф с подключенной локальной моделью. LangGraph Server под капотом дал нам не только сгенерированный REST API, но и чекпоинтер с трейсингом сессий из коробки. Studio, как раз вызывает этот API и визуализирует все, что происходит внутри.
Открываем браузер — Studio должна была подняться автоматически после langgraph dev. Если нет, переходим по адресу.
Вкладка Graph
Это основной инструмент отладки. Здесь вы видите ваш граф визуально: узлы, ребра, текущее активное состояние. Справа — панель для отправки сообщений и просмотра состояния на каждом шаге.
Отправляем первое сообщение и наблюдаем: какой узел сейчас активен (подсвечивается в реальном времени), что лежит в state до и после каждого узла и полный ответ модели с трейсом вызова.
Здесь же доступна история чекпоинтов — можно кликнуть на любой предыдущий шаг и посмотреть состояние графа в тот момент. Или откатиться и запустить снова с другим вводом. Для отладки сложных графов это незаменимо.
Вкладка Chat
Это упрощенный интерфейс в стиле обычного чата — без визуализации графа, просто диалог. Удобно для быстрой проверки, что модель отвечает адекватно.
Вкладка Chat работает не всегда — и это нормально.
Chat доступен, только если состояние вашего графа содержит поле messages со списком сообщений в формате LangChain. Studio смотрит на схему State и если видит совместимую структуру, активирует вкладку. Если State устроен иначе, Chat просто не появится или будет недоступен.
Это не баг и не ошибка конфигурации. Chat — это удобный бонус для чат-ориентированных графов. Если ваш граф занимается чем-то другим (обработкой документов, роутингом, аналитикой), Chat вам и не нужен. Работайте через вкладку Graph, там доступно все то же самое и даже больше.
В нашем случае, поскольку State содержит messages: List[BaseMessage], вкладка Chat должна быть доступна — можно пользоваться обоими режимами.
Для отладки простых сценариев очень удобно. То, что нужно, чтобы понять общую логику работы.
Теперь давайте наш граф усилим и прикрутим к нашему графовому чату «руки».
Как получить агента
Если вы уже сталкивались с агентами, ReAct-агентами, субагентами или супервизорами — вы наверняка слышали про инструменты и MCP-серверы. Но, возможно, не задумывались, как это все работает под капотом.
Сейчас разберем механику, чтобы вы четко понимали, как агент получает доступ к базе данных, интернету или любому произвольному скрипту.
Что нужно для агента с инструментами
Первое — модель должна поддерживать tool calling (вызов инструментов). Это не универсальная фича, а конкретная возможность, которую модель либо умеет, либо нет. Большинство современных моделей — Qwen, LLaMA, Mistral, GPT — поддерживают.
Второе — сами инструменты должны просто существовать. Тут два пути:
Написать самому. Удобно когда нужно что-то быстрое и узкоспециализированное. Буквально несколько строк Python — и инструмент готов;
Подключить MCP-сервер. MCP-сервер — это набор готовых инструментов, объединенных под одной крышей и направленных в одну сторону.
Например, работа с PostgreSQL, Redis, поиск в интернете, обертка над внешним API. Подключили сервер — получили сразу весь набор инструментов одним блоком. Про написание собственных MCP-серверов через FastMCP я уже писал в предыдущих статьях — там все подробно разобрано.
Как связать модель и инструменты
Итак, модель умеет вызывать инструменты, инструменты есть. Логичный вопрос — как их соединить?
И вот тут нужно знать два ключевых понятия: ToolNode и bind_tools.
bind_tools — это коннектор. Вы берете список инструментов и буквально привязываете их к модели. После этого модель знает, какие инструменты существуют, что они делают и когда их стоит вызвать — все это передается через системный промпт автоматически.
ToolNode — это специальный узел графа, который перехватывает решение модели вызвать инструмент, выполняет его и возвращает результат обратно в состояние. То есть модель говорит «хочу вызвать поиск с таким запросом» — ToolNode это исполняет и кладет результат в messages.
Вместе они образуют классический агентный цикл: модель думает → решает вызвать инструмент → ToolNode исполняет → модель получает результат → думает снова → отвечает пользователю.
Давайте посмотрим, как это выглядит в коде.
Пишем свои тулзы и прикручиваем к графу
До этого момента наш граф умел одно — звать LLM и возвращать ее ответ. Модель была замкнута сама в себе: спросил про погоду — получил галлюцинацию, спросил курс биткоина — получил данные «на момент обучения». Пора это исправить и дать агенту руки.
Обратите внимание на схему выше. Мы начали отходить от линейности — теперь у нас двунаправленный маршрут между call_model и tools. Граф больше не идет строго сверху вниз, а умеет возвращаться назад в зависимости от решения модели. Таким образом, мы получили классического ReAct-агента. Если вы уже писали агентов раньше, наверняка увидели знакомую аналогию.
И тут внимательный читатель справедливо заметит: зачем городить граф, если ReAct-агент вызывается буквально одной командой через create_react_agent, а инструменты биндятся еще проще — декоратор @tool вообще не обязателен?
Отвечу честно: для такого простого сценария граф действительно избыточен. Я привел этот пример намеренно, чтобы вы увидели механику изнутри. Потому что только в графе вы сможете объединить десятки таких агентов — запустить их параллельно, выстроить между ними иерархию, добавить супервизора, который будет решать, кому передать задачу. Один ReAct-агент — это кубик. Граф — это конструктор из этих кубиков.
Так что давайте разберемся, как такое описать в коде.
Создаем tools.py
Далее я приведу пример описания только одной тулзы для экономии времени, вы сможете ознакомиться с полным кодом в проекте.
Рядом с graph.py в папке src/agent/ создаем новый файл — tools.py. Никакой попсы типа калькулятора — нам нужны инструменты, которые реально ходят в интернет. Именно так проверяется, что связка «LLM ↔ инструменты» работает.
Кладем туда четыре инструмента — все через публичные API, без ключей и регистраций:
get_weather(city) — погода через Open-Meteo. Сначала геокодинг (город → координаты), затем запрос прогноза. Возвращает температуру, влажность и ветер;
get_crypto_price(coin_id, vs_currency) — цена криптовалюты через CoinGecko. По умолчанию в USD, можно в EUR или RUB. Плюс изменение за 24 часа со стрелочкой;
search_wikipedia(query, lang) — краткая выжимка статьи через Wikipedia REST API. Заголовок, первый абзац и ссылка на полную статью;
get_iss_location() — координаты МКС прямо сейчас и список людей на борту. Просто потому, что это круто.
Для лучшего погружения в тему — предлагаю вам написать собственные инструменты.
Анатомия одного инструмента
Каждая функция — это обычная асинхронная корутина с декоратором @tool из langchain_core.tools:
@tool async def get_weather(city: str) -> str: """Получить текущую погоду в городе. Args: city: Название города на любом языке. """ async with httpx.AsyncClient(timeout=10) as client: # геокодинг geo = await client.get( "https://geocoding-api.open-meteo.com/v1/search", params={"name": city, "count": 1, "language": "ru"} ) geo_data = geo.json() if not geo_data.get("results"): return f"Город '{city}' не найден." lat = geo_data["results"][0]["latitude"] lon = geo_data["results"][0]["longitude"] name = geo_data["results"][0]["name"] # прогноз weather = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": lat, "longitude": lon, "current": "temperature_2m,relative_humidity_2m,wind_speed_10m", } ) w = weather.json()["current"] return ( f"Погода в {name}: {w['temperature_2m']}°C, " f"влажность {w['relative_humidity_2m']}%, " f"ветер {w['wind_speed_10m']} км/ч." )
Два момента которые важно понять.
Docstring — это контракт с моделью. LangChain на основе docstring и сигнатуры генерирует JSON-схему, которая улетает в LLM. Модель читает описание и названия аргументов и на их основе решает, стоит вызвать инструмент или нет. Пишите осмысленно.
Тип возврата — str. Модель получит это как ToolMessage и прочитает как обычный текст. Можно возвращать и словари — они сериализуются в JSON — но строка человечнее для LLM.
Внизу файла собираем все в один список:
TOOLS = [get_weather, get_crypto_price, search_wikipedia, get_iss_location]
Биндим тулзы к модели
Открываем graph.py и меняем инициализацию LLM — добавляем .bind_tools(TOOLS):
llm = ChatOpenAI( base_url=os.getenv("LLM_BASE_URL"), api_key=os.getenv("LLM_KEY", "not-needed"), model=os.getenv("LLM_NAME", "local-model"), temperature=0.7, extra_body={"chat_template_kwargs": {"enable_thinking": False}}, ).bind_tools(TOOLS)
Что происходит под капотом: bind_tools возвращает обернутый Runnable, который при каждом ainvoke добавляет в тело запроса поле tools с JSON-схемой всех наших функций. Модель видит: «есть четыре инструмента, вот их сигнатуры» — и в ответе может прислать не текст, а tool_calls с тем, что хочет вызвать.
Добавляем ToolNode и ветвление
Одного bind_tools мало — модель только просит вызвать инструмент, а кто-то должен это исполнить. За это отвечает ToolNode — готовый узел из langgraph.prebuilt. Он берет список тулзов, разбирает tool_calls из последнего AIMessage, параллельно их выполняет и возвращает в стейт ToolMessage с результатами.
from langgraph.prebuilt import ToolNode, tools_condition graph = ( StateGraph(State, context_schema=Context) .add_node(call_model) .add_node("tools", ToolNode(TOOLS)) .add_edge("__start__", "call_model") .add_conditional_edges("call_model", tools_condition) .add_edge("tools", "call_model") .compile(name="Chat Graph") )
Ключевой элемент здесь — add_conditional_edges("call_model", tools_condition). tools_condition — это готовая функция-маршрутизатор: смотрит в последнее сообщение стейта, и если там есть tool_calls — идем в узел tools, если нет — end.
Получается цикл: модель → вызов инструмента → результат → модель → финальный ответ.
Редьюсер add_messages — без него все сломается
Тут есть подводный камень. В предыдущей версии узел возвращал {"messages": messages + [response]} — весь список целиком. Для простого чата это работало. Но ToolNode возвращает только новые ToolMessage без истории. Если у поля messages нет редьюсера — LangGraph просто заменит список и вся история улетит в трубу.
Решение — аннотация Annotated[..., add_messages]:
from langgraph.graph.message import add_messages from typing import Annotated @dataclass class State: messages: Annotated[List[BaseMessage], add_messages] = field(default_factory=list)
add_messages — стандартный редьюсер LangGraph, который умно мерджит сообщения: не просто конкатенирует, но и умеет заменять по id если сообщение обновилось. После этого любой узел возвращает только новые сообщения, а фреймворк сам дописывает их в общую историю.
Соответственно в call_model тоже упрощаем возврат:
return {"messages": [response]}
Как это работает на живом запросе
Спрашиваем агента: «Какая погода в Краснодаре и сколько стоит биткоин?», в это время:
call_model отправляет запрос в LLM. Модель видит описания четырех инструментов, понимает, что нужно дернуть два из них, и возвращает AIMessage с tool_calls: get_weather(city="Ташкент") и get_crypto_price(coin_id="bitcoin") — без текста;
tools_condition видит tool_calls → маршрутизирует в узел tools;
ToolNode запускает оба HTTP-вызова параллельно, собирает результаты в два ToolMessage и кладет в стейт;
Возвращаемся в call_model. LLM видит в истории: вопрос → свой tool-call → ответы тулзов. Теперь пишет человеческий ответ: «В Краснодаре сейчас 11.6°C, влажность 67%, ветер 8.0 км/ч. Цена биткоина составляет $70,833 (изменение за 24 часа: -1.17%).»;
tools_condition смотрит в последнее сообщение — tool_calls нет → end.
Граф замкнулся, пользователь получил реальные данные вместо выдуманных.
Включаем демонстрацию вызова тулзов для наглядности
Обратите внимание на конечный (единый) ответ. Выше был пример вызова через вкладку чат, но теперь попробуем вызвать через граф.
И сделаем вызов:
Прекрасно отработано!
Подключаем MCP-серверы и объединяем с нашими тулзами
В прошлой секции мы написали четыре своих инструмента и научили агента ими пользоваться. Теперь расширим его возможности — подключим готовые MCP-серверы, не написав внутри них ни строчки кода.
Идея MCP простая: другие разработчики (или мы сами) уже написали инструменты, упаковали их по общему протоколу, а мы просто подключаемся, как к плагинам.
Для примера берем два минималистичных сервера без внешних зависимостей:
mcp-server-fetch — ходит по произвольным URL и возвращает контент страницы. Примитивно, но универсально — агент может читать любой сайт в интернете;
mcp-server-time — текущее время, таймзоны и конвертация между ними. Никакого хранения состояния, никакой лишней инфраструктуры — только чистая функциональность.
Ставим адаптер
Чтобы LangGraph подружился с MCP нужен мост между двумя протоколами. Его роль играет langchain-mcp-adapters — он поднимает MCP-сервер и оборачивает каждый его инструмент в привычный BaseTool, с которым уже умеют работать bind_tools и ToolNode.
uv add langchain-mcp-adapters
Сами MCP-серверы устанавливать заранее не нужно — адаптер запускает их через uvx, который скачивает пакеты из PyPI и запускает в изолированных окружениях. При первом старте займет пару секунд, дальше — из кэша.
Создаем mcp.py
Рядом с tools.py кладем новый файл — src/agent/mcp.py:
from langchain_mcp_adapters.client import MultiServerMCPClient MCP_SERVERS = { "fetch": { "command": "uvx", "args": ["mcp-server-fetch"], "transport": "stdio", }, "time": { "command": "uvx", "args": ["mcp-server-time", "--local-timezone=Asia/Tashkent"], "transport": "stdio", }, } client = MultiServerMCPClient(MCPSERVERS) async def load_mcp_tools(): return await client.gettools()
Разберем по частям, что здесь происходит.
MultiServerMCPClient — менеджер, который держит несколько MCP-серверов сразу. У каждого сервера свое имя и конфиг.
command + args — как запустить процесс сервера. uvx это аналог npx, только для Python-пакетов. uvx mcp-server-fetch означает: скачай пакет если еще не скачан и запусти. Никаких pip install заранее.
transport: «stdio» — способ общения с сервером. Stdio поднимает локальный процесс и разговаривает через stdin/stdout. Никаких портов, никаких сетевых задержек. Для удаленных серверов есть sse и streamable_http — но это уже другая история.
--local-timezone=Asia/Tashkent — аргумент самого time-сервера, не протокола. Говорим ему, что считать локальным временем. У каждого MCP-сервера свои флаги — смотрите в README пакета.
load_mcp_tools() — при вызове поднимает все серверы, делает handshake по MCP-протоколу, запрашивает список инструментов и оборачивает каждый в LangChain-совместимый BaseTool. После этого они неотличимы от того, что мы писали в tools.py.
Объединяем все в graph.py
Поскольку load_mcp_tools асинхронный, грузим MCP один раз при загрузке модуля:
import asyncio from agent.mcp import load_mcp_tools from agent.tools import TOOLS mcptools = asyncio.run(load_mcp_tools()) ALL_TOOLS = TOOLS + mcptools llm = ChatOpenAI( base_url=os.getenv("LLM_BASE_URL"), api_key=os.getenv("LLM_KEY", "not-needed"), model=os.getenv("LLM_NAME", "local-model"), temperature=0.7, streaming=True, ).bind_tools(ALL_TOOLS)
Обратите внимание — список ALL_TOOLS один. В нем бок о бок лежат наши четыре функции и инструменты из MCP. Модель видит их всех одинаково: по имени, описанию и JSON-схеме аргументов. Никакой разницы «наше vs чужое» на уровне графа нет.
То же самое в ToolNode:
.add_node("tools", ToolNode(ALL_TOOLS))
Проверяем, что получилось. При первом старте langgraph dev в логах увидите, как uvx тянет пакеты:
Installed 4 packages in 120ms # mcp-server-fetch Installed 2 packages in 80ms # mcp-server-time
А если спросить, что в итоге попало в граф:
from agent.graph import ALL_TOOLS [t.name for t in ALL_TOOLS] ['get_weather', 'get_crypto_price', 'search_wikipedia', 'get_iss_location', 'fetch', 'get_current_time', 'convert_time']
Семь инструментов — четыре локальных плюс три из двух MCP-серверов. fetch пришел один, а time-сервер раскрылся в get_current_time и convert_time — один MCP-сервер вполне может предоставлять несколько тулзов, LangGraph импортирует их все.
Теперь агенту можно задавать вопросы совсем другого уровня. «Зайди на Hacker News, найди самую обсуждаемую статью и перескажи» — и он честно сходит в интернет через fetch. Или «Сейчас 15:00 в Ташкенте, сколько это в Токио?» — позовет convert_time.
Протестируем.
И давайте проверим, что наши кастомные инструменты тоже поддерживаются.
Проблема реактивных агентов
Сейчас все выглядит футуристично — мы написали скрипты или взяли готовые, и наша локальная модель сама решает, какие тулзы вызывать, в каком порядке и сколько итераций делать. Без единой строчки управляющей логики с нашей стороны. Так в чем проблема?
К сожалению, проблем сразу несколько:
Потеря контроля. Вы можете думать, что модель будет вызывать одни инструменты, а она будет вызывать другие — или вообще не те. Модель принимает решения самостоятельно, и эти решения не всегда совпадают с вашими ожиданиями. В простых сценариях это терпимо. В продакшене — уже нет.
Бесконечные циклы. Реактивный агент теоретически может гонять цикл call_model → tools → call_model бесконечно, если модель не может прийти к финальному ответу. Без явного ограничения итераций это реальная проблема.
Контекстное окно забивается инструментами. Каждый привязанный инструмент — это JSON-схема, которая улетает в модель при каждом запросе. 10 инструментов — 10 схем, 20 инструментов — 20 схем и так далее. Контекстное окно не безгранично, а значит, чем больше инструментов, тем меньше места остается для реального диалога. И тем хуже модель начинает в них ориентироваться.
Непредсказуемая стоимость. Если вы работаете с облачной моделью, каждая итерация цикла — это токены, а токены — это деньги. Агент, который решил сделать семь вызовов вместо двух — неприятный сюрприз в счете.
Что с этим делать?
Советую смотреть на задачу так: если есть возможность обойтись без реактивного агента и супервизора — обходимся. Линейный граф с предсказуемым маршрутом всегда лучше управляемого хаоса.
Если же задача действительно требует динамического поведения, добавляем роутинг. Это когда модель принимает решение не «какой инструмент вызвать», а «в какой узел графа перейти». Разница принципиальная: вы по-прежнему контролируете, что вообще доступно на каждом шаге, а модель лишь выбирает из заранее определенных вами вариантов.
Именно это мы и сделаем дальше — напишем граф с роутингом и посмотрим, как он решает большинство описанных проблем.
Добавляем второй граф с роутингом
До этого момента у нас в langgraph.json жил один граф — agent. Но LangGraph Server устроен так, что один CLI-инстанс может держать сколько угодно графов одновременно. Каждый со своим именем, логикой и состоянием. В Studio они появляются как отдельные вкладки — переключаться можно за секунду.
Это удобно: рядом живут «простой чат с тулзами» и «чат с роутингом», первый переписывать не нужно.
Как подвесить новый граф к серверу
Дописываем одну строчку в langgraph.json:
{ "graphs": { "agent": "./src/agent/graph.py:graph", "router_agent": "./src/agent/router_graph.py:graph" } }
Формат значения: путь_к_файлу:имя_переменной. В переменной должен лежать скомпилированный CompiledGraph. Перезапускаем langgraph dev — в Studio появляется новая карточка router_agent с собственной историей тредов. Графы полностью изолированы друг от друга, никаких пересечений стейтов и конфигов.
Идея роутинга
Вернемся к проблеме которую обсуждали раньше. Простой «чат + тулзы» — это когда модель на каждом вызове видит все инструменты сразу. Пока их семь — терпимо. А если тридцать? Промпт раздуется, модель начнет путаться и может позвать get_weather, когда ее спросили про криптовалюту. Решение классическое — разделить агента на роутер и специалистов.
Роутер — маленький быстрый узел, который делает ровно одно: смотрит на запрос пользователя и решает в какую ветку его отправить. Ничего не вызывает, тулзов не трогает.
Специалисты — полноценные ReAct-агенты, каждый со своим суженным набором инструментов.
Ключевой момент: роутер не должен уметь вызывать тулзы
Это важно и часто упускают. Роутер — чистый классификатор, не агент. Ему не нужен bind_tools, ему не нужен ToolNode. Все что он должен уметь — посмотреть на последнее сообщение и вернуть одну строку: «это чат», «это веб-задача», «это запрос данных».
В коде это реализуется через with_structured_output и Pydantic-схему с Literal:
from pydantic import BaseModel, Field from typing import Literal class Route(BaseModel): destination: Literal["chat", "web", "data"] = Field( description="chat — обычный разговор; web — поиск в интернете; data — актуальные данные" ) routerllm = ChatOpenAI(temperature=0.0, **_llm_kwargs).with_structured_output(Route)
Разберем по частям:
Literal[...] гарантирует, что LLM вернет только одно из трех значений. Фреймворк подсунет модели JSON-схему с enum, и провайдер через function calling жестко это обеспечит.
temperature=0.0 — роутер должен быть детерминированным. Один и тот же вопрос должен стабильно идти в одну и ту же ветку.
description в Field — это прямо промпт для модели. Именно на него она смотрит выбирая куда направить запрос. Чем точнее описание — тем надежнее роутинг.
Никаких bind_tools. Роутер не знает о существовании get_weather или fetch. Он знает только имена веток.
Как это собирается в граф
После того как роутер вернул строку, ее нужно направить в нужную ветку. Для этого добавляем add_conditional_edges с маппингом:
.add_conditional_edges( "router", pick_route, {"chat": "chat", "web": "web_agent", "data": "data_agent"}, ) def pick_route(state: State) -> str: return state.route
pick_route — одна строка. Просто достает значение которое роутер записал в стейт. Можно было бы подумать, что разделение на «узел который решает» и «функцию-селектор которая маршрутизирует» — это дублирование. Но нет, на самом деле такой подход просто дает нужную гибкость.
Каждый специалист собирается, как мини-ReAct-агент — все по той же схеме, что и в первом графе:
.add_conditional_edges("web_agent", tools_condition, {"tools": "web_tools", END: END}) .add_edge("web_tools", "web_agent")
У каждого специалиста свой ToolNode со своим набором тулзов. web_agent физически не может вызвать get_crypto_price — такой функции нет в его bind_tools, модель ее просто не увидит.
Снижаем цены на выделенные серверы в реальном времени
Успейте арендовать со скидкой до 35%, пока лот не ушел другому.
Что в итоге получилось
┌── chat ─────────────────────→ END start → router ──┤ ├── web_agent ⇄ web_tools ──→ END └── data_agent ⇄ data_tools ──→ END
Роутер на входе — один узел, три ветки на выход, два из которых раскрываются в ReAct-циклы. Симметрично, расширяется по готовому шаблону.
Что выигрываем:
в web_agent улетает схема только трех тулзов вместо семи;
одна короткая генерация на десятки токенов;
можно добавить специалиста, это один новый Literal-вариант, новый узел и одна строка в маппинге, а существующие ветки не трогаются;
роутер и специалисты тестируются независимо.
Роутинг — это про разделение обязанностей, а не про вызовы тулзов. Роутер принимает решение, специалисты исполняют. Каждый на своем уровне.
Обратите внимание на подсветку узлов при вызове. Удобно для отладки.
Дополнительные возможности
К сожалению, в формате статьи даже косвенно рассмотреть все, что открывают графы в связке с LangGraph Server, не получится. Поэтому пробежимся теоретически и вскользь, просто чтобы вы понимали, куда можно двигаться дальше.
Граф внутри графа — субграфы
Любой скомпилированный граф можно вставить, как узел в другой граф. Берете compiled_graph и передаете его в .add_node(), как обычную функцию. Снаружи это выглядит, как обычный узел, а внутри прячется полноценный граф со своим состоянием и логикой.
Это основа для построения модульных систем: написали граф для обработки документов, граф для поиска, граф для генерации отчета — и собрали их в один мастер-граф, как конструктор.
Супервизор
Развитие идеи роутера. Только роутер отправляет задачу один раз и забывает, а супервизор следит за выполнением, получает результаты от специалистов и решает, достаточно или надо дослать еще кому-то.
Классический сценарий: пользователь задает сложный вопрос, супервизор раздает подзадачи трем специалистам параллельно, собирает результаты, оценивает и либо формирует финальный ответ, либо запускает еще один круг. Все это — просто узлы и ребра в графе.
Граф, как инструмент
Помните, как мы писали тулзы через @tool? Точно так же можно обернуть целый граф. Декоратор @tool над функцией, которая вызывает graph.ainvoke() — и ваш граф становится инструментом для другого агента.
Это открывает интересный паттерн: верхнеуровневый агент видит инструмент research_agent и просто его вызывает, не зная, что внутри целый граф с несколькими узлами, чекпоинтером и своими тулзами.
Параллельное выполнение узлов
LangGraph поддерживает параллельные ветки через Send — можно запустить несколько узлов одновременно и дождаться результатов всех. Удобно когда нужно одновременно сходить в несколько источников данных и потом собрать все в один ответ.
Human-in-the-loop
Граф можно поставить на паузу в любой точке и дождаться подтверждения от человека. Узел выполнился, записал в стейт, что требует одобрения — и граф встал. Пользователь смотрит, подтверждает или правит — граф продолжает с того места, где остановился. Чекпоинтер как раз для этого и нужен.
Это особенно полезно в сценариях, где агент что-то пишет в базу данных, отправляет письма или делает любые необратимые действия. В таком сценарии результат можно сначала показывать человеку, потом выполнить (или не выполнить).
Персистентность и долгосрочная память
По умолчанию чекпоинтер хранит историю в памяти — при перезапуске сервера все теряется. Но LangGraph поддерживает персистентные чекпоинтеры через PostgreSQL или Redis. Подключил — и треды живут между перезапусками, пользователь может вернуться к диалогу через неделю и продолжить с того места, где остановился.
Это лишь верхушка айсберга. Графовый подход по сути не ограничивает вас ничем — любую логику взаимодействия агентов можно выразить через узлы и ребра. Главное понять принцип, а дальше конкретная реализация упирается только в пределы вашей фантазии и текущую задачу.
Оборачиваем граф в продакшен-API
С графами разобрались. Теперь пора сделать из этого настоящий продукт.
Напомню общую картину. LangGraph Server дает нам готовый REST API вокруг графов, но напрямую его наружу не выставляем — это внутренний сервис. Поверх него мы пишем собственный FastAPI-сервис, который и будет точкой входа для пользователей. Там живет авторизация, биллинг, собственные токены, бизнес-логика — все, что нужно для реального продукта.
Общение между FastAPI и LangGraph Server происходит через LangGraph SDK — тот самый Python-клиент, который мы разбирали в теории. Настало время посмотреть на него в деле.
FastAPI + LangGraph SDK — оборачиваем граф в продакшен-сервис
Итак, у нас есть работающий LangGraph Server с двумя графами. Теперь пишем поверх него свой FastAPI-сервис — тот самый слой, который будет смотреть наружу и принимать запросы от пользователей.
Структура проекта минимальная:
├── main.py # FastAPI app + lifespan ├── router.py # эндпоинты /health, /agent, /router ├── utils.py # авторизация + вызов графа через SDK ├── schemas.py # ChatRequest / ChatResponse ├── deps.py # FastAPI dependencies └── config.py # настройки через pydantic-settings
Полный код лежит на GitHub. Разберем ключевые части.
config.py — настройки
from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore") langgraph_url: str = "http://127.0.0.1:2024" # адрес LangGraph Server access_token: str = "change-me" # наш собственный токен для авторизации agent_assistant_id: str = "agent" # имя первого графа router_assistant_id: str = "router_agent" # имя второго графа app_host: str = "127.0.0.1" app_port: int = 8000 @lru_cache def get_settings() -> Settings: return Settings()
Все тянется из .env, дефолты прописаны прямо в классе. lru_cache гарантирует, что настройки читаются один раз при старте, а не при каждом запросе.
main.py — инициализация клиента
from contextlib import asynccontextmanager from fastapi import FastAPI from langgraph_sdk import get_client from config import get_settings from router import router as graph_router @asynccontextmanager async def lifespan(app: FastAPI): # при старте приложения создаем SDK-клиент и кладем его в app.state settings = get_settings() app.state.client = get_client(url=settings.langgraph_url) yield def create_app() -> FastAPI: app = FastAPI(title="FastAPI + LangGraph SDK demo", lifespan=lifespan) app.include_router(graph_router) return app app = create_app()
Ключевой момент — lifespan. Это механизм FastAPI для кода который должен выполниться один раз при старте и один раз при остановке. Мы создаем SDK-клиент здесь и кладем его в app.state, откуда потом достанем через dependency injection в любом эндпоинте.
get_client(url=...) — это и есть точка входа в LangGraph SDK. Указываем адрес нашего LangGraph Server, и получаем полноценный асинхронный клиент.
utils.py — авторизация и вызов графа
import secrets from uuid import uuid4 from fastapi import HTTPException, status from schemas import ChatRequest, ChatResponse def check_token(req: ChatRequest, settings: Settings) -> None: # secrets.compare_digest защищает от timing-атак if not secrets.compare_digest( req.access_token.encode("utf-8"), settings.access_token.encode("utf-8") ): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid access token") async def run_graph(client: Any, assistant_id: str, req: ChatRequest) -> ChatResponse: # если thread_id не передан — создаем новый, иначе продолжаем существующий диалог thread_id = req.thread_id or uuid4() input_payload = {"messages": [{"role": "user", "content": req.message}]} final_state: Any = None try: async for chunk in client.runs.stream( thread_id=str(thread_id), assistant_id=assistant_id, input=input_payload, stream_mode="values", # получаем полное состояние после каждого шага if_not_exists="create", # если треда нет — создать автоматически ): if chunk.event == "values": final_state = chunk.data # берем последнее состояние — это финальный ответ except Exception as e: raise HTTPException(status_code=502, detail=f"LangGraph error: {e}") return ChatResponse(thread_id=thread_id, output=final_state)
Два момента которые важно понять:
thread_id — это идентификатор сессии. Если клиент передает его в запросе, SDK подтянет чекпоинт и продолжит тот же диалог. Если нет, создаем новый uuid4(). Именно так работает память между сообщениями: каждый следующий вопрос пользователя передает thread_id который пришел в предыдущем ответе.
stream_mode="values" — режим стриминга. В этом режиме мы получаем полное состояние графа после каждого шага. Нас интересует последний чанк — там финальный стейт со всей историей сообщений. Есть и другие режимы: updates (только дельты), messages (токены LLM в реальном времени) — выбирайте под задачу.
router.py — эндпоинты
from fastapi import APIRouter, HTTPException from deps import ClientDep, SettingsDep from schemas import ChatRequest, ChatResponse from utils import check_token, run_graph router = APIRouter(tags=["graph"]) @router.get("/health") async def health(client: ClientDep): try: # проверяем связь с LangGraph Server и возвращаем список доступных графов assistants = await client.assistants.search(limit=20) return {"status": "ok", "assistants": [a["graph_id"] for a in assistants]} except Exception as e: raise HTTPException(status_code=502, detail=str(e)) @router.post("/agent", response_model=ChatResponse) async def agent(req: ChatRequest, client: ClientDep, settings: SettingsDep): check_token(req, settings) return await run_graph(client, settings.agent_assistant_id, req) @router.post("/router", response_model=ChatResponse) async def router_agent(req: ChatRequest, client: ClientDep, settings: SettingsDep): check_token(req, settings) return await run_graph(client, settings.router_assistant_id, req)
/health проверяет, что LangGraph Server живой и возвращает список доступных графов. /agent и /router — два наших графа, каждый за своим эндпоинтом. Авторизация происходит через check_token — статический токен из .env, сравнивается через secrets.compare_digest для защиты от timing-атак.
Как это все запустить
Ставим зависимости:
pip install -r requirements.txt
Копируем и заполняем .env:
cp .env.example .env
Запускаем в двух терминалах параллельно:
# терминал 1 — LangGraph Server langgraph dev # терминал 2 — наш FastAPI python main.py
Проверяем, что все живо:
curl http://127.0.0.1:8000/health # {"status": "ok", "assistants": ["agent", "router_agent"]}
Отправляем первый запрос:
curl -X POST http://127.0.0.1:8000/agent \ -H 'Content-Type: application/json' \ -d '{ "access_token": "ваш-токен", "message": "кто сейчас на МКС?" }'
В ответе придет thread_id — сохраняем его и передаем в следующем запросе чтобы продолжить диалог.
На что еще способен LangGraph SDK
В текущем коде мы использовали самый базовый сценарий — отправили сообщение, получили ответ, вернули клиенту. Но SDK умеет значительно больше, и было бы нечестно об этом не упомянуть.
Фоновые запуски. client.runs.create() запускает граф асинхронно — клиент сразу получает run_id и не ждет ответа. Граф крутится на сервере сам по себе. Удобно для долгих задач: запустили, вернули пользователю идентификатор, он потом сам подтянул результат.
Отложенные запуски. Тот же runs.create() с параметром after_seconds — граф запустится через указанное время. Никакого Celery, никаких очередей — просто параметр в запросе.
Подписка на уже идущий ран. client.runs.join() позволяет подключиться к фоновому запуску который уже выполняется и получать события с самого начала. Полезно если клиент отвалился и переподключился.
Управление тредами и состоянием. client.threads.get_state() возвращает текущий чекпоинт треда — весь стейт графа, как он есть. update_state() позволяет вручную поправить стейт: переписать последнее сообщение, подставить tool-result в обход графа. get_history() дает список всех чекпоинтов — можно откатиться в любую точку и продолжить оттуда.
Мультизадачность. Параметр multitask_strategy определяет, что делать если по треду уже идет активный запуск: отклонить новый (reject), прервать текущий (interrupt), откатить и начать заново (rollback) или поставить в очередь (enqueue).
Долгосрочная память через store. client.store — это key-value хранилище между тредами. В отличие от чекпоинтов которые живут внутри одного диалога, store персистентен глобально. Поддерживает даже векторный поиск если на сервере настроен embedder — для долговременной памяти агента про пользователя это именно то, что нужно.
Расписания. client.crons.create() запускает граф по cron-расписанию. Мониторинг, отчеты, регулярные задачи — без отдельного планировщика.
Human-in-the-loop. Если в графе есть interrupt() — стрим отдаст событие interrupt и встанет на паузу. Продолжить можно передав command={"resume": payload} в следующем запуске. Так реализуется подтверждение действий перед тем, как агент что-то сделает необратимое.
Ну а пока мы разбирались с кодом, оба проекта живут у нас локально. Пора это исправить и вынести все на настоящий сервер. Арендуем VPS, настраиваем окружение и поднимаем стек в продакшен-режиме.
Арендуем сервер
В прошлой части мы уже работали с Selectel — там мы арендовали сервер с GPU на 16 ГБ видеопамяти под запуск локальной модели. Сегодня возвращаемся туда же, но задача другая: поднять LangGraph Server и FastAPI-сервис, а они железа почти не едят. Поэтому берем самый базовый VPS без GPU — дешево, быстро и более чем достаточно для наших целей.
Процесс аренды:
Заходим в панель управления и регистрируемся, если еще нет аккаунта;
Переходим в раздел Продукты → Облачные серверы;
Создаем новый сервер. Мой конфиг для этой задачи:
Параметр | Значение |
|---|---|
Локация | ru-2a |
Операционная система | Ubuntu 24 (без графического драйвера) |
CPU / RAM | 1 vCPU / 2 ГБ |
Диск | SSD 10 ГБ |
Для LangGraph Server и FastAPI этого более чем достаточно — никакой тяжелой математики там нет, все упирается в сеть и память, а не в вычисления.
Настраиваем сервер
Сервер арендован — подключаемся по SSH и готовим окружение.
Подключаемся
ssh root@ваш_ip
При аренде сервера мы заполняли поле с SSH-ключем. Поэтому, если все было настроено корректно, то команды ssh root@ваш_ip будет достаточно для входа на сервер.
Обновляем систему
Первое, что делаем на любом свежем сервере — обновляем пакеты:
apt update && apt upgrade -y
Ставим Docker и Docker Compose
LangGraph CLI в продакшен-режиме работает через Docker — поднимает полный стек через Docker Compose. Поэтому Docker нам обязателен.
# ставим зависимости apt install -y ca-certificates curl gnupg # добавляем официальный репозиторий Docker install -m 0755 -d /etc/apt/keyrings curl -fsSL https://download.docker.com/linux/ubuntu/gpg | gpg --dearmor -o /etc/apt/keyrings/docker.gpg chmod a+r /etc/apt/keyrings/docker.gpg echo \ "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] \ https://download.docker.com/linux/ubuntu \ $(. /etc/os-release && echo "$VERSION_CODENAME") stable" | \ tee /etc/apt/sources.list.d/docker.list > /dev/null # устанавливаем apt update apt install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
Все одной командой можно. Проверяем, что все встало:
docker --version docker compose version
Ставим Python и зависимости.
apt install -y python3 python3-pip python3-venv git
Поднимаем LangGraph CLI проект на сервере
Клонируем репу:
mkdir ~/langgraph && cd ~/langgraph git clone https://github.com/Yakvenalex/HabrGraphCLI .
Можете выполнять пул, как моего репозитория, так и собственного.
Заполняем переменные окружения:
cp .env.example .env nano .env
Заполняем по аналогии с локальным запуском — LANGSMITH_API_KEY, LLM_BASE_URL, LLM_KEY, LLM_NAME.
Ставим CLI если еще не стоит:
pip install "langgraph-cli[inmem]" --break-system-packages
Собираем Docker-образ
В отличие от langgraph dev — продакшен-режим работает через Docker. Сначала добавим важную строку в файл langraph.json:
"dockerfile_lines": [ "RUN apk add --no-cache curl && curl -Ls https://astral.sh/uv/install.sh | UV_INSTALL_DIR=/usr/local/bin sh" ]
Эта инструкция при сборке образа установит в образ curl и uv, Curl нужен для установки uv, а uv нужен для запуска наших MCP серверов.
Теперь можно собрать образ:
langgraph build -t habr-graph-cli
CLI прочитает langgraph.json, подтянет зависимости из pyproject.toml и соберет образ. При первой сборке это займет несколько минут.
Запускаем в прод-режиме
langgraph up
Под капотом CLI генерирует docker-compose.yml и поднимает полный стек: сам LangGraph Server, Redis для очередей и PostgreSQL для персистентного чекпоинтера. В отличие от langgraph dev — здесь уже настоящий продакшен с персистентностью между перезапусками.
Проверяем, что все поднялось:
Остановим и запустим сервер в фоне.
CTRL + C langgraph up --wait
--wait дожидается старта и возвращает управление
Проверим, что все контейнеры успешно запущены.
docker ps
Обратите внимание на важный момент. По умолчанию langgraph up пробрасывает наружу порты LangGraph Server (8123) и Postgres (5433). Для демки и тестирования это допустимо — удобно быстро проверить, что все работает. Но в продакшене это недопустимо: база данных и внутренний API не должны быть доступны из интернета.
Самое простое решение — закрыть лишнее через ufw:
ufw enable ufw allow 22 # SSH — обязательно, иначе потеряете доступ к серверу ufw allow 8000 # порт вашего FastAPI — единственное, что смотрит наружу (и то не всегда, иногда достаточно чтоб наружу торчал только 80й и 443й порты через Nginx proxy manager, но это тема отдельного разговора)
После этого 8123 и 5433 будут недоступны снаружи, но внутри сервера FastAPI по-прежнему сможет обращаться к LangGraph Server через localhost:8123 — файрвол локальный трафик не блокирует. Снаружи остается только одна точка входа — ваш собственный API.
Проверяем сам сервер:
curl http://localhost:8123/ok
Если пришло {"status": "ok"} — LangGraph Server живой и принимает запросы на порту 8123.
Поднимаем FastAPI + LangGraph SDK проект
Возвращаемся в корень и клонируем репу:
cd ~ git clone https://github.com/Yakvenalex/FastApiGraphSDKHabr
cd FastApiGraphSDKHabr
Шаг 1. Создаем виртуальное окружение
python3 -m venv venv source venv/bin/activate
Шаг 2. Устанавливаем зависимости
pip install -r requirements.txt
Шаг 3. Добавляем переменные окружения
cp .env.example .env nano .env
Заполняем — главное указать правильный LANGGRAPH_URL (у нас это http://127.0.0.1:8123) и задать свой ACCESS_TOKEN.
Шаг 4. Тестовый запуск
python3 main.py
Если все поднялось и в логах нет ошибок — останавливаем Ctrl+C и настраиваем автозапуск через systemd.
Шаг 5. Настраиваем systemd
Создаем файл сервиса:
nano /etc/systemd/system/fastapi-graph.service
Содержимое в моем случае:
[Unit] Description=FastAPI + LangGraph SDK After=network.target [Service] Type=simple User=root WorkingDirectory=/root/FastApiGraphSDKHabr ExecStart=/root/FastApiGraphSDKHabr/venv/bin/python main.py Restart=on-failure RestartSec=5 [Install] WantedBy=multi-user.target
Активируем и запускаем:
# перечитываем конфиги systemd systemctl daemon-reload # включаем автозапуск при старте сервера systemctl enable fastapi-graph # запускаем сервис systemctl start fastapi-graph # проверяем статус systemctl status fastapi-graph
Проверяем, что сервис живой:
curl http://localhost:8000/health
Смотреть логи в реальном времени:
journalctl -u fastapi-graph -f
Теперь оба сервиса работают в фоне и будут автоматически подниматься после перезагрузки сервера — LangGraph через Docker, FastAPI через systemd.
Перенос и продление домена по 1 ₽
С легкостью переходите в Selectel от любого другого провайдера. Сайт продолжит работать без остановки.
Прикручиваем доменное имя
Оба сервиса запущены, порты закрыты. Логичный следующий шаг — доменное имя, чтобы в Swagger можно было ходить по человеческой ссылке, а не по IP. Для этого нам понадобится домен с A-записью, указывающей на IP нашего VPS.
Домен берем у Selectel — там же, где и сервер, все в одном месте. В прошлой статье я уже приобретал домен, поэтому использую существующий.
Переходим: Продукты → Домены → Доменные зоны
Кликаем на нужную зону, затем Добавить запись:
тип — A;
имя — @ или нужный поддомен, например api;
значение — IP вашего VPS;
TTL — оставляем по умолчанию.
Сохраняем и ждем несколько минут пока DNS распространится. Проверить можно так:
ping ваш_домен
Устанавливаем Nginx
apt install -y nginx
Создаем конфиг для нашего сервиса
nano /etc/nginx/sites-available/fastapi-graph
Содержимое:
server { listen 80; server_name ваш_домен; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }
Активируем конфиг:
ln -s /etc/nginx/sites-available/fastapi-graph /etc/nginx/sites-enabled/ nginx -t systemctl reload nginx
Открываем порт 80 в файрволе:
ufw allow 80 ufw allow 443
Получаем SSL-сертификат через Certbot
apt install -y certbot python3-certbot-nginx certbot --nginx -d ваш_домен
Certbot сам найдет конфиг Nginx, получит сертификат и перепишет конфиг добавив HTTPS. Следуем инструкциям — вводим email, соглашаемся с условиями.
После успешного получения сертификата конфиг обновится автоматически и будет выглядеть примерно так:
server { listen 443 ssl; server_name ваш_домен; ssl_certificate /etc/letsencrypt/live/ваш_домен/fullchain.pem; ssl_certificate_key /etc/letsencrypt/live/ваш_домен/privkey.pem; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } } server { listen 80; server_name ваш_домен; return 301 https://$host$request_uri; }
Проверяем, что все работает:
systemctl status nginx curl https://ваш_домен/health
Если пришел ответ {"status": "ok"} — все готово. Swagger теперь доступен по адресу https://ваш_домен/docs.
Сертификат автоматически обновляется через cron, Certbot настраивает это сам при установке. Можно проверить:
certbot renew --dry-run
Итог
Сегодня мы прошли большой путь — от голого графа до полноценного продакшен-стека.
Разобрались с экосистемой LangGraph: что такое узлы, ребра, состояние и чекпоинтер. Подняли LangGraph Server через CLI, написали графы с реальными инструментами и MCP-серверами, потестировали все через LangGraph Studio. Разобрали роутинг и поняли почему реактивные агенты — не серебряная пуля. Написали FastAPI-сервис с LangGraph SDK, задеплоили оба проекта на VPS, закрыли лишние порты, прикрутили домен и SSL.
На выходе получили то с чего начали разговор в самом начале: модель есть — теперь есть и продукт.
Весь код из статьи доступен на GitHub:
LangGraph CLI проект — HabrGraphCLI;
FastAPI + SDK — FastApiGraphSDKHabr.
Спасибо, что дочитали — увидимся в следующей части.
