LangGraph: граф состояний, память и человек в контуре

Зачем это нужно

В уроке 8.5.1 вы написали мини-движок графа состояний (задание stage-8/8-5-state-graph): add_node, add_edge, add_conditional_edges, compile, invoke, редьюсеры, лимит шагов, контрольные точки. Теперь то же самое, но настоящий LangGraphLangGraphEN LangGraphОткрытый (MIT) фреймворк LangChain Inc. для агентов с состоянием: граф узлов над общим состоянием, чекпоинты после каждого шага, пауза для человека (interrupt), стриминг, подграфы. Работает и без LangChain.. Имена совпадают намеренно. Новое — то, что в учебном движке писать долго: чекпоинты в SQLite и Postgres, пауза для человека, которая переживает перезапуск процесса, стриминг токенов, история шагов, готовые узлы для инструментов и интеграция с моделями LangChain.

Цена тоже есть. Это ещё одна зависимость со своей экосистемой, а API меняется: в версии 1.0 готовый агент create_react_agent переехал в другой пакет, в 1.1 лимит шагов по умолчанию вырос с 25 до 10 000. Ошибки внутри движка читать тяжелее, чем ошибки своего цикла на 100 строк. Поэтому урок построен вокруг вопросов «какую проблему решает», «как отлаживать и тестировать» и «как выглядит в продакшене», а не вокруг списка классов.

Сквозной пример — ассистент онлайн-школы из программ DDD и SDD. Он отвечает на вопросы по каталогу и записывает на поток, но только после подтверждения человеком. Правила (вместимость, повторная запись) проверяет код, а не модель.

Установка (проверено на langgraph 1.2.14, langchain-core 1.6.7, langchain 1.4.3, langchain-openai 1.6.7, langgraph-checkpoint-sqlite 3.1.1, Python 3.11, октябрь 2026):

python -m venv .venv && source .venv/bin/activate
pip install "langgraph==1.2.14" "langchain==1.4.3" "langchain-openai==1.6.7" "langgraph-checkpoint-sqlite==3.1.1" pytest

Весь код урока запускается без ключей и сети: вместо LLM работает фейковая модель со сценарием ответов.

От мини-движка к LangGraph

Ваш движок из 8.5.1	LangGraph 1.2	Что добавлено
состояние-словарь, редьюсеры	TypedDict + Annotated[тип, редьюсер]	add_messages для сообщений чата
add_node, add_edge, add_conditional_edges	те же методы StateGraph, START, END	контекст запуска Runtime, политики повторов и таймаутов на узле
compile, invoke	compile(checkpointer=...), invoke, stream	режимы стриминга stream_mode
контрольные точки в словаре	InMemorySaver, SqliteSaver, PostgresSaver	история, перемотка, пауза между процессами
лимит шагов	config["recursion_limit"]	по умолчанию 10 007 — задавайте сами
возобновление	invoke(None, config), Command(resume=...)	interrupt() внутри узла
get_state → словарь, next — строка	get_state → StateSnapshot, next — кортеж узлов	параллельные узлы, get_state_history

Модель исполнения та же, что в вашем движке, только с параллельностью. Скомпилированный граф работает супершагами (идея Pregel, системы Google для обработки графов): на каждом супершаге выполняются все активные узлы, их частичные обновления сливаются в состояние через редьюсеры, затем чекпоинтерчекпоинтер (хранилище чекпоинтов)EN checkpointerОбъект, который сохраняет снимок состояния графа после каждого супершага и отдаёт его по thread_id. В LangGraph: InMemorySaver для тестов, SqliteSaver и PostgresSaver (отдельные пакеты) для работы. сохраняет снимок. Два узла, запущенные из START параллельно, попадут в один супершаг. Если оба пишут ключ без редьюсераредьюсер (правило слияния состояния)EN reducerФункция (старое значение, обновление) → новое значение для одного ключа состояния графа. Без редьюсера обновление перезаписывает ключ; с operator.add списки сообщений дописываются., граф остановится с ошибкой InvalidUpdateError: At key 'found': Can receive only one value per step. Мы проверили это на версии 1.2.14.

Состояние, узлы и рёбра

Сначала домен. Он вынесен в отдельный модуль и о LangGraph ничего не знает, поэтому в уроке 8.5.3 его без изменений подключат два других SDK. Поиск — упрощённый лексический поиск из урока 8.2.1: вместо TF-IDF считаем общие «основы» слов. Поток — упрощённый агрегатагрегатEN aggregateКластер сущностей и объектов-значений, который меняется только целиком через корень и после каждой операции соблюдает свои инварианты. Граница агрегата — граница транзакции: одна транзакция меняет один агрегат. В онлайн-школе — поток с записями и листом ожидания. из DDD.4.4. Его инвариантинвариантEN invariantПравило предметной области, которое должно выполняться всегда, после каждой завершённой операции: «записей, занимающих место, не больше вместимости потока». Настоящий инвариант защищают внутри одного агрегата; то, что может быть временно нарушено, — забота политик и согласованности в конечном счёте. — вместимость, плюс правило «нельзя записаться дважды». Возврат за 7 суток — правило операции отмены, а не инвариант потока; здесь он только строка каталога для ответов.

school_domain.py

"""Домен онлайн-школы: каталог, поиск и поток с инвариантом вместимости. О LangGraph ничего не знает."""
import re
from dataclasses import dataclass, field

CATALOG = [
    "Курс «Python с нуля»: 8 недель, цена 24 000 руб. Поток python-2026-11 стартует 10 ноября.",
    "Курс «Машинное обучение»: 12 недель, нужен Python. Поток ml-2026-12 стартует 1 декабря.",
    "Отмена записи за 7 суток до старта и раньше: оплата возвращается полностью, позже — без возврата.",
    "Сертификат выдаётся, если сданы все домашние задания и итоговый проект.",
]


def stems(text: str) -> set[str]:
    """Грубые «основы»: первые 5 букв слов длиннее двух букв (курс, курса, курсы → курс)."""
    return {w[:5] for w in re.findall(r"\w+", text.lower()) if len(w) > 2}


def search(query: str, k: int = 2) -> list[str]:
    """k фрагментов каталога с наибольшим числом общих основ с вопросом."""
    q = stems(query)
    scored = sorted(((len(q & stems(doc)), doc) for doc in CATALOG), key=lambda p: -p[0])
    return [doc for score, doc in scored[:k] if score > 0]


class DomainError(Exception):
    """Нарушено правило Набора — агрегат отказал."""


@dataclass
class Cohort:
    """Поток курса (упрощённый агрегат из DDD.4.4)."""
    cohort_id: str
    capacity: int
    enrolled: set[str] = field(default_factory=set)

    def enroll(self, student_id: str) -> None:
        if student_id in self.enrolled:
            raise DomainError(f"студент {student_id} уже записан на {self.cohort_id}")
        if len(self.enrolled) >= self.capacity:
            raise DomainError(f"на {self.cohort_id} нет мест, можно встать в лист ожидания")
        self.enrolled.add(student_id)


def enroll_student(cohorts: dict, cohort_id: str, student_id: str) -> str:
    """Инструмент «записать на поток». Правила проверяет агрегат, а не модель."""
    if cohort_id not in cohorts:
        return f"Ошибка: потока {cohort_id} нет"
    try:
        cohorts[cohort_id].enroll(student_id)
    except DomainError as e:
        return f"Отказ: {e}"
    return f"Готово: {student_id} записан на {cohort_id}, статус PENDING_PAYMENT"

Теперь граф. Состояние — TypedDict: messages с редьюсером add_messages, остальные ключи без редьюсера (новое значение заменяет старое). add_messages умнее, чем operator.add из 8.5.1: дописывает новые сообщения, но сообщение с уже известным id заменяет старое. Так можно поправить сообщение в истории, а RemoveMessage удаляет его. Готовый класс MessagesState — это состояние из одного такого ключа.

flowchart TD
S(["START"]) --> R["router<br/>вопрос или действие?"]
R -->|"question"| RET["retrieve<br/>поиск по каталогу"]
RET --> A["answer<br/>ответ со ссылками"]
A --> E(["END"])
R -->|"action"| P["plan_action<br/>модель предлагает вызов"]
P -->|"есть tool_calls"| C["confirm<br/>interrupt: спросить человека"]
P -->|"уточняющий вопрос"| E
C --> RP["report<br/>сообщить результат"]
RP --> E

Граф ассистента: сплошные стрелки — обычные рёбра, подписанные — условные; на confirm граф встаёт на паузу

school_graph.py

"""Ассистент онлайн-школы на LangGraph: маршрутизатор, RAG-узел, действие с подтверждением."""
from dataclasses import dataclass
from typing import Annotated, TypedDict

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AnyMessage, SystemMessage, ToolMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.runtime import Runtime
from langgraph.types import interrupt

from school_domain import enroll_student, search


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]   # редьюсер: дописывать, а не заменять
    route: str                                            # "question" или "action"
    context: list[str]                                    # найденные фрагменты каталога


@dataclass
class User:
    """Кто говорит с ассистентом. Берётся из аутентификации, а не из текста сообщения."""
    student_id: str


def request_enrollment(cohort_id: str) -> str:
    """Записать текущего студента на поток cohort_id, например python-2026-11."""
    # Только схема для модели: запись выполняет узел confirm после подтверждения человеком.


ROUTER = "Просят записать, отменить или поставить в лист ожидания — ответь action, иначе question."
ANSWER = "Отвечай только по источникам, ставь ссылки [1], [2]. Ответа нет — скажи «не знаю».\n"
ACTION = "Ты оформляешь запись. Вызови request_enrollment или уточни, на какой поток записать."
REPORT = "Коротко сообщи пользователю результат действия."


def build_graph(llm, cohorts: dict, checkpointer=None):
    def router(state: State):
        verdict = llm.invoke([SystemMessage(ROUTER), state["messages"][-1]]).content
        return {"route": "action" if "action" in verdict.lower() else "question"}

    def retrieve(state: State):
        return {"context": search(state["messages"][-1].content)}

    def answer(state: State):
        sources = "\n".join(f"[{i}] {doc}" for i, doc in enumerate(state["context"], 1))
        return {"messages": [llm.invoke([SystemMessage(ANSWER + sources), *state["messages"]])]}

    def plan_action(state: State):
        model = llm.bind_tools([request_enrollment])
        return {"messages": [model.invoke([SystemMessage(ACTION), *state["messages"]])]}

    def confirm(state: State, runtime: Runtime[User]):
        call = state["messages"][-1].tool_calls[0]
        cohort_id, student_id = call["args"]["cohort_id"], runtime.context.student_id
        # Здесь граф останавливается; после возобновления interrupt вернёт ответ человека
        decision = interrupt({"action": "записать на поток", "cohort_id": cohort_id,
                              "student_id": student_id})
        if decision == "да":
            result = enroll_student(cohorts, cohort_id, student_id)
        else:
            result = "Отменено: человек не подтвердил действие"
        return {"messages": [ToolMessage(result, tool_call_id=call["id"])]}

    def report(state: State):
        return {"messages": [llm.invoke([SystemMessage(REPORT), *state["messages"]])]}

    g = StateGraph(State, context_schema=User)
    for node in (router, retrieve, answer, plan_action, confirm, report):
        g.add_node(node)                                  # имя узла = имя функции
    g.add_edge(START, "router")
    g.add_conditional_edges("router", lambda s: s["route"],
                            {"question": "retrieve", "action": "plan_action"})
    g.add_edge("retrieve", "answer")
    g.add_edge("answer", END)
    g.add_conditional_edges(                              # модель уточняет вместо вызова → END
        "plan_action", lambda s: "confirm" if s["messages"][-1].tool_calls else END,
        ["confirm", END])
    g.add_edge("confirm", "report")
    g.add_edge("report", END)
    return g.compile(checkpointer=checkpointer)


class ScriptedChatModel(GenericFakeChatModel):
    """Фейковая модель со сценарием ответов; bind_tools ничего не меняет — ответы уже записаны."""

    def bind_tools(self, tools, **kwargs):
        return self

На что обратить внимание.

• Узел — обычная функция «состояние → частичное обновление». Она возвращает только изменённые ключи.
• Условное ребро — функция от состояния плюс карта «ответ → узел» (path_map). Карта нужна и для проверки графа при компиляции, и для рисунка: graph.get_graph().draw_mermaid() печатает схему в формате Mermaid.
• Маршрутизатор устойчив к мусору. Модель может ответить «Action.», «действие» или длинной фразой. Всё, что не содержит action, идёт в безопасную ветку вопросов.
• student_id модели не принадлежит. Его передают при запуске через context=User(...), узел читает runtime.context. В схеме инструмента для модели есть только cohort_id, поэтому записать «от имени соседа» нельзя никаким текстом: это минимальные привилегиипринцип наименьших привилегийEN principle of least privilege из урока 8.3.2.
• Фейковая модель. GenericFakeChatModel из langchain_core возвращает ответы из итератора по порядку и умеет стримить их по словам. Её bind_tools не реализован (NotImplementedError), поэтому подкласс в три строки. Это фейковая модель со сценариемфейковая модель со сценариемEN scripted (fake) modelТестовая замена LLM, которая возвращает заранее записанные ответы и вызовы инструментов по порядку. Делает тесты агента детерминированными, быстрыми и бесплатными; проверяет код вокруг модели, но не качество самой модели.: тесты не зависят от сети, денег и случайности.

Запускаем ветку вопросов:

from langchain_core.messages import HumanMessage
from school_graph import ScriptedChatModel, User, build_graph

llm = ScriptedChatModel(messages=iter([
    "question",                                                      # ответ маршрутизатора
    "Курс «Python с нуля» стоит 24 000 руб., старт 10 ноября [1].",     # ответ узла answer
]))
graph = build_graph(llm, cohorts={})
out = graph.invoke({"messages": [HumanMessage("Сколько стоит курс по Python и когда старт?")]},
                   context=User("s-002"))
print(out["route"], len(out["context"]))
print(out["context"][0])
print(out["messages"][-1].content)

question 2
Курс «Python с нуля»: 8 недель, цена 24 000 руб. Поток python-2026-11 стартует 10 ноября.
Курс «Python с нуля» стоит 24 000 руб., старт 10 ноября [1].

Инструменты и готовый агент

У нашего графа маршрут жёсткий: это workflowworkflow (конвейер с заданными шагами)EN workflowПо определению Anthropic («Building effective agents»): система, где LLM и инструменты связаны заранее заданными путями в коде. В агенте путь выбирает сама модель. Обе разновидности Anthropic называет агентными системами. с одной агентной веткой. Классический агент из урока 8.3.1 — цикл «модель → инструменты → модель». В LangGraph это два узла и условное ребро. ToolNode выполняет все tool_calls последнего сообщения и возвращает ToolMessage, а tools_condition ведёт в узел "tools", если вызовы есть, и в END, если нет.

from langchain.agents import create_agent
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from school_graph import ScriptedChatModel


@tool
def free_seats(cohort_id: str) -> int:
    """Сколько свободных мест на потоке cohort_id."""
    return {"python-2026-11": 3}.get(cohort_id, 0)


def scripted():
    return ScriptedChatModel(messages=iter([
        AIMessage("", tool_calls=[{"name": "free_seats", "id": "c1",
                                   "args": {"cohort_id": "python-2026-11"}}]),
        "На потоке python-2026-11 осталось 3 места.",
    ]))


llm = scripted().bind_tools([free_seats])

def agent(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

g = StateGraph(MessagesState)
g.add_node(agent)
g.add_node("tools", ToolNode([free_seats]))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition)   # есть tool_calls → "tools", иначе → END
g.add_edge("tools", "agent")                        # результат инструмента — снова модели
question = {"messages": [HumanMessage("Есть места на python-2026-11?")]}
for m in g.compile().invoke(question)["messages"]:
    print(type(m).__name__, m.content or m.tool_calls[0]["args"])

react = create_agent(scripted(), [free_seats], system_prompt="Ты ассистент онлайн-школы.")
print(react.invoke(question)["messages"][-1].content, list(react.get_graph().nodes))

HumanMessage Есть места на python-2026-11?
AIMessage {'cohort_id': 'python-2026-11'}
ToolMessage 3
AIMessage На потоке python-2026-11 осталось 3 места.
На потоке python-2026-11 осталось 3 места. ['__start__', 'model', 'tools', '__end__']

Последние две строки кода — тот же агент одной функцией. Обратите внимание на импорт. Готовый ReAct-агент langgraph.prebuilt.create_react_agent в LangGraph 1.0 объявлен устаревшим. При вызове он печатает LangGraphDeprecatedSinceV10: create_react_agent has been moved to langchain.agents, а удалить его обещают в 2.0. Замена — create_agent из пакета langchain. Внутри это тот же граф из узлов model и tools, а поведение настраивают middleware: HumanInTheLoopMiddleware(interrupt_on={"enroll": True}) ставит вызов инструмента на подтверждение (продолжение — Command(resume={"decisions": [{"type": "approve"}]})). Есть и middleware для лимита вызовов модели, повторов и сжатия истории.

Когда что брать: готовый агент — если задача и правда «модель сама решает, какие инструменты звать». Свой граф — если маршрут известен заранее или нужны узлы без LLM (поиск, проверки, подтверждения). Ассистенту школы нужен свой граф: модели незачем решать, искать ли в каталоге перед ответом.

Память: чекпоинтер и потоки

Без чекпоинтера граф забывает всё после invoke. С ним каждый супершаг сохраняется в потокепоток выполнения графа (thread)EN thread (thread_id)Цепочка чекпоинтов одного разговора или задачи, адресуется строкой thread_id в config. Новый вызов с тем же thread_id продолжает состояние, с другим — начинает с нуля. Не путать с потоками операционной системы и с потоком курса (Cohort). — цепочке снимков с идентификатором thread_id. Новый вызов с тем же thread_id продолжает разговор: add_messages допишет новый вопрос к старой истории. Это краткосрочная память из урока 8.3.3. Долгосрочную память между потоками (профиль студента) LangGraph хранит отдельно, в Store (compile(store=...)).

from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from school_graph import ScriptedChatModel, User, build_graph

llm = ScriptedChatModel(messages=iter([
    "question", "Поток python-2026-11 стартует 10 ноября [1].",
    "question", "При отмене за 7 суток до старта деньги вернут полностью [1].",
    "Если отменить запись за неделю до старта, оплату вернут целиком [1].",   # для «перемотки»
]))
graph = build_graph(llm, cohorts={}, checkpointer=InMemorySaver())
anna, boris = ({"configurable": {"thread_id": t}} for t in ("anna", "boris"))
for text in ["Когда старт курса по Python?", "А если я отменю запись, вернут деньги?"]:
    graph.invoke({"messages": [HumanMessage(text)]}, anna, context=User("s-002"))

print(len(graph.get_state(anna).values["messages"]), graph.get_state(boris).values)
history = list(graph.get_state_history(anna))                  # от новых чекпоинтов к старым
for snap in history[:4]:
    print(snap.metadata["step"], snap.next, len(snap.values["messages"]))

before = next(s for s in history if s.next == ("answer",))     # последний вход в answer
out = graph.invoke(None, before.config, context=User("s-002"))   # «перемотка» и новая ветка
print(out["messages"][-1].content)
print(len(history), "->", len(list(graph.get_state_history(anna))))

4 {}
8 () 4
7 ('answer',) 3
6 ('retrieve',) 3
5 ('router',) 3
Если отменить запись за неделю до старта, оплату вернут целиком [1].
10 -> 12

Как читать вывод. У Анны 4 сообщения (два вопроса и два ответа), у Бориса пусто: потоки изолированы. get_state_history отдаёт снимки от новых к старым: на шаге 7 граф стоял перед answer, в состоянии три сообщения. invoke(None, snapshot.config) перезапускает граф с этого снимка. Это и есть «путешествие во времени»«путешествие во времени» (перезапуск с прошлого чекпоинта)EN time travelЗапуск графа из любого прошлого чекпоинта потока: повторить шаги после него (replay) или сначала поправить состояние (update_state) и пойти по новой ветке (fork). Старые чекпоинты сохраняются.. Конфиг снимка содержит checkpoint_id, узел answer выполнился заново и дал другой ответ. Старые снимки не стёрты, появилась новая ветка: их было 10, стало 12. Чтобы сначала поправить состояние (например, подсунуть другой фрагмент каталога), вызывают graph.update_state(snapshot.config, {"context": [...]}) и продолжают с вернувшегося конфига. Так в продакшене разбирают инциденты: воспроизводят разговор до плохого шага и проверяют исправление.

InMemorySaver живёт до конца процесса и годится для тестов. Для работы есть SqliteSaver (пакет langgraph-checkpoint-sqlite) и PostgresSaver (langgraph-checkpoint-postgres). Ещё одна настройка — durability в invoke/stream. По умолчанию "async": чекпоинт пишется параллельно со следующим шагом. "sync" дожидается записи, "exit" пишет только в конце. Как граф продолжает после сбоя и почему побочные эффекты узлов должны быть идемпотентными, вы уже видели в 8.5.1. Здесь те же правила, поэтому повторять не будем.

Человек в контуре: interrupt и Command

Запись на поток меняет данные, и её подтверждает человекчеловек в контуре (подтверждение человеком)EN human-in-the-loop. В узле confirm вызов interrupt(значение)прерывание графа (interrupt)EN interruptВызов interrupt(значение) внутри узла останавливает граф и сохраняет чекпоинт; приложение показывает значение человеку и продолжает граф через Command(resume=ответ). При возобновлении узел выполняется заново с начала, а interrupt возвращает ответ. останавливает граф, сохраняет чекпоинт и отдаёт значение наружу, в ключ __interrupt__ результата. Приложение показывает его человеку. Граф продолжают вызовом invoke(Command(resume=ответ), config), и тогда interrupt возвращает этот ответ.

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from school_domain import Cohort
from school_graph import ScriptedChatModel, User, build_graph

cohorts = {"python-2026-11": Cohort("python-2026-11", capacity=20)}
llm = ScriptedChatModel(messages=iter([
    "action",
    AIMessage("", tool_calls=[{"name": "request_enrollment", "id": "call_1",
                               "args": {"cohort_id": "python-2026-11"}}]),
    "Готово: вы записаны на поток python-2026-11. Осталось оплатить курс.",
]))
graph = build_graph(llm, cohorts, checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "anna-1"}}     # один разговор = один поток

out = graph.invoke({"messages": [HumanMessage("Запишите меня на python-2026-11")]},
                   config, context=User("s-002"))
print(out["__interrupt__"][0].value)
print(graph.get_state(config).next, cohorts["python-2026-11"].enrolled)   # стоим перед confirm

out = graph.invoke(Command(resume="да"), config, context=User("s-002"))
print(out["messages"][-2].content)                     # ToolMessage — результат инструмента
print(out["messages"][-1].content, cohorts["python-2026-11"].enrolled)

{'action': 'записать на поток', 'cohort_id': 'python-2026-11', 'student_id': 's-002'}
('confirm',) set()
Готово: s-002 записан на python-2026-11, статус PENDING_PAYMENT
Готово: вы записаны на поток python-2026-11. Осталось оплатить курс.

До ответа человека никто не записан (set()), граф стоит перед confirm. Главное преимущество перед input() из учебного цикла: пауза хранится в чекпоинте, а не в памяти процесса. Человек может ответить через час, а сервер к тому времени перезапустится. Проверим это на SqliteSaver и двух разных запусках Python:

# resume_demo.py: python resume_demo.py start, затем python resume_demo.py resume
import sys
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command
from school_domain import Cohort
from school_graph import ScriptedChatModel, User, build_graph

cohorts = {"python-2026-11": Cohort("python-2026-11", capacity=20)}
config = {"configurable": {"thread_id": "anna-1"}}
call = AIMessage("", tool_calls=[{"name": "request_enrollment", "id": "call_1",
                                  "args": {"cohort_id": "python-2026-11"}}])

with SqliteSaver.from_conn_string("school.db") as saver:        # чекпоинты — в файле
    if sys.argv[1] == "start":                                  # процесс 1: дошли до вопроса человеку
        graph = build_graph(ScriptedChatModel(messages=iter(["action", call])), cohorts, saver)
        graph.invoke({"messages": [HumanMessage("Запишите меня на python-2026-11")]},
                     config, context=User("s-002"))
        print("ждём подтверждения:", graph.get_state(config).next)
    else:                                                       # процесс 2: человек ответил позже
        llm = ScriptedChatModel(messages=iter(["Готово, вы записаны на python-2026-11."]))
        graph = build_graph(llm, cohorts, saver)
        out = graph.invoke(Command(resume="да"), config, context=User("s-002"))
        print(out["messages"][-1].content, cohorts["python-2026-11"].enrolled)

$ python resume_demo.py start
ждём подтверждения: ('confirm',)
$ python resume_demo.py resume
Готово, вы записаны на python-2026-11. {'s-002'}

Второй процесс не вызывал ни маршрутизатор, ни plan_action: в сценарии его модели один ответ, и его хватило. Состояние разговора приехало из school.db. Данные домена (cohorts) здесь живут в памяти и создаются заново, в реальной системе это своя база.

Три правила, которые ломают новички (каждое мы проверили на 1.2.14):

1. При возобновлении узел выполняется заново с самого начала. Код до interrupt выполнится дважды. Поэтому запись в базу, письмо и оплата — только после interrupt. Если до паузы нужен побочный эффект, вынесите его в отдельный узел.
2. Не оборачивайте interrupt в try/except Exception. Пауза реализована исключением GraphInterrupt (наследник Exception). Перехватите его — граф не остановится, а молча пойдёт дальше.
3. Инварианты — в инструменте, подтверждение — в графе. Человек сказал «да», но мест нет — агрегат откажет. Человек нужен для необратимого и чувствительного, а не вместо правил. Отказ возвращается модели как ToolMessage, и она объясняет его пользователю.

Стриминг

Пользователь не должен смотреть на пустой экран, пока граф ищет и думает. graph.stream отдаёт события по ходу выполнения, режим задаёт stream_mode:

Режим	Что приходит	Зачем
"values"	полное состояние после каждого шага	отладка
"updates"	{узел: обновление} после каждого узла	индикатор «ищу в каталоге…»
"messages"	токены LLM и метаданные, из какого узла	печать ответа по мере генерации
"custom"	всё, что узел отправил через get_stream_writer()	прогресс долгого инструмента
"debug"	подробные события задач и чекпоинтов	глубокая отладка

Есть ещё узкие режимы "checkpoints" и "tasks", всего в 1.2 их семь.

from langchain_core.messages import HumanMessage
from school_graph import ScriptedChatModel, User, build_graph

def fresh():
    return ScriptedChatModel(messages=iter(["question", "Поток python-2026-11 стартует 10 ноября [1]."]))

inp = {"messages": [HumanMessage("Когда старт курса по Python?")]}
for chunk in build_graph(fresh(), {}).stream(inp, stream_mode="updates", context=User("s-002")):
    print({node: list(update) for node, update in chunk.items()})      # какой узел что обновил

for token, meta in build_graph(fresh(), {}).stream(inp, stream_mode="messages", context=User("s-002")):
    if meta["langgraph_node"] == "answer":                              # токены только ответа
        print(repr(token.content), end=" ")

{'router': ['route']}
{'retrieve': ['context']}
{'answer': ['messages']}
'Поток' ' ' 'python-2026-11' ' ' 'стартует' ' ' '10' ' ' 'ноября' ' ' '[1].'

Фильтр по langgraph_node обязателен: маршрутизатор тоже вызывает модель, и без фильтра пользователь увидит слово «question». Несколько режимов сразу (stream_mode=["updates", "messages"]) приходят парами (режим, данные). В сервисе из урока 8.4.3 каждое такое событие становится строкой SSE, у stream есть асинхронный двойник astream.

Подграфы — кратко

Скомпилированный граф можно добавить в другой граф как обычный узел. Так команда «Каталога» владеет RAG-веткой, а команда «Набора» — веткой действий. Это те же ограниченные контексты из DDD, только в виде графов. Общие ключи состояния (здесь messages) передаются между родителем и подграфом, остальные остаются внутри.

from typing import Annotated, TypedDict
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from school_domain import search


class RagState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]   # общий ключ с родителем
    context: list[str]                                    # внутренний ключ подграфа


def retrieve(state: RagState):
    return {"context": search(state["messages"][-1].content)}

def answer(state: RagState):
    return {"messages": [AIMessage(f"Нашёл фрагментов: {len(state['context'])}")]}

rag = StateGraph(RagState)
rag.add_node(retrieve); rag.add_node(answer)
rag.add_edge(START, "retrieve"); rag.add_edge("retrieve", "answer")

class Parent(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

parent = StateGraph(Parent)
parent.add_node("rag", rag.compile())                     # скомпилированный граф — обычный узел
parent.add_edge(START, "rag")
for ns, chunk in parent.compile().stream({"messages": [HumanMessage("Когда старт по Python?")]},
                                         stream_mode="updates", subgraphs=True):
    print(ns, list(chunk))

('rag:aa9eb8be-5852-60e8-5cd7-285c0a7c9727',) ['retrieve']
('rag:aa9eb8be-5852-60e8-5cd7-285c0a7c9727',) ['answer']
() ['rag']

С subgraphs=True события подграфа приходят с «пространством имён» (rag:<id задачи>, идентификатор у вас будет другим). Чекпоинтер достаточно передать родителю, и interrupt внутри подграфа работает так же.

Как тестировать граф

Граф — это код, и тестируют его как код: быстро, детерминированно, на каждый коммит. Фейковая модель «играет» нужный сценарий, а тест проверяет, что сделал ваш код: куда пошёл маршрут, где граф остановился, что записано в домен. Репозиторий потоков передаётся в build_graph аргументом, поэтому каждый тест получает свой и тесты не мешают друг другу.

test_school.py

"""Детерминированные тесты графа: модель — сценарий, проверяем маршруты, остановки и инварианты."""
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from school_domain import Cohort
from school_graph import ScriptedChatModel, User, build_graph

CALL = AIMessage("", tool_calls=[{"name": "request_enrollment", "id": "c1",
                                  "args": {"cohort_id": "python-2026-11"}}])


def start(script, text, cohorts=None):
    """Собрать граф со сценарием модели и прогнать первое сообщение в отдельном потоке."""
    cohorts = cohorts if cohorts is not None else {"python-2026-11": Cohort("python-2026-11", 20)}
    graph = build_graph(ScriptedChatModel(messages=iter(script)), cohorts, InMemorySaver())
    config = {"configurable": {"thread_id": "test"}}
    out = graph.invoke({"messages": [HumanMessage(text)]}, config, context=User("s-002"))
    return graph, config, out, cohorts


def test_question_goes_to_rag():
    """Вопрос о курсе идёт через поиск, ответ опирается на найденный фрагмент."""
    graph, config, out, _ = start(["question", "Старт 10 ноября [1]."], "Когда старт курса по Python?")
    assert out["route"] == "question"
    assert "10 ноября" in out["context"][0]


def test_action_waits_for_human():
    """Запись не происходит, пока человек не ответил: граф стоит перед узлом confirm."""
    graph, config, out, cohorts = start(["action", CALL], "Запишите меня на python-2026-11")
    assert graph.get_state(config).next == ("confirm",)
    assert cohorts["python-2026-11"].enrolled == set()


def test_rejection_writes_nothing():
    """Отказ человека — запись не создана, модель получает сообщение об отмене."""
    graph, config, out, cohorts = start(["action", CALL, "Хорошо, не записываю."], "Запишите меня")
    out = graph.invoke(Command(resume="нет"), config, context=User("s-002"))
    assert cohorts["python-2026-11"].enrolled == set()
    assert out["messages"][-2].content.startswith("Отменено")


def test_full_cohort_is_refused_by_invariant():
    """Даже после «да» агрегат не пускает на заполненный поток."""
    full = {"python-2026-11": Cohort("python-2026-11", capacity=1, enrolled={"s-001"})}
    graph, config, out, cohorts = start(["action", CALL, "Мест нет."], "Запишите меня", full)
    out = graph.invoke(Command(resume="да"), config, context=User("s-002"))
    assert "нет мест" in out["messages"][-2].content
    assert cohorts["python-2026-11"].enrolled == {"s-001"}

$ pytest -v test_school.py
test_school.py::test_question_goes_to_rag PASSED                         [ 25%]
test_school.py::test_action_waits_for_human PASSED                       [ 50%]
test_school.py::test_rejection_writes_nothing PASSED                     [ 75%]
test_school.py::test_full_cohort_is_refused_by_invariant PASSED          [100%]
============================== 4 passed in 0.41s ===============================

Чего такие тесты не проверяют: хорошо ли настоящая модель отличает вопрос от действия и правильно ли заполняет cohort_id. Это вопрос качества модели, и на него отвечают эвалы на реальных логах (урок 8.4.2). Хорошая привычка — держать оба слоя: быстрые сценарные тесты в CI на каждый коммит и регрессионный прогон с настоящей моделью перед сменой модели или промпта. Ещё два приёма для продакшена (подробно — в уроке 8.5.4): у add_node есть retry_policy и timeout, а трассы LangGraph по умолчанию никуда не отправляет. Отправка в LangSmith включается переменными окружения LANGSMITH_TRACING=true и LANGSMITH_API_KEY, открытые альтернативы — Langfuse и Phoenix.

Практика

Соберите school_domain.py, school_graph.py и test_school.py из урока, запустите pytest -v, затем выполните упражнения. Решения — под спойлерами, все проверены на langgraph 1.2.14.

Упражнение 1. Бюджет шагов. Модель «зациклилась» и вечно зовёт инструмент. Соберите агент из раздела «Инструменты» с моделью, которая бесконечно возвращает вызов free_seats. Добейтесь, чтобы через 6 шагов пользователь получил вежливый отказ, а не зависший запрос.

Решение

import itertools
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.errors import GraphRecursionError

stuck = (AIMessage("", tool_calls=[{"name": "free_seats", "id": f"c{i}", "args": {"cohort_id": "x"}}])
         for i in itertools.count())        # каждый раз новое сообщение: одинаковый id add_messages заменил бы
llm = ScriptedChatModel(messages=stuck)
# ... граф agent ⇄ tools как в разделе «Инструменты» ...
try:
    app.invoke({"messages": [HumanMessage("Есть места?")]}, {"recursion_limit": 6})
except GraphRecursionError:
    print("За 6 шагов не справились — передаю вопрос оператору.")

Вывод: За 6 шагов не справились — передаю вопрос оператору. Без recursion_limit в версии 1.2 граф сделал бы 10 007 супершагов, а платная модель — тысячи вызовов. Ловушка в комментарии реальна. Если генератор отдаёт один и тот же объект AIMessage, add_messages получает сообщение с известным id. Он заменяет старое сообщение вместо того, чтобы дописать новое, последним в истории остаётся ToolMessage, и граф тихо завершается.

Упражнение 2. Человек правит действие. Оператор видит, что студенту нужен поток ml-2026-12, а не python-2026-11. Пусть Command(resume={"cohort_id": "ml-2026-12"}) означает «да, но на этот поток». Напишите тест. Подумайте, что сказать модели: её вызов отличается от выполненного.

Решение

В confirm после interrupt:

        note = ""
        if isinstance(decision, dict):               # {"cohort_id": ...} — человек поправил поток
            cohort_id, decision = decision["cohort_id"], "да"
            note = f"Человек заменил поток на {cohort_id}. "
        if decision == "да":
            result = note + enroll_student(cohorts, cohort_id, student_id)

Тест: два потока в cohorts, сценарий ["action", CALL, "Записал на ML."], после invoke(Command(resume={"cohort_id": "ml-2026-12"}), ...) проверить, что s-002 записан на ml-2026-12, python-2026-11 пуст, а ToolMessage начинается с «Человек заменил поток». Пометка в ToolMessage нужна, чтобы модель не сообщила студенту старый поток и не повторила свой вызов. Так же поступает HumanInTheLoopMiddleware: к результату отредактированного вызова он добавляет пояснение для модели.

Упражнение 3. Регрессионный набор сценариев. Превратите тесты в таблицу сценариев через pytest.mark.parametrize: сообщение, сценарий модели, ожидаемый get_state(config).next. Добавьте краевые случаи: модель уточняет поток вместо вызова инструмента, маршрутизатор отвечает мусором.

Решение

SCENARIOS = [  # (сообщение, сценарий модели, ожидаемый следующий узел)
    ("Когда старт по Python?", ["question", "10 ноября [1]."], ()),
    ("Запишите меня на python-2026-11", ["action", CALL], ("confirm",)),
    ("Запишите меня куда-нибудь", ["action", "На какой поток?"], ()),     # уточнение без вызова
    ("Привет!", ["что-то непонятное", "Здравствуйте!"], ()),             # мусор → вопрос
]

@pytest.mark.parametrize("text,script,expected_next", SCENARIOS)
def test_scenario(text, script, expected_next):
    """Каждый сценарий доходит до ожидаемой точки и никого не записывает без подтверждения."""
    graph, config, out, cohorts = start(script, text)
    assert graph.get_state(config).next == expected_next
    assert cohorts["python-2026-11"].enrolled == set()

4 passed. Каждый баг, найденный в логах, становится новой строкой таблицы. Это регрессионный набор из 8.4.2, только для траекторий графа.

🖥 На вашей машине (4×GPU)

Замените фейковую модель настоящей. build_graph принимает любой чат-объект LangChain, а ChatOpenAI говорит с любым OpenAI-совместимым сервером из урока 7.2.2. Поднимите vLLM с разбором вызовов инструментов, как в 8.3.1:

CUDA_VISIBLE_DEVICES=0 vllm serve Qwen/Qwen3-8B --port 8000 \
    --enable-auto-tool-choice --tool-call-parser hermes --reasoning-parser qwen3

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.sqlite import SqliteSaver

llm = ChatOpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY",
                 model="Qwen/Qwen3-8B", temperature=0)
# Ollama: base_url="http://localhost:11434/v1", api_key="ollama", model="qwen3:8b"
with SqliteSaver.from_conn_string("school.db") as saver:
    graph = build_graph(llm, cohorts, saver)    # дальше — как в примерах урока

bind_tools у ChatOpenAI настоящий: функция request_enrollment превратится в JSON-схему с одним полем cohort_id, описание возьмётся из докстринга.

Эксперимент. Прогоните 40 сообщений (20 вопросов, 20 просьб записать, часть с опечатками и без номера потока) по три раза. Сравните модели в роли маршрутизатора: Qwen3-1.7B на карте 1 и Qwen3-8B на карте 0. Для этого добавьте в build_graph параметр router_llm. Считайте долю верных маршрутов, долю валидных cohort_id в plan_action и задержку узла по режиму stream_mode="updates". Маленькая модель на маршрутизации — частый приём экономии, и в практикуме 8.5.5 он понадобится. Модели 8B в BF16 нужно около 16 ГБ под веса, 1.7B — около 4 ГБ, обе помещаются на свои карты с запасом под KV-кэш (оценки ориентировочные). Подробнее о машине — на странице «Ваша лаборатория».

Китайский угол: LangGraph в hello-agents и AgentScope

В учебнике hello-agents от Datawhale (CC BY-NC-SA) глава 6 «Практика разработки на фреймворках» показывает LangGraph рядом с AutoGen и AgentScope. AgentScope — открытый фреймворк (Apache-2.0), который развивают в Alibaba. После урока полезно сравнить, как один и тот же агент выглядит в графе состояний и в фреймворках, где агенты переписываются сообщениями. Главы 4 и 7 того же учебника повторяют принцип нашего курса: сначала свой цикл и свой мини-фреймворк, потом готовые.

В README AgentScope 2.0 прямо названы знакомые по уроку возможности: прерывание и возобновление агента в реальном времени, система прав с подтверждением человеком, песочницы для исполнения инструментов. Это те же идеи, только в другой обёртке.

Проверь себя

Проверь себя6 вопросов·

1 Узлы search_catalog и search_faq стартуют из START параллельно и оба возвращают {"found": [...]}, а в состоянии found: list[str] без редьюсера. Что произойдёт в LangGraph 1.2?

Победит обновление узла, завершившегося последнимСписки склеятся автоматическиГраф остановится с InvalidUpdateError: один ключ получил два значения за супершагГраф выполнит узлы по очереди

2 В узле подтверждения нужно отправить письмо «ваша заявка у оператора» и записать студента после «да». Где поставить отправку письма?

Перед interrupt — так письмо уйдёт сразуВ отдельный узел перед узлом с interrupt или после interrupt, но не перед ним в том же узлеВ try/except вокруг interrupt

3 Что даёт чекпоинтер (выберите все верные)?Можно выбрать несколько ответов

Продолжение разговора по тому же thread_idПаузу для человека, которая переживает перезапуск процесса (с SqliteSaver или PostgresSaver)Историю шагов и перезапуск с прошлого чекпоинтаАвтоматическую идемпотентность инструментовОбщую долгосрочную память о студенте между всеми его разговорами

4 Студент просит записать его, человек отвечает «да», запись проходит. Сколько раз за весь разговор вызывается модель в графе урока?

5 Почему student_id передаётся через context=User(...), а не аргументом инструмента, который заполняет модель?

Так меньше токеновИначе текст в разговоре (или инъекция в документе) мог бы заставить модель записать другого студента: личность берётся из аутентификации, а не из генерацииLangGraph запрещает строки в аргументах инструментов

6 Какое значение stream_mode выдаёт токены ответа модели вместе с именем узла, который её вызвал?

Проверить ответы

Итоги

• LangGraph — тот же граф состояний, что ваш движок из 8.5.1: StateGraph, TypedDict с редьюсерами (add_messages для чата), узлы-функции, обычные и условные рёбра, START/END, compile. Модель исполнения — супершаги с чекпоинтом после каждого.
• Инструменты — ToolNode и tools_condition. Готовый агент теперь create_agent из langchain (create_react_agent устарел в 1.0), его поведение настраивают middleware.
• Чекпоинтер и thread_id дают память разговора, историю шагов и перемотку. SqliteSaver или PostgresSaver переносят паузу между процессами. recursion_limit задавайте сами: по умолчанию 10 007.
• interrupt + Command(resume=...) — подтверждение необратимого действия человеком. Узел при возобновлении выполняется заново, поэтому побочные эффекты ставят после interrupt. Инварианты проверяет код домена.
• Тестируйте граф с фейковой моделью: маршруты, паузы, инварианты, отказы. Качество настоящей модели проверяют эвалы. Настоящая модель подключается одной строкой через OpenAI-совместимый ChatOpenAI к vLLM или Ollama.

Дальше: урок 8.5.3 — тот же ассистент школы без явного графа: агенты и передачи управления в OpenAI Agents SDK и агентный цикл Claude Code в Claude Agent SDK, сравнение трёх подходов.