17. Персистентность состояния и чекпоинтинг

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

Граф хранит состояние(state) и управляет им только на время одного вызова invoke(), и ни секундой дольше. Когда вы вызываете его, LangGraph создаёт свежее состояние, запускает узлы, объединяет каждое возвращаемое значение в состояние согласно правилам редьюсеров и возвращает финальное состояние вызывающей стороне. Как только он передал это состояние, граф больше его не помнит. Следующий вызов invoke() начинается с абсолютно нового состояния, не связанного с предыдущим вызовом.

Отсюда вытекают две проблемы. Во-первых, messages тоже является частью состояния, поэтому агент не может запомнить ничего из того, что вы сказали ранее. Во-вторых, если запуск завершается сбоем на полпути, всё, что было выполнено до этого момента, исчезает. Допустим, третий узел выбрасывает исключение: результаты, полученные первыми двумя узлами, пропадают вместе с ним, и вам приходится начинать всё сначала. Ещё в разделе 15.1 мы перечислили «восстановление после прерываний» как одну из причин обратиться к LangGraph — именно эту проблему мы тогда имели в виду.

LangGraph решает это на уровне фреймворка. Прикрепите к графу чекпоинтер(checkpointer), и LangGraph будет автоматически сохранять снимок состояния на каждом шаге выполнения. Эти сохранённые снимки переживают вызов invoke(), поэтому следующий вызов может продолжить с того места, где остановился предыдущий. Это свойство — сохранение состояния за пределами одного запуска — называется персистентностью(persistence).

Вы уже пользовались чекпоинтером раньше. В главе 11, когда мы наделили диалогового RAG-агента памятью на несколько запросов, мы передали create_agent(..., checkpointer=InMemorySaver()) и thread_id. Тогда вам достаточно было знать, что чекпоинтер хранит историю диалога для каждого thread_id; мы так и не объяснили, как именно. А в главе 16, когда мы представили параметр checkpointer, мы сказали: «о том, как это работает, мы расскажем в главе 17». Это и есть та самая глава.

Она состоит из трёх частей. В разделе 17.1 мы прикрепляем чекпоинтер к графу и ведём многошаговые диалоги с помощью thread_id. В разделе 17.2 мы открываем сохранённые чекпоинты, чтобы увидеть, что агент знал в тот или иной момент, — чекпоинты являются вашим главным инструментом для выяснения того, почему агент повёл себя неправильно. В разделе 17.3 мы берём граф, который завершился сбоем на середине выполнения, и возобновляем его с места остановки, а не с начала.

17.1) Перенос состояния между вызовами

17.1.1) Граф, который забывает

Мы начали эту главу с утверждения, что граф не сохраняет своё состояние. Давайте подтвердим это на коде.

Граф ниже имеет такую же форму, как граф say_hello из раздела 15.2. Единственное отличие в том, что узел возвращает ответ LLM вместо фиксированной строки.

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
graph = builder.compile()   # без чекпоинтера

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

# Первый вызов — сообщаем своё имя
graph.invoke({"messages": [{"role": "user", "content": "Привет, меня зовут Боб."}]})
 
# Второй вызов — спрашиваем его обратно
result = graph.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]})
print(result["messages"][-1].content)

Вывод:

Извините, но я не знаю вашего имени. Не могли бы вы сказать, как вас зовут?

Единственным сообщением, которое модель получила при втором вызове, было "Как меня зовут?". Состояние из первого вызова больше не находится в графе, поэтому более ранние сообщения — те, что несли имя, — так и не дошли до модели.

Мы, конечно, могли бы исправить это сами. Сохранить messages, возвращённые первым вызовом, и передать их вместе со вторым. Именно так мы управляли историей диалога ещё в главе 8. Но тогда нам пришлось бы писать свой собственный код для хранения и извлечения истории для каждого диалога и каждого пользователя. Это и есть та работа, которую чекпоинтер берёт на себя.

17.1.2) Чекпоинты и чекпоинтеры

Чекпоинтер(checkpointer) — это объект, задача которого — сохранять состояние. Вы создаёте экземпляр — например, InMemorySaver() — и передаёте его в builder.compile(checkpointer=...), чтобы прикрепить его к вашему графу.

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

Автосохранение в видеоигре — правильная мысленная картина. Игра тихо записывает ваш прогресс каждый раз, когда вы проходите значимую точку, чтобы вы могли выйти и вернуться позже или погибнуть, не начиная всё сначала. Чекпоинтер делает ровно это для графа.

Так когда же наступает «значимая точка»? LangGraph делит выполнение графа на этапы, и каждый этап называется суперштагом(super-step). Чекпоинт сохраняется каждый раз, когда завершается суперштаг.

Причина, по которой это суперштаг, а не просто шаг, в том, что один этап может запускать несколько узлов одновременно. В графе вроде того, что из раздела 17.1.1, где узлы выстроены в прямую линию, запуск одного узла — это один суперштаг. Но в графе, где несколько узлов выполняются параллельно, все эти узлы вместе составляют один суперштаг.

Таким образом, даже один вызов invoke() оставляет после себя несколько чекпоинтов. Мы извлечём их и посмотрим, что именно содержит каждый из них, в разделе 17.2.

InMemorySaver, который мы использовали в качестве примера, — это самый простой из существующих чекпоинтеров. Как следует из названия, он хранит чекпоинты в памяти процесса (RAM). Ничего не нужно устанавливать и ничего не нужно настраивать, что делает его хорошим выбором для обучения и локальной разработки. Компромисс в том, что каждый сохранённый чекпоинт исчезает при перезапуске процесса. Мы рассмотрим продакшн-альтернативы в разделе 17.1.5.

17.1.3) Добавление чекпоинтера

Для прикрепления чекпоинтера требуется всего две вещи.

1. Создать экземпляр чекпоинтера и передать его в compile().
2. Передавать config, содержащий thread_id, при каждом вызове invoke().

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

Давайте применим оба пункта к графу из раздела 17.1.1.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
# 1. Создаём чекпоинтер и передаём его в compile()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 2. Передаём в invoke() config, несущий thread_id
config = {"configurable": {"thread_id": "1"}}
 
graph.invoke(
    {"messages": [{"role": "user", "content": "Привет, меня зовут Боб."}]},
    config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Как меня зовут?"}]},
    config,
)
print(result["messages"][-1].content)

Вывод:

Вас зовут Боб.

Те же два вызова, что и в разделе 17.1.1, но результат другой. На этот раз имя закрепляется.

Вот почему. Граф с прикреплённым чекпоинтером загружает сохранённое состояние перед запуском узла llm_call. Это состояние уже содержит первый обмен репликами. Новое сообщение, которое мы передали, затем объединяется с ним. Как мы видели в разделе 15.2.2, поле messages несёт редьюсер add_messages, поэтому новое сообщение добавляется к существующему списку. В итоге модель получает три сообщения: приветствие, свой первый ответ и новый вопрос.

Мы отправляем только новое сообщение, а LangGraph загружает более раннюю часть диалога из последнего чекпоинта. История диалога, которой мы управляли вручную в главе 8, теперь управляется LangGraph.

17.1.4) thread_id: идентификатор, разделяющий диалоги

thread_id — это идентификатор, который отличает один диалог от другого. Значение выбираете вы. В разделе 17.1.3 мы использовали "1", но подойдёт любая строка. Вызовите граф с тем же thread_id — и вы продолжите этот диалог; вызовите его с другим — и вы начнёте отдельный диалог.

Давайте это подтвердим. Мы устроим так, чтобы Алиса и Боб вели разные диалоги через один и тот же граф.

def send(thread_id: str, text: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config,
    )
    return result["messages"][-1].content
 
# Диалог Алисы
send("alice", "Мой любимый цвет — бирюзовый.")
 
# Диалог Боба — другой thread_id
send("bob", "Мой любимый цвет — оранжевый.")
 
# Спрашиваем у каждого из них снова
print("Алиса:", send("alice", "Какой мой любимый цвет?"))
print("Боб:  ", send("bob", "Какой мой любимый цвет?"))

Вывод:

Алиса: Ваш любимый цвет — бирюзовый.
Боб:   Ваш любимый цвет — оранжевый.

Оба диалога прошли через один и тот же объект graph и один и тот же чекпоинтер, и всё же они никогда не смешивались. thread_id — это первичный ключ, который чекпоинтер использует для хранения и поиска состояния. Разные ключи — совершенно раздельное хранилище.

Так что же произойдёт, если оставить значение без указания?

graph.invoke({"messages": [{"role": "user", "content": "Привет"}]})

Вывод:

ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

Граф вообще не запускается. Как только чекпоинтер прикреплён, thread_id перестаёт быть необязательным — он обязателен.

Учитывая то, что мы только что увидели, это логично. Перед запуском узла чекпоинтер должен загрузить сохранённое состояние — и thread_id — это то, что сообщает ему, состояние какого диалога загружать.

Это базовая форма сервиса чат-бота: один граф, один чекпоинтер и один thread_id на пользователя или на чат-комнату.

17.1.5) Ограничения InMemorySaver и продакшн-альтернативы

Ранее мы сказали, что InMemorySaver хранит чекпоинты в памяти. С этим выбором связаны два ограничения.

Перезапустите процесс — и всё пропало. Переразверните сервис или поднимите сервер заново — и все накопленные до этого диалоги исчезают.

Отдельные процессы не могут его совместно использовать. Реальный сервис распределяет входящие запросы по нескольким процессам. У каждого процесса своя память, поэтому диалог, сохранённый процессом A, невидим для процесса B. Пользователь может каждый раз отправлять один и тот же thread_id и всё равно наблюдать, как диалог рассыпается, в зависимости от того, какой процесс окажется тем, что подхватит запрос.

Вот почему в продакшене используются чекпоинтеры, которые хранят чекпоинты в базе данных.

• SqliteSaver / AsyncSqliteSaver (langgraph-checkpoint-sqlite) — хранит всё в одном файле. Хороший выбор для небольшого сервиса, работающего на одном сервере, или для локального прототипа.
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres) — хранит чекпоинты на сервере базы данных. Добавьте больше серверов — и каждый процесс по-прежнему видит одни и те же чекпоинты. Это тот чекпоинтер, который документация LangGraph рекомендует для продакшена.

Все они реализуют тот же интерфейс, что и InMemorySaver. Код вашего графа, ваши узлы и то, как вы используете thread_id, остаются в точности такими же. Меняется лишь то, как вы создаёте чекпоинтер.

create_agent, с которым вы познакомились в главе 16, использует чекпоинтеры точно так же. Передайте чекпоинтер в его параметр checkpointer и передайте в invoke() config, несущий thread_id. Именно это заставило диалогового RAG-агента из главы 11 помнить более ранние шаги.

Мы продолжим использовать InMemorySaver до конца этой главы. Где бы ни хранились чекпоинты, способ работы с ними одинаков.

17.2) Изучение состояния и отладка

В разделе 17.1.2 мы сказали, что чекпоинтер сохраняет состояние на каждом суперштаге. Давайте извлечём эти чекпоинты и посмотрим на них.

Открывать сохранённые чекпоинты — это не праздное любопытство. Агенты ведут себя неправильно. Они вызывают инструменты в цикле, который никогда не заканчивается, они теряют нить более ранних шагов, они выбирают ветвь, которую вы никогда не ожидали. Чтобы выяснить почему, вам нужно знать, как выглядело состояние в тот момент, — и ответ есть в чекпоинтах.

LangGraph предоставляет вам два метода.

• graph.get_state(config) — возвращает самый недавний чекпоинт для этого диалога.
• graph.get_state_history(config) — возвращает каждый чекпоинт для этого диалога, начиная с самого нового.

Обоим нужен thread_id в config, поскольку они должны знать, чьи чекпоинты вы запрашиваете.

Чекпоинты, которые возвращают эти два метода, представлены объектами StateSnapshot.

17.2.1) StateSnapshot: что внутри чекпоинта

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

У состояния будет два вида полей. У foo нет редьюсера, поэтому оно перезаписывается; у bar он есть, поэтому оно накапливается. Это та же схема, что и с редьюсером add_messages, который мы прикрепили к messages в разделе 15.2.2, — здесь мы используем в качестве редьюсера operator.add из Python для конкатенации списков.

from operator import add
from typing_extensions import TypedDict, Annotated
 
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    foo: str                        # нет редьюсера → перезапись
    bar: Annotated[list[str], add]  # редьюсер add → накопление
 
def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}
 
def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}
 
builder = StateGraph(State)
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar": []}, config)
 
snapshot = graph.get_state(config)
print(snapshot)

Вывод:

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                             'checkpoint_id': '1f17da03-9654-65d8-8002-9e59231bb481'}},
    metadata={'source': 'loop', 'step': 2, 'parents': {}},
    created_at='2026-07-12T03:17:33.637368+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                                    'checkpoint_id': '1f17da03-9653-6ca0-8001-7c24c8ec66f2'}},
    tasks=(),
    interrupts=()
)

Восемь полей. Давайте разберём их по одному.

• values — состояние в том виде, в каком оно было на этом чекпоинте. bar равно ['a', 'b'], потому что редьюсер add накопил то, что вернули оба узла; foo равно 'b', потому что у него нет редьюсера, поэтому победила последняя запись. Это то поле, которое отвечает на вопрос «как выглядело состояние в тот момент?»
• next — кортеж имён узлов, которые должны быть запущены после этого чекпоинта. Пустой () означает, что запускать больше нечего, то есть граф завершился. ('node_b',) означает, что впереди ещё node_b.
• config — адрес этого чекпоинта. thread_id идентифицирует диалог, а checkpoint_id идентифицирует, какой момент внутри него. LangGraph присваивает checkpoint_id автоматически каждый раз, когда сохраняет чекпоинт.
• metadata — служебная информация о запуске. source сообщает вам, откуда взялся чекпоинт: "input" означает, что он был построен из входных данных, которые вы передали в invoke(), а "loop" означает, что он был создан, пока граф выполнялся. step — это номер суперштага.
• created_at — когда чекпоинт был сохранён. Удобно, когда вы сопоставляете вещи с вашими логами.
• parent_config — config чекпоинта, непосредственно предшествующего этому. Следуйте за ним — и вы сможете идти назад по запуску. Для самого первого чекпоинта он равен None.
• tasks — запись выполнения для узлов, перечисленных в next. В момент сохранения чекпоинта эти узлы ещё не запускались; как только они запускаются, их результат прикрепляется к этому чекпоинту. Узел, который успешно завершился, оставляет своё возвращаемое значение в result, а узел, который завершился сбоем, оставляет своё исключение в error.
• interrupts — место, где граф приостановился, чтобы передать управление человеку. LangGraph может остановиться посреди запуска и подождать, пока кто-то одобрит шаг или предоставит значение, и это поле записывает такие паузы.

Поля вы читаете как атрибуты. metadata — это словарь, поэтому значения из него вы извлекаете по ключу.

snapshot = graph.get_state(config)
 
print(snapshot.values)            # {'foo': 'b', 'bar': ['a', 'b']}
print(snapshot.next)              # ()
print(snapshot.metadata["step"])  # 2

Из них поле, к которому вы будете обращаться чаще всего при отладке, — это next. Если next не пустое, граф не дошёл до конца — он остановился где-то посередине. И когда мы будем возобновлять граф, завершившийся сбоем, в разделе 17.3, именно с этого поля мы начнём.

17.2.2) Прохождение по истории чекпоинтов

get_state() показывает вам только самый недавний чекпоинт. Но отладка часто означает вопрос «как мы здесь оказались?» — и для этого вам нужна вся траектория запуска. get_state_history() даёт вам её.

for snap in graph.get_state_history(config):
    print(f"step={snap.metadata['step']:>2}  "
          f"next={str(snap.next):<16}  values={snap.values}")

Вывод:

step= 2  next=()                values={'foo': 'b', 'bar': ['a', 'b']}
step= 1  next=('node_b',)       values={'foo': 'a', 'bar': ['a']}
step= 0  next=('node_a',)       values={'foo': '', 'bar': []}
step=-1  next=('__start__',)    values={'bar': []}

Самый новый чекпоинт идёт первым, поэтому работайте снизу вверх, чтобы проследить запуск по порядку.

• step -1 — сразу после того, как invoke() получил входные данные. Обратите внимание, что {"foo": "", "bar": []}, которые мы передали, не появляются в values. Помещение входных данных в состояние само по себе является этапом, и этот этап ещё не запущен. __start__ в next — это внутренний узел, который это делает.

  bar показывается как [], но это не то значение, которое мы передали. Поле с редьюсером стартует с пустым значением, чтобы записи в него накапливались. У foo нет редьюсера, поэтому у него вообще нет стартового значения — вот почему оно здесь не появляется.

• step 0 — __start__ запустился, и входные данные теперь в состоянии. foo='' и bar=[] — это значения, которые мы передали. Следующим на очереди node_a.
• step 1 — результат запуска node_a. foo теперь 'a', а bar — ['a'], следующим node_b.
• step 2 — результат запуска node_b. next пустое, поэтому граф завершён.

Непустое next на step 0 или step 1 не означает, что граф там остановился. Чекпоинт, взятый пока граф ещё выполняется, естественно имеет узел, выстроенный на очередь. Когда в разделе 17.2.1 говорилось, что «непустое next означает, что граф остановился», речь шла о чекпоинте в точке, где что-то пошло не так. В середине истории next просто показывает вам, какой путь прошёл граф.

17.3) Возобновление с места сбоя

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

17.3.1) Возобновление с помощью invoke(None, config)

Возобновление простое: передайте None там, где идут входные данные.

graph.invoke(None, config)

Это означает «нового ввода нет; продолжай с сохранённого состояния». config, разумеется, по-прежнему нуждается в thread_id, поскольку LangGraph должен знать, какой диалог продолжать.

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

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    step_1_done: bool
    step_2_done: bool
 
first_try = True   # флаг, чтобы первый запуск завершился сбоем
 
def step_1(state: State):
    print("step_1 выполняется (дорогая работа)")
    return {"step_1_done": True}
 
def step_2(state: State):
    global first_try
    if first_try:
        first_try = False
        print("step_2 завершился сбоем (таймаут API)")
        raise RuntimeError("Внешний API превысил время ожидания")
    print("step_2 выполняется")
    return {"step_2_done": True}
 
builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "job-42"}}
 
try:
    graph.invoke({"step_1_done": False, "step_2_done": False}, config)
except RuntimeError as e:
    print("Сбой:", e)

Вывод:

step_1 выполняется (дорогая работа)
step_2 завершился сбоем (таймаут API)
Сбой: Внешний API превысил время ожидания

step_1 успешно завершился, а step_2 выбросил исключение. Давайте выясним, где остановился граф, используя get_state(), который мы изучили в разделе 17.2.

snapshot = graph.get_state(config)
print("next   =", snapshot.next)
print("values =", snapshot.values)

Вывод:

next   = ('step_2',)
values = {'step_1_done': True, 'step_2_done': False}

next равно ('step_2',), что говорит нам, что граф остановился посреди выполнения step_2. А в values step_1_done равно True — результат step_1 всё ещё там, в чекпоинте.

Теперь мы возобновляем с помощью None.

result = graph.invoke(None, config)
print("Final =", result)

Вывод:

step_2 выполняется
Final = {'step_1_done': True, 'step_2_done': True}

step_1 выполняется (дорогая работа) так и не напечаталось. step_1 не запустился во второй раз. LangGraph загрузил сохранённое состояние и продолжил с step_2. Мы не заплатили за этот дорогой первый шаг дважды.

17.3.2) На что обращать внимание при возобновлении

Когда вы возобновляете, узел, который завершился сбоем, запускается снова. Если этот узел вызывает LLM или обращается к внешнему API, эти вызовы тоже происходят снова — и они могут вернуться с чем-то другим.

Вот где кроется ловушка. Если step_2 отправил письмо, а затем завершился сбоем, возобновление отправляет второе письмо. LangGraph гарантирует только то, что он не перезапустит узлы, которые успешно завершились.

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