Стабільність LangGraph Частина 2: PostgresSavers, HITL та продакшн-бекенди
Перемістіть LangGraph з InMemorySaver у PostgresSaver із використанням пулів PostgreSQL та асинхронних зберігачів, порівняйте SQLite та Redis, а також реалізуйте функції переривання та відновлення роботи у продакшні.
У першій частині було розглянуто причини існування механізму збереження даних у LangGraph — стан, контрольні точки, засоби їх створення, потоки та InMemorySaver. У цій частині ці ідеї адаптуються для використання у продакшені. Засоби збереження даних у пам’яті зникають при перезапуску процесу; справжні агенти потребують сховища, яке буде існувати довше, ніж сам процес Python.
База даних знаходиться поза процесом додатку. Збої, перезапуски та заміна серверів залишають історію контрольних точок недоторканою. PostgresSaver зберігає ці контрольні точки у PostgreSQL, а не у RAM.
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() для паузи з даними, необмежене очікування без використання CPU, поки стан зберігається у пам’яті, та Command(resume=...) для продовження роботи. Без контрольних точок механізм паузи та очікування не має де зберігати заморожену структуру даних.
Реальні сценарії затвердження — рішення щодо кредитів, які перевищують певний поріг, клінічні рекомендації перед записуванням даних, перевірка юридичних умов, перевищення бюджету — часто потребують очікування від кількох хвилин до тижнів. Завдяки сталості це очікування залишається ефективним та безпечним.
Сталість між можливостями агентів
Довгі розмови, багатоетапні плани, які залишаються актуальними після збоїв, цикли інструментів, які не повинні забувати попередні результати, та робочі процеси з переходами між варіантами — усе це залежить від надійних контрольних точок. ID потоків є інструментом, який оператори та інтерфейси використовують для продовження правильного виконання завдань.
Помилки, які постійно з’являються
- Виклик
setup()при кожному запиті.
thread_id.interrupt().thread_id під час продовження роботи.Практики, які добре працюють
Виконуйте налаштування схеми як міграцію. Віддавайте перевагу пулам. Використовуйте асинхронні зберігачі в асинхронних додатках. Зберігайте ідентифікатори потоків короткими та стабільними. Документуйте, які середовища використовують який бекенд. Додавайте тести на аварійне завершення та продовження роботи. Слідкуйте за зростанням та збереженням даних у таблиці контрольних точок.
Підсумок
Зберігання даних LangGraph у продакшені здебільшого полягає у використанні „того самого API графу, але різного надійного сховища“. Postgres є стандартним вибором для корпоративних проектів; SQLite та Redis покривають окремі сегменти ринку; спеціальні механізми зберігання використовуються для задоволення обов’язкових інфраструктурних вимог. Участь людини та відновлення після збоїв — це не факультативні функції; саме тому існують механізми створення контрольних точок. Підключіть бекенд якомога раніше, щоб переривання та продовження роботи були реальністю ще до того, як агент почне працювати з користувачами.
Операційний посібник для бекендів з контрольними точками
Ставіться до зберігання контрольних точок так само, як до основних даних додатку. Графіки резервного копіювання, відпрацювання процедур відновлення та політики зберігання мають обговорюватися разом із записами клієнтів. Чат-продукт, який не може відновити історію розмов після регіонального переходу, насправді не поставив пам’ять — він поставив лише надію.
Планування пропускної здатності має враховувати контрольні точки за кожною активною ниткою щодня, середній розмір блоків даних у станах з великою кількістю повідомлень та зростання обсягу даних під час піку одночасних операцій. Великі об’єми даних, які зберігаються у стані бази даних, значно збільшують її розмір ще до того, як кількість рядків стане тривожною. Краще зберігати великогабаритні дані у об’єктному сховищі, а посилання залишати у стані бази даних.
Контроль доступу має велике значення: користувач бази даних, який виконує функції контрольного показника, не повинен мати тих самих прав, що й користувач, який використовує потужні функції DDL для міграції непов’язаних схем. Необхідно регулярно змінювати облікові дані, використовувати механізми автентифікації IAM, де це дозволяють постачальники хмарних послуг, та перевіряти, хто може виконувати операції SELECT щодо стану розмов. Ідентифікатори ниток часто відповідають користувачам; дані, отримані шляхом експорту, слід вважати конфіденційною інформацією особистого характеру.
Під час міграції з InMemorySaver на Postgres у діючій системі подвійне записування рідко є необхідним, якщо ви погоджуєтесь на те, що потоки в оперативній пам’яті скидаються один раз. Запишіть час проведення технічного обслуговування, впровадьте новий механізм контролю стану та стежте за рівнем помилок під час відновлення роботи. При переході типу «синій/зелений» переконайтеся, що обидві версії вказують на одну й ту саму базу даних, інакше виникне розбіжність у історії потоків.
Поради щодо тестування: запустіть Postgres у середовищі CI (або у тимчасовому контейнері), виконайте функцію setup() один раз, запустіть п’ять прикладних сценаріїв як інтеграційні тести та переконайтеся, що другий процес може продовжити роботу потоку, створеного першим. Саме цей тест міжпроцесного взаємодії виявляє більше помилок у реальних умовах, ніж будь-який тест типу unit у пам’яті.
Нарешті, навчіть особу, яка працює у режимі наряду, різниці між „graph failed“ та „checkpointer unreachable“. Симптоми схожі — виникає помилка — але способи її усунення відрізняються. Панелі керування, які розділяють винятки програми та періоди очікування на з’єднання з базою даних, економлять години під час інцидентів.
Ідентичність потоку та користувацький інтерфейс продукту
Ідентифікатори потоків — це не лише ключі бази даних; вони є об’єктами продукту. Навмисно прив’язуйте їх до сеансів користувачів, заявок на підтримку чи ідентифікаторів документів. Для можливості „продовжити розмову“ у користувацькому інтерфейсі використовуйте цей ідентифікатор знову. Для „початку заново“ створюйте новий ідентифікатор. Змішування цих функцій у одному ідентифікаторі призводить до ситуацій, коли старі результати інструментів знову з’являються під час розмов.
Для користувачів кількох пристроїв стабільність ідентифікаторів потоків між клієнтами вимагає пошуків з урахуванням облікових даних. Ніколи не приймайте ідентифікатор потоку, наданий клієнтом, без перевірки його належності. Чекпоїнтери зберігають все, що ви їм надасте; вони не контролюють приватність даних.
Об’єкти TypedDict у стані постійно еволюціонують. Додавати необов’язкові поля легко; перейменування чи видалення полів руйнує старі контрольні точки. Краще використовувати додаткові зміни, параметри за замовчуванням для відсутніх ключів та чіткі міграції під час руйнівних змін. Зберігайте schema_version у стані та використовуйте логіку вузлів для роботи з ним за потреби.
Великі списки повідомлень вимагають політик скорочення: узагальнюйте старі повідомлення в одне поле, зберігайте останні необроблені повідомлення та перевіряйте можливість продовження роботи після скорочення. Неконтрольована історія — найпоширеніша причина розмноження таблиць контрольних точок у Postgres.
Інтерфейси схвалення мають відображати дані переривання, відповідну частину стану та терміни. Ідентифікатор схвалюючої особи слід зберігати у стані під час відновлення роботи. Якщо терміни схвалення закінчуються, необхідно запустити механізм спостереження, який скасує процес або переведе його на вищий рівень, замість того щоб залишати потоки у стані призупинки назавжди. Збереження даних у довгостроковій пам’яті дозволяє тривало чекати; політика продукту має встановлювати обмеження щодо цього.
Тренування з відновлення після катастроф
Щокварталу необхідно відновити резервну копію у тестове середовище та продовжити роботу відомого ідентифікатора потоку. Перевірити, чи повідомлення, статус переривання та залишки даних від інструментів збігаються. Документувати значення RPO/RTO для бази даних контрольних точок разом із основною базою даних додатку. Агенти, які пам’ятають користувачів, мають дотримуватися тих самих вимог до безперервності роботи, що й системи електронної пошти.
Примітки щодо продуктивності
Пулі з’єднань слід налаштовувати відповідно до кількості одночасних робочих процесів графа, а не лише веб-обработувачів. Слідкуйте за схемою N+1, коли кожен вузол зайво створює контрольні точки — налаштовуйте використання контрольників відповідно до стандартних параметрів версії LangGraph. Скомпресуйте або виведіть накопичені результати роботи бінарних інструментів у зовнішні формати.
Примітки щодо безпеки
Шифруйте диски, використовуйте TLS для зв’язку з Postgres, обмежуйте мережеві шляхи від робочих процесів до сховища та видаляйте конфіденційну інформацію зі стану перед створенням контрольної точки, якщо інструменти можуть отримати нешифровані ключі. Розгляньте використання проміжних засобів для маскування даних на рівні полів. Під час перевірок на відповідність будуть ставитися запитання; підготуйте відповіді з діаграмами потоку даних від користувача → граф → контрольник → диск.
Ці операційні шари перетворюють приклади коду на систему, якою можна спокійно користуватися. API є простими; справжньою перевагою у продакшені є належна організація роботи всієї системи.
Координація контрольників з горизонтальним автоскалінгом
Коли Kubernetes масштабує робочі процеси графа з двох подів до двадцяти, кожен под повинен взаємодіяти з однаковим стабільним чекпоїнтером. Стикі-сесії не є необхідними, якщо ідентифікатори потоків передаються з кожним запитом. Пули готових процесів не повинні покладатися на локальні файли SQLite. Перевірки стану мають перевіряти підключення до бази даних, а не лише життєздатність процесу. Під час розгортання процеси, які виводяться з експлуатації, повинні завершитися або зробити чекпоїнт під час виконання, щоб перерви залишалися можливими до відновлення. Для тестування стійкості вбивайте поди під час очікування функції interrupt() та перевіряйте можливість відновлення роботи на іншому поді. Цей один експеримент підтверджує архітектуру краще, ніж десяток тестів на одиницю.
Тиск у зворотному напрямку від документа: якщо зберігання контрольних точок уповільнюється, чи повинні працівники зменшити навантаження чи вставити його у чергу? Краще явна помилка, ніж тихий внутрішній резервний варіант, який не узгоджується між подами. Оператори ніколи не повинні „тимчасово“ переходити на InMemorySaver у продакшені, щоб „виправити затримку“ — цей перехід скасовує обіцянку продукту.
Чек-лист перед тим, як вважати зберігання завершеним
Підтверджено, що setup() виконується як процес міграції; налаштовані пули; асинхронний та синхронний режими узгоджені з веб-фреймворком; дотримується принцип власності ідентифікаторів потоків; перевірено можливість відновлення після переривання між процесами; резервні копії були успішно відновлені один раз; задокументовано процес зберігання даних; переглянуто правила обробки персональних даних; панелі керування дозволяють розрізняти помилки БД та графіків; посібники з використання пов’язують ідентифікатори потоків із інструментами підтримки користувачів. Коли кожна галочка поставлена, PostgresSaver (або обраний вами бекенд) більше не є просто фрагментом коду — він стає частиною гарантій щодо стабільності продукту. До того часу сприймайте демонстрації як демонстрації та не ввімкнюйте флаги функцій для процесів з тривалим існуванням типу HITL.
Коротка історія про невдале відновлення
Один із інженерів, який працює у режимі чергування, колись бачив, як схвалення застрягали протягом годин. Граф вів себе нормально; пароль checkpointer оновлювався, а працівники зберігали старі облікові дані в кеші. Нові переривання зазнавали невдачі, поки старі потоки все ще здійснювали читання даних. Рішенням стало оновлення облікових даних та додавання правила до посібника: перевіряти автентифікацію checkpointer перед перезапуском контейнерів графа. Збережіть цей урок разом із фрагментами коду PostgresSaver, щоб наступне оновлення не призвело до повторної перерви у роботі. Постійна пам’ять включає операційну пам’ять — людські знання, зафіксовані у посібниках.
Цей інцидент також прояснив питання моніторингу: необхідно створювати сповіщення про помилки автентифікації checkpointer на окремій сторінці, а не в загальному категорії 500. Коли продуктом є пам’ять, шлях до бази даних є залежністю нульового рівня. Завдяки такій дисципліні заміна InMemorySaver на PostgresSaver перестає бути страшною міграцією та стає звичайним кроком у процесі переходу від ноутбук-записника до сервісу.