Разработка надежных AI-агентов на Python требует решения проблемы сохранения контекста. LLM по умолчанию не хранят состояние между вызовами, что делает невозможным восстановление после сбоев сети, превышения лимитов API или перезапуска сервера. LangGraph решает эту задачу через механизмы персистентности и контрольных точек. В этой статье мы разберем, как настроить checkpointer на SQLite и PostgreSQL, и рассмотрим практические аспекты их использования в российских проектах.
Проблема stateless LLM и решение LangGraph
В классических цепочках LangChain память реализуется через `ChatMessageHistory` — массив сообщений, передаваемый при каждом вызове модели. Это приводит к двум проблемам: превышение контекстного окна и отсутствие гранулярности шагов. Если агент ошибся на 15-м шаге, откатить выполнение к 12-му без потери всего сеанса невозможно.
LangGraph рассматривает приложение как граф состояний (StateGraph). Каждый узел принимает текущее состояние, выполняет логику и возвращает обновления. Checkpointer перехватывает эти обновления после завершения работы каждого узла и сохраняет их в хранилище. Это превращает граф в отказоустойчивую машину состояний, способную возобновлять работу с любой точки.
Настройка MemorySaver для локальной разработки
Для тестирования в изолированной среде используйте `MemorySaver` из пакета `langgraph-checkpoint-memory`. Он хранит историю в оперативной памяти процесса, что удобно для отладки, но данные исчезают при перезагрузке.
Пример инициализации графа с поддержкой сохранения состояния:
python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AnyMessage
class AgentState(TypedDict):
messages: list[AnyMessage]
def call_model(state: AgentState):
return {«messages»: [f»Обработано сообщений: {len(state[‘messages’])}»]}
workflow = StateGraph(AgentState)
workflow.add_node(«agent», call_model)
workflow.add_edge(START, «agent»)
workflow.add_edge(«agent», END)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
При вызове графа передайте `thread_id` в конфигурации:
python
config = {«configurable»: {«thread_id»: «session_user_001»}}
response = app.invoke({«messages»: [«Привет, агент!»]}, config)
Повторный вызов с тем же `thread_id` извлечет предыдущее состояние и продолжит работу с накопленным контекстом.
Переход на SQLite для продакшен-прототипов
`MemorySaver` не подходит для продакшена из-за потери данных при перезапуске. LangGraph поддерживает бэкенды на базе реляционных баз данных через пакеты `langgraph-checkpoint-sqlite` и `langgraph-checkpoint-postgres`.
Подключение SQLite:
python
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
conn = sqlite3.connect(«checkpoints.db», check_same_thread=False)
memory_db = SqliteSaver(conn)
app_persistent = workflow.compile(checkpointer=memory_db)
Преимущества SQLite checkpointer:
— Отказоустойчивость: состояние сохраняется на диске.
— Инспекция сессий: можно запросить таблицы напрямую.
— Низкие накладные расходы для однопоточных приложений.
Недостаток: SQLite не поддерживает конкурентную запись из нескольких процессов. Для веб-серверов на FastAPI с несколькими воркерами используйте PostgreSQL.
Настройка PostgreSQL checkpointer для высоконагруженных систем
PostgreSQL обеспечивает масштабирование и конкурентный доступ. Установите пакет:
bash
pip install langgraph-checkpoint-postgres
Пример подключения:
python
from langgraph.checkpoint.postgres import PostgresSaver
import asyncpg
connection_string = «postgresql://user:password@localhost:5432/langgraph»
conn = await asyncpg.connect(connection_string)
checkpointer = PostgresSaver(conn)
app = workflow.compile(checkpointer=checkpointer)
Для синхронного использования применяйте `psycopg`:
python
import psycopg
from langgraph.checkpoint.postgres import PostgresSaver
conn = psycopg.connect(connection_string)
checkpointer = PostgresSaver(conn)
Таблица `checkpoints` создается автоматически. Вы можете настроить пул соединений через `psycopg_pool` для асинхронных веб-серверов.
Сравнение checkpointer: SQLite vs PostgreSQL vs MemorySaver
Ниже приведена таблица сравнения основных характеристик:
| Параметр | MemorySaver | SQLite | PostgreSQL |
|---|---|---|---|
| Хранение данных | Оперативная память | Файл на диске | СУБД |
| Сохранение после перезапуска | Нет | Да | Да |
| Конкурентный доступ | Один поток | Один процесс | Много процессов |
| Производительность | Высокая | Средняя | Средняя |
| Простота настройки | Минимальная | Простая | Требует настройки СУБД |
| Подходит для | Разработка, тесты | Прототипы, малые проекты | Продакшен, микросервисы |
Выбор зависит от требований к отказоустойчивости и масштабированию. Для российских проектов с ограниченными ресурсами SQLite — оптимальный стартовый вариант.
Управление размером контекста и очистка старых checkpoints
Бесконтрольный рост таблиц контрольных точек — частая проблема. Со временем база данных раздувается, а запросы на выборку истории замедляются.
Архитектурные подходы для предотвращения:
1. Ограничение глубины истории (TTL): настройте периодическую очистку записей для сессий, неактивных более 30 дней. Например, через SQL-запрос:
sql
DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days';
2. Редукция состояния (State Reducers): используйте функции агрегации, например `add_messages` из LangChain, для суммаризации старых сообщений.
3. Изоляция баз данных: выделите отдельный инстанс PostgreSQL под персистентность агентов, чтобы сбои не затрагивали основную базу приложения.
Путешествие во времени и откат состояния
LangGraph поддерживает «путешествие во времени» через `get_state_history`. Это позволяет получить список всех контрольных точек и перезапустить выполнение из любой точки прошлого.
Пример получения истории:
python
config = {«configurable»: {«thread_id»: «session_user_001»}}
history = list(app_persistent.get_state_history(config))
for state in history:
print(f»Checkpoint ID: {state.config[‘configurable’][‘checkpoint_id’]}»)
print(f»Next nodes: {state.next}»)
print(f»State values: {state.values}\n»)
Если агент зациклился или принял неверное решение, возьмите `checkpoint_id` успешного шага до сбоя, передайте его в конфигурации и продолжите выполнение с измененными входными данными. Это критически важно для реализации безопасных циклов самокоррекции в кодинг-агентах.
Практические рекомендации для российских разработчиков
Перед внедрением персистентности в рабочий контур проверьте:
— Убедитесь, что все объекты в состоянии графа корректно сериализуются. Избегайте передачи открытых сокетов, файловых дескрипторов или сложных лямбда-функций.
— Для многопоточных веб-серверов на FastAPI или Celery используйте пулы соединений с БД. Для PostgreSQL применяйте `psycopg_pool`, для SQLite — `check_same_thread=False`.
— Протестируйте поведение агента принудительным завершением процесса (SIGKILL) посреди выполнения шага. Транзакционность бэкенда должна защищать данные от повреждения.
— Настройте мониторинг размера таблиц `checkpoints` и `checkpoint_writes`. При превышении лимита в 1 ГБ для SQLite или 10 ГБ для PostgreSQL — запускайте очистку.
Эти подходы помогут создать надежных AI-агентов с персистентной памятью, которые не теряют контекст даже при сбоях инфраструктуры.
Источники:
— github.com/langchain-ai/langgraph
— langchain-ai.github.io/langgraph/
— langchain-ai.github.io/langgraph/how-tos/persistence/
— python.langchain.com/docs/concepts/langgraph/