戰(zhàn)：PostgreSQL 持久化與中斷恢復(fù))
1. 為什么手寫(xiě) Loop 終究會(huì)撞上恢復(fù)這堵墻我最早做對(duì)話式 Agent 的時(shí)候和大多數(shù)人一樣寫(xiě)了一個(gè)while True循環(huán)讀用戶輸入、拼 prompt、調(diào)模型、解析工具調(diào)用、執(zhí)行工具、把結(jié)果塞回消息列表、再調(diào)模型直到模型不再請(qǐng)求工具為止。這套東西跑 demo 特別爽幾十行代碼就能讓模型查天氣、算數(shù)學(xué)、讀文件。但只要業(yè)務(wù)稍微真實(shí)一點(diǎn)問(wèn)題就全冒出來(lái)了。最典型的一個(gè)場(chǎng)景用戶讓 Agent 幫忙做一份包含十幾個(gè)步驟的數(shù)據(jù)處理任務(wù)跑到第七步的時(shí)候用戶突然關(guān)掉了瀏覽器或者網(wǎng)絡(luò)抖了一下或者后端服務(wù)重啟了。這時(shí)候你那個(gè)while True循環(huán)里的所有狀態(tài)——已經(jīng)執(zhí)行到哪一步、中間產(chǎn)出了什么、下一步該干嘛——全部隨著進(jìn)程一起蒸發(fā)了。用戶回來(lái)一看得從頭再來(lái)。這種體驗(yàn)在真實(shí)產(chǎn)品里是致命的。手寫(xiě) Loop 的第二個(gè)死穴是人機(jī)協(xié)同Human-in-the-loop。很多任務(wù)需要中途讓人確認(rèn)一下比如我要給這個(gè)客戶發(fā)郵件了你確認(rèn)下內(nèi)容或者這個(gè)數(shù)據(jù)庫(kù)刪除操作要不要執(zhí)行。在裸循環(huán)里你只能靠input()阻塞或者自己造一套狀態(tài)機(jī)把等待確認(rèn)這個(gè)狀態(tài)存到某個(gè)地方。一旦要支持多個(gè)用戶并發(fā)、要支持確認(rèn)后從斷點(diǎn)繼續(xù)你就會(huì)發(fā)現(xiàn)自己其實(shí)是在重新發(fā)明一個(gè)工作流引擎。第三個(gè)問(wèn)題是可觀測(cè)性。手寫(xiě)循環(huán)里每一步的輸入輸出都散落在各種變量和日志里你很難回答這個(gè)任務(wù)到底卡在哪一步了上一步的模型輸出是什么工具調(diào)用參數(shù)對(duì)不對(duì)這類問(wèn)題。排查線上問(wèn)題時(shí)你只能靠翻日志猜。所以當(dāng)我第一次接觸 LangGraph 的時(shí)候最打動(dòng)我的不是它能畫(huà)流程圖而是它把Checkpoint檢查點(diǎn)做成了框架的一等公民。它本質(zhì)上是在告訴你Agent 的執(zhí)行狀態(tài)不應(yīng)該活在內(nèi)存里而應(yīng)該被持久化下來(lái)隨時(shí)可以暫停、恢復(fù)、回放、甚至分叉。這篇文章我就想把這套東西講透——從 LangGraph 的圖結(jié)構(gòu)到 PostgreSQL 做 Checkpoint 存儲(chǔ)再到用 AG-UI 把中斷恢復(fù)的能力暴露到前端跑通一條完整的鏈路。適合讀這篇的人已經(jīng)用 LangChain 或裸 API 寫(xiě)過(guò) Agent、現(xiàn)在被狀態(tài)管理折磨、想上 LangGraph 但被 checkpoint 概念繞暈的開(kāi)發(fā)者。如果你還沒(méi)寫(xiě)過(guò) Agent建議先補(bǔ)一下基礎(chǔ)不然有些坑你體會(huì)不到。2. LangGraph 的圖模型到底和 LangChain 差在哪2.1 從鏈到圖不是換個(gè)名字那么簡(jiǎn)單很多人問(wèn) LangChain 和 LangGraph 的區(qū)別網(wǎng)上答案一大堆但大多停留在一個(gè)是鏈一個(gè)是圖這種廢話層面。我用一句話概括我的理解LangChain 解決的是怎么把組件串起來(lái)LangGraph 解決的是怎么讓一個(gè)有狀態(tài)的流程可控地跑起來(lái)。LangChain 的 Chain 本質(zhì)是線性的、無(wú)狀態(tài)的或者說(shuō)狀態(tài)是隱式傳遞的。你寫(xiě)一個(gè)LLMChain它的輸入輸出是明確的跑完就結(jié)束了。就算你用AgentExecutor它內(nèi)部那個(gè)循環(huán)也是黑盒的你沒(méi)法在中途插手也沒(méi)法把中間狀態(tài)存下來(lái)。LangGraph 的核心抽象是StateGraph。你定義一個(gè) State通常是個(gè) TypedDict 或者 Pydantic 模型然后定義若干個(gè) Node節(jié)點(diǎn)本質(zhì)是函數(shù)接收 State 返回 State 的更新再用 Edge邊把它們連起來(lái)。邊可以是普通的也可以是條件邊conditional edge根據(jù)當(dāng)前 State 決定下一步走哪個(gè)節(jié)點(diǎn)。這個(gè)模型的關(guān)鍵在于State 是顯式聲明的每個(gè) Node 對(duì) State 的修改是可追蹤的。這就為 Checkpoint 提供了基礎(chǔ)——因?yàn)榭蚣苤烂恳徊街?State 長(zhǎng)什么樣它就能把 State 存下來(lái)。2.2 State 與 Reducer狀態(tài)合并的隱藏規(guī)則這里有個(gè)特別容易踩的坑我必須單獨(dú)拎出來(lái)講。LangGraph 里每個(gè) Node 返回的不是完整的 State而是對(duì) State 的部分更新。框架會(huì)把這些更新合并進(jìn)當(dāng)前 State。默認(rèn)的合并策略是覆蓋——你返回{messages: [新消息]}它就直接把messages字段覆蓋掉。但對(duì)話場(chǎng)景里我們想要的是追加不是覆蓋。這時(shí)候就要用到Reducer歸約函數(shù)。LangGraph 提供了Annotated配合add_messages來(lái)聲明某個(gè)字段用追加的方式合并from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] current_step: str task_result: dictadd_messages這個(gè) reducer 不只是簡(jiǎn)單 append它還會(huì)處理消息 ID 去重、處理RemoveMessage這類特殊指令。我見(jiàn)過(guò)太多人因?yàn)闆](méi)加Annotated[list, add_messages]導(dǎo)致每輪對(duì)話歷史都被覆蓋模型永遠(yuǎn)失憶然后花半天時(shí)間懷疑是不是模型的問(wèn)題。提示自定義 reducer 時(shí)函數(shù)簽名是(current_value, new_update) - merged_value。如果你有復(fù)雜的合并邏輯比如字典深合并完全可以自己寫(xiě)一個(gè) reducer 傳進(jìn)去。2.3 條件邊與循環(huán)Agent 的思考-行動(dòng)骨架一個(gè) ReAct 風(fēng)格的 Agent在 LangGraph 里通常長(zhǎng)這樣一個(gè)agent節(jié)點(diǎn)負(fù)責(zé)調(diào)模型一個(gè)tools節(jié)點(diǎn)負(fù)責(zé)執(zhí)行工具然后用條件邊判斷——如果模型輸出了工具調(diào)用就去tools如果模型直接給了最終答案就END。from langgraph.graph import StateGraph, END def should_continue(state: State): last_message state[messages][-1] if last_message.tool_calls: return tools return END builder StateGraph(State) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.set_entry_point(agent) builder.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) builder.add_edge(tools, agent) graph builder.compile()注意tools - agent這條邊它構(gòu)成了一個(gè)循環(huán)。這就是為什么 LangGraph 能表達(dá) Agent 的迭代行為而普通 DAG 不行。這個(gè)循環(huán)也是 Checkpoint 最有價(jià)值的地方——因?yàn)檠h(huán)意味著執(zhí)行到一半是常態(tài)你必須能隨時(shí)把循環(huán)的當(dāng)前狀態(tài)存下來(lái)。3. Checkpoint 機(jī)制中斷恢復(fù)的地基3.1 Checkpoint 到底存了什么很多人以為 Checkpoint 就是存了個(gè)消息列表其實(shí)遠(yuǎn)不止。一個(gè)完整的 Checkpoint 包含channel_values當(dāng)前所有 State 字段的值也就是你的messages、current_step這些。channel_versions每個(gè)字段的版本號(hào)用于并發(fā)控制和增量更新。versions_seen每個(gè)節(jié)點(diǎn)看到的字段版本用來(lái)判斷哪些節(jié)點(diǎn)需要重新執(zhí)行。pending_sends待處理的消息發(fā)送用于 map-reduce 之類的并行場(chǎng)景。除此之外還有一個(gè)CheckpointTuple的概念它把 checkpoint 本身、它的 ID、父 checkpoint ID、以及 metadata比如step、source、writes打包在一起。這個(gè)父子鏈結(jié)構(gòu)非常關(guān)鍵它讓整個(gè)執(zhí)行歷史變成了一棵樹(shù)你可以從任意一個(gè)歷史節(jié)點(diǎn)分叉出去重新跑。3.2 thread_id會(huì)話隔離的鑰匙Checkpoint 是按thread_id隔離的。你可以把它理解成會(huì)話 ID或者任務(wù) ID。同一個(gè) thread_id 下的所有 checkpoint 構(gòu)成一條時(shí)間線graph.get_state(config)拿到的永遠(yuǎn)是這條時(shí)間線上最新的狀態(tài)。config {configurable: {thread_id: user-123-task-456}} result graph.invoke({messages: [(user, 幫我分析這份數(shù)據(jù))]}, config)這里有個(gè)經(jīng)驗(yàn)thread_id 的生成策略要提前想清楚。如果你用用戶 ID 做 thread_id那同一個(gè)用戶的所有對(duì)話會(huì)共享一條時(shí)間線歷史會(huì)無(wú)限增長(zhǎng)如果你用用戶ID 會(huì)話ID那每個(gè)會(huì)話獨(dú)立但跨會(huì)話的上下文就斷了。我的做法是會(huì)話級(jí) thread_id 用 UUID同時(shí)在 State 里存一個(gè) user_id 字段做業(yè)務(wù)關(guān)聯(lián)這樣既隔離了會(huì)話又能按用戶維度查詢。3.3 內(nèi)存 Saver 與持久化 Saver 的分水嶺LangGraph 內(nèi)置了MemorySaver把 checkpoint 存在內(nèi)存里。開(kāi)發(fā)調(diào)試夠用但進(jìn)程一重啟就沒(méi)了而且多實(shí)例部署時(shí)各存各的完全不能用。生產(chǎn)環(huán)境必須換成持久化 Saver。官方支持的有SqliteSaver、PostgresSaver社區(qū)還有 Redis、MongoDB 的實(shí)現(xiàn)。選哪個(gè)我的判斷標(biāo)準(zhǔn)很簡(jiǎn)單場(chǎng)景推薦 Saver理由本地開(kāi)發(fā)、單機(jī) demoMemorySaver零配置重啟即清空單機(jī)小規(guī)模、邊緣部署SqliteSaver單文件無(wú)需額外服務(wù)生產(chǎn)、多實(shí)例、高并發(fā)PostgresSaver事務(wù)保證、并發(fā)安全、可水平擴(kuò)展已有 Redis 基礎(chǔ)設(shè)施社區(qū) Redis Saver復(fù)用現(xiàn)有組件讀寫(xiě)快我選 PostgreSQL 的原因很實(shí)在Agent 的狀態(tài)數(shù)據(jù)天然需要事務(wù)和一致性。你不想看到工具執(zhí)行了但狀態(tài)沒(méi)存上這種臟數(shù)據(jù)吧Postgres 的 ACID 能兜住這個(gè)底。而且 checkpoint 表會(huì)隨著對(duì)話增長(zhǎng)變得很大Postgres 的分區(qū)、索引、歸檔能力都是現(xiàn)成的。4. 用 PostgreSQL 落地 Checkpoint 存儲(chǔ)4.1 環(huán)境準(zhǔn)備與依賴安裝先把依賴裝齊。注意版本LangGraph 迭代很快不同版本 API 有差異pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-poolpsycopg是 psycopg3PostgresSaver 用的是它不是老的psycopg2。這點(diǎn)很多人會(huì)搞混裝錯(cuò)了會(huì)報(bào)連接相關(guān)的奇怪錯(cuò)誤。數(shù)據(jù)庫(kù)這邊你需要一個(gè)可用的 PostgreSQL 實(shí)例本地 Docker 起一個(gè)最省事docker run -d --name langgraph-pg \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBlanggraph \ -p 5432:5432 postgres:164.2 初始化 checkpoint 表結(jié)構(gòu)PostgresSaver 需要幾張表來(lái)存 checkpoint、writes、migrations。官方提供了setup()方法自動(dòng)建表from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://postgres:yourpasswordlocalhost:5432/langgraph?sslmodedisable pool ConnectionPool(conninfoDB_URI, max_size20, kwargs{autocommit: True}) checkpointer PostgresSaver(pool) checkpointer.setup() # 建表只需執(zhí)行一次注意setup()會(huì)創(chuàng)建checkpoints、checkpoint_writes、checkpoint_migrations等表。生產(chǎn)環(huán)境建議用獨(dú)立的 migration 流程管理而不是每次啟動(dòng)都調(diào)setup()避免權(quán)限和并發(fā)問(wèn)題。建完之后你可以連上去看看表結(jié)構(gòu)理解一下數(shù)據(jù)是怎么落的SELECT thread_id, checkpoint_id, parent_checkpoint_id, metadata-step AS step, type FROM checkpoints WHERE thread_id user-123-task-456 ORDER BY checkpoint_id;你會(huì)看到每個(gè) checkpoint 都有parent_checkpoint_id這就是那條時(shí)間線的鏈。metadata里的step告訴你這是第幾步source告訴你是loop節(jié)點(diǎn)執(zhí)行還是input用戶輸入觸發(fā)的。4.3 把 checkpointer 掛到圖上編譯圖的時(shí)候把 checkpointer 傳進(jìn)去整個(gè)圖就具備了持久化能力graph builder.compile(checkpointercheckpointer)之后每次invoke或stream只要 config 里帶了thread_id狀態(tài)就會(huì)自動(dòng)落庫(kù)。這里有個(gè)細(xì)節(jié)checkpointer 是編譯期綁定的不是調(diào)用期。也就是說(shuō)你不能一個(gè)圖實(shí)例用兩個(gè)不同的 checkpointer需要不同存儲(chǔ)就編譯兩個(gè)圖。4.4 連接池與并發(fā)別讓數(shù)據(jù)庫(kù)成為瓶頸生產(chǎn)環(huán)境一定要用連接池。我上面用了ConnectionPoolmax_size根據(jù)你的并發(fā)量調(diào)。有個(gè)坑PostgresSaver 的寫(xiě)入是同步的如果你的圖節(jié)點(diǎn)里有大量并發(fā)checkpoint 寫(xiě)入可能成為瓶頸。我的優(yōu)化經(jīng)驗(yàn)是連接池max_size至少設(shè)為預(yù)期并發(fā)數(shù)的 1.5 倍。給checkpoints表的thread_id建索引setup()默認(rèn)會(huì)建但你要確認(rèn)。定期歸檔老 checkpoint。對(duì)話歷史無(wú)限增長(zhǎng)的話單表幾個(gè) G 是遲早的事?？梢园碿reated_at分區(qū)或者寫(xiě)個(gè)定時(shí)任務(wù)把超過(guò) N 天的 thread 數(shù)據(jù)遷到冷存儲(chǔ)。CREATE INDEX IF NOT EXISTS idx_checkpoints_thread_id ON checkpoints (thread_id, checkpoint_id DESC);5. 中斷恢復(fù)的完整實(shí)現(xiàn)鏈路5.1 用 interrupt 實(shí)現(xiàn)暫停等人確認(rèn)LangGraph 提供了interrupt()函數(shù)在節(jié)點(diǎn)里調(diào)用它就能暫停執(zhí)行把控制權(quán)交還給調(diào)用方。這是實(shí)現(xiàn) Human-in-the-loop 最優(yōu)雅的方式from langgraph.types import interrupt, Command def send_email_node(state: State): draft state[email_draft] # 暫停把草稿拋給前端確認(rèn) decision interrupt({ action: confirm_email, draft: draft }) if decision[approved]: # 真正發(fā)送 send_email(draft) return {status: sent} else: return {status: rejected, feedback: decision.get(feedback)}當(dāng)圖執(zhí)行到這個(gè)節(jié)點(diǎn)時(shí)interrupt()會(huì)拋出一個(gè)特殊信號(hào)LangGraph 把當(dāng)前狀態(tài)存成 checkpoint然后invoke返回。此時(shí)你可以通過(guò)graph.get_state(config)看到state.next指向send_email_node說(shuō)明它卡在這里等人。5.2 恢復(fù)執(zhí)行Command 的正確用法用戶確認(rèn)之后怎么讓它繼續(xù)跑用Command(resume...)from langgraph.types import Command # 用戶點(diǎn)了批準(zhǔn) resumed graph.invoke( Command(resume{approved: True}), config{configurable: {thread_id: user-123-task-456}} )resume的值會(huì)作為interrupt()的返回值注入到節(jié)點(diǎn)里節(jié)點(diǎn)從暫停的那一行繼續(xù)往下執(zhí)行。這里有個(gè)極其容易踩的坑恢復(fù)時(shí)必須用同一個(gè) thread_id而且圖實(shí)例必須是同一個(gè)編譯產(chǎn)物節(jié)點(diǎn)定義一致。如果你改了節(jié)點(diǎn)代碼再恢復(fù)LangGraph 會(huì)嘗試從 checkpoint 恢復(fù)但節(jié)點(diǎn)邏輯變了行為可能不符合預(yù)期。提示interrupt()之前的代碼在恢復(fù)時(shí)會(huì)重新執(zhí)行一遍。所以interrupt()之前不要放有副作用的操作比如已經(jīng)發(fā)了郵件、已經(jīng)扣了款。把副作用放在interrupt()之后或者用冪等設(shè)計(jì)兜底。5.3 從任意歷史點(diǎn)分叉時(shí)間旅行Checkpoint 的父子鏈結(jié)構(gòu)讓你可以做時(shí)間旅行——拿到某個(gè)歷史 checkpoint 的 ID從那里重新跑history list(graph.get_state_history(config)) # history[0] 是最新history[-1] 是最早 target history[3] # 回到第 4 個(gè)狀態(tài) forked graph.invoke( None, # 不傳新輸入從該點(diǎn)繼續(xù) config{configurable: {thread_id: user-123-task-456, checkpoint_id: target.config[configurable][checkpoint_id]}} )這個(gè)能力在調(diào)試時(shí)簡(jiǎn)直是神器。線上出問(wèn)題你可以把那個(gè) thread 的歷史拉出來(lái)從出錯(cuò)前的狀態(tài)重放看看到底哪一步的模型輸出跑偏了。也可以做如果當(dāng)時(shí)用戶選了另一個(gè)選項(xiàng)會(huì)怎樣的假設(shè)分析。5.4 狀態(tài)查詢與前端對(duì)接的接口設(shè)計(jì)前端要展示當(dāng)前卡在哪一步、等什么確認(rèn)需要后端暴露查詢接口。核心就是get_statestate graph.get_state(config) # state.values: 當(dāng)前 State # state.next: 下一步要執(zhí)行的節(jié)點(diǎn)元組 # state.tasks: 當(dāng)前待處理的任務(wù)含 interrupt 信息state.tasks里會(huì)包含 interrupt 拋出的 payload前端拿到就能渲染確認(rèn)彈窗?；謴?fù)時(shí)前端把用戶選擇 POST 回來(lái)后端調(diào)Command(resume...)。這套接口設(shè)計(jì)的關(guān)鍵是把 thread_id 和 checkpoint_id 都透?jìng)鹘o前端前端每次請(qǐng)求帶上后端就能精確定位到該恢復(fù)哪個(gè)狀態(tài)。6. AG-UI 把中斷恢復(fù)搬到前端6.1 AG-UI 解決的是什么問(wèn)題前面講的都是后端能力。但用戶是在瀏覽器里操作的后端暫停了等人確認(rèn)這件事怎么實(shí)時(shí)、結(jié)構(gòu)化地傳到前端傳統(tǒng)做法是前端輪詢或者自己定一套 WebSocket 協(xié)議每個(gè)項(xiàng)目重復(fù)造輪子。AG-UI 是一套面向 Agent 前端的協(xié)議它定義了 Agent 和后端之間的事件流格式——文本增量、工具調(diào)用開(kāi)始/結(jié)束、狀態(tài)更新、中斷請(qǐng)求等等。它和 LangGraph 的配合點(diǎn)在于LangGraph 的 stream 事件可以被適配成 AG-UI 事件前端用統(tǒng)一的協(xié)議消費(fèi)。6.2 事件流從 graph.stream 到前端LangGraph 的stream支持多種模式updates模式會(huì)在每個(gè)節(jié)點(diǎn)執(zhí)行后吐出更新messages模式會(huì)流式吐 tokencustom模式可以吐你自定義的事件。做 AG-UI 對(duì)接時(shí)通常組合使用async for event in graph.astream(input, config, stream_mode[updates, messages, custom]): mode, payload event if mode messages: # 適配成 AG-UI 的 TEXT_MESSAGE_CONTENT yield to_agui_text(payload) elif mode updates: # 節(jié)點(diǎn)級(jí)更新適配成 STATE_DELTA 或 STEP 事件 yield to_agui_step(payload) elif mode custom: # 自定義事件比如工具調(diào)用進(jìn)度 yield to_agui_custom(payload)當(dāng)圖執(zhí)行到interrupt()時(shí)stream 會(huì)結(jié)束當(dāng)前輪次前端收到一個(gè)中斷信號(hào)彈出確認(rèn) UI。用戶操作后前端發(fā)一個(gè)新的請(qǐng)求后端用Command(resume...)繼續(xù) stream。6.3 前端如何渲染等待確認(rèn)狀態(tài)前端拿到 interrupt payload 后需要根據(jù)action類型渲染不同的 UI。我的做法是約定一套 action 到組件的映射action 類型前端組件用戶操作confirm_email郵件預(yù)覽 批準(zhǔn)/拒絕按鈕返回{approved: bool}confirm_delete危險(xiǎn)操作警告彈窗返回{confirmed: bool}select_option選項(xiàng)列表返回{selected: string}input_text文本輸入框返回{text: string}這套映射讓后端可以自由地插入人工確認(rèn)點(diǎn)前端只需要按 action 渲染?；謴?fù)時(shí)把用戶的選擇作為resume值傳回節(jié)點(diǎn)里interrupt()的返回值就是它。6.4 斷線重連前端刷新后怎么找回狀態(tài)這是實(shí)際產(chǎn)品里繞不開(kāi)的問(wèn)題。用戶確認(rèn)彈窗彈出來(lái)結(jié)果他刷新了頁(yè)面彈窗沒(méi)了任務(wù)卡在那。解決辦法是前端啟動(dòng)時(shí)用 thread_id 調(diào)一次狀態(tài)查詢接口后端返回state.tasks如果里面有 pending 的 interrupt前端就重新渲染確認(rèn) UI。app.get(/threads/{thread_id}/state) async def get_thread_state(thread_id: str): config {configurable: {thread_id: thread_id}} state graph.get_state(config) return { values: state.values, next: state.next, interrupts: [ task.interrupts for task in state.tasks if task.interrupts ] }這樣無(wú)論用戶怎么刷新、換設(shè)備只要 thread_id 還在通常存在 URL 或 localStorage狀態(tài)就能恢復(fù)。這也是 Checkpoint 持久化最直接的價(jià)值體現(xiàn)。7. 實(shí)測(cè)中那些文檔不會(huì)告訴你的坑7.1 checkpoint 表膨脹與清理策略跑了一段時(shí)間后你會(huì)發(fā)現(xiàn)checkpoints表漲得飛快。每執(zhí)行一個(gè)節(jié)點(diǎn)就寫(xiě)一條一個(gè) 20 步的任務(wù)就是 20 條記錄加上checkpoint_writes表數(shù)據(jù)量是翻倍的。我的處理方式對(duì)不再活躍的 thread保留最近 N 個(gè) checkpoint老的刪掉。但注意不能隨便刪中間的因?yàn)楦缸渔湐嗔藭?huì)導(dǎo)致時(shí)間旅行失效。要?jiǎng)h就刪整條 thread 的早期記錄或者按時(shí)間歸檔。用checkpointer.delete_thread(thread_id)刪整個(gè) thread這是官方提供的干凈方式。定期跑歸檔任務(wù)把超過(guò) 30 天的 thread 數(shù)據(jù)導(dǎo)出到對(duì)象存儲(chǔ)然后從主表刪除。7.2 恢復(fù)時(shí)節(jié)點(diǎn)重復(fù)執(zhí)行的副作用前面提過(guò)interrupt()之前的代碼會(huì)重跑。我踩過(guò)一個(gè)具體的坑在interrupt()之前調(diào)了一次外部 API 記錄待確認(rèn)日志結(jié)果恢復(fù)時(shí)又記了一次日志里出現(xiàn)兩條。解決辦法是把這類副作用移到interrupt()之后或者用冪等鍵去重。還有一個(gè)更隱蔽的如果你的節(jié)點(diǎn)里有time.sleep()或者耗時(shí)計(jì)算恢復(fù)時(shí)會(huì)重新執(zhí)行導(dǎo)致響應(yīng)變慢。所以節(jié)點(diǎn)設(shè)計(jì)要盡量輕重活拆到獨(dú)立節(jié)點(diǎn)讓 checkpoint 的粒度更細(xì)。7.3 thread_id 設(shè)計(jì)不當(dāng)導(dǎo)致的串話我見(jiàn)過(guò)一個(gè)事故開(kāi)發(fā)時(shí)圖省事用固定的default做 thread_id結(jié)果所有用戶的對(duì)話全串在一起A 用戶看到了 B 用戶的歷史。這在生產(chǎn)環(huán)境是災(zāi)難級(jí)的。正確做法是 thread_id 必須全局唯一且和用戶綁定。我的生成規(guī)則是f{user_id}:{session_uuid}既保證唯一又方便按用戶查詢。同時(shí)在后端做一層校驗(yàn)請(qǐng)求里的 user_id 必須和 thread_id 里的 user_id 匹配防止越權(quán)訪問(wèn)別人的會(huì)話。7.4 版本升級(jí)導(dǎo)致的 checkpoint 不兼容LangGraph 的 checkpoint 格式在不同版本間可能變化。我遇到過(guò)一次升級(jí)后老 checkpoint 反序列化失敗的情況。經(jīng)驗(yàn)是升級(jí) LangGraph 前先在測(cè)試環(huán)境用生產(chǎn)數(shù)據(jù)的副本驗(yàn)證老 checkpoint 能否正常讀取。大版本升級(jí)時(shí)考慮把活躍 thread 跑完再升級(jí)或者寫(xiě)遷移腳本。生產(chǎn)環(huán)境鎖定版本別用latest。7.5 并發(fā)寫(xiě)同一 thread 的競(jìng)態(tài)如果同一個(gè) thread_id 同時(shí)有兩個(gè)請(qǐng)求進(jìn)來(lái)比如用戶手快點(diǎn)了兩次兩個(gè)請(qǐng)求都會(huì)嘗試寫(xiě) checkpoint可能產(chǎn)生沖突。PostgresSaver 內(nèi)部有版本控制但業(yè)務(wù)層面最好加鎖——用 Redis 分布式鎖按 thread_id 加鎖保證同一時(shí)刻只有一個(gè)執(zhí)行流在跑。lock_key flanggraph:lock:{thread_id} if not acquire_lock(lock_key, timeout30): raise HTTPException(409, 該任務(wù)正在處理中請(qǐng)稍候)8. 從裸 Loop 遷移到 LangGraph 的實(shí)操建議如果你手上已經(jīng)有一個(gè)跑得還行的手寫(xiě) Loop想遷移到 LangGraph我的建議是別一次性重寫(xiě)。分三步走第一步先把 Loop 里的狀態(tài)抽出來(lái)定義成顯式的 State 結(jié)構(gòu)。這一步不改邏輯只是把散落的變量收攏。你會(huì)發(fā)現(xiàn)很多隱式狀態(tài)之前根本沒(méi)意識(shí)到。第二步把 Loop 的每一輪拆成節(jié)點(diǎn)。調(diào)模型是一個(gè)節(jié)點(diǎn)執(zhí)行工具是一個(gè)節(jié)點(diǎn)判斷是否繼續(xù)是一個(gè)條件邊。這時(shí)候先不接 checkpointer用 MemorySaver 跑通確認(rèn)行為和原來(lái)一致。第三步接上 PostgresSaver加上 thread_id測(cè)試中斷恢復(fù)。這一步最容易暴露問(wèn)題——你會(huì)發(fā)現(xiàn)有些節(jié)點(diǎn)有副作用、有些狀態(tài)沒(méi)被正確 reducer 合并。逐個(gè)修。遷移完之后你會(huì)獲得幾個(gè)原來(lái)沒(méi)有的能力任務(wù)可以暫?；謴?fù)、可以時(shí)間旅行調(diào)試、可以多人協(xié)同確認(rèn)、可以水平擴(kuò)展因?yàn)闋顟B(tài)在數(shù)據(jù)庫(kù)不在內(nèi)存。這些能力在裸 Loop 里要么做不了要么得自己造一套成本遠(yuǎn)高于直接用框架。最后分享一個(gè)我自己的判斷標(biāo)準(zhǔn)當(dāng)一個(gè) Agent 任務(wù)的執(zhí)行時(shí)間超過(guò) 30 秒或者需要人工介入或者需要跨請(qǐng)求保持上下文就該上 LangGraph Checkpoint 了。低于這個(gè)復(fù)雜度裸 Loop 反而更直接。工具是拿來(lái)解決問(wèn)題的不是拿來(lái)炫技的。 LangGraph Checkpoint 實(shí)戰(zhàn)：PostgreSQL 持久化與中斷恢復(fù)
1. 為什么手寫(xiě) Loop 終究會(huì)撞上恢復(fù)這堵墻我最早做對(duì)話式 Agent 的時(shí)候和大多數(shù)人一樣寫(xiě)了一個(gè)while True循環(huán)讀用戶輸入、拼 prompt、調(diào)模型、解析工具調(diào)用、執(zhí)行工具、把結(jié)果塞回消息列表、再調(diào)模型直到模型不再請(qǐng)求工具為止。這套東西跑 demo 特別爽幾十行代碼就能讓模型查天氣、算數(shù)學(xué)、讀文件。但只要業(yè)務(wù)稍微真實(shí)一點(diǎn)問(wèn)題就全冒出來(lái)了。最典型的一個(gè)場(chǎng)景用戶讓 Agent 幫忙做一份包含十幾個(gè)步驟的數(shù)據(jù)處理任務(wù)跑到第七步的時(shí)候用戶突然關(guān)掉了瀏覽器或者網(wǎng)絡(luò)抖了一下或者后端服務(wù)重啟了。這時(shí)候你那個(gè)while True循環(huán)里的所有狀態(tài)——已經(jīng)執(zhí)行到哪一步、中間產(chǎn)出了什么、下一步該干嘛——全部隨著進(jìn)程一起蒸發(fā)了。用戶回來(lái)一看得從頭再來(lái)。這種體驗(yàn)在真實(shí)產(chǎn)品里是致命的。手寫(xiě) Loop 的第二個(gè)死穴是人機(jī)協(xié)同Human-in-the-loop。很多任務(wù)需要中途讓人確認(rèn)一下比如我要給這個(gè)客戶發(fā)郵件了你確認(rèn)下內(nèi)容或者這個(gè)數(shù)據(jù)庫(kù)刪除操作要不要執(zhí)行。在裸循環(huán)里你只能靠input()阻塞或者自己造一套狀態(tài)機(jī)把等待確認(rèn)這個(gè)狀態(tài)存到某個(gè)地方。一旦要支持多個(gè)用戶并發(fā)、要支持確認(rèn)后從斷點(diǎn)繼續(xù)你就會(huì)發(fā)現(xiàn)自己其實(shí)是在重新發(fā)明一個(gè)工作流引擎。第三個(gè)問(wèn)題是可觀測(cè)性。手寫(xiě)循環(huán)里每一步的輸入輸出都散落在各種變量和日志里你很難回答這個(gè)任務(wù)到底卡在哪一步了上一步的模型輸出是什么工具調(diào)用參數(shù)對(duì)不對(duì)這類問(wèn)題。排查線上問(wèn)題時(shí)你只能靠翻日志猜。所以當(dāng)我第一次接觸 LangGraph 的時(shí)候最打動(dòng)我的不是它能畫(huà)流程圖而是它把Checkpoint檢查點(diǎn)做成了框架的一等公民。它本質(zhì)上是在告訴你Agent 的執(zhí)行狀態(tài)不應(yīng)該活在內(nèi)存里而應(yīng)該被持久化下來(lái)隨時(shí)可以暫停、恢復(fù)、回放、甚至分叉。這篇文章我就想把這套東西講透——從 LangGraph 的圖結(jié)構(gòu)到 PostgreSQL 做 Checkpoint 存儲(chǔ)再到用 AG-UI 把中斷恢復(fù)的能力暴露到前端跑通一條完整的鏈路。適合讀這篇的人已經(jīng)用 LangChain 或裸 API 寫(xiě)過(guò) Agent、現(xiàn)在被狀態(tài)管理折磨、想上 LangGraph 但被 checkpoint 概念繞暈的開(kāi)發(fā)者。如果你還沒(méi)寫(xiě)過(guò) Agent建議先補(bǔ)一下基礎(chǔ)不然有些坑你體會(huì)不到。2. LangGraph 的圖模型到底和 LangChain 差在哪2.1 從鏈到圖不是換個(gè)名字那么簡(jiǎn)單很多人問(wèn) LangChain 和 LangGraph 的區(qū)別網(wǎng)上答案一大堆但大多停留在一個(gè)是鏈一個(gè)是圖這種廢話層面。我用一句話概括我的理解LangChain 解決的是怎么把組件串起來(lái)LangGraph 解決的是怎么讓一個(gè)有狀態(tài)的流程可控地跑起來(lái)。LangChain 的 Chain 本質(zhì)是線性的、無(wú)狀態(tài)的或者說(shuō)狀態(tài)是隱式傳遞的。你寫(xiě)一個(gè)LLMChain它的輸入輸出是明確的跑完就結(jié)束了。就算你用AgentExecutor它內(nèi)部那個(gè)循環(huán)也是黑盒的你沒(méi)法在中途插手也沒(méi)法把中間狀態(tài)存下來(lái)。LangGraph 的核心抽象是StateGraph。你定義一個(gè) State通常是個(gè) TypedDict 或者 Pydantic 模型然后定義若干個(gè) Node節(jié)點(diǎn)本質(zhì)是函數(shù)接收 State 返回 State 的更新再用 Edge邊把它們連起來(lái)。邊可以是普通的也可以是條件邊conditional edge根據(jù)當(dāng)前 State 決定下一步走哪個(gè)節(jié)點(diǎn)。這個(gè)模型的關(guān)鍵在于State 是顯式聲明的每個(gè) Node 對(duì) State 的修改是可追蹤的。這就為 Checkpoint 提供了基礎(chǔ)——因?yàn)榭蚣苤烂恳徊街?State 長(zhǎng)什么樣它就能把 State 存下來(lái)。2.2 State 與 Reducer狀態(tài)合并的隱藏規(guī)則這里有個(gè)特別容易踩的坑我必須單獨(dú)拎出來(lái)講。LangGraph 里每個(gè) Node 返回的不是完整的 State而是對(duì) State 的部分更新。框架會(huì)把這些更新合并進(jìn)當(dāng)前 State。默認(rèn)的合并策略是覆蓋——你返回{messages: [新消息]}它就直接把messages字段覆蓋掉。但對(duì)話場(chǎng)景里我們想要的是追加不是覆蓋。這時(shí)候就要用到Reducer歸約函數(shù)。LangGraph 提供了Annotated配合add_messages來(lái)聲明某個(gè)字段用追加的方式合并from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] current_step: str task_result: dictadd_messages這個(gè) reducer 不只是簡(jiǎn)單 append它還會(huì)處理消息 ID 去重、處理RemoveMessage這類特殊指令。我見(jiàn)過(guò)太多人因?yàn)闆](méi)加Annotated[list, add_messages]導(dǎo)致每輪對(duì)話歷史都被覆蓋模型永遠(yuǎn)失憶然后花半天時(shí)間懷疑是不是模型的問(wèn)題。提示自定義 reducer 時(shí)函數(shù)簽名是(current_value, new_update) - merged_value。如果你有復(fù)雜的合并邏輯比如字典深合并完全可以自己寫(xiě)一個(gè) reducer 傳進(jìn)去。2.3 條件邊與循環(huán)Agent 的思考-行動(dòng)骨架一個(gè) ReAct 風(fēng)格的 Agent在 LangGraph 里通常長(zhǎng)這樣一個(gè)agent節(jié)點(diǎn)負(fù)責(zé)調(diào)模型一個(gè)tools節(jié)點(diǎn)負(fù)責(zé)執(zhí)行工具然后用條件邊判斷——如果模型輸出了工具調(diào)用就去tools如果模型直接給了最終答案就END。from langgraph.graph import StateGraph, END def should_continue(state: State): last_message state[messages][-1] if last_message.tool_calls: return tools return END builder StateGraph(State) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.set_entry_point(agent) builder.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) builder.add_edge(tools, agent) graph builder.compile()注意tools - agent這條邊它構(gòu)成了一個(gè)循環(huán)。這就是為什么 LangGraph 能表達(dá) Agent 的迭代行為而普通 DAG 不行。這個(gè)循環(huán)也是 Checkpoint 最有價(jià)值的地方——因?yàn)檠h(huán)意味著執(zhí)行到一半是常態(tài)你必須能隨時(shí)把循環(huán)的當(dāng)前狀態(tài)存下來(lái)。3. Checkpoint 機(jī)制中斷恢復(fù)的地基3.1 Checkpoint 到底存了什么很多人以為 Checkpoint 就是存了個(gè)消息列表其實(shí)遠(yuǎn)不止。一個(gè)完整的 Checkpoint 包含channel_values當(dāng)前所有 State 字段的值也就是你的messages、current_step這些。channel_versions每個(gè)字段的版本號(hào)用于并發(fā)控制和增量更新。versions_seen每個(gè)節(jié)點(diǎn)看到的字段版本用來(lái)判斷哪些節(jié)點(diǎn)需要重新執(zhí)行。pending_sends待處理的消息發(fā)送用于 map-reduce 之類的并行場(chǎng)景。除此之外還有一個(gè)CheckpointTuple的概念它把 checkpoint 本身、它的 ID、父 checkpoint ID、以及 metadata比如step、source、writes打包在一起。這個(gè)父子鏈結(jié)構(gòu)非常關(guān)鍵它讓整個(gè)執(zhí)行歷史變成了一棵樹(shù)你可以從任意一個(gè)歷史節(jié)點(diǎn)分叉出去重新跑。3.2 thread_id會(huì)話隔離的鑰匙Checkpoint 是按thread_id隔離的。你可以把它理解成會(huì)話 ID或者任務(wù) ID。同一個(gè) thread_id 下的所有 checkpoint 構(gòu)成一條時(shí)間線graph.get_state(config)拿到的永遠(yuǎn)是這條時(shí)間線上最新的狀態(tài)。config {configurable: {thread_id: user-123-task-456}} result graph.invoke({messages: [(user, 幫我分析這份數(shù)據(jù))]}, config)這里有個(gè)經(jīng)驗(yàn)thread_id 的生成策略要提前想清楚。如果你用用戶 ID 做 thread_id那同一個(gè)用戶的所有對(duì)話會(huì)共享一條時(shí)間線歷史會(huì)無(wú)限增長(zhǎng)如果你用用戶ID 會(huì)話ID那每個(gè)會(huì)話獨(dú)立但跨會(huì)話的上下文就斷了。我的做法是會(huì)話級(jí) thread_id 用 UUID同時(shí)在 State 里存一個(gè) user_id 字段做業(yè)務(wù)關(guān)聯(lián)這樣既隔離了會(huì)話又能按用戶維度查詢。3.3 內(nèi)存 Saver 與持久化 Saver 的分水嶺LangGraph 內(nèi)置了MemorySaver把 checkpoint 存在內(nèi)存里。開(kāi)發(fā)調(diào)試夠用但進(jìn)程一重啟就沒(méi)了而且多實(shí)例部署時(shí)各存各的完全不能用。生產(chǎn)環(huán)境必須換成持久化 Saver。官方支持的有SqliteSaver、PostgresSaver社區(qū)還有 Redis、MongoDB 的實(shí)現(xiàn)。選哪個(gè)我的判斷標(biāo)準(zhǔn)很簡(jiǎn)單場(chǎng)景推薦 Saver理由本地開(kāi)發(fā)、單機(jī) demoMemorySaver零配置重啟即清空單機(jī)小規(guī)模、邊緣部署SqliteSaver單文件無(wú)需額外服務(wù)生產(chǎn)、多實(shí)例、高并發(fā)PostgresSaver事務(wù)保證、并發(fā)安全、可水平擴(kuò)展已有 Redis 基礎(chǔ)設(shè)施社區(qū) Redis Saver復(fù)用現(xiàn)有組件讀寫(xiě)快我選 PostgreSQL 的原因很實(shí)在Agent 的狀態(tài)數(shù)據(jù)天然需要事務(wù)和一致性。你不想看到工具執(zhí)行了但狀態(tài)沒(méi)存上這種臟數(shù)據(jù)吧Postgres 的 ACID 能兜住這個(gè)底。而且 checkpoint 表會(huì)隨著對(duì)話增長(zhǎng)變得很大Postgres 的分區(qū)、索引、歸檔能力都是現(xiàn)成的。4. 用 PostgreSQL 落地 Checkpoint 存儲(chǔ)4.1 環(huán)境準(zhǔn)備與依賴安裝先把依賴裝齊。注意版本LangGraph 迭代很快不同版本 API 有差異pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-poolpsycopg是 psycopg3PostgresSaver 用的是它不是老的psycopg2。這點(diǎn)很多人會(huì)搞混裝錯(cuò)了會(huì)報(bào)連接相關(guān)的奇怪錯(cuò)誤。數(shù)據(jù)庫(kù)這邊你需要一個(gè)可用的 PostgreSQL 實(shí)例本地 Docker 起一個(gè)最省事docker run -d --name langgraph-pg \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBlanggraph \ -p 5432:5432 postgres:164.2 初始化 checkpoint 表結(jié)構(gòu)PostgresSaver 需要幾張表來(lái)存 checkpoint、writes、migrations。官方提供了setup()方法自動(dòng)建表from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://postgres:yourpasswordlocalhost:5432/langgraph?sslmodedisable pool ConnectionPool(conninfoDB_URI, max_size20, kwargs{autocommit: True}) checkpointer PostgresSaver(pool) checkpointer.setup() # 建表只需執(zhí)行一次注意setup()會(huì)創(chuàng)建checkpoints、checkpoint_writes、checkpoint_migrations等表。生產(chǎn)環(huán)境建議用獨(dú)立的 migration 流程管理而不是每次啟動(dòng)都調(diào)setup()避免權(quán)限和并發(fā)問(wèn)題。建完之后你可以連上去看看表結(jié)構(gòu)理解一下數(shù)據(jù)是怎么落的SELECT thread_id, checkpoint_id, parent_checkpoint_id, metadata-step AS step, type FROM checkpoints WHERE thread_id user-123-task-456 ORDER BY checkpoint_id;你會(huì)看到每個(gè) checkpoint 都有parent_checkpoint_id這就是那條時(shí)間線的鏈。metadata里的step告訴你這是第幾步source告訴你是loop節(jié)點(diǎn)執(zhí)行還是input用戶輸入觸發(fā)的。4.3 把 checkpointer 掛到圖上編譯圖的時(shí)候把 checkpointer 傳進(jìn)去整個(gè)圖就具備了持久化能力graph builder.compile(checkpointercheckpointer)之后每次invoke或stream只要 config 里帶了thread_id狀態(tài)就會(huì)自動(dòng)落庫(kù)。這里有個(gè)細(xì)節(jié)checkpointer 是編譯期綁定的不是調(diào)用期。也就是說(shuō)你不能一個(gè)圖實(shí)例用兩個(gè)不同的 checkpointer需要不同存儲(chǔ)就編譯兩個(gè)圖。4.4 連接池與并發(fā)別讓數(shù)據(jù)庫(kù)成為瓶頸生產(chǎn)環(huán)境一定要用連接池。我上面用了ConnectionPoolmax_size根據(jù)你的并發(fā)量調(diào)。有個(gè)坑PostgresSaver 的寫(xiě)入是同步的如果你的圖節(jié)點(diǎn)里有大量并發(fā)checkpoint 寫(xiě)入可能成為瓶頸。我的優(yōu)化經(jīng)驗(yàn)是連接池max_size至少設(shè)為預(yù)期并發(fā)數(shù)的 1.5 倍。給checkpoints表的thread_id建索引setup()默認(rèn)會(huì)建但你要確認(rèn)。定期歸檔老 checkpoint。對(duì)話歷史無(wú)限增長(zhǎng)的話單表幾個(gè) G 是遲早的事?？梢园碿reated_at分區(qū)或者寫(xiě)個(gè)定時(shí)任務(wù)把超過(guò) N 天的 thread 數(shù)據(jù)遷到冷存儲(chǔ)。CREATE INDEX IF NOT EXISTS idx_checkpoints_thread_id ON checkpoints (thread_id, checkpoint_id DESC);5. 中斷恢復(fù)的完整實(shí)現(xiàn)鏈路5.1 用 interrupt 實(shí)現(xiàn)暫停等人確認(rèn)LangGraph 提供了interrupt()函數(shù)在節(jié)點(diǎn)里調(diào)用它就能暫停執(zhí)行把控制權(quán)交還給調(diào)用方。這是實(shí)現(xiàn) Human-in-the-loop 最優(yōu)雅的方式from langgraph.types import interrupt, Command def send_email_node(state: State): draft state[email_draft] # 暫停把草稿拋給前端確認(rèn) decision interrupt({ action: confirm_email, draft: draft }) if decision[approved]: # 真正發(fā)送 send_email(draft) return {status: sent} else: return {status: rejected, feedback: decision.get(feedback)}當(dāng)圖執(zhí)行到這個(gè)節(jié)點(diǎn)時(shí)interrupt()會(huì)拋出一個(gè)特殊信號(hào)LangGraph 把當(dāng)前狀態(tài)存成 checkpoint然后invoke返回。此時(shí)你可以通過(guò)graph.get_state(config)看到state.next指向send_email_node說(shuō)明它卡在這里等人。5.2 恢復(fù)執(zhí)行Command 的正確用法用戶確認(rèn)之后怎么讓它繼續(xù)跑用Command(resume...)from langgraph.types import Command # 用戶點(diǎn)了批準(zhǔn) resumed graph.invoke( Command(resume{approved: True}), config{configurable: {thread_id: user-123-task-456}} )resume的值會(huì)作為interrupt()的返回值注入到節(jié)點(diǎn)里節(jié)點(diǎn)從暫停的那一行繼續(xù)往下執(zhí)行。這里有個(gè)極其容易踩的坑恢復(fù)時(shí)必須用同一個(gè) thread_id而且圖實(shí)例必須是同一個(gè)編譯產(chǎn)物節(jié)點(diǎn)定義一致。如果你改了節(jié)點(diǎn)代碼再恢復(fù)LangGraph 會(huì)嘗試從 checkpoint 恢復(fù)但節(jié)點(diǎn)邏輯變了行為可能不符合預(yù)期。提示interrupt()之前的代碼在恢復(fù)時(shí)會(huì)重新執(zhí)行一遍。所以interrupt()之前不要放有副作用的操作比如已經(jīng)發(fā)了郵件、已經(jīng)扣了款。把副作用放在interrupt()之后或者用冪等設(shè)計(jì)兜底。5.3 從任意歷史點(diǎn)分叉時(shí)間旅行Checkpoint 的父子鏈結(jié)構(gòu)讓你可以做時(shí)間旅行——拿到某個(gè)歷史 checkpoint 的 ID從那里重新跑history list(graph.get_state_history(config)) # history[0] 是最新history[-1] 是最早 target history[3] # 回到第 4 個(gè)狀態(tài) forked graph.invoke( None, # 不傳新輸入從該點(diǎn)繼續(xù) config{configurable: {thread_id: user-123-task-456, checkpoint_id: target.config[configurable][checkpoint_id]}} )這個(gè)能力在調(diào)試時(shí)簡(jiǎn)直是神器。線上出問(wèn)題你可以把那個(gè) thread 的歷史拉出來(lái)從出錯(cuò)前的狀態(tài)重放看看到底哪一步的模型輸出跑偏了。也可以做如果當(dāng)時(shí)用戶選了另一個(gè)選項(xiàng)會(huì)怎樣的假設(shè)分析。5.4 狀態(tài)查詢與前端對(duì)接的接口設(shè)計(jì)前端要展示當(dāng)前卡在哪一步、等什么確認(rèn)需要后端暴露查詢接口。核心就是get_statestate graph.get_state(config) # state.values: 當(dāng)前 State # state.next: 下一步要執(zhí)行的節(jié)點(diǎn)元組 # state.tasks: 當(dāng)前待處理的任務(wù)含 interrupt 信息state.tasks里會(huì)包含 interrupt 拋出的 payload前端拿到就能渲染確認(rèn)彈窗?；謴?fù)時(shí)前端把用戶選擇 POST 回來(lái)后端調(diào)Command(resume...)。這套接口設(shè)計(jì)的關(guān)鍵是把 thread_id 和 checkpoint_id 都透?jìng)鹘o前端前端每次請(qǐng)求帶上后端就能精確定位到該恢復(fù)哪個(gè)狀態(tài)。6. AG-UI 把中斷恢復(fù)搬到前端6.1 AG-UI 解決的是什么問(wèn)題前面講的都是后端能力。但用戶是在瀏覽器里操作的后端暫停了等人確認(rèn)這件事怎么實(shí)時(shí)、結(jié)構(gòu)化地傳到前端傳統(tǒng)做法是前端輪詢或者自己定一套 WebSocket 協(xié)議每個(gè)項(xiàng)目重復(fù)造輪子。AG-UI 是一套面向 Agent 前端的協(xié)議它定義了 Agent 和后端之間的事件流格式——文本增量、工具調(diào)用開(kāi)始/結(jié)束、狀態(tài)更新、中斷請(qǐng)求等等。它和 LangGraph 的配合點(diǎn)在于LangGraph 的 stream 事件可以被適配成 AG-UI 事件前端用統(tǒng)一的協(xié)議消費(fèi)。6.2 事件流從 graph.stream 到前端LangGraph 的stream支持多種模式updates模式會(huì)在每個(gè)節(jié)點(diǎn)執(zhí)行后吐出更新messages模式會(huì)流式吐 tokencustom模式可以吐你自定義的事件。做 AG-UI 對(duì)接時(shí)通常組合使用async for event in graph.astream(input, config, stream_mode[updates, messages, custom]): mode, payload event if mode messages: # 適配成 AG-UI 的 TEXT_MESSAGE_CONTENT yield to_agui_text(payload) elif mode updates: # 節(jié)點(diǎn)級(jí)更新適配成 STATE_DELTA 或 STEP 事件 yield to_agui_step(payload) elif mode custom: # 自定義事件比如工具調(diào)用進(jìn)度 yield to_agui_custom(payload)當(dāng)圖執(zhí)行到interrupt()時(shí)stream 會(huì)結(jié)束當(dāng)前輪次前端收到一個(gè)中斷信號(hào)彈出確認(rèn) UI。用戶操作后前端發(fā)一個(gè)新的請(qǐng)求后端用Command(resume...)繼續(xù) stream。6.3 前端如何渲染等待確認(rèn)狀態(tài)前端拿到 interrupt payload 后需要根據(jù)action類型渲染不同的 UI。我的做法是約定一套 action 到組件的映射action 類型前端組件用戶操作confirm_email郵件預(yù)覽 批準(zhǔn)/拒絕按鈕返回{approved: bool}confirm_delete危險(xiǎn)操作警告彈窗返回{confirmed: bool}select_option選項(xiàng)列表返回{selected: string}input_text文本輸入框返回{text: string}這套映射讓后端可以自由地插入人工確認(rèn)點(diǎn)前端只需要按 action 渲染?；謴?fù)時(shí)把用戶的選擇作為resume值傳回節(jié)點(diǎn)里interrupt()的返回值就是它。6.4 斷線重連前端刷新后怎么找回狀態(tài)這是實(shí)際產(chǎn)品里繞不開(kāi)的問(wèn)題。用戶確認(rèn)彈窗彈出來(lái)結(jié)果他刷新了頁(yè)面彈窗沒(méi)了任務(wù)卡在那。解決辦法是前端啟動(dòng)時(shí)用 thread_id 調(diào)一次狀態(tài)查詢接口后端返回state.tasks如果里面有 pending 的 interrupt前端就重新渲染確認(rèn) UI。app.get(/threads/{thread_id}/state) async def get_thread_state(thread_id: str): config {configurable: {thread_id: thread_id}} state graph.get_state(config) return { values: state.values, next: state.next, interrupts: [ task.interrupts for task in state.tasks if task.interrupts ] }這樣無(wú)論用戶怎么刷新、換設(shè)備只要 thread_id 還在通常存在 URL 或 localStorage狀態(tài)就能恢復(fù)。這也是 Checkpoint 持久化最直接的價(jià)值體現(xiàn)。7. 實(shí)測(cè)中那些文檔不會(huì)告訴你的坑7.1 checkpoint 表膨脹與清理策略跑了一段時(shí)間后你會(huì)發(fā)現(xiàn)checkpoints表漲得飛快。每執(zhí)行一個(gè)節(jié)點(diǎn)就寫(xiě)一條一個(gè) 20 步的任務(wù)就是 20 條記錄加上checkpoint_writes表數(shù)據(jù)量是翻倍的。我的處理方式對(duì)不再活躍的 thread保留最近 N 個(gè) checkpoint老的刪掉。但注意不能隨便刪中間的因?yàn)楦缸渔湐嗔藭?huì)導(dǎo)致時(shí)間旅行失效。要?jiǎng)h就刪整條 thread 的早期記錄或者按時(shí)間歸檔。用checkpointer.delete_thread(thread_id)刪整個(gè) thread這是官方提供的干凈方式。定期跑歸檔任務(wù)把超過(guò) 30 天的 thread 數(shù)據(jù)導(dǎo)出到對(duì)象存儲(chǔ)然后從主表刪除。7.2 恢復(fù)時(shí)節(jié)點(diǎn)重復(fù)執(zhí)行的副作用前面提過(guò)interrupt()之前的代碼會(huì)重跑。我踩過(guò)一個(gè)具體的坑在interrupt()之前調(diào)了一次外部 API 記錄待確認(rèn)日志結(jié)果恢復(fù)時(shí)又記了一次日志里出現(xiàn)兩條。解決辦法是把這類副作用移到interrupt()之后或者用冪等鍵去重。還有一個(gè)更隱蔽的如果你的節(jié)點(diǎn)里有time.sleep()或者耗時(shí)計(jì)算恢復(fù)時(shí)會(huì)重新執(zhí)行導(dǎo)致響應(yīng)變慢。所以節(jié)點(diǎn)設(shè)計(jì)要盡量輕重活拆到獨(dú)立節(jié)點(diǎn)讓 checkpoint 的粒度更細(xì)。7.3 thread_id 設(shè)計(jì)不當(dāng)導(dǎo)致的串話我見(jiàn)過(guò)一個(gè)事故開(kāi)發(fā)時(shí)圖省事用固定的default做 thread_id結(jié)果所有用戶的對(duì)話全串在一起A 用戶看到了 B 用戶的歷史。這在生產(chǎn)環(huán)境是災(zāi)難級(jí)的。正確做法是 thread_id 必須全局唯一且和用戶綁定。我的生成規(guī)則是f{user_id}:{session_uuid}既保證唯一又方便按用戶查詢。同時(shí)在后端做一層校驗(yàn)請(qǐng)求里的 user_id 必須和 thread_id 里的 user_id 匹配防止越權(quán)訪問(wèn)別人的會(huì)話。7.4 版本升級(jí)導(dǎo)致的 checkpoint 不兼容LangGraph 的 checkpoint 格式在不同版本間可能變化。我遇到過(guò)一次升級(jí)后老 checkpoint 反序列化失敗的情況。經(jīng)驗(yàn)是升級(jí) LangGraph 前先在測(cè)試環(huán)境用生產(chǎn)數(shù)據(jù)的副本驗(yàn)證老 checkpoint 能否正常讀取。大版本升級(jí)時(shí)考慮把活躍 thread 跑完再升級(jí)或者寫(xiě)遷移腳本。生產(chǎn)環(huán)境鎖定版本別用latest。7.5 并發(fā)寫(xiě)同一 thread 的競(jìng)態(tài)如果同一個(gè) thread_id 同時(shí)有兩個(gè)請(qǐng)求進(jìn)來(lái)比如用戶手快點(diǎn)了兩次兩個(gè)請(qǐng)求都會(huì)嘗試寫(xiě) checkpoint可能產(chǎn)生沖突。PostgresSaver 內(nèi)部有版本控制但業(yè)務(wù)層面最好加鎖——用 Redis 分布式鎖按 thread_id 加鎖保證同一時(shí)刻只有一個(gè)執(zhí)行流在跑。lock_key flanggraph:lock:{thread_id} if not acquire_lock(lock_key, timeout30): raise HTTPException(409, 該任務(wù)正在處理中請(qǐng)稍候)8. 從裸 Loop 遷移到 LangGraph 的實(shí)操建議如果你手上已經(jīng)有一個(gè)跑得還行的手寫(xiě) Loop想遷移到 LangGraph我的建議是別一次性重寫(xiě)。分三步走第一步先把 Loop 里的狀態(tài)抽出來(lái)定義成顯式的 State 結(jié)構(gòu)。這一步不改邏輯只是把散落的變量收攏。你會(huì)發(fā)現(xiàn)很多隱式狀態(tài)之前根本沒(méi)意識(shí)到。第二步把 Loop 的每一輪拆成節(jié)點(diǎn)。調(diào)模型是一個(gè)節(jié)點(diǎn)執(zhí)行工具是一個(gè)節(jié)點(diǎn)判斷是否繼續(xù)是一個(gè)條件邊。這時(shí)候先不接 checkpointer用 MemorySaver 跑通確認(rèn)行為和原來(lái)一致。第三步接上 PostgresSaver加上 thread_id測(cè)試中斷恢復(fù)。這一步最容易暴露問(wèn)題——你會(huì)發(fā)現(xiàn)有些節(jié)點(diǎn)有副作用、有些狀態(tài)沒(méi)被正確 reducer 合并。逐個(gè)修。遷移完之后你會(huì)獲得幾個(gè)原來(lái)沒(méi)有的能力任務(wù)可以暫?；謴?fù)、可以時(shí)間旅行調(diào)試、可以多人協(xié)同確認(rèn)、可以水平擴(kuò)展因?yàn)闋顟B(tài)在數(shù)據(jù)庫(kù)不在內(nèi)存。這些能力在裸 Loop 里要么做不了要么得自己造一套成本遠(yuǎn)高于直接用框架。最后分享一個(gè)我自己的判斷標(biāo)準(zhǔn)當(dāng)一個(gè) Agent 任務(wù)的執(zhí)行時(shí)間超過(guò) 30 秒或者需要人工介入或者需要跨請(qǐng)求保持上下文就該上 LangGraph Checkpoint 了。低于這個(gè)復(fù)雜度裸 Loop 反而更直接。工具是拿來(lái)解決問(wèn)題的不是拿來(lái)炫技的。