LangGraph и PostgreSQL: checkpoint локального агента

Что именно хранит checkpoint?

Checkpointer сохраняет состояние графа внутри одного потока (thread_id) на границах шагов. Это не хранилище общих знаний между потоками: для межпоточных данных LangGraph предлагает Store. В памяти процесса InMemorySaver теряет состояние после перезапуска; PostgresSaver записывает checkpoint в PostgreSQL. s2 s7

Намерение этого гайда — не «как показать кнопку подтверждения» и не выбор локальной LLM. Здесь проверяем, переживает ли состояние тестового потока закрытие одного Python-процесса и запуск второго. Для реального агента модель и инструменты подключаются отдельно: checkpoint не скачивает модель, не предоставляет сервер инференса и не проверяет разрешения на внешние действия. Это вывод из границы API графа и checkpointer, а не измерение. s2 s5 s7

Что подготовить до запуска?

Нужны доступный тестовый PostgreSQL с отдельной БД и минимально необходимыми правами, Python-окружение с langgraph и langgraph-checkpoint-postgres, постоянный непустой ID потока и DSN в переменной среды PG_DSN. README пакета указывает, что он устанавливает Psycopg 3 без extras; если окружению нужен готовый binary backend, документация отдельно допускает psycopg[binary]. Не копируйте пароль БД в статью или Git. s1 s6

Проверять существующий рабочий процесс этим примером не надо: выберите тестовый поток, который не выполняет побочных действий. Строгая десериализация через LANGGRAPH_STRICT_MSGPACK=true или явный список разрешённых модулей снижает риск исполнения кода при чтении недоверенного содержимого БД; это не замена доступа к БД, шифрования и резервных копий. Последние три меры — редакционное ограничение области угроз, а не обещание встроенной функции LangGraph. s1 s6

Как собрать минимальный граф с PostgreSQL?

Документация фиксирует PostgresSaver.from_conn_string(...), а его исходный код сам открывает соединение с autocommit=True и row_factory=dict_row. Если вы передаёте собственное соединение Psycopg, задайте эти параметры вручную: без autocommit таблицы после setup() могут не сохраниться, а кортежи вместо словарей сломают чтение. setup() создаёт необходимые таблицы и миграции; в отдельном проекте вызывайте его при первичной подготовке схемы. s1 s8

Ниже учебный скрипт checkpoint_demo.py. У него только два узла: первый готовит значение состояния, второй делает управляемую паузу без доступа к сети, файлам или корпоративным инструментам. Команда resume не создаёт новый поток: она открывает тот же DSN и thread_id. durability="sync" запрашивает синхронную запись перед следующим шагом; для async документация допускает малое окно потери checkpoint при аварии процесса. s4 s5 s7

import os
import sys
from typing import TypedDict

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class State(TypedDict, total=False):
prepared: str
resumed: bool

def prepare(state: State) -> State:
return {"prepared": "test-document-v1"}

def pause(state: State) -> State:
# Код до interrupt() может повториться при возобновлении.
signal = interrupt("Продолжить тестовый поток?")
return {"resumed": signal == "continue"}

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

if len(sys.argv) != 2 or sys.argv[1] not in {"start", "resume"}:
raise SystemExit("Использование: python checkpoint_demo.py start|resume")
if not os.getenv("PG_DSN"):
raise SystemExit("Укажите PG_DSN в окружении; не сохраняйте пароль в файле")

config = {"configurable": {"thread_id": "modelium-example-001"}}
with PostgresSaver.from_conn_string(os.environ["PG_DSN"]) as saver:
saver.setup() # Для примера; первая настройка создаёт таблицы.
graph = builder.compile(checkpointer=saver)
if sys.argv[1] == "start":
result = graph.invoke({"prepared": ""}, config, durability="sync")
assert result.get("__interrupt__"), "Ожидалась пауза"
print("paused:", graph.get_state(config).values)
else:
assert graph.get_state(config).next, "Нет приостановленного потока"
result = graph.invoke(Command(resume="continue"), config, durability="sync")
assert not result.get("__interrupt__"), "Поток снова приостановлен"
assert result["prepared"] == "test-document-v1" and result["resumed"]
print("resumed:", graph.get_state(config).values)

Команды по документации, в вашем собственном тестовом окружении (не результат прогона Modelium): s1 s4 s6

python -m pip install langgraph langgraph-checkpoint-postgres 'psycopg[binary]'
export LANGGRAPH_STRICT_MSGPACK=true
# Задайте PG_DSN безопасным способом вне shell history и репозитория.
python checkpoint_demo.py start
# Закройте первый процесс; только затем запустите второй с тем же PG_DSN.
python checkpoint_demo.py resume

После start должен появиться признак __interrupt__ и состояние с prepared. После запуска другого процесса resume читает из той же БД и проверяет сохранённое значение prepared и новое resumed. Сравните thread_id и адрес БД; новый ID запускает другой поток. Одинаковый процесс не проверяет стойкость к рестарту. s4 s7

Что делать при ошибках или неожиданном повторе?

Если таблиц нет — проверьте доступ к БД и вызов setup(). При ручном открытии соединения проверьте autocommit=True, row_factory=dict_row; у from_conn_string они заданы в исходном коде. Для асинхронного графа используйте AsyncPostgresSaver и его await ...setup(); не смешивайте синхронный пример с ainvoke. Если thread_id чрезмерно длинный, документация предлагает держать его короче 255 символов. s1 s2 s8

При interrupt() узел на возобновлении выполняется с начала, поэтому побочное действие до точки паузы может повториться. Checkpoint обеспечивает восстановление состояния, но не «ровно один раз» для платежа, записи CRM или команды на сервере. Для реального агента выделяйте внешние действия в отдельные узлы/задачи, используйте идемпотентные ключи у внешней системы и журнал подтверждений. LangGraph описывает replay и pending writes, но итоговая корректность сторонней системы требует отдельной проверки. s3 s4 s7

Если поток должен пережить принудительное отключение в середине шага, режим exit не подходит: документация прямо указывает отсутствие промежуточных checkpoint; async имеет окно риска, sync ждёт записи перед продолжением. Даже sync не гарантирует, что внешний побочный эффект и запись checkpoint совершились как одна транзакция. Оцените права доступа к БД, секреты в состоянии и правила хранения данных до работы с корпоративной информацией. s3 s7

Второй график проверки — отказ: в отдельном тестовом потоке вызовите Command(resume="stop") и убедитесь, что resumed остаётся False. Следующий шаг: откройте гайд по подтверждению действий человеком и проверьте паузу/отказ/возобновление на тестовых действиях без доступа к рабочим данным. s4