LangGraph Persistence Частка 2: PostgresSavers, HITL і серверы для працы ў працэсе
Перамясціце LangGraph з InMemorySaver у PostgresSaver з пуламі PostgreSQL і асінхроннымі сэўварамі, парабялейце SQLite і Redis, а таксама запрацаваце функцыі перарыву і продажу ў працоўнай асяроддзя.
З кэш-пунктав RAM да стойкіх бэкендаў
У першай частцы было розглянуто, чаму існуе функцыя зберагчэння дадзеных у LangGraph — стан, кэш-пункты, прыстроі для стварэння кэш-пунктав, ніткі і InMemorySaver. У гэтай частцы тыя ж ідэі адпрацоўваны для практычнага вжывання. Зберагчыкі дадзеных у памяці знікаюць пасля перазапуску процэса; справжнім агентам патрэбны магчымасці зберагчэння, якія будуць актуальныя пасля завершэння працы процэса Python.
PostgresSaver для стойкасці у працэсе віробніцтва
База дадзеных знаходзіцца параду ў процэсу прыложэння. Збоі, перэзапускі і змяны апарату не паграждаюць історыі кэш-пунктав. 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
Шырокі, працюе пераважна з памяцю, з можлівасцю зберагчэння на дыску — падходзіць для продуктав чату з высокай праўоцею, якія выкарыстоўваюць многа адночасных тыльнякаў.
Управляемы постгрэс у воблаку
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']}
Зупініце скрыпт, запускайце яго завтра з тым самым thread id, і размова будзе продаважвацца, таму што база дадзеных зберагчыць пункты перапыткі.
Чаму HITL залежыць ад стойкасці
Тры ўзелі: interrupt() для паўзы з пакетам дадзеных, несканчатлівая чаканне без выкарыстоўвання CPU, калі стан знаходзіцца ў запамяці, і Command(resume=...) для продажу працы. Без пункта перапаказа паўза-чаканне не мае куды зберагчы «заморожаную» структуру дадзеных.
Рэальныя патэрны затверджэння — рашэнні пра крэдыты, якія перавышаюць певны порог, клінічныя рэкамендацыі перад фіксацыяй даных, аналіз прававых клauзул, перакрыцця расходаў — часта трываюць ад калькі хвілін да нясколькох тыдзён. Перазбіраннае зберагчэнне робіць такую чаканне дышаўным і безпечным.
Перазбіраннае зберагчэнне між можлівасцямі агента
Дзейныя размовы, багатоэтапныя планы, якія выжываюць пасля збоёў, ціклы інструментаў, якія не павінны забываць паканальныя рэзультаты, а таксама рабочыя працэсы з розгалужэннямі і паўторным прыўязам усе завісяць ад надзейных пунктов перапаказа. ID-ы потокаў ёсць інструментамі, якія аператары і інтэрфейсы выкарыстоўваюць для продажу правільнага варыянта працы.
Памілкі, якія продовжваюць з’являцца
- Выкліканне
setup()за кожны запіт.
thread_id.interrupt().thread_id пад прызначэнне працы.Практыкі, якія добра функцыонуюць
Выканайце настройку схемы як міграцыю. Вярніцеся да пулів. Оывайце асінхронныя сэйверы ў асінхронных дапрыемах. Робіце значэння thread ids короткімі і стабільнымі. Документавайце, якія сераўысы вядуць які бэкенд. Дадайце тэсты, якія спрычыняюць збой і прызначэнне працы. Следзіце за зростанням і зберагчэннем табелі контрольных пунктав.
Падсумак
Зберагачванне дадзейнаў LangGraph адбываецца пераважна за дапамою «той самыя API для графа, але іншага надзейнога сховішча». Postgres ёсць стандартны выбар для корпоратыўных прыемлів; SQLite і Redis заполняюць спецыяльныя нішы; спецыяльныя зберагачвальнікі выкарыстоўваюцца для абавязковай інфраструктуры. Участка чалавека ў процэсе і вярненне пасля збою не ёсць функцыямі на выбар — гэта прычына існавання механізмаў стварэння контрольных пунктав. Прыłąчыце бэкенд якомога раней, каб перэрывы і продажчэнне работы былі рэальныя ўжо да таго, як агент зустрэнёцца з корыстувальнікамі.
Практычныя інструкцыі для бэкендоў з контрольнымі пунктамі
Ставіцеся да сховішча контрольных пунктав як да основных дадзейнаў прыемлі. Графікі рэзервавання, трэніроўкі вярнення дадзейнаў і правілы зберагачвання павінны быць часткай тых сабе розмов, што і запісы кліяў. Продукт чату, які не можа вярнуць дыялогі пасля рэгіональнага перэключэння, на самай працоўцы не выклікае памяць — ён выклікае надзею.
Планаванне прыемкі павінна фіксаваць пункты контролю за кожным акtyўным потокам ў дзень, сярэдній розмер блока для станоў з велікай колькасцю паведамленняў, а таксама зростанне пад максымальным роўнем паралельнасці. Вялікія пакеты дадзеных, якія зберагаюцца ў стане, значна збільшуць розмер базы дадзеных Postgres задоўга пры тым, як колькасць рэядоў ўсё ўсё не стане трывожной. Лепшая зберагаць вялікія файлы ў сховішчы об’ектаў, а канкрэтныя паказывачы залішаць у стане.
Контроль даступу мае важлівэ значэнне: паводлы для перагляду пунктам контролю не павінен быць той жа адміністратар, які выкарбавае некалязаныя схемы за дапамогою магчымасцей DDL. Неабходна зменяць кранты, пры можлівасці вікорыстоўваць аўтэнтыфікацыю IAM, якую надаюць прадавцы хмарных сервісаў, а таксама пераглядаць, хто можа выканаць запіт SELECT да стану кансэрвацыі. Ідэнтыфікаторы потакоў часта апошньняе павязаны з ідэнтыфікаторамі корыстувачаў; даны, якія выкарбавляюцца, трэба спрыятаць як персональныя даны.
Калі перахоўваецеся з InMemorySaver на Postgres у робочай сістэме, двойная запісь рэдкая разлучна, калі вы пагодзіцеся з тым, што потокі RAM перазначаюцца аднойчы. Запісайце час перабудовы, выкладзіце новы checkpointer і сачыце за колькасцю памылак пасля вярнення працы. Для перэходу тыпу blue/green паказваце, каб оба варыянты вядалі да адной і той жа базы дадзеных, інакш будзе плутанне ў історыі потокаў.
Парацы па тэставанні: запускайце Postgres у CI (або у тымчасовым кантэйнеры), аднойчы выконайце setup(), запрацоўваюце пяць прыкладных фарматаванняя як тэсты інтэграціі і пераканаецеся, што другі працэс можа продыржыць поток, створаны першым. Гэты просты тэст між працэсамі ловіць больш багоў у рэальных умовах, чым будзь-які тэст на памяці.
На завершанне неабяжна научыць адпаведальных працавачаў разліку межу “graph failed” і “checkpointer unreachable”. Сімптамы супадаюць — выклікаецца адказ — але способы ўсунення разныя. Дашборды, якіе аддзеляюць выключэнні прыемлівае программы ад таймаў выключэння базы дадзенаў, зарабляюць гадзіны часу пад час інцыдэнаў.
Ідэнтыфікаторы нітак і корыстнае відношэнне да продукту
Ідэнтыфікаторы нітак — гэта не толькі ключы базы дадзенаў; гэта таксама об’екты продукту. Няхай яны прывязваюцца да сесій корыстнікаў, заявак на падтрымку чы ўпорадкованых ідэнтыфікатораў дакументаў у свядомы спосаб. У інтэрфейсе корыстніка неабяжна адобразыць можлівасць “працягнуць размову” паўтарным выкарыстоўваннем ідэнтыфікатора. Можлівасць “пачаць занова” неабяжна адобразыць стварэнням новага ідэнтыфікатора. Сумешчанне гэтых функцый у аднам ідэнтыфікаторэ стварае проблемы, калі старыя рэзультаты знаходжаць праз новыя чаты.
Для корыстнікаў, якія выкарыстоўваюць калькулятары разных прыстроёў, стабільнасць ідэнтыфікатораў нітак межу кліентамі выкалікае неабяжнае адшукванне з урахоўваннем автэнтыкацыі. Ніколі не прыймайце ідэнтыфікатор нітак, які падае кліент, без пераканання ў яго прыналежнасці. Checkpointers зберагаюць усё, што ў яных падае; яны не контролююць прыналежнасць.
Эвалюяцыя схемы для стану з кампіят-пунктамі
Тыпаваныя словнікі стану развіваюцца. Даадзенне неабяжных поль ёсць простым; перайменаванне чы развінуць польлі наражае на абяранне старыях кампіят-пунктах. Лепш выконваць дадатковыя змены, настаўляць чытальнікі на выкарыстанне значэння за замовчаннем для відсутных ключоў і выкарыстоўваць явныя міграцыі для руйнівных змян. Зберагаеце schema_version у стане і, калі трэба, адлічвайце логіку вузла па яму.
Большыя спісы прыемаў выказваюць патрэбу ў правілах адсортавання: падсумавваеце старэя прыемы ў адно поле, захоўваеце няўжылі прыемы і пераканаецеся, чы можна продаваць працэс пасля адсортавання. Неконтрольваная історыя — гэта найчастэйша прычына, чаму табліцы кампіят-пунктов у Postgres заполняюцца да краю.
Шаблоны перарыву UX
Інтэрфейсы для затверджэння должны паказваць даныя перэрыву, адпраўнае становішча і канечныя тэрміны. Айдэнтыцю апраўніка трэба зберагаць у стане пад прызначэнне працы. Калі тэрміны затверджэння выйшлі, неабходна запускаць механізм нагляду, які анулюець задачу або паўзіць ескалецыю, а не заставляе потокі вечна чакаць. Перзберагачванне дадзеных дазволяе дужа давалі чакаць; правілы продукту должны гэта обмежваць.
Разыгранні для вярнення пасля катастроф
Кожны квартал трэба вярнуць рэзервную копію ў прыготавальны рэжым і практычна запусціць вядомы ідэнтыфікатор потока. Пераканацца, што паведамленні, статус перэрыву і залішкі інструментаў збягаюцца. Документаваць показнікі RPO/RTO для базы дадзеных контрольных пунктав разам з основной базай дадзеных прыкладнага програмнага забезпечэння. Агенты, які памятаюць корыстувачаў, успадковуюць тыя ж апэкты пра неперывнасць, што і сістэмы электранайпісаў.
Зьместкі пра выдатнасць
Пулы з’яўленняў должны быць налаштаваны па колькасці адночасна працюючых элементаў графа, а не толькі працоўнікаў веб-сервісу. Следзіце за патэранам N+1, калі кожны вузол непатрэбна запісвае пункт перапаўнення — налаштавайце вікорыстоўванне інструменту перапаўнення па стандартным налаштаванням версіі LangGraph. Компрэсуйце або выводзіце за межы сістэмы вялікія рэзультаты бінарных інструментаў.
Прымечанні па безпецы
Шыфруйце дыскі, вікорыстоўвайце TLS для з’яўленняў у Postgres, обмежыце меры канектавання ад працоўнікаў да базы дадзенаў, а таксама удаліце секрэты з стану пры пункты перапаўнення, калі інструменты калі-небудзь бачаць нешыфраваныя ключы. Разглядзіце можлівасць вікорыстоўвання проміжных рашэнняў для маскавання дадзенаў на рывень поля. Пры пераглядах выпанення стандатаў будуць задаваць запытанні; падготавіце адповедзі з дыяграмамі трактавання дадзенаў ад карыстальніка → графа → інструменту перапаўнення → дыска.
Этыя оператывныя слоі ператвараюць прыклады коду на сістэму, якой можна даверываць. API-і ў меру простыя; справжнім прызнакам продакшна ёсць правільная організацыя працы всей сістэмы.
Коордынацыя інструментаў перапаўнення з горызонтальным автаскаліваннем
Калі Kubernetes расшыроўвае колькасць працоўнікаў графу з двух падов да двадцяці, кожны пад павінен выступаць у зв’язку з тым самым стойкім чэкпоінтам. Неякія «лепкія» сесіі не патрэбны, калі ідэнтыфікаторы нітак перавозяцца разам з кожным запросам. Базы дадзеных у фармаце Warm pools не должны выкорыстоўваць локальныя файлы SQLite. Перакананні пра стане должны пераканваліцца ў наяўнасці з’язку з базай дадзеных, а не толькі ў жывучасці працоўніка. Падчас развяртання працоўнікі, якія выключаюцца, должны завершыць свою працу або створыць чэкпоінт падчас ўжывання, каб перэрывы заставаліся можлівымі да вярнення. Адрабоўваце тэсты на абранасць, выключаючы пады пад час чакання на interrupt() і пераканваліваючыся ў можлівасць вярнення працы на іншы пад. Які-то такі експерымент пераканвае ў правільнасці архітэктуры больш, чым дзесяткі тэстаў на елементарныя функцыі.
Натыск дакументаў: калі сховішча пунктавых пераглядоў сповалкаюцца, чы хадзібударыя должны зменшыць навантажэння чы стаціявацца у чэргу? Лепш явны прычыненне адказа чы мовчанкавыя варыянты ў памяці, якія не суспільна працуюць у розных падзеях. Аператары ніколі не должны “тэмпаратывна” перайсці на InMemorySaver у працоўным рэжыме, каб “вылечыць затрымку” — такі перэход унічтожае абяцанне продукту.
Спіс пераконтрацыяў пры адзначэнні, што перазберагач нараджання завершаны
Паказваець, чы роўна setup() выканана як частка міграцыі; пулы настроены; асінхронна і сінхронная праця падтрымлена ў співпаўненні з веб-фрэймворкам; адпаведная прыналежнасць ID ніткі гарантуецца; тэставанне паўтарнага запуску пасля перарыву між процэсамі; архіўы восстанавлены ўспэшна аднойчы; задачы з збераганням дакументаваны; адпрацоўка персональных даных пераглянута; паказвачы даступнасці разлічваюць бяговыя прычыны бяд у БД і графіках; інструкцыі з запуску прыяўляюць ID ніткі да інструментаў падтрымкі корыстувача. Калі кожна галочка паказваецца, PostgresSaver (או выбраны вам бэкенд) больш не ёсць фрагментом коду — яна стае часткай гарантый памяці продукту. Даколе гэта не будзе рэалізавана, спрыягайце дэмы як дэмы і не включайце флагі функцый для процэсаў HITL з дугім трыманнем.
Короткая історыя пра паўтарны запуск, які не выйшоў
Адын інжынер, які быў у рэжыме адказу, аднойчы бачыў, як затрымка ў затверджэннях трывала гадзіны. Графік працы быў нормальны; пароль checkpointer перывалілі, а працавальнікі зберагалі старыя кранты. Новыя перывадкі не маглі працаваць, пакуль старыя потокі ўсё яшчэ чыталі даныя. Рашэнням была аднова крантов плюс пункт у інструкцыі: пераканацца ў автэнтыкацыі checkpointer пры пачатку запуску графа падзеяў. Зберагачыце гэты урок разам з фрагментамі PostgresSaver, каб наступная перывалка не павтарылася. Трывалая памэц включае ў сябе таксонамічную памэц — людскія інструкцыі, запісаныя ў карэспандэнціях.
Гэты інцыдэнт таксама дапамогў яснаякі ў паводзе манітарынгу: апавешчэнне пра неудачы автэнтыкаціі checkpointer павінна выдаўацца на адной зусім окрэмай сторанцы, а не ў загальнай катэгорыі 500. Калі памэц являе ся продуктам, тады шлях да базы дадзенаў ёсць залежнасцю нулявага рангу. З такой дысціпліной замена InMemorySaver на PostgresSaver перастае быць страшным міграцыйным процэсам і стае звычайным крокам у пераходзе ад нотбуку да сервісу.