Проблема: граф без памяти

Вспомним, как работает состояние (урок про State): оно живёт в пределах одного прогона. Подали вход → состояние эволюционировало по шагам → invoke вернул финал и всё забыл. Следующий invoke не знает ничего о предыдущем.

Без checkpointer агент не помнит диалог
python 
graph = builder.compile()      # без checkpointer

graph.invoke({"messages": [{"role": "user", "content": "Меня зовут Иван"}]})
# → "Приятно познакомиться, Иван!"

graph.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]})
# → "Не знаю, вы не представились"   ← память обнулилась между вызовами

Для одноразовой задачи это нормально. Но агенту-собеседнику нужна непрерывность: помнить контекст диалога, продолжить начатое после перезапуска процесса, поставить выполнение на паузу для подтверждения человеком. Всё это даёт persistence — сохранение состояния через checkpointer.

Как устроен checkpointing

Идея простая: после каждого супершага (одного «такта» графа) LangGraph сохраняет полный снимок состояния — checkpoint. Снимки складываются в хранилище через объект checkpointer. Цепочка снимков одной сессии называется тредом (thread) и адресуется идентификатором thread_id.

100%
колёсико — масштаб  ·  зажать и тянуть — перемещение
Thread · thread_id: "user-42" checkpoint 0 ["Иван"] checkpoint 1 [.., "привет"] checkpoint 2 [.., "как зовут?"] ckpt 3 invoke #2 Checkpointer MemorySaver (RAM) · SqliteSaver (диск) · PostgresSaver каждый супершаг сохраняется; invoke #2 с тем же thread_id продолжает с последнего снимка

Когда ты снова запускаешь граф с тем же thread_id, LangGraph загружает последний checkpoint этого треда и продолжает с него — состояние не теряется. Разные thread_id — это независимые сессии (разные пользователи, разные диалоги), они не видят состояние друг друга.

ℹ️ Checkpoint — это снимок ВСЕГО состояния

Сохраняется не только messages, а вся схема состояния целиком: счётчики, флаги, промежуточные данные. Поэтому возобновление честное — граф продолжает ровно с того места и с теми данными, что были на момент снимка.

Включаем память: checkpointer + thread_id

Чтобы граф стал «помнящим», нужны две вещи: передать checkpointer в compile() и указывать thread_id в config при каждом запуске. Без thread_id граф с checkpointer'ом работать не будет — он не поймёт, в какую сессию сохранять.

MemorySaver + thread_id — память между вызовами
python 
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)   # ← подключаем память

# thread_id выбирает сессию
config = {"configurable": {"thread_id": "user-42"}}

graph.invoke({"messages": [{"role": "user", "content": "Меня зовут Иван"}]}, config)
# → "Приятно познакомиться, Иван!"

graph.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]}, config)
# → "Вас зовут Иван"   ✅ память сохранилась — тот же thread_id

Обрати внимание на две детали. Во-первых, во втором вызове мы передали только новое сообщение — старую историю подтянул checkpointer, а reducer add_messages дописал новое к загруженному. Во-вторых, ключ thread_id лежит внутри configurable — это стандартное место для рантайм-параметров в LangChain.

⚠️ Нет thread_id — будет ошибка

Если граф скомпилирован с checkpointer'ом, но ты вызвал invoke без thread_id в config — получишь ошибку. Checkpointer обязан знать, в какой тред писать. Запомни связку: есть checkpointer → обязателен thread_id.

MemorySaver: для разработки

MemorySaver хранит чекпойнты в оперативной памяти процесса (обычный словарь под капотом). Он мгновенный и не требует настройки — идеален для прототипов, тестов и примеров.

Но у него есть фундаментальное ограничение: память живёт ровно столько, сколько живёт процесс. Перезапустил приложение — все треды исчезли. Для продакшена, где диалоги должны переживать рестарты и делиться между воркерами, нужен персистентный saver.

❌ MemorySaver не для прода

Никогда не держи MemorySaver в продакшене: при рестарте сервиса вся история пользователей пропадёт, а при нескольких воркерах каждый будет иметь свою копию памяти. Это инструмент разработки. Для боевой системы — SqliteSaver (один сервер) или PostgresSaver (масштаб).

SqliteSaver: персистентность на диск

SqliteSaver хранит чекпойнты в файле SQLite. Интерфейс тот же — меняется только объект, переданный в compile(). Зато теперь история переживает перезапуск: остановил процесс, запустил снова с тем же thread_id — диалог на месте.

SqliteSaver — память переживает рестарт
python 
# pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver

# from_conn_string — контекстный менеджер; путь к файлу БД
with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "user-42"}}
    graph.invoke({"messages": [{"role": "user", "content": "Меня зовут Иван"}]}, config)

# ... перезапустили процесс, открыли тот же файл ...
with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "user-42"}}
    graph.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]}, config)
    # → "Вас зовут Иван"   ✅ данные прочитаны из файла
ℹ️ Async и Postgres

Для async-приложений есть AsyncSqliteSaver (пакет тот же, импорт из langgraph.checkpoint.sqlite.aio) — его методы ainvoke/astream-совместимы. Для масштаба и многих воркеров — PostgresSaver из langgraph-checkpoint-postgres: тот же API, но хранилище — PostgreSQL. Менять придётся одну строку — создание saver'а.

Checkpointer Хранилище Когда использовать
MemorySaver RAM процесса прототипы, тесты, примеры
SqliteSaver файл SQLite один сервер, локальные приложения, MVP
PostgresSaver PostgreSQL прод, много воркеров, высокая нагрузка

Просмотр состояния и путешествие во времени

Раз все снимки сохранены, их можно читать. Скомпилированный граф с checkpointer'ом умеет показать текущее состояние треда и всю его историю — это бесценно для отладки и лежит в основе human-in-the-loop.

get_state и get_state_history
python 
config = {"configurable": {"thread_id": "user-42"}}

# Текущее состояние треда
snapshot = graph.get_state(config)
print(snapshot.values)      # значения состояния (messages, счётчики, ...)
print(snapshot.next)        # какие узлы выполнятся следующими (() если END)

# Вся история чекпойнтов — от свежего к старому
for snap in graph.get_state_history(config):
    print(snap.config["configurable"]["checkpoint_id"], "→", snap.values)

Каждый снимок (StateSnapshot) хранит не только значения, но и свой checkpoint_id. Указав конкретный id в config, можно вернуться к прошлому состоянию и запустить граф оттуда — это «time travel». А update_state позволяет вручную поменять состояние треда (например, поправить ответ перед продолжением).

Time travel и ручная правка состояния
python 
# Откатиться к конкретному прошлому чекпойнту и продолжить с него
past = {"configurable": {"thread_id": "user-42",
                         "checkpoint_id": "1ef..."}}
graph.invoke(None, past)        # None = «не добавляй вход, продолжи с снимка»

# Вручную дописать в состояние треда (применяются reducer'ы)
graph.update_state(config, {"messages": [{"role": "user", "content": "правка"}]})
✅ Фундамент для human-in-the-loop

Возможность остановиться, посмотреть состояние, поправить его и продолжить с того же места — это ровно то, на чём строится human-in-the-loop (следующий раздел). Прерывание графа interrupt_before + сохранённый checkpoint = пауза на подтверждение человеком. Checkpointing — обязательное условие для этих сценариев.

Полный пример: чат-бот с памятью

Минимальный диалоговый агент, который помнит контекст
python 
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

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

llm = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# Память подключается одной строкой
graph = builder.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "dialog-1"}}

# Диалог в несколько реплик — на каждом шаге шлём только новое сообщение
for text in ["Привет! Меня зовут Иван.", "Посоветуй книгу.", "А как меня зовут?"]:
    answer = graph.invoke({"messages": [{"role": "user", "content": text}]}, config)
    print("User:", text)
    print("Bot :", answer["messages"][-1].content, "\n")

# Последний ответ корректно назовёт имя — история жива внутри треда dialog-1

Типичные ошибки

❌ Ошибка 1: checkpointer есть, thread_id нет

Скомпилировал с checkpointer'ом, а в invoke не передал {"configurable": {"thread_id": ...}} — будет ошибка. Память всегда привязана к треду.

❌ Ошибка 2: шлёшь всю историю заново

С памятью на каждом шаге передавай только новое сообщение, а не весь диалог. История подтянется из checkpoint, а add_messages допишет новое. Если послать всю историю снова — сообщения задвоятся.

❌ Ошибка 3: MemorySaver в продакшене

После рестарта вся память исчезнет, а при нескольких воркерах у каждого будет своя. Для прода — персистентный saver (SQLite/Postgres).

❌ Ошибка 4: один thread_id на всех пользователей

thread_id — это идентификатор сессии. Захардкодишь общий — и все пользователи будут видеть чужие диалоги. Генерируй уникальный thread_id на пользователя/беседу.

Шпаргалка

Checkpointing — всё в одном месте
python 
# 1. Выбираем checkpointer
from langgraph.checkpoint.memory import MemorySaver        # RAM — разработка
from langgraph.checkpoint.sqlite import SqliteSaver        # файл — один сервер
# from langgraph.checkpoint.postgres import PostgresSaver  # БД — прод

# 2. Подключаем в compile()
graph = builder.compile(checkpointer=MemorySaver())

# 3. Указываем thread_id в config (ОБЯЗАТЕЛЬНО при checkpointer)
config = {"configurable": {"thread_id": "user-42"}}

# 4. Запускаем — шлём только НОВОЕ сообщение, история подтянется сама
graph.invoke({"messages": [msg]}, config)

# 5. Инспекция и time travel
snap = graph.get_state(config)            # текущее состояние (.values, .next)
for s in graph.get_state_history(config): ...   # вся история снимков
graph.invoke(None, {"configurable": {"thread_id": "user-42",
                                     "checkpoint_id": "1ef..."}})  # откат
graph.update_state(config, {"messages": [msg]})  # ручная правка

# Правила:
#  • есть checkpointer → обязателен thread_id
#  • разные thread_id = независимые сессии
#  • MemorySaver — только dev; прод → Sqlite/Postgres
#  • с памятью шли только новое сообщение, не весь диалог

Практическое задание

Научи граф помнить:

Задание: диалоговый агент с персистентной памятью

  1. Возьми чат-граф из примера выше и подключи MemorySaver. Проведи диалог из трёх реплик в одном thread_id и убедись, что агент помнит имя из первой реплики.
  2. Запусти второй диалог с другим thread_id и проверь, что он «чистый» — не видит первую сессию.
  3. Замени MemorySaver на SqliteSaver с файлом. Запусти скрипт, заверши процесс, запусти снова с тем же thread_id — память должна сохраниться.
  4. Выведи graph.get_state(config).values и .next после диалога — посмотри, что внутри снимка.
  5. Пройди по get_state_history(config) и распечатай, как менялось число сообщений от снимка к снимку.
  6. Со звёздочкой: через update_state вручную добавь системное сообщение в середину треда и проверь, как это повлияет на следующий ответ.

Что дальше

⚙️

Entry point, END, компиляция

Где подключается checkpointer

 🧵

Thread management

Несколько одновременных сессий

 ⏸️

Human-in-the-loop

Пауза и подтверждение на сохранённом состоянии

 🧩

Все уроки модуля 03

LangGraph целиком