做完十幾個(gè) Agent demo 之后你會(huì)發(fā)現(xiàn)一個(gè)殘酷的事實(shí)在 Jupyter Notebook 里跑得挺順的智能體一上生產(chǎn)就原形畢露。用戶刷新一下頁(yè)面任務(wù)重跑一遍錢扣了兩次凌晨三點(diǎn)模型調(diào)用超時(shí)整個(gè)流程從頭再來(lái)前面寫(xiě)進(jìn)數(shù)據(jù)庫(kù)的狀態(tài)全得重建更頭疼的是人工審批環(huán)節(jié)——Agent 運(yùn)行到一半要等領(lǐng)導(dǎo)點(diǎn)個(gè)按鈕結(jié)果進(jìn)程一重啟圖的狀態(tài)沒(méi)了。LangGraph 斷點(diǎn)恢復(fù)和冪等執(zhí)行就是專門治這三類毛病的。這篇文章我把這套東西從原理到落地一次講透代碼能直接抄。在正式動(dòng)手之前有必要先說(shuō)說(shuō) LangGraph 在整個(gè)技術(shù)棧里的位置。現(xiàn)在搜 LangGraph 教程十個(gè)里有八個(gè)會(huì)糾結(jié)它和 LangChain 的區(qū)別。我的理解很簡(jiǎn)單LangChain 是一把瑞士軍刀里面全是工具函數(shù)、模型封裝、Prompt 模板這些零件而 LangGraph 是裝刀的戰(zhàn)術(shù)背心它關(guān)心的是你身上這些裝備怎么按順序掏出來(lái)、掏到一半被打斷能不能塞回去、事后能不能從某個(gè)位置繼續(xù)掏。它通過(guò)狀態(tài)圖的方式組織 Agent 的執(zhí)行流讓每一步都有跡可循、可停可續(xù)而且把狀態(tài)和檢查點(diǎn)作為一等公民內(nèi)置進(jìn)了框架。本文會(huì)從斷點(diǎn)恢復(fù)的底層機(jī)制講起然后進(jìn)入冪等執(zhí)行這個(gè)工程化繞不開(kāi)的命題最后用一個(gè) FastAPI LangGraph 的實(shí)戰(zhàn)項(xiàng)目把它們串起來(lái)。這個(gè)方案適合正在把 Agent 從 Demo 推向生產(chǎn)的開(kāi)發(fā)者也適合被AI 下地干活折磨得懷疑人生的后端工程師。1. 先搞清楚一個(gè)前提LangGraph 到底在解決什么問(wèn)題1.1 LangGraph 和 LangChain 的本質(zhì)區(qū)別很多人問(wèn) LangChain 和 LangGraph 的面試題怎么答其實(shí)就是一行話的事LangChain 提供了與模型、工具、文檔交互的抽象能力LangGraph 則是把這些能力組織成一張可執(zhí)行、可暫停、可恢復(fù)的狀態(tài)圖。LangChain 的 Chain 也有順序執(zhí)行的邏輯但這種執(zhí)行是一把梭的——鏈一旦啟動(dòng)要么跑完要么失敗重來(lái)中間不保留可以被外部干預(yù)的狀態(tài)。LangGraph 則把執(zhí)行過(guò)程建模成由節(jié)點(diǎn)Node和邊Edge構(gòu)成的有向圖每個(gè)節(jié)點(diǎn)就是一次計(jì)算或一個(gè)工具調(diào)用邊定義了流轉(zhuǎn)規(guī)則。關(guān)鍵差異在于LangGraph 里的圖在每次節(jié)點(diǎn)運(yùn)行前后都會(huì)生成一個(gè)狀態(tài)快照也就是 checkpoint你可以通過(guò)它實(shí)現(xiàn)時(shí)間旅行、條件分支重放以及人工介入。這種差異帶來(lái)的直接感受是用 LangChain 寫(xiě) Agent像在流水線上干活從頭到尾一氣呵成用 LangGraph 寫(xiě) Agent像在玩有存檔的游戲任何時(shí)刻都能存一檔、讀一檔、改一檔再繼續(xù)玩。建議學(xué)習(xí)路徑也很直接先用pip install langgraph跑通官方手冊(cè)中文版里的 ReAct 示例理解節(jié)點(diǎn)、邊、狀態(tài)這三個(gè)核心概念再看本篇文章涉及的狀態(tài)持久化和中斷機(jī)制最后再上手真實(shí)業(yè)務(wù)改造。1.2 斷點(diǎn)恢復(fù)的本質(zhì)把圖執(zhí)行變成可中斷事務(wù)斷點(diǎn)恢復(fù)聽(tīng)起來(lái)很高端本質(zhì)上就是數(shù)據(jù)庫(kù)事務(wù)里 Savepoint 和 Rollback 的思想搬到了 Agent 編排層。LangGraph 在執(zhí)行一個(gè)節(jié)點(diǎn)前會(huì)檢查是否有可恢復(fù)的 checkpoint如果有它不會(huì)從頭跑而是直接恢復(fù)到上次中斷后的節(jié)點(diǎn)繼續(xù)執(zhí)行。這里涉及三個(gè)核心概念thread_id、checkpointer、checkpoint_id。thread_id 是會(huì)話標(biāo)識(shí)LangGraph 用它區(qū)分不同的對(duì)話和任務(wù)同一個(gè) thread 的多次執(zhí)行共享狀態(tài)checkpointer 是存儲(chǔ)后端決定 checkpoint 和狀態(tài)持久化到哪里比如內(nèi)存、SQLite、Postgrescheckpoint_id 則是每次圖執(zhí)行生成的狀態(tài)版本號(hào)相當(dāng)于游戲存檔的時(shí)間戳。你可以這樣理解thread_id 是游戲存檔的文件名checkpointer 是硬盤(pán)checkpoint_id 是存檔的時(shí)間點(diǎn)。一個(gè) Agent 應(yīng)用對(duì)應(yīng)多個(gè) thread_id每個(gè) thread 有自己的一系列 checkpoint_id當(dāng)你用同一個(gè) thread_id 再次調(diào)用圖時(shí)LangGraph 會(huì)找最新的 checkpoint 作為起點(diǎn)。這套機(jī)制天然適合 human-in-the-loop 場(chǎng)景Agent 執(zhí)行到需要人工審批的節(jié)點(diǎn)先暫停把狀態(tài)寫(xiě)好存檔然后告訴外部我需要人來(lái)確認(rèn)人類做出決定后應(yīng)用再用同一個(gè) thread_id 調(diào)用續(xù)跑接口LangGraph 會(huì)從斷點(diǎn)接著走而不是把之前的工作推倒重來(lái)。2. 斷點(diǎn)恢復(fù)的落地姿勢(shì)從圖級(jí)斷點(diǎn)到節(jié)點(diǎn)級(jí)中斷2.1 編譯期斷點(diǎn) interrupt_before / interrupt_after用 LangGraph 實(shí)現(xiàn)斷點(diǎn)最直觀的方式是在編譯圖的時(shí)候指定斷點(diǎn)位置。interrupt_before表示進(jìn)入指定節(jié)點(diǎn)前暫停interrupt_after表示離開(kāi)指定節(jié)點(diǎn)后暫停。這種方式適合流程固定的場(chǎng)景比如必須先審核、后下單。代碼層面非常簡(jiǎn)單from langgraph.graph import StateGraph, START, END # 假設(shè)我們定義了一個(gè)簡(jiǎn)單的 Agent 圖 graph StateGraph(MyState) graph.add_node(plan, plan_node) graph.add_node(execute, execute_node) graph.add_node(review, review_node) graph.add_edge(START, plan) graph.add_edge(plan, execute) graph.add_edge(execute, review) graph.add_edge(review, END) checkpointer SqliteSaver.from_conn_string(checkpoints.db) # 編譯時(shí)指定斷點(diǎn) app graph.compile( checkpointercheckpointer, interrupt_before[review], # 進(jìn)入 review 前停住 interrupt_after[execute], # execute 完成且寫(xiě)入狀態(tài)后停住 )這樣編譯完之后你調(diào)用app.invoke(input, config{configurable: {thread_id: order_001}})圖會(huì)一路執(zhí)行到 execute 節(jié)點(diǎn)然后停在進(jìn)入 review 之前。此時(shí)圖的狀態(tài)保存在 checkpoints.db 里進(jìn)程掛了也不怕。要恢復(fù)只需要再次調(diào)用同一個(gè)圖不需要重新傳完整輸入。很多人這里會(huì)踩坑以為恢復(fù)是重新調(diào)用一次原來(lái)的輸入其實(shí) LangGraph 的邏輯是只要 thread_id 一致且該 thread 存在未完成的執(zhí)行調(diào)用就會(huì)嘗試從最近的 checkpoint 續(xù)跑?？梢韵胂蟪梢粋€(gè)斷點(diǎn)續(xù)傳的過(guò)程恢復(fù)時(shí)只需要傳你需要注入的外部決定比如審批結(jié)果result app.invoke( None, # 或傳審批結(jié)果取決于業(yè)務(wù)邏輯 config{configurable: {thread_id: order_001}} )執(zhí)行會(huì)從interrupt_before指定的節(jié)點(diǎn)重新開(kāi)始先執(zhí)行 review再繼續(xù)后續(xù)邊。編譯期斷點(diǎn)的缺點(diǎn)是死板一旦編譯就固定了如果業(yè)務(wù)流程是動(dòng)態(tài)的這種方式就不夠靈活。2.2 節(jié)點(diǎn)內(nèi)中斷 interrupt() 與 Command(resume)LangGraph 提供了更靈活的動(dòng)態(tài)中斷方式在節(jié)點(diǎn)內(nèi)部調(diào)用interrupt()函數(shù)。這個(gè)函數(shù)的作用相當(dāng)于在任意指定的執(zhí)行點(diǎn)舉手暫停它會(huì)把一個(gè) payload 暴露給外部系統(tǒng)比如前端頁(yè)面等待外部通過(guò)Command(resume...)把決定塞回來(lái)。說(shuō)一個(gè)真實(shí)業(yè)務(wù)Agent 在幫用戶下單前需要確認(rèn)價(jià)格是否可接受。你在確認(rèn)價(jià)格這個(gè)節(jié)點(diǎn)里寫(xiě)from langgraph.types import interrupt, Command def confirm_price_node(state): # 準(zhǔn)備好給用戶看的報(bào)價(jià)信息 payload { order_id: state[order_id], total_price: state[total_price], items: state[items], } # 中斷把 payload 暴露出去等待用戶確認(rèn) decision interrupt(payload) if decision.get(approved): return {status: approved} else: return {status: rejected, reason: decision.get(reason)}interrupt()被調(diào)用后圖的執(zhí)行立刻暫停不會(huì)繼續(xù)往下走。此時(shí)如果你用app.get_state(config)查看狀態(tài)會(huì)發(fā)現(xiàn)圖處于中斷狀態(tài)且可以拿到interrupt里的 payload。外部應(yīng)用比如 FastAPI 接口把這個(gè) payload 展示給用戶用戶點(diǎn)同意或拒絕應(yīng)用把決定作為參數(shù)傳回from langgraph.types import Command # 用戶點(diǎn)了“同意” app.invoke( Command(resume{approved: True}), config{configurable: {thread_id: order_001}} )Command(resume...)會(huì)喚醒中斷的圖把值傳給interrupt()的返回值。執(zhí)行會(huì)接著 confirm_price_node 往下走。這種方式實(shí)現(xiàn)了真正意義上的動(dòng)態(tài)暫停和外部介入LangGraph 官方叫它 human-in-the-loop是斷點(diǎn)恢復(fù)最強(qiáng)的一個(gè)用法。2.3 狀態(tài)檢查、修改與跳過(guò)執(zhí)行除了暫停和恢復(fù)我們經(jīng)常需要在恢復(fù)前修改圖的狀態(tài)或者臨時(shí)跳過(guò)某個(gè)節(jié)點(diǎn)。LangGraph 提供了get_state和update_state兩個(gè)接口用途很像數(shù)據(jù)庫(kù)里的查詢和 UPDATE。config {configurable: {thread_id: order_001}} # 查看當(dāng)前 state 和 checkpoint snapshot app.get_state(config) print(snapshot.values) # 當(dāng)前狀態(tài)字段 print(snapshot.next) # 下一步會(huì)執(zhí)行哪些節(jié)點(diǎn) # 手動(dòng)修改 state app.update_state(config, {customer_name: 張三}, as_nodeplan)修改 state 的值會(huì)生成一個(gè)新的 checkpoint后續(xù)執(zhí)行基于新的狀態(tài)繼續(xù)。這里有個(gè)細(xì)節(jié)update_state時(shí)你可以通過(guò)as_node指定以哪個(gè)節(jié)點(diǎn)的名義寫(xiě)入這會(huì)影響圖的狀態(tài)更新規(guī)則。如果你希望跳過(guò)某個(gè)節(jié)點(diǎn)可以在中斷后手動(dòng) update_state把該節(jié)點(diǎn)要寫(xiě)入的狀態(tài)直接寫(xiě)進(jìn)去再恢復(fù)執(zhí)行即可。日常開(kāi)發(fā)里我通常把查看狀態(tài) 人工修改 恢復(fù)執(zhí)行三個(gè)動(dòng)作做成三個(gè)獨(dú)立 API方便前端根據(jù)業(yè)務(wù)場(chǎng)景自由組合。從實(shí)踐來(lái)看這種設(shè)計(jì)比把所有邏輯都塞進(jìn)一次 invoke 調(diào)用更可靠。3. 冪等執(zhí)行讓 Agent 重復(fù)跑也不會(huì)出事3.1 為什么 LangGraph 不保證冪等斷點(diǎn)恢復(fù)解決了流程中斷的問(wèn)題但引出了下一個(gè)問(wèn)題既然執(zhí)行可以被暫停、恢復(fù)、甚至重放那么同一個(gè)節(jié)點(diǎn)被重復(fù)執(zhí)行怎么辦LangGraph 本身沒(méi)有內(nèi)置冪等機(jī)制。原因很直接框架層面無(wú)法判斷你的工具調(diào)用是不是冪等的。比如一個(gè)節(jié)點(diǎn)里調(diào)用了發(fā)送短信的接口這個(gè)接口天然不是冪等的但另一個(gè)節(jié)點(diǎn)里只是做一次本地計(jì)算重復(fù)執(zhí)行也沒(méi)關(guān)系?？蚣懿恢肋@些所以它選擇把決定權(quán)交給你。但是 LangGraph 提供了實(shí)現(xiàn)冪等的關(guān)鍵信息。每次 invoke 都會(huì)生成一個(gè) runtime run_id每次節(jié)點(diǎn)執(zhí)行后生成 checkpoint_id。你可以把它們當(dāng)作執(zhí)行某段邏輯的流水號(hào)結(jié)合業(yè)務(wù)側(cè)的冪等鍵就能判斷某段副作用是否已經(jīng)執(zhí)行過(guò)。這里強(qiáng)調(diào)一句config里傳入的thread_id是會(huì)話維度不是請(qǐng)求維度同一 session 里有多次 invokerun_id 每次都會(huì)變。所以不要拿 run_id 或 checkpoint_id 當(dāng)業(yè)務(wù)冪等鍵它們只適合做內(nèi)部調(diào)試和狀態(tài)追蹤。3.2 冪等設(shè)計(jì)三件套冪等鍵、副作用登記、唯一約束要讓 Agent 的執(zhí)行變得冪等我的實(shí)戰(zhàn)經(jīng)驗(yàn)是用三件套入口生成冪等鍵、副作用處先登記后執(zhí)行、數(shù)據(jù)庫(kù)加唯一約束。第一步在 Agent 任務(wù)創(chuàng)建時(shí)生成全局唯一的冪等鍵這個(gè)鍵要貫穿整個(gè)執(zhí)行流通常叫run_id但注意是自己生成的業(yè)務(wù) run_id不是 LangGraph 內(nèi)部的 run?？梢苑旁?state 最外層一路往下傳。第二步在節(jié)點(diǎn)執(zhí)行副作用發(fā)消息、扣積分、調(diào)用外部下單接口之前先往 operation_log 表插一條記錄字段包括 run_id、操作類型、操作參數(shù)、狀態(tài)。如果插入的時(shí)候拋唯一約束沖突說(shuō)明同一 run 下這個(gè)操作已經(jīng)執(zhí)行過(guò)直接跳過(guò)副作用邏輯并復(fù)用上次的結(jié)果。第三步在數(shù)據(jù)庫(kù)里給(run_id, op_key)建立唯一索引。這里的 op_key 可以是通知用戶扣減庫(kù)存這樣的操作標(biāo)識(shí)也可以是 action 名稱加參數(shù)哈希。這樣即使代碼邏輯漏判了數(shù)據(jù)庫(kù)也會(huì)攔住第二次執(zhí)行。落地到 LangGraph 節(jié)點(diǎn)里基礎(chǔ)設(shè)施可以做成一個(gè)裝飾器def idempotent_node(op_key): def decorator(func): def wrapper(state): run_id state[run_id] log lookup_operation(run_id, op_key) if log: return log[result] # 已執(zhí)行過(guò)直接返回緩存結(jié)果 result func(state) # 首次執(zhí)行副作用 record_operation(run_id, op_key, result) # 寫(xiě)執(zhí)行記錄 return result return wrapper return decorator idempotent_node(deduct_inventory) def deduct_inventory_node(state): # 真正的扣減邏輯 return {inventory_left: state[inventory] - 1}這套方案的巧妙之處在于它把冪等從框架層下沉到了業(yè)務(wù)層。無(wú)論 LangGraph 怎么重放節(jié)點(diǎn)、怎么斷點(diǎn)續(xù)跑只要 run_id 不變?nèi)魏胃弊饔枚紩?huì)被數(shù)據(jù)庫(kù)的唯一約束擋住。3.3 工具調(diào)用層的冪等LangGraph 里最常見(jiàn)的副作用集中在工具調(diào)用上。函數(shù)調(diào)用tool calling模型會(huì)給每個(gè)工具調(diào)用生成一個(gè)唯一的tool_call_id這個(gè) ID 在單輪對(duì)話內(nèi)是唯一的。但問(wèn)題是如果圖執(zhí)行重放同一個(gè)工具調(diào)用可能會(huì)被再次觸發(fā)LangGraph 自帶的 ToolNode 會(huì)再次執(zhí)行該工具。好在 LangGraph 的 ToolNode 有內(nèi)置的消息去重機(jī)制。當(dāng)你用langgraph.prebuilt.ToolNode時(shí)如果某個(gè)tool_call_id已經(jīng)出現(xiàn)在歷史消息里ToolNode 會(huì)直接使用歷史結(jié)果而不會(huì)再次執(zhí)行工具。這個(gè)行為依賴于 checkpointer 保存的消息列表。所以只要你的圖啟用了 checkpointer并且工具節(jié)點(diǎn)用的是官方 ToolNode一定程度上已經(jīng)具備按 tool_call_id 去重的能力。但如果你沒(méi)用 ToolNode而是自己寫(xiě)節(jié)點(diǎn)處理工具調(diào)用那就必須自己實(shí)現(xiàn)冪等。我的做法是在工具執(zhí)行前檢查當(dāng)前 tool_call_id 是否已經(jīng)在狀態(tài)里的 tool_results 字段中出現(xiàn)過(guò)出現(xiàn)過(guò)就直接拿結(jié)果沒(méi)出現(xiàn)過(guò)才執(zhí)行真正的調(diào)用。def smart_tool_executor(state): messages state[messages] last_ai_msg messages[-1] results [] for tool_call in last_ai_msg.tool_calls: existed state[tool_results].get(tool_call[id]) if existed is not None: results.append(existed) # 復(fù)用歷史結(jié)果 continue result real_executor.invoke(tool_call) # 真正執(zhí)行 state[tool_results][tool_call[id]] result results.append(result) return {tool_results: state[tool_results], messages: results}這套邏輯在你需要給工具調(diào)用加緩存、加審計(jì)、加限流的場(chǎng)景下更實(shí)用官方 ToolNode 的自動(dòng)去重就不夠用了。特別是當(dāng)你對(duì)接的資金、積分系統(tǒng)有自己一套冪等憑證體系時(shí)用 tool_call_id 做聯(lián)動(dòng)往往比重新建一套更直接。4. 落地實(shí)戰(zhàn)FastAPI LangGraph 的生產(chǎn)級(jí)斷點(diǎn)恢復(fù)服務(wù)4.1 架構(gòu)與存儲(chǔ)選型從 SQLite 到 Postgres斷點(diǎn)恢復(fù)依賴于 checkpointer所以選什么存儲(chǔ)就決定了你能恢復(fù)到什么程度以及能扛多大并發(fā)。我把常用三個(gè)存儲(chǔ)后端放在一起對(duì)比根據(jù)部署規(guī)模直接挑存儲(chǔ)后端典型場(chǎng)景并發(fā)能力生產(chǎn)可用性備注MemorySaverDemo、本地調(diào)試單進(jìn)程單線程低進(jìn)程重啟狀態(tài)全丟慎用于生產(chǎn)SqliteSaver單機(jī)小規(guī)模服務(wù)受限于單機(jī)中注意線程鎖和文件鎖配套代碼簡(jiǎn)單門檻低PostgresSaver多副本、高可用環(huán)境高高是生產(chǎn)推薦方案需要數(shù)據(jù)庫(kù)連接池和遷移腳本單機(jī)場(chǎng)景直接上 SQLite 就夠了連接字符串傳一個(gè)路徑即可。多副本部署、要做負(fù)載均衡的話建議直接上 Postgres。LangGraph 提供了 AsyncPostgresSaver異步接口配合 FastAPI 的 async 路由非常順滑。需要強(qiáng)調(diào)一點(diǎn)postgres_saver 使用前必須創(chuàng)建表結(jié)構(gòu)官方提供了checkpoint_ddl腳本不要用 SQLite 的表結(jié)構(gòu)去 Postgres 里跑字段類型和索引都對(duì)不上踩一次坑少說(shuō)浪費(fèi)半小時(shí)。4.2 核心代碼構(gòu)建帶審批的人類介入 Agent用 FastAPI LangChain LangGraph 組合做一個(gè)審批后通知的 Agent。業(yè)務(wù)路徑是創(chuàng)建任務(wù) - 模擬扣減積分 - 人工審批 - 審批通過(guò)后發(fā)通知。重點(diǎn)演示斷點(diǎn)恢復(fù)與冪等抑制。# app.py —— 完整示例骨架 import os from typing import TypedDict, Annotated from fastapi import FastAPI from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.types import interrupt, Command import sqlite3 class AgentState(TypedDict): run_id: str user_id: str points_deducted: bool approval: str notified: bool conn sqlite3.connect(agent_state.db, check_same_threadFalse) conn.execute( CREATE TABLE IF NOT EXISTS operation_log ( run_id TEXT, op_key TEXT, result TEXT, PRIMARY KEY (run_id, op_key) ) ) conn.commit() checkpointer SqliteSaver(conn)注意連接池的check_same_threadFalse默認(rèn)的 SQLite 連接不能跨線程用FastAPI 是并發(fā)模型不關(guān)掉會(huì)各種報(bào) thread error。下面定義三個(gè)節(jié)點(diǎn)。第一個(gè)節(jié)點(diǎn)扣積分有工程量第二個(gè)節(jié)點(diǎn)做人工審批中斷第三個(gè)節(jié)點(diǎn)發(fā)通知冪等保護(hù)。def deduct_points_node(state: AgentState): run_id state[run_id] # 冪等查操作日志 row conn.execute( SELECT result FROM operation_log WHERE run_id? AND op_key?, (run_id, deduct_points), ).fetchone() if row: return {points_deducted: True} # 已扣過(guò)不重復(fù)扣 # 真正扣減積分的代碼這里用 print 模擬 print(f真實(shí)扣減積分user{state[user_id]}) conn.execute( INSERT INTO operation_log (run_id, op_key, result) VALUES (?, ?, ?), (run_id, deduct_points, ok), ) conn.commit() return {points_deducted: True} def approval_node(state: AgentState): decision interrupt({ message: 是否允許扣減積分并發(fā)送通知, user_id: state[user_id], run_id: state[run_id], }) return {approval: decision.get(decision, denied)} def notify_node(state: AgentState): if state[approval] ! approved: return {notified: False} run_id state[run_id] row conn.execute( SELECT result FROM operation_log WHERE run_id? AND op_key?, (run_id, send_notify), ).fetchone() if row: return {notified: True} print(f真實(shí)發(fā)送通知user{state[user_id]}) conn.execute( INSERT INTO operation_log (run_id, op_key, result) VALUES (?, ?, ?), (run_id, send_notify, ok), ) conn.commit() return {notified: True}構(gòu)建圖并編譯builder StateGraph(AgentState) builder.add_node(deduct, deduct_points_node) builder.add_node(approval, approval_node) builder.add_node(notify, notify_node) builder.add_edge(START, deduct) builder.add_edge(deduct, approval) builder.add_edge(approval, notify) builder.add_edge(notify, END) app builder.compile(checkpointercheckpointer)FastAPI 路由部分提供三個(gè)接口from fastapi import FastAPI, HTTPException server FastAPI() server.post(/runs) def create_run(user_id: str): import uuid run_id str(uuid.uuid4()) # 啟動(dòng)圖執(zhí)行運(yùn)行到 approval 節(jié)點(diǎn)會(huì)自動(dòng)中斷 app.invoke( {run_id: run_id, user_id: user_id}, config{configurable: {thread_id: run_id}}, ) return {run_id: run_id} server.get(/runs/{run_id}/state) def get_run_state(run_id: str): snap app.get_state({configurable: {thread_id: run_id}}) return { status: snap.next, values: snap.values, interrupts: [i.payload for i in snap.interrupts] if snap.interrupts else [], } server.post(/runs/{run_id}/resume) def resume_run(run_id: str, decision: str): snap app.get_state({configurable: {thread_id: run_id}}) if not snap.interrupts: raise HTTPException(status_code400, detail當(dāng)前狀態(tài)不可恢復(fù)) app.invoke( Command(resume{decision: decision}), config{configurable: {thread_id: run_id}}, ) return {status: resumed}這樣前后端就能完成創(chuàng)建任務(wù) - 查詢審批信息 - 審批 - 恢復(fù)執(zhí)行。整個(gè)生命周期里SQLite 里的 checkpoint 保存了每一步狀態(tài)operation_log 確保了扣減積分和發(fā)通知這兩個(gè)副作用不會(huì)被重復(fù)執(zhí)行。4.3 完整調(diào)用流程演示用 curl 走一遍完整流程你會(huì)很直觀看到斷點(diǎn)恢復(fù)和冪等是怎么協(xié)同的。創(chuàng)建一次任務(wù)圖會(huì)一路跑到 approval 節(jié)點(diǎn)然后自動(dòng)暫停curl -X POST http://localhost:8000/runs?user_idu_100 # 返回 {run_id: abc-123}查詢狀態(tài)你會(huì)看到interrupts里有審批的 payloadnext指向 approval 之后要執(zhí)行的下一個(gè)節(jié)點(diǎn)curl http://localhost:8000/runs/abc-123/state人工審批通過(guò)恢復(fù)執(zhí)行curl -X POST http://localhost:8000/runs/abc-123/resume \ -H Content-Type: application/json \ -d {decision: approved}此刻的關(guān)鍵點(diǎn)來(lái)了如果你再次手動(dòng)調(diào)用扣積分節(jié)點(diǎn)比如誤操作operation_log 里已經(jīng)存在(abc-123, deduct_points)這條記錄節(jié)點(diǎn)會(huì)直接返回{points_deducted: True}不會(huì)真的扣第二次積分。同理send_notify也不會(huì)重復(fù)發(fā)通知。這個(gè)就是業(yè)務(wù)層的冪等兜底。如果把整條鏈路里的每一步都記錄到數(shù)據(jù)庫(kù)你甚至可以做重放審計(jì)——某個(gè) run 到底扣了多少次積分、哪一步是冪等跳過(guò)的、哪一步真實(shí)執(zhí)行了一查便知。5. 實(shí)戰(zhàn)中的坑與排查經(jīng)驗(yàn)5.1 斷點(diǎn)恢復(fù)常見(jiàn)的報(bào)錯(cuò)與解決辦法工程落地會(huì)遇到各種邊界情況我把自己常遇到的幾個(gè)問(wèn)題整理成了速查表報(bào)錯(cuò)/現(xiàn)象根因處理方式Checkpointer required圖里使用了 interrupt 但在 compile 時(shí)沒(méi)傳 checkpointercompile(checkpointer...)恢復(fù)調(diào)用不生效從頭開(kāi)始執(zhí)行thread_id 不一致或使用了不同的 config key確認(rèn) token 不變檢查 config 拼寫(xiě)Cannot update non-existent node...update_state 時(shí) as_node 傳了一個(gè)圖里不存在的節(jié)點(diǎn)名查看圖節(jié)點(diǎn)列表確保 as_node 合法resume 后狀態(tài)不是預(yù)期值Command(resume) 傳參結(jié)構(gòu)不對(duì)檢查 interrupt() 返回值的接收語(yǔ)義SQLitedatabase is locked多線程并發(fā)寫(xiě)同一個(gè) SQLite 文件開(kāi)啟 WAL 模式或改用 PostgresSaver恢復(fù)后節(jié)點(diǎn)重復(fù)執(zhí)行業(yè)務(wù)副作用沒(méi)有做冪等保護(hù)用 3.2 節(jié)的三件套去攔截SQLite 的鎖是最常見(jiàn)的坑尤其是 FastAPI 多 worker 部署時(shí)。建議在初始化連接后執(zhí)行PRAGMA journal_modeWAL;能明顯減少并發(fā)讀寫(xiě)的鎖沖突。但說(shuō)到底多 worker 部署就別用 SQLite 了上 PostgresSaver 是正道。5.2 冪等沒(méi)生效的幾個(gè)隱蔽原因冪等邏輯看著簡(jiǎn)單實(shí)際有幾種隱蔽情況會(huì)導(dǎo)致失效。最常見(jiàn)的是冪等鍵沒(méi)有貫穿整個(gè)調(diào)用鏈。比如你的 run_id 在 create_run 接口生成了但某個(gè)工具節(jié)點(diǎn)內(nèi)部又自己 new 了一個(gè) uuid那節(jié)點(diǎn)側(cè)查 operation_log 時(shí)主鍵永遠(yuǎn)對(duì)不上每次都會(huì)執(zhí)行真實(shí)副作用。建議把冪等鍵放到 state 的頂層字段節(jié)點(diǎn)里只管 read不許 write。第二個(gè)隱蔽原因是副作用執(zhí)行成功但記錄寫(xiě)入失敗。比如真實(shí)扣減積分接口調(diào)用成功了但 operation_log 的 INSERT 因?yàn)槟撤N異常沒(méi)提交此時(shí) Agent 拋錯(cuò)重試業(yè)務(wù)邏輯發(fā)現(xiàn)日志里沒(méi)有記錄又扣了一次。解決辦法是把副作用執(zhí)行和副作用登記放在同一個(gè)事務(wù)邊界里最好的方式是把扣減積分與記錄日志放到同一個(gè)數(shù)據(jù)庫(kù)事務(wù)要么都成功要么都失敗。第三個(gè)隱蔽原因是參數(shù)變化導(dǎo)致的冪等繞過(guò)。比如 op_key 只取了操作名但操作里包含了金額參數(shù)第一次扣 100第二次改成扣 50唯一約束認(rèn)為這是同一條記錄直接跳過(guò)了扣 50 的請(qǐng)求。這就要求 op_key 的設(shè)計(jì)要包含關(guān)鍵的參數(shù)指紋一般是用action hashlib.md5(sorted_params)。5.3 生產(chǎn)部署檢查清單最后給一份我每次上線 Agent 服務(wù)前都會(huì)過(guò)一遍的清單照著做能少走很多彎路生產(chǎn)環(huán)境 checkpointer 是否選擇了 PostgresSaver并創(chuàng)建了正確的 checkpoint 表所有有外部副作用的節(jié)點(diǎn)是否都經(jīng)過(guò)冪等裝飾器或等效邏輯保護(hù)operation_log表是否建了(run_id, op_key)唯一索引關(guān)鍵工具調(diào)用是否用 tool_call_id 做了去重或緩存恢復(fù)接口是否做了并發(fā)控制避免同一個(gè) thread_id 被兩個(gè)人同時(shí) resume是否對(duì)get_state、update_state做了操作審計(jì)方便排查人工干預(yù)的記錄是否有兜底超時(shí)機(jī)制比如 Agent 長(zhǎng)期處于中斷狀態(tài)或恢復(fù)失敗時(shí)有守護(hù)任務(wù)負(fù)責(zé)清理或告警關(guān)于第 4 條里提到的并發(fā)控制簡(jiǎn)單做法是在恢復(fù)接口層加一把 Redis 鎖鎖的 key 是resume:{thread_id}保證同一時(shí)刻只有一個(gè) resume 請(qǐng)求真正驅(qū)動(dòng)圖繼續(xù)執(zhí)行。否則兩個(gè)請(qǐng)求同時(shí)Command(resume...)狀態(tài)會(huì)亂成一鍋粥。6. 斷點(diǎn)狀態(tài)設(shè)計(jì)與冪等鍵命名的一些心得這里講一個(gè)容易被忽略的細(xì)節(jié)斷點(diǎn)恢復(fù)時(shí)interrupt()的返回值本質(zhì)上是從Command(resume)里透?jìng)鬟^(guò)來(lái)的它不會(huì)自動(dòng)做校驗(yàn)。如果你在審批節(jié)點(diǎn)里期望收到一個(gè) dict但恢復(fù)端傳了字符串節(jié)點(diǎn)代碼可能直接報(bào)TypeError。所以我習(xí)慣在 interrupt 節(jié)點(diǎn)里加一層簡(jiǎn)單的 schema 校驗(yàn)比如用 Pydantic 解析解析失敗就拋一個(gè)自定義異常FastAPI 層捕獲后返回 400。這能避免生產(chǎn)環(huán)境出現(xiàn)匪夷所思的恢復(fù)錯(cuò)誤。冪等鍵命名同樣有講究。業(yè)界慣例會(huì)區(qū)分request_id客戶端發(fā)起請(qǐng)求時(shí)生成的 ID用于端到端全鏈路追蹤。idempotency_key專門用于冪等控制的業(yè)務(wù)鍵同一業(yè)務(wù)動(dòng)作多次重試時(shí)保持不變。run_idAgent 任務(wù)內(nèi)部流轉(zhuǎn)用的標(biāo)識(shí)可以就是idempotency_key但注意它不是 LangGraph 框架的 runtime run_id。我通常直接讓入口生成的 UUID 同時(shí)充當(dāng)request_id、idempotency_key和thread_id三合一減少概念數(shù)量降低溝通成本。你可以在日志里同時(shí)打印這個(gè)值和 LangGraph 內(nèi)部的 checkpoint_id方便追蹤狀態(tài)對(duì)應(yīng)關(guān)系。# 日志示例 [RUN abc-123] checkpoint 9f2e... 到達(dá)人工審批 [RUN abc-123] checkpoint 9f2e... 恢復(fù)執(zhí)行審批結(jié)果 approved [RUN abc-123] 冪等跳過(guò) send_notify原因已執(zhí)行最后再分享一個(gè)我在實(shí)戰(zhàn)中堅(jiān)持的習(xí)慣所有 LangGraph 節(jié)點(diǎn)都寫(xiě)成純函數(shù)風(fēng)格不直接操作外部全局狀態(tài)只通過(guò) state 的輸入輸出做數(shù)據(jù)流轉(zhuǎn)。副作用統(tǒng)一收斂到獨(dú)立的工具層或服務(wù)層節(jié)點(diǎn)只做編排。這樣做的好處是當(dāng)你要加斷點(diǎn)、加冪等、加審計(jì)時(shí)改動(dòng)面非常小每個(gè)節(jié)點(diǎn)就像積木一樣可以自由插拔組合。這套理念配合 LangGraph 的 checkpointer基本能滿足絕大多數(shù)生產(chǎn)級(jí) Agent 服務(wù)的需求。