LangGraph: ИИ-агенты с графом, памятью и контролем исполнения

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

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

Здесь рассматривается Python-репозиторий. Для JavaScript и TypeScript есть отдельный LangGraph.js. Если приложение уже написано на JS, обзор LangChain.js в связанных материалах поможет разобраться с соседним уровнем экосистемы.

Граф требует заранее продумать данные и переходы. Это оправдано для повторяемого процесса с несколькими исходами. Если нужен один ответ на вопрос, начните с обычного вызова модели или готового агентного API; потом выделите граф для тех мест, где появился настоящий порядок действий.

State описывает поля текущего процесса. Node получает состояние и возвращает изменения. Edge ведёт к следующему узлу; условная связь выбирает путь по данным. StateGraph собирает описание, а compile() превращает его в исполняемый граф.

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

Основной пакет langgraph требует Python 3.10 или новее. Для отдельного проекта создайте окружение и установите библиотеку. Для показанного ниже графа не нужен ключ модели: узлы работают с обычными данными.

uv init graph-demo
cd graph-demo
uv add langgraph

В репозитории несколько пакетов, которые выпускаются независимо. Метка cli==0.4.33 в статистике относится к последнему GitHub-релизу командной строки, а не к версии основной библиотеки langgraph. Версии установленных зависимостей смотрите в lock-файле проекта и фиксируйте при развёртывании.

Первый пример проверяет строку по простому правилу и назначает очередь. Он показывает порядок шагов без затрат на модель. Сохраните его как main.py и выполните uv run python main.py.

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

class TicketState(TypedDict):
    ticket: str
    category: str
    queue: str

def classify(state: TicketState):
    category = "billing" if "оплат" in state["ticket"].lower() else "support"
    return {"category": category}

def billing(state: TicketState):
    return {"queue": "Финансовый отдел"}

def support(state: TicketState):
    return {"queue": "Техническая поддержка"}

builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("billing", billing)
builder.add_node("support", support)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda state: state["category"])
builder.add_edge("billing", END)
builder.add_edge("support", END)
graph = builder.compile()
print(graph.invoke({"ticket": "Не прошла оплата", "category": "", "queue": ""}))

Здесь ожидается очередь «Финансовый отдел», потому что входная строка содержит «оплат». Для реальной классификации замените функцию classify вызовом модели или собственного классификатора. Переходы и финальная обработка при этом останутся частью графа.

Checkpointer сохраняет состояние исполнения в контрольных точках. Отдельную цепочку запусков связывает thread_id. Один идентификатор нужно передавать при чтении состояния и продолжении той же работы. Наличие checkpointer позволяет использовать остановки для человека и просматривать историю шагов.

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo-ticket-1"}}
graph.invoke({"ticket": "Не прошла оплата", "category": "", "queue": ""}, config)
print(graph.get_state(config).values)

InMemorySaver хранит данные в оперативной памяти. После завершения процесса они исчезают. Для переживания перезапуска выбирайте постоянное хранилище, например SQLite для локальной разработки или PostgreSQL для сервиса, и настраивайте соответствующий пакет checkpointer.

Хранилище со временем накапливает историю. Заранее задайте срок хранения, удаление старых данных и правила доступа к заявкам. thread_id нужен для группировки исполнения, но проверка права пользователя на эту заявку выполняется приложением.

interrupt() приостанавливает узел и отдаёт приложению данные для человека. После ответа тот же запуск продолжают через Command(resume=...). Нужны checkpointer и постоянный thread_id: без них процесс не сможет связать ответ с сохранённой работой.

from typing import TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt

class ReviewState(TypedDict):
    draft: str
    approved: bool

def review(state: ReviewState):
    approved = interrupt({"question": "Подтвердить ответ?", "draft": state["draft"]})
    return {"approved": bool(approved)}

builder = StateGraph(ReviewState)
builder.add_node("review", review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "review-1"}}
print(graph.invoke({"draft": "Уточните номер заказа.", "approved": False}, config))
# После получения решения человека:
print(graph.invoke(Command(resume=True), config))

При возобновлении код узла до interrupt() выполняется повторно. Не ставьте перед остановкой необратимое действие без защиты от повторов. Ответ в примере приходит прямо из кода; в приложении его должен передать авторизованный оператор.

Store хранит данные отдельно от состояния одного запуска. Это подходит для предпочтений пользователя и сведений, которые нужны в разных диалогах. Checkpointer отвечает за ход конкретного процесса, Store за долгосрочные записи. Их можно подключить одновременно.

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

LangGraph не распознаёт сложный скан только потому, что граф называется RAG. Для подготовки PDF пригодится отдельный обработчик вроде Docling. Проверяйте отдельно извлечение текста, попадание нужного фрагмента в поиск и то, как модель использовала найденные сведения.

Узел графа может вызвать модель напрямую или использовать компоненты LangChain. Framework управляет исполнением, а модель и её инструменты выбираются отдельно. Если задача требует вызова функций, убедитесь, что выбранная модель поддерживает этот способ работы.

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

Для стандартного цикла «модель выбирает инструмент, инструмент возвращает результат» часто удобнее готовый агент LangChain. Сам граф становится полезен, когда нужно управлять этим циклом вместе с проверками, дополнительными этапами и сохранённым состоянием.

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

for part in graph.stream(
    {"ticket": "Не прошла оплата", "category": "", "queue": ""},
    stream_mode="updates", version="v2",
):
    print(part["type"], part["data"])

Формат version="v2" требует LangGraph 1.1 или новее и даёт единообразные поля type, ns и data. Перед отправкой событий в браузер выберите разрешённые поля. Поток состояния может включать внутренние данные, которые не должны появляться в клиентском интерфейсе.

Граф можно встроить в собственный Python-сервис. Для простого локального запуска достаточно библиотеки; способ приёма HTTP-запросов, очередь и постоянное хранилище вы выбираете отдельно. У локального графа нет обязательной подписки на LangSmith.

LangSmith Deployment добавляет инфраструктуру для агентных приложений: API-сервер, выполнение длительных задач и работу с сохранённым состоянием. Есть разные варианты размещения, включая управляемый сервис и развёртывание в своей инфраструктуре. Это отдельный продукт со своими условиями.

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

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

Начинайте тестирование с функций без модели, как в быстром старте. Затем отдельно проверяйте ответы модели на небольшой набор примеров. Так станет видно, ошибка находится в переходах графа, инструменте или содержании ответа.

LangGraph распространяется по MIT: разрешены изменения и коммерческое использование с сохранением уведомления об авторских правах и лицензии. Гарантии качества и пригодности лицензия не даёт. У моделей, подключённых сервисов и LangSmith собственные условия.

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

Выбирайте LangGraph, когда условия переходов и восстановление исполнения важнее готового описания команды по ролям. Сравните один процесс с CrewAI и Agent Framework: сколько кода потребовали остановка для человека, сохранение состояния и обработка ошибки. Их подробные обзоры доступны в таблице альтернатив.

• LangGraph на GitHub
• Официальный сайт LangGraph
• Документация LangGraph