LangGraph на Python: первый граф с моделью через шлюз по шагам

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

Как работает LangGraph на Python

В документации LangGraph у графа три составляющих:

• State: схема данных, которая передаётся по графу. Обычно это TypedDict.
• Nodes: функции Python. Узел получает состояние и возвращает частичное обновление, а не всё состояние целиком.
• Edges: переходы. Обычные ведут из одного узла в другой, условные выбирают следующий узел функцией.

Каждое обновление применяется к состоянию через редьюсер ключа. Если редьюсер не задан, новое значение просто заменяет старое. Для накопления, например списка сообщений, редьюсер указывают через Annotated, как operator.add. Выполнение идёт шагами (super-steps): узлы одного шага отрабатывают, их обновления сливаются в состояние, затем стартует следующий шаг. Этого достаточно, чтобы понять поведение примера ниже.

Установка и ключ

pip install -U langgraph langchain-openai
export ZUKKO_API_KEY="fetch_ваш_ключ"

Ключ создаётся в кабинете↗. Для моделей используем ChatOpenAI из langchain-openai: он принимает base_url и ключ, значит работает с любым OpenAI-совместимым сервером. Адрес для Zukko заканчивается на /v1. Про сам пакет и вызовы SDK см. справочник по Python↗.

Первый граф: ответ службы поддержки

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

import os
from typing import Literal, TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

BASE_URL = "https://api.zukko.pro/v1"
KEY = os.environ["ZUKKO_API_KEY"]

cheap = ChatOpenAI(model="deepseek-v4.1-flash", base_url=BASE_URL, api_key=KEY, temperature=0)
strong = ChatOpenAI(model="claude-sonnet-5-5", base_url=BASE_URL, api_key=KEY)  # без temperature: новые Claude принимают только 1.0


class State(TypedDict):

Разберём ключевые места.

Узлы возвращают только изменения. classify отдаёт {"category": ...}, остальные ключи остаются как были. Поэтому в draft мы сами увеличиваем attempts: счётчик живёт в состоянии.

Условные рёбра. route возвращает имя следующего узла, а аннотация Literal[...] подсказывает LangGraph допустимые варианты. Для review мы используем словарь, чтобы вернуть смысловые метки retry и ok и сопоставить их с узлом draft и константой END.

Цикл. Условное ребро из draft через функцию review обратно в draft образует петлю. Она ограничена дважды: условием attempts < 2 в самой логике и параметром recursion_limit при запуске.

Где ставить ограничения

В документации сказано: предел рекурсии задаёт максимум шагов графа за один запуск, при превышении выбрасывается GraphRecursionError, а начиная с версии 1.0.6 значение по умолчанию равно 1000. Для платного API это слишком щедро: цикл с ошибкой в условии выполнит сотни вызовов модели. Поэтому:

1. Всегда передавайте recursion_limit в config, как в примере. Это отдельный ключ config, а не часть configurable.
2. Держите свой счётчик попыток в состоянии, как attempts. Он нагляднее и позволяет завершиться штатно, а не исключением.
3. Назначьте ключу API дневной лимит в $: если ограничения в коде подведут, расход остановится на стороне шлюза. Как это настроить, описано в статье про лимиты расхода для агентов↗.

Смотрим выполнение по шагам

Чтобы увидеть, какой узел что вернул, используйте стриминг по обновлениям:

for update in app.stream(
    {"question": "Не приходит код подтверждения при входе", "attempts": 0},
    config={"recursion_limit": 10},
    stream_mode="updates",
):
    print(update)

Каждый элемент показывает имя узла и то, что он добавил в состояние. Это лучший отладчик для первого графа. Если нужно, чтобы диалог продолжался между запусками, при компиляции передают чекпойнтер, например InMemorySaver из langgraph.checkpoint.memory, а при вызове указывают thread_id в configurable. Для продакшена выбирают постоянное хранилище; подробности в документации LangGraph.

Сколько стоит один прогон

Цены на 18 октября 2026, $ за 1M токенов: deepseek-v4.1-flash вход 0.013 / выход 0.038, claude-sonnet-5-5 вход 0.4 / выход 2. Допустим, на классификацию и проверку вместе уходит 550 токенов входа и 8 токенов выхода, а на черновик 200 токенов входа и 250 токенов выхода.

• дешёвая модель: 550 × 0.013 / 1 000 000 + 8 × 0.038 / 1 000 000 ≈ $0.0000072 + $0.0000003;
• сильная модель: 200 × 0.4 / 1 000 000 + 250 × 2 / 1 000 000 = $0.00008 + $0.0005;
• итого около $0.0006 за обращение.

Если проверка отклонит первый черновик и граф сделает второй, сильная часть удвоится и сумма составит около $0.0012. Обратите внимание, что почти вся стоимость приходится на сильную модель, поэтому каскад и окупается. Цифры показывают порядок величины, на реальных текстах токенов будет больше.

Актуальные цены и список моделей: GET https://api.zukko.pro/v1/models или страница «Модели» в кабинете. Какая модель лучше подойдёт под ваши обращения, проверяйте на своих данных.

Что дальше

Если вы идёте по курсу LangGraph и дошли до второй части, обычно дальше добавляют инструменты (узел вызывает функции, как в quickstart из документации), хранение состояния между запусками и участие человека в цикле. Всё это строится на тех же трёх понятиях: состояние, узлы, рёбра. А общий способ подключения LangChain к шлюзу описан в статье про LangChain и шлюз↗.

Частые вопросы

Нужен ли LangChain, чтобы использовать LangGraph?

Сам граф от LangChain не зависит, но модели удобно брать из langchain-openai. Можно вызывать и обычный openai SDK прямо внутри узла: узел это просто функция.

Почему граф не завершается?

Скорее всего, условное ребро всегда возвращает узел-петлю. Добавьте счётчик в состояние и recursion_limit в config, затем посмотрите стриминг stream_mode="updates".

Какой base_url указывать?

https://api.zukko.pro/v1. LangChain использует OpenAI-совместимый клиент, поэтому суффикс /v1 нужен.

Работает ли это из России?

Оплата из России — СБП, MasterCard или криптовалюта. Код графа не меняется.

Создайте ключ в кабинете↗, задайте ZUKKO_API_KEY и запустите пример: за минуту будет видно, как граф проходит по узлам.