Разработка надежных LLM-агентов упирается в одну фундаментальную проблему: сохранение состояния между шагами выполнения. Когда граф вычислений делает десятки итераций, обращается к инструментам и ждет пользовательского ввода, стандартная оперативная память процесса становится узким горлышком. Сбой во время выполнения или перезапуск сервиса стирают всю предысторию сессии, заставляя пользователя начинать диалог с нуля.
Фреймворк LangGraph решает эту задачу с помощью встроенного механизма персистентности (Persistence) и системы checkpoints. В этой статье мы сосредоточимся на практической настройке: как подключить SQLiteSaver, управлять историей сообщений без раздувания контекстного окна и избежать типичных ошибок при миграции на PostgreSQL.
Архитектура персистентности в LangGraph: как работают checkpoints
В основе персистентности LangGraph лежит концепция неизменяемых состояний (immutable states) и графов переходов, управляемых через `StateGraph`. Каждый раз, когда узел (node) завершает работу и возвращает обновленный словарь состояния, компилятор графа создает точку сохранения — checkpoint.
Checkpoint фиксирует три ключевых элемента:
1. Config (конфигурация потока, включая `thread_id` для разделения сессий).
2. Values (актуальные данные состояния на текущем шаге: история сообщений, промежуточные артефакты, переменные).
3. Next (указатель на следующие узлы, которые должны выполниться в графе).
Без подключения внешнего `checkpointer` эти данные хранятся исключительно в оперативной памяти и исчезают сразу после завершения скрипта. Для production-окружения требуется подключить персистентный бэкенд.
Подключение SQLiteSaver для локального хранения состояний
Для локальных разработок, тестирования и автономных демо-стендов оптимально подходит встроенный `SQLiteSaver`. Он позволяет сохранять историю всех сессий в локальном файле базы данных, обеспечивая мерж-контроль и возобновление прерванных сессий.
Ниже приведен пример инициализации графа с привязкой к файловой базе данных SQLite:
python
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
Открываем соединение с базой данных в режиме автокоммита
db_path = «agent_states.db»
conn = sqlite3.connect(db_path, check_same_thread=False)
Создаем менеджер сохранений
memory = SqliteSaver(conn)
Компилируем граф с передачей checkpointer
app = workflow.compile(checkpointer=memory)
Теперь при каждом запуске графа необходимо передавать уникальный идентификатор потока (`thread_id`) в конфигурации вызова (`config`). Это позволяет вести параллельные диалоги с разными пользователями в рамках одного процесса:
python
config = {«configurable»: {«thread_id»: «session_user_404»}}
Первый запуск
for event in app.stream({«messages»: [«Привет, составь план миграции базы данных»]}, config):
print(event)
Если процесс завершится аварийно, повторный вызов `app.stream` с тем же `thread_id` мгновенно восстановит последнюю валидную точку из `agent_states.db` и продолжит выполнение с шага, на котором произошел срыв.
Управление историей сообщений и ограничение контекста
Сохранение каждого шага в базе данных решает проблему надежности, но порождает другую — экспоненциальный рост истории сообщений (`messages`). Если агент выполняет многошаговую задачу с привлечением инструментов (Tool Calling), контекстное окно LLM быстро забивается промежуточными техническими логами.
Для предотвращения деградации качества генерации и роста затрат на API применяются две практики:
— Обрезка старых сообщений (Message Trimming) на уровне узла подготовки запроса.
— Суммаризация (Summarization) длинных диалогов перед записью в основное состояние.
Пример функции обрезки истории перед передачей в модель:
python
from langchain_core.messages import RemoveMessage
def trim_messages(state: State):
messages = state[«messages»]
# Оставляем системный промпт и последние 10 сообщений
if len(messages) > 12:
old_messages = messages[1:-10]
# Пометка старых сообщений на удаление из текущего состояния
return {«messages»: [RemoveMessage(id=m.id) for m in old_messages]}
return {}
Этот подход гарантирует, что база данных не раздувается гигабайтами устаревших технических вызовов, а агент сохраняет фокус на текущей задаче.
Сравнение бэкендов: SQLiteSaver vs PostgresSaver
Выбор бэкенда зависит от масштаба системы. В таблице ниже приведены ключевые различия.
| Бэкенд | Тип хранения | Конкурентный доступ | Асинхронная поддержка | Рекомендуемый сценарий |
|---|---|---|---|---|
| SQLiteSaver | Файловая БД | Ограничен (один процесс) | Нет | Локальная разработка, тестирование, демо-стенды |
| AsyncPostgresSaver | PostgreSQL | Полная поддержка (пул соединений) | Да (asyncio) | Production: сотни параллельных пользователей |
| MemorySaver | Оперативная память | Ограничен | Нет | Быстрое прототипирование, отладка |
Для промышленных систем, обслуживающих сотни параллельных пользователей, использование SQLite накладывает ограничения по конкурентной записи. Официальный пакет `langgraph-checkpoint-postgres` предоставляет масштабируемую замену с поддержкой асинхронных пулов соединений (`AsyncPostgresSaver`).
Миграция: от SQLite к PostgresSaver
Миграция требует минимальных изменений в коде инициализации:
python
import asyncio
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool
DB_URI = «postgresql://user:password@localhost:5432/agents_db»
async def main():
async with AsyncConnectionPool(
conninfo=DB_URI,
max_size=20,
kwargs={«autocommit»: True}
) as pool:
checkpointer = AsyncPostgresSaver(pool)
await checkpointer.setup()
# Компиляция асинхронного графа
app = workflow.compile(checkpointer=checkpointer)
config = {«configurable»: {«thread_id»: «prod_session_101»}}
result = await app.ainvoke({«messages»: [«Запуск асинхронного агента»]}, config)
print(result)
if name == «main»:
asyncio.run(main())
Пул соединений PostgreSQL обеспечивает изоляцию транзакций, предотвращает коллизии при одновременном обновлении состояния одного `thread_id` несколькими рабочими воркерами и упрощает бэкап данных.
Возможные ограничения и подводные камни
Настройка персистентности в LangGraph требует учета специфических архитектурных нюансов:
Сериализуемость данных в состоянии. Все объекты, сохраняемые в `State`, должны поддерживать сериализацию (JSON или pickle в зависимости от выбранного бэкенда). Передача живых сетевых соединений, открытых файловых дескрипторов или сложных несериализуемых экземпляров классов в состояние вызовет фатальную ошибку сохранения checkpoint.
2. Миграция схемы при изменении структуры State. Если вы добавили новое поле в `TypedDict` состояния, старые checkpoints в базе данных не будут содержать это поле, что может привести к ошибкам `KeyError` в существующих сессиях. Требуется версионирование схемы или написание миграционных скриптов для базы данных.
3. Размер базы данных. При интенсивном использовании агентов размер таблиц checkpoints растет лавинообразно. Рекомендуется настроить политику очистки (TTL) для завершенных или заброшенных сессий.
Практические рекомендации по проверке
Перед внедрением персистентных checkpoints в production выполните следующие шаги:
— Проверьте поведение агента при искусственном прерывании процесса (`Ctrl+C` или убийство системного процесса) во время выполнения вызова инструмента.
— Замерьте время восстановления сессии из базы данных при длине истории более 50 сообщений.
— Убедитесь, что все пользовательские объекты внутри `State` корректно сериализуются без потери данных.
Источники