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