給 Agent 加超時(shí)重試本身不難難的是每次執(zhí)行失敗之后怎么把現(xiàn)場(chǎng)干干凈凈地收掉。這個(gè)教訓(xùn)我是在線上被真實(shí)流量教育過(guò)之后才徹底想明白的。今天把整個(gè)設(shè)計(jì)和踩坑過(guò)程整理出來(lái)希望能幫你少走一段彎路。1. 項(xiàng)目背景一個(gè) AI Agent 的穩(wěn)定性改造先交代一下背景。我負(fù)責(zé)的項(xiàng)目是一個(gè)基于 LangChain FastAPI 的 AI Agent 服務(wù)核心能力是讓大模型自主完成多步任務(wù)比如查詢數(shù)據(jù)庫(kù)、調(diào)用內(nèi)部 API、生成報(bào)表、發(fā)通知甚至跨系統(tǒng)編排動(dòng)作。外層接口是同步 HTTP內(nèi)部走的是循環(huán)式 Agent 執(zhí)行LLM 推理 → 工具調(diào)用 → 結(jié)果回填 → 再次推理直到模型判斷任務(wù)結(jié)束。項(xiàng)目上線初期穩(wěn)定運(yùn)行但流量一上來(lái)問(wèn)題就暴露了。典型癥狀包括用戶請(qǐng)求偶爾 504、LLM 服務(wù)偶發(fā)超時(shí)導(dǎo)致整個(gè)請(qǐng)求卡死、工具調(diào)用失敗后 Agent 像“失憶”一樣重復(fù)做無(wú)用功。于是我們啟動(dòng)了一輪穩(wěn)定性改造核心目標(biāo)就兩個(gè)字超時(shí)和重試。聽(tīng)起來(lái)非常常規(guī)做起來(lái)也確實(shí)能解決一部分問(wèn)題但改完之后真正折磨我的是第三個(gè)字狀態(tài)清理。這里也順便說(shuō)清楚這篇文章適合誰(shuí)看。如果你正在做 AI Agent 相關(guān)的工程化落地尤其是涉及多輪工具調(diào)用、狀態(tài)持久化、甚至異步任務(wù)編排的這篇文章提到的方案和坑會(huì)很有參考價(jià)值。如果你只是調(diào)通了一個(gè)基于 LangChain 的 demo對(duì)超時(shí)重試的認(rèn)知還停留在 requests 庫(kù)加個(gè) timeout 參數(shù)那更建議從頭讀一遍因?yàn)槟氵t早會(huì)踩到同樣的坑。2. 超時(shí)與重試的落地實(shí)現(xiàn)2.1 三層超時(shí)設(shè)計(jì)先說(shuō)說(shuō)超時(shí)。這是最容易做錯(cuò)的地方很多人給 Agent 加超時(shí)就是在 HTTP 客戶端上設(shè)置一下或者給 FastAPI 接口加個(gè)超時(shí)中間件但實(shí)際上 Agent 的超時(shí)體系是分層的至少拆成三層才夠用。第一層是網(wǎng)絡(luò)請(qǐng)求超時(shí)。這是最底層的針對(duì) Agent 依賴的所有外部服務(wù)LLM 供應(yīng)商 API、數(shù)據(jù)庫(kù)連接、內(nèi)部工具服務(wù)的 HTTP 接口。這里我統(tǒng)一走的是 httpx 或 aiohttp 的 timeout 配置連接超時(shí)一般給 5 秒讀超時(shí)根據(jù)服務(wù)特點(diǎn)差異化配置。LLM 調(diào)用這類“慢服務(wù)”讀超時(shí)給 60 秒因?yàn)榇竽Ｐ土魇捷敵霰緛?lái)就慢內(nèi)部 API 讀超時(shí)給 10 秒超過(guò)這個(gè)閾值基本就是服務(wù)有問(wèn)題了。第二層是動(dòng)作級(jí)超時(shí)。這一層很多人會(huì)忽略。Agent 的“一個(gè)動(dòng)作”不只是調(diào)用一次外部 API它包含一次完整的工具調(diào)用周期LLM 生成工具參數(shù) → 執(zhí)行工具 → 返回結(jié)果給 LLM。這個(gè)周期里任何一個(gè)環(huán)節(jié)都可能卡住尤其是工具本身內(nèi)部有重試邏輯的時(shí)候整個(gè)動(dòng)作可能被拖到幾分鐘。所以我會(huì)給單個(gè)工具動(dòng)作設(shè)置獨(dú)立超時(shí)稱為 action_timeout默認(rèn) 30 秒。第三層是會(huì)話級(jí)超時(shí)。這是最高層的兜底對(duì)應(yīng)整個(gè) Agent 任務(wù)的總執(zhí)行時(shí)長(zhǎng)?，F(xiàn)實(shí)場(chǎng)景里 LLM 可能陷入死循環(huán)不斷地調(diào)用同一個(gè)工具但參數(shù)永遠(yuǎn)不對(duì)如果只有動(dòng)作超時(shí)沒(méi)有會(huì)話超時(shí)整個(gè)請(qǐng)求就會(huì)耗盡資源。會(huì)話超時(shí)我一般根據(jù)任務(wù)復(fù)雜度配 120 秒到 300 秒超過(guò)直接中斷并返回超時(shí)錯(cuò)誤。三層超時(shí)的關(guān)系用數(shù)字表達(dá)就是網(wǎng)絡(luò)超時(shí) 動(dòng)作超時(shí) 會(huì)話超時(shí)每一層都是上一層的兜底防線。這個(gè)金字塔結(jié)構(gòu)能保證任意一層出現(xiàn)問(wèn)題都不會(huì)讓整個(gè)任務(wù)無(wú)限期掛起。2.2 重試策略不能一刀切重試比超時(shí)更微妙。一開(kāi)始我們天真地對(duì)所有失敗做統(tǒng)一重試結(jié)果發(fā)現(xiàn)效果很差甚至產(chǎn)生了雙倍故障。原因很簡(jiǎn)單重試是有前置條件的不是所有失敗都值得重試。我后來(lái)把所有 Agent 執(zhí)行過(guò)程中的失敗分成了三類。第一類是瞬時(shí)失敗典型代表是網(wǎng)絡(luò)抖動(dòng)、連接池已滿、第三方服務(wù) 503 或 429。這類失敗是值得重試的但要遵循退避策略。我用的方案是初始等待 500ms每次重試等待時(shí)間翻倍最多重試 3 次同時(shí)加上一個(gè)小的隨機(jī)抖動(dòng)來(lái)避免同時(shí)重試造成的驚群效應(yīng)。第二類是永久性失敗比如參數(shù)格式錯(cuò)誤、API key 無(wú)效、業(yè)務(wù)規(guī)則不滿足。這類失敗重試多少次都沒(méi)用必須立刻失敗并把錯(cuò)誤信息原樣返回上層讓 Agent 有機(jī)會(huì)重新規(guī)劃。注意這里有個(gè)關(guān)鍵點(diǎn)工具的永久性失敗要作為“觀察結(jié)果”反饋給 LLM而不是簡(jiǎn)單地終結(jié)整個(gè) Agent 循環(huán)。因?yàn)?LLM 可能根據(jù)錯(cuò)誤信息調(diào)整參數(shù)重來(lái)這在 Agent 里是一次“更高級(jí)別的重試”。第三類是冪等性失敗也就是請(qǐng)求已經(jīng)成功執(zhí)行但響應(yīng)沒(méi)有收到典型例子是網(wǎng)絡(luò)超時(shí)后服務(wù)端其實(shí)已經(jīng)把數(shù)據(jù)庫(kù)記錄寫進(jìn)去了。這種情況下盲目重試會(huì)造成重復(fù)寫入。后面我也會(huì)說(shuō)這類問(wèn)題光靠重試參數(shù)解決不了必須在工具層做冪等設(shè)計(jì)。實(shí)現(xiàn)重試時(shí)我強(qiáng)烈建議不要自己手寫 while 循環(huán)直接用 tenacity 這類庫(kù)它支持指數(shù)退避、重試條件函數(shù)、最大重試次數(shù)等配置代碼極其簡(jiǎn)潔。提示重試策略必須區(qū)分“值得重試的錯(cuò)誤”和“不值得重試的錯(cuò)誤”。統(tǒng)一重試是新手最容易犯的錯(cuò)誤它在瞬時(shí)失敗場(chǎng)景下有效但在 LLM 工具調(diào)用場(chǎng)景下會(huì)放大副作用。3. 真正的坑狀態(tài)清理3.1 狀態(tài)清理是什么問(wèn)題加完超時(shí)和重試后系統(tǒng)表面上是穩(wěn)定了但沒(méi)過(guò)多久就出現(xiàn)了更隱蔽的癥狀。比如用戶發(fā)起一個(gè)任務(wù)第一次執(zhí)行超時(shí)了用戶重試之后發(fā)現(xiàn)系統(tǒng)里出現(xiàn)了兩份數(shù)據(jù)再比如 Agent 中途失敗但已經(jīng)調(diào)用過(guò)的工具副效應(yīng)留在了系統(tǒng)里下次跑同樣的任務(wù)時(shí)舊數(shù)據(jù)還在導(dǎo)致結(jié)果錯(cuò)亂。這就是狀態(tài)清理的范疇。超時(shí)和重試解決的是“任務(wù)如何結(jié)束”而狀態(tài)清理解決的是“任務(wù)結(jié)束后系統(tǒng)里不應(yīng)該留下任何實(shí)驗(yàn)痕跡”。Agent 和普通接口最大的區(qū)別在于它有工具調(diào)用副作用。普通接口超時(shí)了客戶端不干了就行最多回滾一個(gè)數(shù)據(jù)庫(kù)事務(wù)但 Agent 超時(shí)時(shí)可能已經(jīng)調(diào)用了三個(gè)工具每個(gè)工具都有外部副作用有些還是不可回滾的——比如發(fā)了郵件、推送了通知、寫了日志。這些副效應(yīng)不會(huì)隨著超時(shí)自動(dòng)消失它們就是狀態(tài)殘留。狀態(tài)殘留本質(zhì)上會(huì)讓系統(tǒng)從“每個(gè)請(qǐng)求獨(dú)立”退化成“請(qǐng)求之間有隱含依賴”。第一次請(qǐng)求失敗留下的殘留數(shù)據(jù)會(huì)污染第二次請(qǐng)求的判定邏輯重試觸發(fā)的同一工具調(diào)用會(huì)因?yàn)榍耙淮蔚臍埩魯?shù)據(jù)而產(chǎn)生重復(fù)副作用。這些問(wèn)題都比“一個(gè)接口超時(shí)”嚴(yán)重得多因?yàn)樗鼈冏屨麄€(gè)系統(tǒng)變得不可預(yù)期。3.2 Agent 狀態(tài)到底包含哪些東西要清理狀態(tài)得先定義狀態(tài)。Agent 執(zhí)行過(guò)程中的狀態(tài)不是單一的我拆出了四層。第一層是消息歷史。也就是 LLM 對(duì)話上下文包括用戶問(wèn)題、Assistant 的思維鏈輸出、工具觀察結(jié)果。這些數(shù)據(jù)在失敗后處理起來(lái)比較微妙如果全都清掉重試時(shí) LLM 就失憶了會(huì)重新犯錯(cuò)如果全保留失敗時(shí)的誤區(qū)會(huì)被帶進(jìn)重試?yán)風(fēng)LM 可能執(zhí)著于錯(cuò)誤的思路。后面我會(huì)給一個(gè)更具體的處理策略。第二層是執(zhí)行軌跡。包括已完成的工具調(diào)用記錄、每輪 LLM 返回的中間 Action、耗時(shí)指標(biāo)、 token 消耗。這個(gè)狀態(tài)主要用于可觀測(cè)性和審計(jì)失敗后不一定要清但要標(biāo)記該軌跡對(duì)應(yīng)的任務(wù)已失敗避免它出現(xiàn)在后續(xù)查詢中。第三層是業(yè)務(wù)副作用狀態(tài)。這是最容易出問(wèn)題的。Agent 已經(jīng)寫進(jìn)數(shù)據(jù)庫(kù)的記錄、已經(jīng)創(chuàng)建的文件、已經(jīng)發(fā)送的通知這些都不在 Agent 進(jìn)程內(nèi)但它們是 Agent 行為的結(jié)果。失敗后這些副作用需要被識(shí)別并根據(jù)業(yè)務(wù)規(guī)則決定是回滾、補(bǔ)償還是標(biāo)記為“孤兒數(shù)據(jù)”。第四層是運(yùn)行期上下文。包括當(dāng)前執(zhí)行到的步驟編號(hào)、臨時(shí)變量、工具參數(shù)緩存、重試計(jì)數(shù)等。這一層最簡(jiǎn)單進(jìn)程內(nèi)對(duì)象請(qǐng)求結(jié)束自然銷毀但如果用了異步任務(wù)架構(gòu)要小心上下文對(duì)象在 worker 里滯留導(dǎo)致后續(xù)任務(wù)讀到上一輪的殘留數(shù)據(jù)。3.3 什么狀態(tài)該清什么不能清狀態(tài)清理最容易踩的坑就是“一刀切”。我們一開(kāi)始的策略很簡(jiǎn)單失敗就清空所有狀態(tài)重新開(kāi)始效果差到令人崩潰。因?yàn)榍謇淼魻顟B(tài)的同時(shí)也清理掉了重試的意義LLM 沒(méi)有任何記憶的情況下重試和第一次執(zhí)行沒(méi)有任何區(qū)別該失敗的還是會(huì)失敗。后來(lái)我總結(jié)出的原則是分情況處理。如果重試策略是“從當(dāng)前 Agent 循環(huán)上下文繼續(xù)”那么消息歷史要保留但只保留到出問(wèn)題步驟之前的部分出錯(cuò)的那一步觀察結(jié)果要替換為新重試的結(jié)果。如果重試策略是“重置 Agent 內(nèi)部循環(huán)但保留用戶目標(biāo)”那么消息歷史要壓縮成一個(gè)整體摘要再把用戶原始需求拼接進(jìn)去。這樣 LLM 有背景知識(shí)又不會(huì)被上一次的具體錯(cuò)誤路徑綁架。真正的死坑在第三層業(yè)務(wù)副作用狀態(tài)。這一層不存在統(tǒng)一的“清理”方案完全取決于業(yè)務(wù)語(yǔ)義。有些副作用是可回滾的比如數(shù)據(jù)庫(kù)操作fail 之后在 catch 塊里執(zhí)行反向操作就行。有些是不可回滾的比如發(fā)送郵件這類只能靠事后補(bǔ)償或人工介入。還有一種很隱蔽的情況Agent 在一次執(zhí)行中先調(diào)用了“創(chuàng)建工單”工具又調(diào)用了“分配負(fù)責(zé)人”工具最后因?yàn)槌瑫r(shí)失敗了。工單已創(chuàng)建但負(fù)責(zé)人分配沒(méi)完成這時(shí)如果簡(jiǎn)單回滾掉“創(chuàng)建工單”會(huì)把一個(gè)本來(lái)部分有效的結(jié)果直接變成無(wú)用工單但如果不回滾用戶重試時(shí)又要重新創(chuàng)建一張工單就出現(xiàn)了重復(fù)數(shù)據(jù)。處理這種半完成業(yè)務(wù)狀態(tài)我最終采用的是快照與恢復(fù)模式。Agent 啟動(dòng)時(shí)先建立業(yè)務(wù)快照記錄當(dāng)前業(yè)務(wù)狀態(tài)指紋執(zhí)行期間每次工具調(diào)用前先提交一次“檢查點(diǎn)”內(nèi)容是當(dāng)前商業(yè)操作的描述和參數(shù)失敗時(shí)根據(jù)檢查點(diǎn)生成一條懸空任務(wù)記錄讓用戶在 UI 里自行決定是繼續(xù)執(zhí)行、回滾還是作廢。這個(gè)模式避免了自動(dòng)清理的粗暴性讓“清不清、怎么清”變成一個(gè)用戶可決策的流程。注意自動(dòng)回滾業(yè)務(wù)副作用是一個(gè)高風(fēng)險(xiǎn)的默認(rèn)行為。如果無(wú)法 100% 確認(rèn)工具操作的語(yǔ)義寧可保留現(xiàn)場(chǎng)并標(biāo)記為異常也不要在業(yè)務(wù)數(shù)據(jù)上執(zhí)行有創(chuàng)造性的“清理動(dòng)作”。4. 實(shí)操中的實(shí)現(xiàn)細(xì)節(jié)4.1 帶狀態(tài)清理的 Agent 執(zhí)行環(huán)境設(shè)計(jì)具體到代碼層面我重構(gòu)了 Agent 的核心執(zhí)行器。整個(gè)設(shè)計(jì)可以用一句話概括所有手動(dòng)“清狀態(tài)”的地方都被顯式化成一個(gè) checkpoint 字段失敗后用統(tǒng)一的事件機(jī)制處理而不是散落在各種 except 塊里。下面給出核心執(zhí)行器的結(jié)構(gòu)代碼這是一個(gè)基于偽代碼的簡(jiǎn)化示例體現(xiàn)的是分層思路實(shí)際工程實(shí)現(xiàn)會(huì)比這復(fù)雜一些但骨架是有效的。# agent_executor.py from dataclasses import dataclass, field, asdict from typing import Any, Optional import uuid, time from enum import Enum class TaskStatus(str, Enum): RUNNING running SUCCEEDED succeeded FAILED failed TIMEOUT timeout SUSPENDED suspended # 半完成狀態(tài)需要人工決策 dataclass class Checkpoint: checkpoint_id: str step_index: int tool_name: str tool_args: dict business_state_fingerprint: str created_at: float dataclass class AgentTaskContext: task_id: str user_goal: str message_history: list[Any] field(default_factorylist) tool_results: list[dict] field(default_factorylist) checkpoints: list[Checkpoint] field(default_factorylist) run_state: dict[str, Any] field(default_factorydict) status: TaskStatus TaskStatus.RUNNING error_info: Optional[dict] None created_at: float 0.0 def push_checkpoint(self, checkpoint: Checkpoint): self.checkpoints.append(checkpoint) def snapshot_fingerprint(self) - str: # 這里實(shí)際會(huì)取業(yè)務(wù)側(cè)關(guān)鍵數(shù)據(jù)表的CRC或版本號(hào) return hash(str(self.tool_results)) dataclass class AgentAction: tool_name: str tool_args: dict thought: str我用 dataclass 定義了執(zhí)行上下文核心是 checkpoints 列表。列表里存放每一步工具調(diào)用的檢查點(diǎn)檢查點(diǎn)是后續(xù)做狀態(tài)清理的原始依據(jù)。然后定義超時(shí)參數(shù)和執(zhí)行器主體dataclass class AgentConfig: session_timeout: float 120.0 action_timeout: float 30.0 max_retries: int 3 base_backoff: float 0.5 class AgentExecutor: def __init__(self, config: AgentConfig): self.config config self.tools {} # 由外部注冊(cè) def execute(self, context: AgentTaskContext) - AgentTaskContext: deadline time.monotonic() self.config.session_timeout started time.monotonic() context.status TaskStatus.RUNNING context.created_at started while time.monotonic() deadline: step_index len(context.message_history) # 1. 調(diào)用 LLM 獲取下一步動(dòng)作 action self._invoke_llm_with_timeout(context, deadline) if action is None: return self._mark_failed(context, llm response none after retries) # LLM 判斷任務(wù)結(jié)束 if action.tool_name FINISH: context.status TaskStatus.SUCCEEDED return context # 2. 執(zhí)行工具帶動(dòng)作級(jí)超時(shí)和重試 if action.tool_name not in self.tools: context.run_state[last_error] ftool not found: {action.tool_name} context.message_history.append( {role: tool, tool_name: action.tool_name, content: Error: tool not found} ) continue tool_result self._call_tool_with_policy( context, action, deadline ) if tool_result[success]: # 執(zhí)行成功記錄檢查點(diǎn)這里的檢查點(diǎn)就是“狀態(tài)清理”的基礎(chǔ) ctx_fp context.snapshot_fingerprint() context.push_checkpoint( Checkpoint( checkpoint_idstr(uuid.uuid4()), step_indexstep_index, tool_nameaction.tool_name, tool_argsaction.tool_args, business_state_fingerprintctx_fp, created_attime.time(), ) ) context.tool_results.append(tool_result[data]) context.run_state[last_error] None else: # 失敗處理標(biāo)記錯(cuò)誤并記錄到觀察結(jié)果讓LLM做更高層決策 if tool_result[retryable]: # 超出重試上限后仍失敗 context.run_state[last_error] tool_result[error] context.message_history.append( {role: tool, tool_name: action.tool_name, content: fError: {tool_result[error]}} ) else: context.run_state[last_error] tool_result[error] context.message_history.append( {role: tool, tool_name: action.tool_name, content: fFatal Error: {tool_result[error]}} ) # 每步結(jié)束后檢查是否需要提前掛起半完成狀態(tài) if self._needs_suspend(context): context.status TaskStatus.SUSPENDED return context # 會(huì)話超時(shí) context.status TaskStatus.TIMEOUT return context def _call_tool_with_policy(self, context, action, deadline): # 執(zhí)行工具調(diào)用帶動(dòng)作級(jí)超時(shí) tool_fn self.tools[action.tool_name] last_error None for attempt in range(self.config.max_retries): try: result self._run_tool_with_action_timeout( tool_fn, action.tool_args, self.config.action_timeout ) return {success: True, data: result} except TimeoutError as exc: last_error faction timeout after {self.config.action_timeout}s # 超時(shí)是最典型的瞬時(shí)錯(cuò)誤做退避重試 self._backoff(attempt 1) except Exception as exc: # 區(qū)分可重試異常和永久異常 last_error str(exc) if not self._is_retryable_exception(exc): return {success: False, retryable: False, error: last_error} self._backoff(attempt 1) return {success: False, retryable: True, error: last_error} def _run_tool_with_action_timeout(self, tool_fn, args, timeout): from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers2) as pool: future pool.submit(tool_fn, **args) return future.result(timeouttimeout) staticmethod def _is_retryable_exception(exc): # 實(shí)際判斷會(huì)看異常類型連接錯(cuò)誤、超時(shí)、限流可重試 # 參數(shù)校驗(yàn)、權(quán)限、業(yè)務(wù)拒絕不可重試。 return connection in str(exc).lower() or timeout in str(exc).lower() staticmethod def _backoff(attempt): import time, random wait 0.5 * (2 ** attempt) random.uniform(0, 0.5) time.sleep(wait) def _mark_failed(self, context, message): context.status TaskStatus.FAILED context.error_info {message: message} return context def _needs_suspend(self, context): # 當(dāng)存在已執(zhí)行工具且最近一步失敗時(shí)評(píng)估是否進(jìn)入掛起 if context.run_state.get(last_error) is not None: # 如果已經(jīng)產(chǎn)生了不可忽略的業(yè)務(wù)用戶流程如創(chuàng)建了工單、發(fā)了通知就掛起 return len(context.checkpoints) 0 return False這里_needs_suspend的邏輯是核心只要之前已經(jīng)有過(guò)成功的業(yè)務(wù)操作后續(xù)再失敗就不直接標(biāo)記 FAILED而是標(biāo)記 SUSPENDED。SUSPENDED 狀態(tài)下系統(tǒng)會(huì)保留所有 checkpoint 和上下文用戶可以查看到達(dá)了哪一步、最后成功的是什么、失敗的是什么然后手動(dòng)選擇繼續(xù)或回滾。4.2 狀態(tài)清理策略的實(shí)現(xiàn)上面的執(zhí)行器只負(fù)責(zé)把狀態(tài)遺漏保留下來(lái)真正的清理動(dòng)作是在收到“確認(rèn)失敗”事件后觸發(fā)的。我實(shí)現(xiàn)了一個(gè)StateCleanupManager它的職責(zé)是根據(jù)業(yè)務(wù)配置決定某個(gè)工具調(diào)用結(jié)果要不要回滾、怎么回滾。# state_cleanup.py from typing import Callable, Optional from dataclasses import dataclass dataclass class CleanupAction: tool_name: str rollback_fn: Optional[Callable] # 如果可回滾則提供反向方法 is_destructive: bool # 是否是破壞性回滾如刪除數(shù)據(jù) requires_manual: bool # 是否需要人工決策 class StateCleanupManager: 狀態(tài)清理策略注冊(cè)中心。 每個(gè)工具在注冊(cè)時(shí)必須同時(shí)聲明自己的清理策略。 這是解決狀態(tài)清理問(wèn)題的最關(guān)鍵設(shè)計(jì)——清理不是萬(wàn)能的后置處理 而是每個(gè)工具的固有屬性必須在工具研發(fā)階段就定義。 def __init__(self): self._cleanup_map: dict[str, CleanupAction] {} def register(self, tool_name: str, rollback_fn: Optional[Callable] None, is_destructive: bool False, requires_manual: bool False): self._cleanup_map[tool_name] CleanupAction( tool_nametool_name, rollback_fnrollback_fn, is_destructiveis_destructive, requires_manualrequires_manual, ) def resolve_cleanup_plan(self, context) - list[dict]: 根據(jù)上下文里的 checkpoints生成清理計(jì)劃。 返回的每個(gè)計(jì)劃項(xiàng)都標(biāo)注了類型供上層做自動(dòng)化或人工處理。 plan [] for cp in reversed(context.checkpoints): # 反序回滾 action self._cleanup_map.get(cp.tool_name) if not action: continue if action.requires_manual or action.rollback_fn is None: plan.append({checkpoint: cp, mode: manual}) else: plan.append({checkpoint: cp, mode: auto, action: action}) return plan def execute_cleanup(self, context) - dict: plan self.resolve_cleanup_plan(context) auto_results [] manual_needed [] for item in plan: if item[mode] manual: manual_needed.append(item[checkpoint]) continue try: item[action].rollback_fn(item[checkpoint].tool_args) auto_results.append({checkpoint_id: item[checkpoint].checkpoint_id, status: rolled_back}) except Exception as exc: auto_results.append({checkpoint_id: item[checkpoint].checkpoint_id, status: rollback_failed, error: str(exc)}) return { auto_rolled_back: auto_results, manual_required: manual_needed, }這個(gè)設(shè)計(jì)的核心是一個(gè)原則工具的清理策略必須和工具本身同時(shí)注冊(cè)。我見(jiàn)過(guò)太多項(xiàng)目先開(kāi)發(fā)幾十個(gè)工具最后才想起來(lái)做狀態(tài)清理結(jié)果每一個(gè)工具都需要靠逆向讀代碼來(lái)猜能不能安全回滾那個(gè)成本高到你想哭。如果從一開(kāi)始每個(gè)工具定義時(shí)就順手聲明一下清理策略這個(gè)事基本是無(wú)縫集成的。清理策略也要有類型區(qū)分。我給每個(gè)清理動(dòng)作打標(biāo)可自動(dòng)回滾的比如修改類操作、可自動(dòng)回退但破壞性的比如刪除數(shù)據(jù)、必須人工介入的比如發(fā)送消息。這三類在系統(tǒng)里走完全不同的流程自動(dòng)回滾直接執(zhí)行破壞性回滾需要二次確認(rèn)人工介入則掛起并通知用戶。這樣既保證了自動(dòng)化程度又避免了自動(dòng)執(zhí)行的二次傷害。4.3 冪等設(shè)計(jì)的落地重試伴隨的另一個(gè)核心問(wèn)題就是冪等。如果我的工具調(diào)用了“寫入數(shù)據(jù)庫(kù)”操作第一次寫成功了但沒(méi)有返回重試時(shí)會(huì)不會(huì)寫兩遍答案是會(huì)除非工具本身做了冪等控制。冪等控制的通用方案是在工具參數(shù)里強(qiáng)制要求一個(gè)request_id這個(gè) ID 在 Agent 生成動(dòng)作時(shí)就注入。數(shù)據(jù)庫(kù)側(cè)用這個(gè) ID 作為唯一鍵重復(fù)寫入時(shí)只返回首次結(jié)果不再插入新記錄。這個(gè)設(shè)計(jì)對(duì)重試的收益是巨大的因?yàn)榻^大多數(shù)重試場(chǎng)景都是“結(jié)果未知但請(qǐng)求可能已成功”冪等可以完美消除重復(fù)副作用。# idempotent_tool.py import hashlib import json from typing import Callable class IdempotencyGuard: 冪等守衛(wèi)給任意工具調(diào)用包一層冪等控制。 核心思路用 request_id 工具名 參數(shù)hash 做唯一索引。 def __init__(self, backend_store): self.store backend_store # 可以是 redis/mysql要求能按唯一鍵存取 def execute(self, tool_name: str, tool_args: dict, tool_fn: Callable, request_id: str): key hashlib.md5( f{request_id}:{tool_name}:{json.dumps(tool_args, sort_keysTrue)}.encode() ).hexdigest() # 嘗試獲取已有結(jié)果 if self.store.exists(key): return self.store.get(key) # 執(zhí)行前先做“預(yù)定”標(biāo)記防止并發(fā)下重復(fù)執(zhí)行 if not self.store.acquire_lock(key): # 另一個(gè)實(shí)例正在執(zhí)行同一請(qǐng)求等待結(jié)果 return self.store.wait_and_get(key, timeout10) try: result tool_fn(**tool_args) self.store.set(key, result) return result finally: self.store.release_lock(key)這樣改造之后之前線上出現(xiàn)的“用戶重試導(dǎo)致雙份數(shù)據(jù)”問(wèn)題基本絕跡。這里說(shuō)一句實(shí)在話冪等設(shè)計(jì)比狀態(tài)清理的任何后置方案都更根本如果工具本身可冪等大量的重試擾動(dòng)都能被自動(dòng)吸收。狀態(tài)清理是用來(lái)處理不可冪等場(chǎng)景的兜底兩者配合才是完整方案。提示給工具調(diào)用加冪等守衛(wèi)的成本遠(yuǎn)低于復(fù)盤事故成本。建議在 Agent 工具開(kāi)發(fā)規(guī)范里把“必須支持 request_id 冪等”定為默認(rèn)要求而不是可有可無(wú)的加分項(xiàng)。5. 超時(shí)重試與狀態(tài)清理的完整聯(lián)動(dòng)5.1 失敗后的完整處理流程當(dāng)一次 Agent 執(zhí)行進(jìn)入失敗路徑后完整流程是這樣的。先看錯(cuò)誤類型。如果是網(wǎng)絡(luò)層瞬時(shí)錯(cuò)誤且尚未產(chǎn)生任何業(yè)務(wù)副作用最簡(jiǎn)單的處理是走自動(dòng)重試重試上限內(nèi)大概率能成功這個(gè)場(chǎng)景不走狀態(tài)清理邏輯。如果是 LLM 層的邏輯死循環(huán)比如反復(fù)調(diào)用相同工具且參數(shù)完全一致此時(shí)啟動(dòng)循環(huán)檢測(cè)截?cái)鄬?duì)話歷史到最近三次提示 LLM “檢測(cè)到重復(fù)動(dòng)作請(qǐng)嘗試完全不同的方法”。這一步很有效因?yàn)楹芏嗨姥h(huán)其實(shí)是 LLM 對(duì)上下文理解偏差導(dǎo)致的只要打斷它的慣性即可。如果已經(jīng)產(chǎn)生了業(yè)務(wù)副作用再分類處理。副作用是可以自動(dòng)回滾的比如工具里有對(duì)應(yīng)的 rollback 函數(shù)就自動(dòng)執(zhí)行回滾并記錄審計(jì)日志。副作用不可回滾那就把任務(wù)標(biāo)記為 SUSPENDED在 UI 上給用戶展示“當(dāng)前任務(wù)已暫?！钡目ㄆㄆ淹瓿傻牟襟E清單和剩余可選操作繼續(xù)執(zhí)行、整單作廢、標(biāo)記人工處理。這一步?jīng)]有任何自動(dòng)化魔法是產(chǎn)品層面必須接受的復(fù)雜度。到這里整個(gè)流程就算閉環(huán)了超時(shí)保護(hù)住最壞情況下的資源占用重試覆蓋掉瞬時(shí)失敗和可恢復(fù)故障狀態(tài)清理處理掉失敗后的副作用殘留。三者組合才敢說(shuō)這個(gè) Agent 真正具備了一點(diǎn)工程化的穩(wěn)定性基礎(chǔ)。5.2 可觀測(cè)性要怎么配合狀態(tài)清理做得好不好很大程度上取決于能不能看到狀態(tài)。我強(qiáng)烈建議在 Agent 執(zhí)行上下文中把每一步的 checkpoint、狀態(tài)指紋、清理計(jì)劃全部暴露到日志系統(tǒng)里。我用過(guò)的最佳組合是結(jié)構(gòu)化日志JSON 格式進(jìn) ELK執(zhí)行軌跡進(jìn) Jaeger業(yè)務(wù)懸空任務(wù)單獨(dú)存一張表。三者配合排查一次線上事故的效率能提高一個(gè)數(shù)量級(jí)。這里有一個(gè)細(xì)節(jié)值得注意checkpoint 里的business_state_fingerprint每一次工具調(diào)用成功后我會(huì)對(duì)整個(gè)業(yè)務(wù)關(guān)鍵數(shù)據(jù)做一個(gè)輕量哈希記錄在檢查點(diǎn)里。后續(xù)如果要做回滾可以直接反查這個(gè)指紋驗(yàn)證現(xiàn)狀是否與當(dāng)時(shí)執(zhí)行時(shí)一致。如果用戶已經(jīng)基于當(dāng)時(shí)的執(zhí)行結(jié)果做了額外操作指紋不匹配這時(shí)“無(wú)腦回滾”就是危險(xiǎn)動(dòng)作。這個(gè)設(shè)計(jì)把自動(dòng)回滾的安全邊界劃得很清楚只允許回滾未被后續(xù)操作污染的數(shù)據(jù)。5.3 什么時(shí)候需要引入異步補(bǔ)償機(jī)制如果 Agent 任務(wù)本身是長(zhǎng)時(shí)間運(yùn)行的比如分鐘級(jí)甚至小時(shí)級(jí)的編排任務(wù)同步執(zhí)行的狀態(tài)清理方案會(huì)變得不適用。因?yàn)橐粋€(gè)任務(wù)的失敗和清理可能相隔幾十分鐘中間用戶可能已經(jīng)查詢過(guò)數(shù)據(jù)、其他任務(wù)可能已經(jīng)依賴了它產(chǎn)生的狀態(tài)。這種場(chǎng)景下必須把狀態(tài)清理做成異步補(bǔ)償機(jī)制通常走一條獨(dú)立的補(bǔ)償隊(duì)列。失敗事件進(jìn)入隊(duì)列之后由專門的補(bǔ)償 worker 根據(jù)檢查點(diǎn)執(zhí)行回滾或者用戶決策。這本質(zhì)上就是 Saga 模式在 Agent 場(chǎng)景的實(shí)現(xiàn)。如果用上了我上面講的StateCleanupManager和IdempotencyGuard遷移到異步隊(duì)列的改動(dòng)成本很小因?yàn)榍謇磉壿嫳緛?lái)就已經(jīng)和 Agent 執(zhí)行器解耦了。6. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄6.1 重試后重復(fù)副作用表現(xiàn)Agent 第一次調(diào)用“創(chuàng)建設(shè)備記錄”工具超時(shí)用戶重試出現(xiàn)了兩條相同的設(shè)備記錄。排查思路先看工具是否有冪等控制。如果是歷史代碼沒(méi)有冪等層優(yōu)先補(bǔ)request_id冪等方案。如果只是偶爾出現(xiàn)且不是高頻路徑可以退而求其次用清理策略做檢測(cè)——在每個(gè)檢查點(diǎn)里記錄創(chuàng)建記錄的唯一業(yè)務(wù)鍵失敗后反查該鍵是否已存在存在則視為“已副作用完成”不再回滾只返回首次結(jié)果。這個(gè)方案雖然不如真正的冪等優(yōu)雅但能解決兼容老工具的問(wèn)題。6.2 會(huì)話超時(shí)真的觸發(fā)了但沒(méi)有用表現(xiàn)配置了 120 秒會(huì)話超時(shí)但請(qǐng)求實(shí)際運(yùn)行了 300 秒才返回。排查思路大概率是 LLM 流式調(diào)用阻塞了主線程而超時(shí)檢查在 while 循環(huán)里只在下一次迭代才生效。如果你調(diào)用 LLM 是阻塞式的且它一直不返回循環(huán)根本走不到超時(shí)檢查。解決方法是把 LLM 調(diào)用也放進(jìn)_run_tool_with_action_timeout的線程池執(zhí)行讓 Action 級(jí)和 Session 級(jí)的超時(shí)真正生效。這是我踩得最痛的坑之一agent 不是所有調(diào)用都會(huì)主動(dòng)感知外層超時(shí)標(biāo)志的。6.3 狀態(tài)清理誤殺了正常數(shù)據(jù)表現(xiàn)任務(wù)失敗后自動(dòng)回滾回滾后另一個(gè)正常任務(wù)的數(shù)據(jù)也消失了。排查思路典型的回滾邊界定義錯(cuò)誤。某個(gè)工具的 rollback 函數(shù)寫得過(guò)于寬泛比如 DELETE 語(yǔ)句只按類型過(guò)濾而沒(méi)帶 request_id。修正方法就是升級(jí)清理策略的精度rollback 操作必須攜帶原調(diào)用的完整參數(shù)和 request_id且回滾范圍必須只能作用于該次調(diào)用產(chǎn)生的最小粒度數(shù)據(jù)。另外這也能用指紋校驗(yàn)擋掉回滾前比對(duì)業(yè)務(wù)指紋不一致就不執(zhí)行自動(dòng)回滾轉(zhuǎn)人工。6.4 清理策略速查表失敗類型是否重試典型工具示例狀態(tài)清理策略是否走人工網(wǎng)絡(luò)瞬斷是指數(shù)退避HTTP調(diào)用、LLM調(diào)用無(wú)需清理否LLM死循環(huán)否打斷重置推理步驟截?cái)嗌舷挛姆穹莾绲葘懭氤瑫r(shí)是需冪等數(shù)據(jù)庫(kù)INSERT冪等去重否已執(zhí)行的通知類工具否發(fā)郵件、推送通知不可回滾記錄懸空是已創(chuàng)建的工單未分配負(fù)責(zé)人否內(nèi)部業(yè)務(wù)API部分回滾/掛起是臨時(shí)文件、分布式鎖是緩存、文件、鎖自動(dòng)清理否這張表是我們?cè)趯?shí)踐中沉淀下來(lái)的分類標(biāo)準(zhǔn)基本上所有工具都能歸入其中某一類。每次新增工具時(shí)第一件事就是想清楚它落在哪一行然后在代碼注冊(cè)時(shí)把清理策略配置好。這會(huì)成為 Agent 穩(wěn)定性的一道重要防線。7. 一點(diǎn)個(gè)人體會(huì)這次改造讓我對(duì) Agent 工程化有了完全不同的理解。一開(kāi)始我以為超時(shí)和重試是嚴(yán)謹(jǐn)性的體現(xiàn)做完了才發(fā)現(xiàn)它們頂多算 App 的入口保護(hù)真正的工程難點(diǎn)全在業(yè)務(wù)副作用的管理上。超時(shí)和重試本質(zhì)上是處理時(shí)間和頻率的問(wèn)題而狀態(tài)清理處理的是真實(shí)世界被 Agent 改變了之后怎么辦的問(wèn)題。后者復(fù)雜得多因?yàn)槟悴荒芾碚摶亍俺蜂N”一封已發(fā)出的郵件不能憑空“復(fù)原”一次外部系統(tǒng)的狀態(tài)修改。根據(jù)我的個(gè)人經(jīng)驗(yàn)做 Agent 穩(wěn)定性的項(xiàng)目建議不要一開(kāi)始就追求全自動(dòng)清理而是先把“檢查點(diǎn) 懸空任務(wù) 人工決策”這套半自動(dòng)方案跑通讓團(tuán)隊(duì)的認(rèn)知上線再逐步把高頻且安全的路徑自動(dòng)化。如果倒過(guò)來(lái)一上來(lái)就寫一堆自動(dòng)回滾代碼大概率會(huì)在生產(chǎn)環(huán)境的某個(gè)深夜搞出一場(chǎng)事故。最后再分享一個(gè)我實(shí)際用得很順手的小技巧給 Agent 上下文的每個(gè) checkpoint 都加一個(gè)created_at字段配合任務(wù)級(jí)created_at可以算出“某個(gè)工具執(zhí)行后到底存活了多久”。當(dāng)一個(gè)任務(wù)失敗但用戶遲遲不處理懸空任務(wù)時(shí)超出 24 小時(shí)的懸空任務(wù)會(huì)被自動(dòng)升級(jí)為特殊狀態(tài)并通知負(fù)責(zé)人。這個(gè)功能救了我好幾次因?yàn)樵谡鎸?shí)的業(yè)務(wù)里懸而不決的狀態(tài)往往比失敗本身更危險(xiǎn)。希望這篇內(nèi)容對(duì)你的 Agent 工程化之路有一點(diǎn)點(diǎn)幫助如果你也在思考類似的問(wèn)題歡迎拿這個(gè)故事去檢驗(yàn)自己的設(shè)計(jì)很多時(shí)候只有在出過(guò)一次事之后才會(huì)真正理解狀態(tài)清理的分量。