Память и сохранение состояния

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

#Две памяти и зачем они обе

Контрольная точка — это снимок состояния графа на конкретном шаге. Она отвечает на вопрос «что происходило в этом диалоге»: история сообщений, промежуточные поля, место остановки. Живёт снимок внутри одного потока, который идентифицируется thread_id.

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

Разница проявляется на простом примере. Пользователь в понедельник сказал «я предпочитаю краткие ответы». Если это осело в контрольной точке, во вторник в новом диалоге агент об этом не вспомнит. Если в хранилище, вспомнит.

#Контрольные точки

Чекпоинтер передаётся при компиляции, а поток задаётся конфигурацией вызова:

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-42-chat-1"}}
result = graph.invoke({"messages": [{"role": "user", "content": "Привет"}]}, config=config)

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

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

#Выбор чекпоинтера

InMemorySaver держит всё в памяти процесса. Подходит для тестов и примеров, теряет данные при перезапуске и не работает, когда воркеров несколько.

SqliteSaver пишет в файл. Годится для локальной разработки и однопроцессных инструментов.

PostgresSaver — это рабочий вариант для продакшена. Перед первым использованием нужно создать схему:

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db")
checkpointer.setup()          # создаёт таблицы, вызывается один раз
graph = builder.compile(checkpointer=checkpointer)

В асинхронном приложении берут AsyncPostgresSaver: синхронный вариант в асинхронном обработчике заблокирует цикл событий.

#Чтение и правка состояния

Скомпилированный граф умеет показывать текущее состояние потока:

config = {"configurable": {"thread_id": "user-42-chat-1"}}
snapshot = graph.get_state(config)

print(snapshot.values)   # само состояние
print(snapshot.next)     # какие узлы выполнятся следующими

Поле next полезно при разборе остановок: пустой кортеж означает, что граф завершился, непустой что он ждёт продолжения.

Полную историю снимков возвращает get_state_history в порядке от свежих к старым:

for snapshot in graph.get_state_history(config):
    print(snapshot.config["configurable"]["checkpoint_id"], snapshot.next)

Состояние можно изменить снаружи. update_state создаёт новую контрольную точку, причём значения проходят через редукторы: список с add_messages пополнится, а не перезапишется.

graph.update_state(config, {"escalated": True})

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

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

Каждая контрольная точка имеет идентификатор. Если передать его в конфигурации, граф продолжит выполнение с того момента: узлы до точки пропускаются, идущие после выполняются заново.

config = {
    "configurable": {
        "thread_id": "user-42-chat-1",
        "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c",
    }
}
graph.invoke(None, config=config)

Приём применяют для отладки («что будет, если на этом шаге ответ модели был бы другим») и для восстановления после неудачной ветки. Передача None на вход означает «ничего не добавляй, просто продолжай».

#Режимы сохранения

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

Режим sync записывает точку до перехода к следующему шагу. Максимальная надёжность, максимальная задержка.

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

Режим exit сохраняет лишь при завершении или прерывании графа. Быстро, но при сбое посреди работы потеряется весь прогресс.

graph.stream({"messages": [...]}, config=config, durability="async")

#Долговременная память: хранилище

Хранилище устроено как пространство имён, ключ и значение. Пространство имён — это кортеж строк, по которому данные разделяются между пользователями и типами записей:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

Доступ из узла идёт через рантайм:

from langgraph.runtime import Runtime

def remember_preference(state: State, runtime: Runtime):
    namespace = (state["user_id"], "preferences")
    runtime.store.put(namespace, "reply_style", {"value": "кратко"})
    item = runtime.store.get(namespace, "reply_style")
    return {"style": item.value["value"]}

Асинхронные версии методов называются aput, aget, asearch, adelete. В асинхронном графе используют именно их.

#Поиск по смыслу

Хранилище умеет искать не только по ключу. Если при создании передать конфигурацию индексации, записи будут векторизоваться, и поиск станет семантическим:

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "dims": 1536,
        "fields": ["food_preference", "$"],
    }
)

memories = store.search((user_id, "memories"), query="что человек любит есть", limit=3)

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

#Что где хранить

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

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

#Эксплуатация

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

checkpointer.delete_thread(thread_id="user-42-chat-1")

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

#Частые ошибки

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

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

Синхронный PostgresSaver в асинхронном приложении. Блокировка цикла событий проявляется как загадочная просадка производительности под нагрузкой.

Ожидание, что хранилище само что-то запомнит. Запись в него — это явное действие узла или инструмента, автоматически туда ничего не попадает.

Устранение неисправностей