Поддержание состояния LangGraph Часть 2: PostgresSavers, HITL и продакшн-бэкенды

От чекпоинтов в ОЗУ к надежным бэкендам

В первой части рассматривалось, зачем существует механизм сохранения состояния LangGraph — состояние, чекпоинты, инструменты для их создания, потоки и InMemorySaver. В этой части эти же принципы адаптируются для использования в производстве. Инструменты сохранения в памяти исчезают при перезапуске процесса; настоящим агентам нужно хранилище, которое будет существовать дольше, чем сам процесс на Python.

PostgresSaver для надежности в производстве

База данных находится вне процесса приложения. Сбои, перезапуски и замена серверов сохраняют историю чекпоинтов нетронутой. PostgresSaver записывает эти чекпоинты в PostgreSQL, а не в ОЗУ.

PostgreSQL является популярным выбором, поскольку он надежен, широко предоставляется в виде управляемых облачных сервисов, безопасен для одновременной работы нескольких серверов приложений и способен масштабироваться по мере увеличения объема чекпоинтов с ростом трафика.

Установите специальный пакет:

pip install langgraph-checkpoint-postgres

Основные настройки:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # This creates the required tables the first time you use it.
    # Only needs to run once per database.
    checkpointer.setup()    graph = builder.compile(checkpointer=checkpointer)    config = {"configurable": {"thread_id": "session-99"}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": "Hello"}]},
        config=config,
    )

from_conn_string открывает соединение, управляемое контекстом. Функция setup() создает такие таблицы, как checkpoints и checkpoint_blobs, только один раз — следует рассматривать её как процедуру миграции, а не как действие, выполняемое при каждом запросе. Функция compile(checkpointer=...) использует ту же API, что и путь в оперативной памяти; смена бэкендов не приводит к переписыванию структуры данных.

Пулы соединений

Открытие нового соединения при каждом запросе тратит время. Лучше использовать пул готовых к работе соединений:

from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/mydb"with ConnectionPool(DB_URI) as pool:
    checkpointer = PostgresSaver(pool)
    checkpointer.setup()  # one-time setup
    graph = builder.compile(checkpointer=checkpointer)

Асинхронная версия

Приложения в стиле FastAPI должны использовать AsyncPostgresSaver:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

Если вы сами создаёте соединения с psycopg, установите autocommit=True, а также row_factory=dict_row; в противном случае функция setup() может не сохранять таблицы. Длина поля thread_id должна быть менее 255 символов; UUID являются безопасным стандартным значением.

Другие движки

SQLite

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

Один локальный файл сохраняется после перезагрузки и подходит для прототипов или скриптов, работающих в одном процессе. Однако блокировка файлов делает его неудобным для веб-приложений с несколькими рабочими процессами.

Redis

Быстрый движок, ориентированный на использование оперативной памяти с возможностью сохранения данных на диске; подходит для продуктов чата с высокой пропускной способностью и большим количеством одновременных потоков.

Управляемый облачный Postgres

RDS, Cloud SQL, Supabase, Neon и другие сервисы используют стандартный PostgreSQL, поэтому PostgresSaver работает корректно, пока провайдер занимается резервными копиями и обеспечением непрерывной работы.

Персонализированные инструменты контроля состояния

Подклассируйте BaseCheckpointSaver и реализуйте методы put, get_tuple, list (а также их асинхронные версии), когда внутренняя память является обязательной.

Быстрый выбор

Локальный долговечный скрипт → SQLite. Работа с несколькими рабочими процессами в продакшене → Postgres/AsyncPostgres. Экстремально высокая нагрузка на создание точек контроля → Redis. Существующий управляемый Postgres → Postgres, размещенный в облаке. Принудительное использование внутренней памяти → пользовательский механизм сохранения.

Пять примеров постепенного развития

Пример 1 — отсутствие сохранения данных (амнезия)

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: intdef increment(state: State) -> dict:
    return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)graph = builder.compile()  # No checkpointer passed in!print(graph.invoke({"count": 0}))  # {'count': 1}
print(graph.invoke({"count": 0}))  # {'count': 1} again — no memory of the last call!

{'count': 1}
{'count': 1}

Второй вызов не может увидеть первый: нет точки контроля, нет идентификатора потока.

Пример 2 — сохранение состояния с помощью InMemorySaver

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
    count: intdef increment(state: State) -> dict:
    return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "counter-1"}}print(graph.invoke({"count": 0}, config))  # {'count': 1}
print(graph.invoke({"count": 0}, config))  # {'count': 2}

{'count': 1}
{'count': 2}

Теперь одинаковые входные данные продолжают обрабатываться с использованием предыдущей точки контроля вместо перезапуска.

Пример 3 — возобновление работы после имитации сбоя

from langgraph.checkpoint.memory import InMemorySaver

# Shared storage dictionary simulates a database that survives across "restarts"
shared_storage = {}checkpointer_1 = InMemorySaver()
checkpointer_1.storage = shared_storage  # pretend this is durable storagegraph = builder.compile(checkpointer=checkpointer_1)
config = {"configurable": {"thread_id": "resume-demo"}}graph.invoke({"count": 0}, config)  # count becomes 1
print("Process 'crashes' here...")# Simulate a restart: a brand new checkpointer object, same underlying storage
checkpointer_2 = InMemorySaver()
checkpointer_2.storage = shared_storage
graph_after_restart = builder.compile(checkpointer=checkpointer_2)# Resume using the SAME thread_id
result = graph_after_restart.invoke({"count": 0}, config)
print(result)  # {'count': 2} — continued, did NOT reset to 0

Обмен словарем внутри процесса имитирует перезапуск; в режиме производства просто направляют новый процесс на ту же базу данных Postgres, после чего продолжается работа по thread_id.

Пример 4 — участие человека с использованием interrupt()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    approved: booldef request_approval(state: State) -> dict:
    decision = interrupt(f"Approve this task? '{state['task']}' (yes/no)")
    return {"approved": decision == "yes"}builder = StateGraph(State)
builder.add_node("request_approval", request_approval)
builder.add_edge(START, "request_approval")
builder.add_edge("request_approval", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "approval-1"}}# Step 1: Start the graph — it will pause at interrupt()
result = graph.invoke({"task": "Send $500 refund", "approved": False}, config)
print(result)  # Contains an '__interrupt__' key with our question# ... hours later, a human reviews and approves ...# Step 2: Resume with the human's answer
final_result = graph.invoke(Command(resume="yes"), config)
print(final_result)  # {'task': 'Send $500 refund', 'approved': True}

interrupt(...) приостанавливает работу и создаёт точки контроля. Запрашивающий код видит данные в поле __interrupt__. Позже с помощью Command(resume=...) продолжается работа. При возобновлении LangGraph перезапускает весь узел с начала — побочные эффекты сохраняются после перерыва, а не до него.

Пример 5 — чат-бот с поддержкой Postgres

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

class State(TypedDict):
    messages: listdef chatbot_node(state: State) -> dict:
    last_message = state["messages"][-1]
    reply = f"You said: {last_message}"
    return {"messages": state["messages"] + [reply]}builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # run once
    graph = builder.compile(checkpointer=checkpointer)    config = {"configurable": {"thread_id": "user-42"}}
    result = graph.invoke({"messages": ["Hi there"]}, config)
    print(result)

{'messages': ['Hi there', 'You said: Hi there']}

Остановите скрипт, запустите его завтра с тем же идентификатором потока — разговор продолжится, поскольку в базе данных сохраняются точки контроля.

Почему HITL зависит от сохранения данных

Три компонента: interrupt() для паузы с передачей данных, бесконечное ожидание без использования процессора при сохранении состояния в памяти, и Command(resume=...) для продолжения работы. Без точек контроля механизм паузы и ожидания не имеет где хранить замороженную структуру данных.

Реальные сценарии утверждения — принятие решений о кредитах при превышении пороговых значений, выдача клинических рекомендаций перед записью в медицинские карты, проверка юридических условий, контроль за превышением расходов — часто требуют ожидания от нескольких минут до недель. Механизм сохранения данных делает такое ожидание экономичным и безопасным.

Сохранение данных независимо от возможностей агента

Длительные диалоги, многоэтапные планы, способные выдержать сбои, циклы инструментов, которые не должны забывать предыдущие результаты, а также рабочие процессы с переходами между ветвями все зависят от надежных точек контроля. Идентификаторы потоков служат инструментами, которые операторы и интерфейсы используют для возобновления правильного запуска процесса.

Часто встречающиеся ошибки

• Вызов функции setup() при каждом запросе.

Использование SQLite с несколькими рабочими процессами Gunicorn/Uvicorn.

Чрезмерно большие значения thread_id.

Выполнение необратимых побочных действий до вызова interrupt().

Забывание передачи того же thread_id при возобновлении работы.

Предположение, что демо-версии в памяти обеспечивают такую же надежность, как и продакшн-системы.

Практики, эффективно работающие в любых условиях

Выполняйте настройку схемы в виде миграций. Предпочитайте использование пулов. В асинхронных приложениях применяйте асинхронные механизмы сохранения данных. Сохраняйте значения thread_id короткими и стабильными. Документируйте, какие среды используют тот или иной бэкенд. Добавляйте тесты на сбои и возобновление работы. Контролируйте рост и хранение таблиц с точками контроля.

Итог

Поддержание состояния LangGraph в производстве сводится в основном к использованию «той же API графа, но другого надежного хранилища». Postgres является стандартным выбором для корпоративных проектов; SQLite и Redis заполняют определенные ниши; пользовательские механизмы сохранения предназначены для соблюдения требований к инфраструктуре. Участие человека и восстановление после сбоев — это не факультативные функции, а причина существования механизмов создания точек контроля. Необходимо заранее настроить взаимодействие с бэкендом, чтобы возможности прерывания и продолжения работы существовали на самом деле еще до того, как агент начнет взаимодействовать с пользователями.

Руководство по эксплуатации бэкендов для точек контроля

Относитесь к хранению точек контроля так же, как к основным данным приложения. Расписания резервного копирования, учения по восстановлению и политики хранения должны обсуждаться вместе с информацией о клиентах. Чат-приложение, которое не может восстановить историю сообщений после регионального сбоя, на самом деле не поставляет память — оно поставляет лишь надежду.

Планирование мощности должно включать отслеживание контрольных точек по активному потоку в день, среднего размера блобов в состояниях с большим объемом сообщений, а также роста нагрузки при максимальной одновременности работы. Большие объемы данных, загружаемые в состояние базы, значительно увеличат её размер задолго до того, как количество строк станет тревожным. Лучше хранить объемные данные в облачном хранилище объектов, а указатели оставлять в состоянии базы.

Контроль доступа имеет большое значение: пользователь базы данных, отвечающий за контрольные точки, не должен иметь тех же полномочий, что и пользователь, выполняющий миграцию несвязанных схем с использованием DDL. Необходимо регулярно менять учетные данные, предпочитать механизмы аутентификации IAM там, где это позволяют поставщики облачных услуг, и контролировать, кто может выполнять операцию SELECT над состоянием общения. Идентификаторы потоков часто соответствуют идентификаторам пользователей; данные, получаемые в результате экспорта, следует считать персональными данными.

При миграции с InMemorySaver на Postgres в рабочей системе двойное запись редко бывает необходима, если согласиться на то, что потоки в оперативной памяти сбрасываются один раз. Задайте время для технического обслуживания, разверните новый механизм контроля состояния и отслеживайте уровень ошибок при возобновлении работы. При переходе по схеме «синий/зеленый» убедитесь, что оба варианта указывают на одну и ту же базу данных, иначе возникнут противоречия в истории потоков.

Советы по тестированию: запустите Postgres в среде CI (или в временном контейнере), выполните функцию setup() один раз, протестируйте пять примеров в качестве интеграционных тестов и убедитесь, что второй процесс может возобновить поток, созданный первым. Такой тест межпроцессного взаимодействия выявляет больше ошибок в реальных условиях, чем любой тест на оперативной памяти.

Наконец, обучите сотрудников дежурной службы различию между состояниями «graph failed» и «checkpointer unreachable». Симптомы схожи — возникает ошибка, — но способы устранения отличаются. Панели управления, разделяющие исключения приложений и проблемы с временем ожидания базы данных, позволяют сэкономить часы во время инцидентов.

Идентификаторы потоков и пользовательский интерфейс

Идентификаторы потоков — это не просто ключи базы данных; они представляют собой объекты продукта. Намеренно связывайте их с сессиями пользователей, заявками на поддержку или идентификаторами документов. Для возможности «продолжить разговор» в интерфейсе используйте тот же идентификатор; для «начала с нуля» создавайте новый. Смешивание этих функций в одном идентификаторе приводит к ситуациям, когда старые результаты инструментов снова появляются в диалогах.

Для пользователей, использующих несколько устройств, необходимы поиски с учетом авторизации из-за стабильности идентификаторов потоков между клиентами. Никогда не принимайте идентификатор потока, предоставленный клиентом, без проверки его принадлежности. Чекпоинтеры хранят всё, что им передается; они не обеспечивают контроля за принадлежностью данных.

Эволюция схемы для сохраняемого состояния

Объекты TypedDict в состоянии могут меняться. Добавление необязательных полей происходит легко; переименование или удаление полей нарушает существующие точки сохранения. Лучше использовать добавочные изменения, настройку по умолчанию для отсутствующих ключей и явные миграции при разрушительных изменениях. Храните значение schema_version в состоянии и используйте его для логики узлов при необходимости.

У больших списков сообщений следует применять политики сокращения: сводите более старые сообщения в одно поле, сохраняйте самые свежие необработанные сообщения и тестируйте возможность продолжения работы после сокращения. Бесконечная история — наиболее частая причина роста размеров таблиц точек сохранения в Postgres.

Шаблоны прерывания пользовательского интерфейса

Интерфейсы утверждения должны отображать данные перехвата, соответствующую часть состояния и сроки выполнения. Идентификатор утверждающего лица следует сохранять в состоянии при возобновлении работы. Если сроки утверждения истекают, необходимо использовать механизм наблюдения, который отменяет процесс или переводит его на более высокий уровень обработки, вместо того чтобы оставлять потоки в состоянии паузы бесконечно. Возможность сохранения данных позволяет длительное ожидание; политика продукта должна устанавливать для этого ограничения.

Упражнения по восстановлению после катастроф

Каждый квартал необходимо восстановить резервную копию в тестовую среду и возобновить работу известного по идентификатору потока. Необходимо подтвердить, что сообщения, статус перехвата и остатки данных в инструментах совпадают. Следует задокументировать показатели RPO/RTO для базы данных контрольных точек наряду с основной базой данных приложения. Агенты, запоминающие пользователей, должны обеспечивать такую же непрерывность работы, как и системы электронной почты.

Замечания по производительности

Размер пулов подключений должен определяться количеством одновременно работающих графовых процессоров, а не только веб-обрабатчиков. Следите за паттерном N+1, при котором каждый узел ненужно записывает точку контроля — настраивайте использование инструментов проверки в соответствии с параметрами по умолчанию версии LangGraph. Сжимайте или выносите за пределы системы большие объемы бинарных данных, генерируемых инструментами.

Примечания по безопасности

Шифруйте диски, используйте TLS для связи с Postgres, ограничивайте сетевые пути от рабочих процессов к хранилищу, а также удаляйте конфиденциальную информацию из состояния системы перед записью точки контроля, если инструменты могут получить доступ к исходным ключам. Рассмотрите возможность использования промежуточных компонентов для маскировки данных на уровне полей. Во время проверок соответствия будут задаваться вопросы — подготовьте ответы с диаграммами потока данных от пользователя → графа → инструмента проверки → диска.

Эти операционные слои превращают примеры кода в систему, на которую можно положиться. API имеют небольшой размер; настоящей особенностью продакшн-среды является соблюдение определенных правил работы.

Согласование работы инструментов проверки с горизонтальным автоскейлингом

Когда Kubernetes масштабирует рабочие процессы графа с двух подов до двадцати, каждый под должен взаимодействовать с одним и тем же надежным контрольным механизмом. Стейки-сессии не нужны, если идентификаторы потоков передаются вместе с каждым запросом. Пулы готовых процессов не должны полагаться на локальные файлы SQLite. Проверки работоспособности должны подтверждать соединение с базой данных, а не только живучесть процесса. Во время развертывания процессы должны завершаться или сохранять состояние в промежуточный момент, чтобы возможно было продолжить их работу после прерывания. Для тестирования устойчивости убирайте поды во время ожидания вызова interrupt() и проверяйте возможность продолжения работы на другом поде. Один такой эксперимент проверяет архитектуру лучше, чем десяток тестов на единицы.

Нагрузка на систему хранения документов: если скорость работы хранилища чекпоинтов снижается, должны ли рабочие процессы снизить нагрузку или ожидать в очереди? Лучше явно обработать сбои, чем использовать скрытые внутренние решения, которые могут приводить к несоответствиям между подами. Операторы никогда не должны «временно» переключаться на InMemorySaver в производственной среде для устранения задержек — такой переключение нарушает обещания, данные продуктом.

Чек-лист перед тем, как считать процесс сохранения завершённым

Подтверждено, что функция setup() выполняется в режиме миграции; настроены пулы ресурсов; режим асинхронной и синхронной обработки согласован с веб-фреймворком; обеспечена привязка идентификаторов потоков к владельцам; протестировано возобновление работы при перерывах между процессами; резервные копии восстановлены с успехом один раз; задачи по хранению данных задокументированы; проверена обработка персональных данных; панели управления различают ошибки базы данных и графиков; руководства по эксплуатации связывают идентификаторы потоков с инструментами поддержки пользователей. После того как каждый пункт будет отмечен, PostgresSaver (или выбранный вами бэкенд) перестанет быть просто фрагментом кода — он станет частью гарантий качества продукта. До этого момента рассматривайте демонстрации как демонстрации и оставляйте флаги функций выключенными для долгосрочных процессов типа HITL.

Короткая история о неудачном возобновлении работы

Один из инженеров дежурной службы однажды столкнулся с ситуацией, когда утверждения застревали в ожидании часами. График работы работал нормально; пароль контроллера проверки менялся, но сотрудники сохраняли старые учетные данные в кэше. Новые прерывания не срабатывали, пока старые потоки продолжали читать данные. Решением стало обновление учетных данных, а также добавление правила в руководство: проверять авторизацию контроллера перед перезапуском подов графика. Сохраните этот урок вместе с фрагментами кода PostgresSaver, чтобы при следующем обновлении не повторилась проблема. Постоянная память включает в себя операционную память — знания, зафиксированные в руководствах по эксплуатации.

Этот инцидент также прояснил вопросы мониторинга: необходимо отправлять уведомления о сбоях авторизации контроллера проверки на отдельной странице, а не в общую категорию ошибок 500. Когда память является ключевым ресурсом, путь к базе данных считается зависимостью первого уровня. Благодаря такой дисциплине замена InMemorySaver на PostgresSaver перестает быть страшной миграцией и превращается в обычный шаг на пути от ноутбука к сервису.