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