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

Источник: https://modelium.tech/zhurnal/langgraph-postgres-checkpoint/

Для сохранения состояния локального агента LangGraph подключите PostgresSaver при компиляции графа, один раз создайте таблицы через setup() и передавайте постоянный thread_id. Проверяйте восстановление на тестовом потоке: остановите процесс после interrupt(), запустите новый и продолжите тот же поток через Command(resume=...). PostgreSQL хранит checkpoint графа, но сам по себе не делает вызовы внешних инструментов идемпотентными и не гарантирует изоляцию данных.

**Срез на 30 сентября 2026 года:** код LangGraph закреплён по commit `07b3318`, документация — по commit `2fcd119`. Ниже приведён пример для собственного тестового экземпляра PostgreSQL, а не результат запуска Modelium. [s1](#src-s1) [s2](#src-s2) [s8](#src-s8)

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

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

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

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

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

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

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

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

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

import osimport sysfrom typing import TypedDictfrom langgraph.checkpoint.postgres import PostgresSaverfrom langgraph.graph import END, START, StateGraphfrom langgraph.types import Command, interruptclass State(TypedDict, total=False):    prepared: str    resumed: booldef 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](#src-s1) [s4](#src-s4) [s6](#src-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](#src-s4) [s7](#src-s7)

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

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

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

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

Второй график проверки — отказ: в отдельном тестовом потоке вызовите `Command(resume="stop")` и убедитесь, что `resumed` остаётся `False`. **Следующий шаг:** откройте [гайд по подтверждению действий человеком](/zhurnal/langgraph-human-in-the-loop/) и проверьте паузу/отказ/возобновление на тестовых действиях без доступа к рабочим данным. [s4](#src-s4)

## Источники

1. [LangGraph checkpoint-postgres — README и setup](https://github.com/langchain-ai/langgraph/blob/07b33185eab893be2ed031eedae52f09314bf77c/libs/checkpoint-postgres/README.md) — получено 2026-09-30
2. [LangGraph — persistence and checkpointer vs store](https://github.com/langchain-ai/docs/blob/2fcd119c6cb031ba897b72931a175e2869dbb7b1/src/oss/langgraph/persistence.mdx) — получено 2026-09-30
3. [LangGraph — fault tolerance and recovery](https://github.com/langchain-ai/docs/blob/2fcd119c6cb031ba897b72931a175e2869dbb7b1/src/oss/langgraph/fault-tolerance.mdx) — получено 2026-09-30
4. [LangGraph — interrupts and resume semantics](https://github.com/langchain-ai/docs/blob/2fcd119c6cb031ba897b72931a175e2869dbb7b1/src/oss/langgraph/interrupts.mdx) — получено 2026-09-30
5. [LangGraph — Graph API, StateGraph and node behavior](https://github.com/langchain-ai/docs/blob/2fcd119c6cb031ba897b72931a175e2869dbb7b1/src/oss/langgraph/use-graph-api.mdx) — получено 2026-09-30
6. [LangGraph checkpoint — serialization security and thread IDs](https://github.com/langchain-ai/langgraph/blob/07b33185eab893be2ed031eedae52f09314bf77c/libs/checkpoint/README.md) — получено 2026-09-30
7. [LangGraph — checkpointers, durability modes and pending writes](https://github.com/langchain-ai/docs/blob/2fcd119c6cb031ba897b72931a175e2869dbb7b1/src/oss/langgraph/checkpointers.mdx) — получено 2026-09-30
8. [PostgresSaver — pinned implementation of connection and setup](https://github.com/langchain-ai/langgraph/blob/07b33185eab893be2ed031eedae52f09314bf77c/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py) — получено 2026-09-30