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