LangGraph

LangGraph это фреймворк для приложений и ИИ-агентов на базе больших языковых моделей (large language model, LLM), в котором процесс выполнения описан как граф состояний и переходов. Узлы графа выполняют работу, рёбра решают, куда идти дальше, а общее состояние живёт между шагами и переживает перезапуск процесса. Такой подход нужен там, где простой последовательности запросов к модели уже не хватает. Агент обращается к внешним инструментам, возвращается к предыдущим этапам, ждёт подтверждения человека.

Что такое LangGraph и какую задачу он решает

LangGraph относится к классу низкоуровневых фреймворков оркестрации агентных приложений. Он не прячет логику агента за одним вызовом, а даёт примитивы, из которых эта логика собирается: разделяемое состояние, узлы-функции и рёбра-переходы. Библиотеку разрабатывает команда LangChain. Версия 1.0 стала общедоступной 29 октября 2025 года. Актуальный релиз на момент подготовки статьи это 1.2.11 от 28 июля 2026 года.

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

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

Архитектура LangGraph, состояние и узлы графа

Архитектура держится на трёх понятиях. State это разделяемая структура данных, снимок приложения на текущий момент. Nodes это обычные функции, которые принимают состояние, что-то делают и возвращают его обновление. Edges это функции или фиксированные переходы, определяющие, какой узел выполнится следующим. Документация формулирует это прямо: узлы делают работу, рёбра говорят, что делать дальше, и внутри узла может быть как вызов LLM, так и обычный код без всякого искусственного интеллекта.

Состояние графа и редьюсеры

Состояние обычно описывают типизированной схемой, чаще всего через TypedDict. Каждый ключ состояния это канал, и у канала может быть редьюсер (reducer) — функция, которая решает, как соединить старое значение с новым. Без редьюсера новое значение затирает старое. С редьюсером add_messages список сообщений накапливается, поэтому история диалога растёт, а не перезаписывается на каждом шаге.

Это не украшение, а несущая конструкция. Именно редьюсеры позволяют нескольким узлам писать в одно поле состояния параллельно и не затирать результаты друг друга.

Условные рёбра и объект Command

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

Второй способ управлять переходом это объект Command, который узел возвращает вместо обычного словаря. В нём можно одновременно обновить состояние и указать следующий узел через параметр goto. Такой приём удобен, когда решение о маршруте естественно принимается внутри узла, а не в отдельной функции-роутере.

Как LangGraph работает под капотом

Под графом лежит модель передачи сообщений (message passing), вдохновлённая системой Pregel от Google. Выполнение идёт дискретными супершагами (super-step). Один супершаг это такт, в котором отрабатывают все узлы, запланированные на этот момент, в том числе параллельные. Завершившийся узел отправляет сообщения по своим рёбрам, получатели попадают в план следующего такта, и так до момента, когда планировать больше нечего.

Чекпоинты, потоки и восстановление после сбоя

На границе каждого супершага чекпоинтер (checkpointer) сохраняет снимок состояния. Снимки группируются в потоки (threads). Идентификатор потока thread_id передаётся в конфигурации запуска и работает как указатель. С тем же значением граф продолжит прерванный прогон, с новым начнёт с чистого листа. Дополнительно LangGraph пишет результаты отдельных узлов внутри супершага. Поэтому при падении одного узла успешные соседи не пересчитываются заново.

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

• Участие человека в процессе. Граф можно остановить, показать состояние человеку и продолжить после его ответа.
• Память между запусками. Follow-up сообщения в тот же поток видят предыдущую переписку.
• Путешествие во времени. Можно вернуться к прошлому чекпоинту, переиграть шаг и разветвить состояние для проверки альтернативы.
• Отказоустойчивость. После падения процесса выполнение продолжается с последнего успешного супершага.

Все четыре пункта опираются на один механизм, поэтому граф без чекпоинтера это, по сути, обычный скрипт с красивой структурой.

Human-in-the-loop и подводные камни функции interrupt

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

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

• Побочные эффекты до паузы должны быть идемпотентными. Вставка записи в базу перед interrupt выполнится повторно на каждом возобновлении и наплодит дублей. Ставьте её после паузы или используйте upsert.
• Не оборачивайте interrupt в широкий блок try except. Пауза реализована через специальное исключение, и перехват всего подряд её просто съест.
• Не меняйте порядок вызовов interrupt внутри узла. Значения для возобновления сопоставляются строго по индексу, поэтому условный пропуск паузы или цикл с недетерминированным числом итераций ломают соответствие.

Эти три правила закрывают большинство необъяснимых на первый взгляд эффектов вроде задвоенных заявок и потерянных подтверждений. Зачем агентам вообще такие паузы, разобрано в статье про эволюцию чат-ботов до ИИ-агентов и шаблоны рабочих процессов LLM.

Режимы durability и цена надёжности

Персистентность стоит времени, поэтому LangGraph даёт выбрать, насколько часто фиксировать состояние. Режим задаётся параметром durability при запуске графа.

Вторая статья расходов это объём хранилища. По умолчанию чекпоинт содержит полное значение каждого канала состояния, поэтому длинный диалог с накапливающимся списком сообщений раздувает базу. Для таких случаев в версиях начиная с 1.2 появился DeltaChannel, который хранит только приращения и восстанавливает значение проигрыванием записей. Механизм помечен как бета, то есть API ещё может измениться.

Выбор хранилища тоже влияет на результат. InMemorySaver держит чекпоинты в оперативной памяти и теряет их при перезапуске. SqliteSaver годится для локальной разработки. В промышленной эксплуатации берут PostgresSaver. И ещё деталь, thread_id стоит держать короче 255 символов, иначе запись упрётся в ограничение колонки.

Когда LangGraph оправдан, а когда избыточен

Фреймворк даёт выигрыш там, где сценарий действительно ветвится и живёт долго.

• Агенты с инструментами. Цикл рассуждение-действие-наблюдение, где число итераций заранее неизвестно.
• Процессы с согласованием. Заявки, платежи, рассылки, где перед необратимым действием нужен человек, причём подтверждение может прийти через сутки.
• Многошаговые RAG-конвейеры. Переформулировка запроса, повторный поиск, проверка ответа критиком и возврат на предыдущий шаг при плохом результате.
• Мультиагентные схемы. Оркестратор и исполнители, где подграфы имеют собственное пространство чекпоинтов.

Общая логика простая — чем больше в сценарии циклов, пауз и точек отказа, тем быстрее окупается явный граф.

Обратная сторона тоже есть. Для одиночного запроса к модели с фиксированным промптом граф это лишний слой: тот же результат даёт прямой вызов клиента. Для строго линейного конвейера из трёх шагов без ветвлений выигрыш тоже сомнительный. И ещё одно. LangGraph управляет потоком выполнения, но не делает модель умнее. Качество решений упирается в саму LLM и в описание инструментов. Практическую сторону темы, от выбора модели до эксплуатации, разбирают на курсе ИИ агенты для оптимизации бизнес-процессов. Сам класс таких систем описан в материале про Agentic AI.

Практика с графом агента на локальной модели

По традиции весь код используемый в статье выкладываем на наш GitHub репозиторий

Стенд предельно скромный: macOS, Python 3.14, Ollama 0.32.9 на локальном порту 11434 и модель qwen2.5:7b с поддержкой вызова инструментов. Ни платных ключей, ни внешних сервисов не нужно. Файл agent_graph.py собирает базовый агентный цикл: состояние с редьюсером, узел модели, узел инструментов и условное ребро между ними.

# протестировано для langgraph 1.2.11, langchain-ollama 1.1.0, Ollama 0.32.9, Python 3.14
"""Минимальный граф агента на LangGraph: состояние, узел модели, узел инструментов и цикл между ними."""

from __future__ import annotations

import json
import os
from typing import Annotated, Any, TypedDict

from langchain.messages import ToolMessage
from langchain.tools import tool
from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

# Модель и адрес Ollama выносим в переменные окружения, чтобы код не был прибит к одной модели
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")

# Локальный справочник вместо похода во внешнюю систему: прогон должен быть воспроизводимым
# Данные по курсу AGENT взяты со страницы курса bigdataschool.ru на 20.04.2026
COURSES = {
    "AGENT": {
        "title": "ИИ агенты для оптимизации бизнес-процессов",
        "price_rub": 66000,
        "days": 6,
    }
}


@tool
def course_info(code: str) -> str:
    """Вернуть данные курса BigDataSchool по коду: название, стоимость в рублях и длительность в днях."""
    item = COURSES.get(code.strip().upper())
    if item is None:
        # Инструмент не бросает исключение, а возвращает читаемую ошибку: модель сможет её обработать
        return json.dumps(
            {"error": f"курс {code} не найден", "known_codes": sorted(COURSES)},
            ensure_ascii=False,
        )
    return json.dumps(item, ensure_ascii=False)


TOOLS = [course_info]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}


class AgentState(TypedDict):
    """Состояние графа. Редьюсер add_messages дописывает новые сообщения, а не затирает список."""

    messages: Annotated[list, add_messages]


def build_model() -> ChatOllama:
    """Локальная модель через Ollama. temperature и seed зафиксированы ради воспроизводимости прогона."""
    return ChatOllama(model=OLLAMA_MODEL, base_url=OLLAMA_URL, temperature=0, seed=0)


def build_graph(model: Any = None):
    """Собрать и скомпилировать граф. Модель передаётся аргументом, чтобы её можно было подменить в тестах."""
    bound = (model or build_model()).bind_tools(TOOLS)

    def call_model(state: AgentState) -> dict:
        # Узел модели: отдаём всю историю сообщений и возвращаем ответ одним элементом списка
        return {"messages": [bound.invoke(state["messages"])]}

    def call_tools(state: AgentState) -> dict:
        # Узел инструментов: выполняем каждый вызов, который запросила модель
        last = state["messages"][-1]
        results = []
        for call in last.tool_calls:
            output = TOOLS_BY_NAME[call["name"]].invoke(call["args"])
            results.append(ToolMessage(content=output, tool_call_id=call["id"]))
        return {"messages": results}

    def should_continue(state: AgentState) -> str:
        # Условное ребро: пока модель просит инструменты, крутимся в цикле, иначе выходим
        return "tools" if getattr(state["messages"][-1], "tool_calls", None) else END

    builder = StateGraph(AgentState)
    builder.add_node("agent", call_model)
    builder.add_node("tools", call_tools)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue, ["tools", END])
    builder.add_edge("tools", "agent")
    return builder.compile()


if __name__ == "__main__":
    graph = build_graph()
    question = "Сколько стоит курс AGENT и сколько дней он идёт?"
    state = graph.invoke({"messages": [{"role": "user", "content": question}]})
    for message in state["messages"]:
        message.pretty_print()

Прогон показывает ровно тот цикл, который нарисован на схеме выше. Модель запрашивает инструмент, получает результат и только после этого формулирует ответ.

Второй файл hitl_approval.py добавляет к тому же графу чекпоинтер SQLite и паузу на подтверждение внутри инструмента. Обратите внимание на порядок: побочный эффект стоит после вызова interrupt, потому что код до паузы выполнится повторно при возобновлении.

# протестировано для langgraph 1.2.11, langgraph-checkpoint-sqlite 3.1.1, langchain-ollama 1.1.0, Ollama 0.32.9, Python 3.14
"""Тот же граф агента, но с чекпоинтером SQLite и паузой на подтверждение человека через interrupt."""

from __future__ import annotations

import os
import sqlite3
from typing import Annotated, Any, TypedDict

from langchain.messages import ToolMessage
from langchain.tools import tool
from langchain_ollama import ChatOllama
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
DB_PATH = os.getenv("LANGGRAPH_DB", "langgraph_demo.db")


@tool
def submit_request(course_code: str, people: int) -> str:
    """Отправить заявку на обучение группы сотрудников по коду курса. Требует подтверждения человека."""
    # Пауза до подтверждения. Значение уедет вызывающей стороне, граф встанет и дождётся ответа
    decision = interrupt(
        {
            "action": "submit_request",
            "course_code": course_code,
            "people": people,
            "question": "Отправляем заявку?",
        }
    )
    # Побочный эффект стоит после interrupt: узел перезапускается целиком, и до паузы код выполнится повторно
    if decision is True:
        return f"Заявка отправлена: курс {course_code}, участников {people}"
    return "Заявка отменена человеком"


TOOLS = [submit_request]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}


class AgentState(TypedDict):
    """Состояние графа. Редьюсер add_messages дописывает новые сообщения в конец истории."""

    messages: Annotated[list, add_messages]


def build_model() -> ChatOllama:
    """Локальная модель через Ollama с зафиксированными temperature и seed."""
    return ChatOllama(model=OLLAMA_MODEL, base_url=OLLAMA_URL, temperature=0, seed=0)


def build_graph(checkpointer: Any, model: Any = None):
    """Собрать граф и скомпилировать его с чекпоинтером: без него interrupt работать не будет."""
    bound = (model or build_model()).bind_tools(TOOLS)

    def call_model(state: AgentState) -> dict:
        return {"messages": [bound.invoke(state["messages"])]}

    def call_tools(state: AgentState) -> dict:
        last = state["messages"][-1]
        results = []
        for call in last.tool_calls:
            output = TOOLS_BY_NAME[call["name"]].invoke(call["args"])
            results.append(ToolMessage(content=output, tool_call_id=call["id"]))
        return {"messages": results}

    def should_continue(state: AgentState) -> str:
        return "tools" if getattr(state["messages"][-1], "tool_calls", None) else END

    builder = StateGraph(AgentState)
    builder.add_node("agent", call_model)
    builder.add_node("tools", call_tools)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue, ["tools", END])
    builder.add_edge("tools", "agent")
    return builder.compile(checkpointer=checkpointer)


if __name__ == "__main__":
    connection = sqlite3.connect(DB_PATH, check_same_thread=False)
    graph = build_graph(SqliteSaver(connection))

    # thread_id это указатель на состояние: с тем же значением граф продолжит прерванный прогон
    config = {"configurable": {"thread_id": "demo-1"}}
    question = "Оформи заявку на курс AGENT для 3 сотрудников"

    first = graph.invoke({"messages": [{"role": "user", "content": question}]}, config)
    print("ПАУЗА НА ПОДТВЕРЖДЕНИЕ:", first["__interrupt__"])

    snapshot = graph.get_state(config)
    print("СЛЕДУЮЩИЙ УЗЕЛ ПОСЛЕ ВОЗОБНОВЛЕНИЯ:", snapshot.next)

    # Возобновляем тот же поток: значение resume станет результатом вызова interrupt внутри инструмента
    final = graph.invoke(Command(resume=True), config)
    for message in final["messages"]:
        message.pretty_print()

    print("ЧЕКПОИНТОВ В ПОТОКЕ:", len(list(graph.get_state_history(config))))
    connection.close()

В выводе видно и объект паузы с параметрами будущей заявки, и подсказку о том, какой узел выполнится после возобновления.

Те же пять чекпоинтов читаются из файла базы отдельным процессом, который ничего не знает о предыдущем запуске. Это и есть практическое подтверждение того, что состояние графа живёт в хранилище, а не в оперативной памяти приложения. Из прогона стоит вынести одно наблюдение. Модели на 7 и 8 миллиардов параметров вызывают инструменты стабильно только на коротких и однозначных запросах. На длинной постановке задачи они иногда отвечают текстом вместо вызова. Проверка на llama3.1:8b дала ту же структуру вывода, что и на qwen2.5:7b, поэтому код не завязан на конкретную модель.

Заключение

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

Референсные ссылки

• LangGraph Graph API overview, официальная документация по состоянию, узлам, рёбрам и супершагам
• Checkpointers, чекпоинты, потоки, режимы durability и DeltaChannel
• Interrupts, правила работы с паузами и возобновлением графа
• Persistence, разделение короткой и долгой памяти агента
• LangGraph 1.0 is now generally available, анонс первого стабильного релиза