Чекпоинтер LangGraph сохраняет состояние графа между шагами выполнения. Благодаря этому AI-агент может продолжить работу после остановки, хранить отдельный контекст для каждого диалога и поддерживать сценарии с подтверждением действий человеком.
Выбор хранилища здесь влияет не только на скорость. SQLite удобен для локального прототипа, но становится ограничением при запуске нескольких процессов. PostgreSQL требует отдельной инфраструктуры, зато поддерживает конкурентные подключения, резервное копирование и эксплуатацию в распределённой среде.
Ниже — рабочая схема перехода от локальной разработки к продакшену без ручного копирования внутренних таблиц и зависимости от неподтверждённых параметров API.
Что именно сохраняет чекпоинтер LangGraph
LangGraph выполняет граф последовательными этапами. После завершения этапа подключённый checkpointer записывает снимок состояния, связанный с конкретным потоком выполнения.
В сохранённые данные могут входить:
- значения каналов и полей состояния;
- следующий узел или набор узлов;
- служебная конфигурация графа;
- метаданные текущего запуска;
- сведения о задачах, ожидающих выполнения;
- промежуточные записи завершённых узлов.
Доступ к истории строится вокруг конфигурации с `thread_id`:
python
config = {
«configurable»: {
«thread_id»: «user-42-dialog-7»
}
}
Все вызовы с одним `thread_id` относятся к одной ветке состояния. Новый идентификатор создаёт независимую ветку.
Важно понимать границу восстановления: LangGraph возвращается к сохранённому этапу графа, а не к произвольной строке внутри Python-функции. Если узел отправил запрос во внешнюю систему, а затем завершился с ошибкой, повторный запуск может ещё раз выполнить этот запрос. Поэтому операции оплаты, отправки сообщений и изменения записей должны быть идемпотентными.
SQLite и PostgreSQL: критерии выбора
Для локального агента и серверного приложения нужны разные режимы хранения.
| Сценарий | Рекомендуемое хранилище | Причина | Основное ограничение |
|---|---|---|---|
| Локальная разработка | SQLite | Нет отдельного сервера, база хранится в файле | Конкурентные записи могут блокироваться |
| Автотесты и демонстрация | SQLite или память | Простое создание изолированного окружения | Не подходит для общего состояния нескольких процессов |
| Один сервер, несколько рабочих процессов | PostgreSQL | Общая база и конкурентные подключения | Нужны пул соединений и мониторинг |
| Распределённый продакшен | PostgreSQL | Резервное копирование, репликация, централизованное хранение | Возникают сетевые задержки и расходы на эксплуатацию |
| Диалоги с критичными данными | PostgreSQL | Управляемый доступ и штатные процедуры восстановления | Требуется отдельная политика удаления данных |
Не стоит выбирать хранилище по искусственному тесту одной операции. Итоговая задержка зависит от размера состояния, числа параллельных запусков, расположения базы и частоты сохранения. Для своего приложения полезнее измерять медиану и p95 при реальном количестве рабочих процессов.
Установка актуальных пакетов
Интеграции с базами данных поставляются отдельными пакетами. Для SQLite установите:
bash
pip install -U langgraph langgraph-checkpoint-sqlite
Для PostgreSQL:
bash
pip install -U langgraph langgraph-checkpoint-postgres «psycopg[binary,pool]»
В рабочем проекте версии следует зафиксировать в `requirements.txt`, `pyproject.toml` или lock-файле. Формат служебных таблиц и API интеграций могут меняться, поэтому обновление LangGraph и пакета checkpointer лучше проводить одновременно и сначала проверять на копии базы.
Пример фиксации совместимого набора:
text
langgraph==<проверенная-версия>
langgraph-checkpoint-sqlite==<проверенная-версия>
langgraph-checkpoint-postgres==<проверенная-версия>
psycopg[binary,pool]==<проверенная-версия>
Конкретные номера нужно брать из протестированного окружения проекта, а не переносить вслепую из примера.
Базовый граф с сохраняемым состоянием
Сначала создадим небольшой граф, который изменяет счётчик и добавляет запись в историю. Один и тот же граф затем можно скомпилировать с SQLite или PostgreSQL.
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class AgentState(TypedDict):
request: str
attempts: int
history: list[str]
def process_request(state: AgentState) -> dict:
current_attempt = state[«attempts»] + 1
return {
«attempts»: current_attempt,
«history»: state[«history»] + [
f»Попытка {current_attempt}: {state[‘request’]}»
],
}
builder = StateGraph(AgentState)
builder.add_node(«process_request», process_request)
builder.add_edge(START, «process_request»)
builder.add_edge(«process_request», END)
Функция узла возвращает только обновления состояния. В более сложном агенте такими полями могут быть история сообщений, промежуточный результат поиска, статус согласования или идентификатор внешней операции.
Не следует помещать в состояние всё содержимое файлов, крупные изображения или ответы API без ограничений. Чем больше снимок, тем выше нагрузка на базу и дольше восстановление. Крупные объекты лучше сохранять в объектном хранилище, а в граф передавать ссылку и необходимые метаданные.
Настройка SQLite для локальной разработки
`SqliteSaver` создаёт файловую базу и позволяет повторно открыть состояние после перезапуска процесса.
python
from langgraph.checkpoint.sqlite import SqliteSaver
DB_PATH = «agent_checkpoints.sqlite»
with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
app = builder.compile(checkpointer=checkpointer)
config = {
«configurable»: {
«thread_id»: «demo-thread-001»
}
}
initial_state = {
«request»: «Проверить документ»,
«attempts»: 0,
«history»: [],
}
result = app.invoke(initial_state, config=config)
print(result)
snapshot = app.get_state(config)
print(snapshot.values)
Файл базы нельзя считать полноценной резервной копией сам по себе. Если он находится только на локальном диске контейнера, пересоздание контейнера уничтожит сохранённое состояние. Для Docker требуется постоянный том, а для Kubernetes — соответствующий persistent volume.
SQLite разумно использовать, когда одновременно выполняются следующие условия:
- приложение работает в одном процессе или имеет низкую конкуренцию записи;
- потеря локальной базы не приводит к критическому инциденту;
- не требуется общий доступ с нескольких серверов;
- разработчику важен быстрый старт без отдельной СУБД.
Режим WAL иногда уменьшает конфликт чтения и записи, но не превращает SQLite в распределённое хранилище. Его поведение необходимо отдельно проверять с используемым драйвером, файловой системой и моделью запуска приложения.
Настройка PostgreSQL для продакшена
Строку подключения нельзя размещать в исходном коде. Передайте её через переменную окружения или менеджер секретов:
bash
export DATABASE_URL=»postgresql://app_user:password@db-host:5432/agents?sslmode=require»
Подключение и создание служебной схемы:
python
import os
from langgraph.checkpoint.postgres import PostgresSaver
database_url = os.environ[«DATABASE_URL»]
with PostgresSaver.from_conn_string(database_url) as checkpointer:
checkpointer.setup()
app = builder.compile(checkpointer=checkpointer)
config = {
«configurable»: {
«thread_id»: «customer-8-session-194»
}
}
result = app.invoke(
{
«request»: «Подготовить сводку»,
«attempts»: 0,
«history»: [],
},
config=config,
)
print(result)
Метод `setup()` создаёт или обновляет необходимые объекты хранения. Его следует выполнять контролируемо: при первичной настройке, во время развёртывания или в отдельной миграционной задаче. Необязательно запускать изменение схемы при каждом пользовательском запросе.
В продакшене также нужны:
- пул соединений с обоснованными минимальным и максимальным размерами;
- TLS для подключения к удалённой базе;
- отдельная роль PostgreSQL с минимальными правами;
- ограничения времени ожидания соединения и выполнения запроса;
- резервное копирование с проверкой восстановления;
- наблюдение за числом подключений, размером таблиц и задержками;
- политика хранения и удаления старых веток.
Интерфейсы настройки пула зависят от версии пакета. Перед внедрением нужно сверить пример с документацией именно установленной версии `langgraph-checkpoint-postgres`.
Как восстановить граф после ошибки
Предположим, узел завершился с исключением. Последний успешно записанный снимок остаётся связанным с `thread_id`. После устранения причины можно повторно вызвать граф с той же конфигурацией.
python
config = {
«configurable»: {
«thread_id»: «job-2025-001»
}
}
try:
app.invoke(
{
«request»: «Обработать пакет документов»,
«attempts»: 0,
«history»: [],
},
config=config,
)
except Exception as error:
print(f»Выполнение остановлено: {error}»)
snapshot = app.get_state(config)
print(«Сохранённые значения:», snapshot.values)
print(«Следующие задачи:», snapshot.next)
Если граф находится в состоянии, из которого допустимо продолжение, используется тот же `thread_id`. В зависимости от структуры графа и типа остановки продолжение может выполняться вызовом с `None`:
python
result = app.invoke(None, config=config)
print(result)
Перед таким запуском проверьте `snapshot.next` и содержимое `snapshot.values`. Автоматическое повторение без анализа особенно опасно для узлов с внешними эффектами.
Для защиты от дублей используйте ключ идемпотентности:
python
def send_to_external_service(state: AgentState) -> dict:
operation_key = (
f»{state[‘request’]}:{state[‘attempts’] + 1}»
)
Внешний сервис должен отклонить повтор операции
# с уже обработанным operation_key.
return {
«attempts»: state[«attempts»] + 1,
«history»: state[«history»] + [operation_key],
}
Сам чекпоинтер не обеспечивает транзакцию между PostgreSQL и сторонним API. Если необходима строгая гарантия, применяют шаблоны outbox, журнал операций или отдельную очередь доставки.
Разделение пользователей и защита данных
`thread_id` разделяет состояния, но не является механизмом авторизации. Если клиент может произвольно передать чужой идентификатор, одной настройки LangGraph недостаточно.
Безопасная схема выглядит так:
Приложение аутентифицирует пользователя.
Сервер находит ветку по собственной таблице соответствий.
3. Проверяется право пользователя на этот диалог или задание.
4. Только после проверки формируется конфигурация LangGraph.
5. Клиент не получает прямой доступ к таблицам checkpointer.
Лучше использовать случайные непрогнозируемые идентификаторы, например UUID, но даже UUID не заменяет проверку прав.
Для персональных и конфиденциальных данных дополнительно определите:
- какие поля разрешено включать в состояние;
- сколько времени хранится история;
- кто может читать снимки;
- как выполняется удаление по запросу пользователя;
- попадают ли значения в резервные копии;
- маскируются ли секреты и токены до сохранения.
Пароли, ключи API и строки подключения нельзя записывать в состояние графа. Они должны поступать из защищённого хранилища во время выполнения узла.
Почему не стоит переносить внутренние таблицы вручную
Прямой экспорт строк из SQLite и вставка в PostgreSQL выглядит просто, но это ненадёжный путь. Схемы двух checkpointer могут различаться, а структура данных зависит от версии пакетов. Кроме основной таблицы могут использоваться связанные записи, двоичные объекты, пространства имён и служебные версии.
Безопаснее выбрать один из трёх вариантов:
- оставить старые ветки в SQLite только для чтения, а новые создавать в PostgreSQL;
- перенести необходимые бизнес-данные через публичное состояние графа и начать новую ветку;
- написать отдельный инструмент миграции на основе публичных методов checkpointer и протестировать его на копии базы.
Если сохранение точной истории обязательно, зафиксируйте версии исходного и целевого пакетов, сравните число веток и снимков, а затем проведите тестовое восстановление нескольких состояний. Простого совпадения количества строк недостаточно.
Перед переключением приложения:
Остановите запись в исходную базу.
Создайте резервную копию.
3. Выполните тестовый перенос.
4. Прочитайте состояния через API LangGraph.
5. Запустите контрольные ветки с разными `thread_id`.
6. Проверьте повторные внешние операции.
7. Подготовьте план возврата к прежнему хранилищу.
Ограничение роста базы
Состояние агента часто увеличивается вместе с историей сообщений. Если каждый узел добавляет полный ответ модели, результаты инструментов и документы, объём чекпоинтов быстро становится значительным.
Практические меры:
- хранить в состоянии только данные, необходимые следующим узлам;
- заменять старую переписку краткой сводкой после заданного порога;
- выносить файлы и крупные результаты в отдельное хранилище;
- ограничивать размер ответа инструментов до записи в граф;
- удалять завершённые тестовые ветки по утверждённой политике;
- отслеживать рост базы по дням, а не только свободное место на диске.
Не удаляйте строки напрямую без проверки документации используемого checkpointer. Нарушение связей между служебными записями может сделать историю нечитаемой. Для регулируемых данных процедуру удаления необходимо проверить вместе с резервными копиями и журналами.
Как проверить производительность на своей нагрузке
Универсальные цифры для SQLite и PostgreSQL мало полезны: локальный SQLite на SSD может оказаться быстрее удалённой базы для одного запроса, но проиграть при нескольких конкурентных записях.
Минимальный тест должен включать:
| Проверка | Что измерять | Условие |
|---|---|---|
| Одиночный диалог | p50 и p95 вызова графа | Типичный размер состояния |
| Параллельные ветки | Ошибки и p95 | Реальное число рабочих процессов |
| Крупная история | Время записи и чтения | Верхняя граница контекста |
| Перезапуск приложения | Время восстановления | Новое соединение с базой |
| Отказ базы | Поведение и повторные операции | Кратковременная недоступность |
Не ограничивайтесь скоростью `app.invoke()`: она включает работу узлов и внешних моделей. Отдельно измеряйте задержку базы, число активных соединений, блокировки и рост объёма хранения.
Для SQLite критичны ошибки блокировки и время ожидания записи. Для PostgreSQL — заполнение пула, сетевые таймауты, длительные транзакции и превышение лимита подключений.
Контрольный список перед запуском
Для локального проекта достаточно проверить:
- установлен отдельный пакет `langgraph-checkpoint-sqlite`;
- файл базы находится в постоянном каталоге;
- каждый тест или диалог получает понятный `thread_id`;
- после перезапуска процесса состояние читается через `get_state()`.
Для продакшена дополнительно проверьте:
- установлен `langgraph-checkpoint-postgres`;
- `setup()` выполнен на целевой базе;
- учётные данные передаются через секреты;
- соединение использует TLS;
- размер пула согласован с лимитом PostgreSQL;
- внешние операции защищены от повторов;
- настроены резервные копии и проверено восстановление;
- есть политика хранения и удаления пользовательских данных;
- метрики показывают задержки, ошибки и рост таблиц;
- версии LangGraph и интеграции зафиксированы.
Практический следующий шаг: подключите SQLite к тестовой версии графа, выполните два вызова с одинаковым `thread_id`, перезапустите процесс и проверьте `app.get_state(config)`. Затем запустите тот же тест на PostgreSQL из двух параллельных процессов. Такой эксперимент быстро показывает, соответствует ли выбранное хранилище реальной модели нагрузки.