清理的工程實踐與踩坑指南)
給 Agent 加超時重試本身不難難的是每次執(zhí)行失敗之后怎么把現(xiàn)場干干凈凈地收掉。這個教訓(xùn)我是在線上被真實流量教育過之后才徹底想明白的。今天把整個設(shè)計和踩坑過程整理出來希望能幫你少走一段彎路。1. 項目背景一個 AI Agent 的穩(wěn)定性改造先交代一下背景。我負(fù)責(zé)的項目是一個基于 LangChain FastAPI 的 AI Agent 服務(wù)核心能力是讓大模型自主完成多步任務(wù)比如查詢數(shù)據(jù)庫、調(diào)用內(nèi)部 API、生成報表、發(fā)通知甚至跨系統(tǒng)編排動作。外層接口是同步 HTTP內(nèi)部走的是循環(huán)式 Agent 執(zhí)行LLM 推理 → 工具調(diào)用 → 結(jié)果回填 → 再次推理直到模型判斷任務(wù)結(jié)束。項目上線初期穩(wěn)定運(yùn)行但流量一上來問題就暴露了。典型癥狀包括用戶請求偶爾 504、LLM 服務(wù)偶發(fā)超時導(dǎo)致整個請求卡死、工具調(diào)用失敗后 Agent 像“失憶”一樣重復(fù)做無用功。于是我們啟動了一輪穩(wěn)定性改造核心目標(biāo)就兩個字超時和重試。聽起來非常常規(guī)做起來也確實能解決一部分問題但改完之后真正折磨我的是第三個字狀態(tài)清理。這里也順便說清楚這篇文章適合誰看。如果你正在做 AI Agent 相關(guān)的工程化落地尤其是涉及多輪工具調(diào)用、狀態(tài)持久化、甚至異步任務(wù)編排的這篇文章提到的方案和坑會很有參考價值。如果你只是調(diào)通了一個基于 LangChain 的 demo對超時重試的認(rèn)知還停留在 requests 庫加個 timeout 參數(shù)那更建議從頭讀一遍因為你遲早會踩到同樣的坑。2. 超時與重試的落地實現(xiàn)2.1 三層超時設(shè)計先說說超時。這是最容易做錯的地方很多人給 Agent 加超時就是在 HTTP 客戶端上設(shè)置一下或者給 FastAPI 接口加個超時中間件但實際上 Agent 的超時體系是分層的至少拆成三層才夠用。第一層是網(wǎng)絡(luò)請求超時。這是最底層的針對 Agent 依賴的所有外部服務(wù)LLM 供應(yīng)商 API、數(shù)據(jù)庫連接、內(nèi)部工具服務(wù)的 HTTP 接口。這里我統(tǒng)一走的是 httpx 或 aiohttp 的 timeout 配置連接超時一般給 5 秒讀超時根據(jù)服務(wù)特點差異化配置。LLM 調(diào)用這類“慢服務(wù)”讀超時給 60 秒因為大模型流式輸出本來就慢內(nèi)部 API 讀超時給 10 秒超過這個閾值基本就是服務(wù)有問題了。第二層是動作級超時。這一層很多人會忽略。Agent 的“一個動作”不只是調(diào)用一次外部 API它包含一次完整的工具調(diào)用周期LLM 生成工具參數(shù) → 執(zhí)行工具 → 返回結(jié)果給 LLM。這個周期里任何一個環(huán)節(jié)都可能卡住尤其是工具本身內(nèi)部有重試邏輯的時候整個動作可能被拖到幾分鐘。所以我會給單個工具動作設(shè)置獨立超時稱為 action_timeout默認(rèn) 30 秒。第三層是會話級超時。這是最高層的兜底對應(yīng)整個 Agent 任務(wù)的總執(zhí)行時長?，F(xiàn)實場景里 LLM 可能陷入死循環(huán)不斷地調(diào)用同一個工具但參數(shù)永遠(yuǎn)不對如果只有動作超時沒有會話超時整個請求就會耗盡資源。會話超時我一般根據(jù)任務(wù)復(fù)雜度配 120 秒到 300 秒超過直接中斷并返回超時錯誤。三層超時的關(guān)系用數(shù)字表達(dá)就是網(wǎng)絡(luò)超時 動作超時 會話超時每一層都是上一層的兜底防線。這個金字塔結(jié)構(gòu)能保證任意一層出現(xiàn)問題都不會讓整個任務(wù)無限期掛起。2.2 重試策略不能一刀切重試比超時更微妙。一開始我們天真地對所有失敗做統(tǒng)一重試結(jié)果發(fā)現(xiàn)效果很差甚至產(chǎn)生了雙倍故障。原因很簡單重試是有前置條件的不是所有失敗都值得重試。我后來把所有 Agent 執(zhí)行過程中的失敗分成了三類。第一類是瞬時失敗典型代表是網(wǎng)絡(luò)抖動、連接池已滿、第三方服務(wù) 503 或 429。這類失敗是值得重試的但要遵循退避策略。我用的方案是初始等待 500ms每次重試等待時間翻倍最多重試 3 次同時加上一個小的隨機(jī)抖動來避免同時重試造成的驚群效應(yīng)。第二類是永久性失敗比如參數(shù)格式錯誤、API key 無效、業(yè)務(wù)規(guī)則不滿足。這類失敗重試多少次都沒用必須立刻失敗并把錯誤信息原樣返回上層讓 Agent 有機(jī)會重新規(guī)劃。注意這里有個關(guān)鍵點工具的永久性失敗要作為“觀察結(jié)果”反饋給 LLM而不是簡單地終結(jié)整個 Agent 循環(huán)。因為 LLM 可能根據(jù)錯誤信息調(diào)整參數(shù)重來這在 Agent 里是一次“更高級別的重試”。第三類是冪等性失敗也就是請求已經(jīng)成功執(zhí)行但響應(yīng)沒有收到典型例子是網(wǎng)絡(luò)超時后服務(wù)端其實已經(jīng)把數(shù)據(jù)庫記錄寫進(jìn)去了。這種情況下盲目重試會造成重復(fù)寫入。后面我也會說這類問題光靠重試參數(shù)解決不了必須在工具層做冪等設(shè)計。實現(xiàn)重試時我強(qiáng)烈建議不要自己手寫 while 循環(huán)直接用 tenacity 這類庫它支持指數(shù)退避、重試條件函數(shù)、最大重試次數(shù)等配置代碼極其簡潔。提示重試策略必須區(qū)分“值得重試的錯誤”和“不值得重試的錯誤”。統(tǒng)一重試是新手最容易犯的錯誤它在瞬時失敗場景下有效但在 LLM 工具調(diào)用場景下會放大副作用。3. 真正的坑狀態(tài)清理3.1 狀態(tài)清理是什么問題加完超時和重試后系統(tǒng)表面上是穩(wěn)定了但沒過多久就出現(xiàn)了更隱蔽的癥狀。比如用戶發(fā)起一個任務(wù)第一次執(zhí)行超時了用戶重試之后發(fā)現(xiàn)系統(tǒng)里出現(xiàn)了兩份數(shù)據(jù)再比如 Agent 中途失敗但已經(jīng)調(diào)用過的工具副效應(yīng)留在了系統(tǒng)里下次跑同樣的任務(wù)時舊數(shù)據(jù)還在導(dǎo)致結(jié)果錯亂。這就是狀態(tài)清理的范疇。超時和重試解決的是“任務(wù)如何結(jié)束”而狀態(tài)清理解決的是“任務(wù)結(jié)束后系統(tǒng)里不應(yīng)該留下任何實驗痕跡”。Agent 和普通接口最大的區(qū)別在于它有工具調(diào)用副作用。普通接口超時了客戶端不干了就行最多回滾一個數(shù)據(jù)庫事務(wù)但 Agent 超時時可能已經(jīng)調(diào)用了三個工具每個工具都有外部副作用有些還是不可回滾的——比如發(fā)了郵件、推送了通知、寫了日志。這些副效應(yīng)不會隨著超時自動消失它們就是狀態(tài)殘留。狀態(tài)殘留本質(zhì)上會讓系統(tǒng)從“每個請求獨立”退化成“請求之間有隱含依賴”。第一次請求失敗留下的殘留數(shù)據(jù)會污染第二次請求的判定邏輯重試觸發(fā)的同一工具調(diào)用會因為前一次的殘留數(shù)據(jù)而產(chǎn)生重復(fù)副作用。這些問題都比“一個接口超時”嚴(yán)重得多因為它們讓整個系統(tǒng)變得不可預(yù)期。3.2 Agent 狀態(tài)到底包含哪些東西要清理狀態(tài)得先定義狀態(tài)。Agent 執(zhí)行過程中的狀態(tài)不是單一的我拆出了四層。第一層是消息歷史。也就是 LLM 對話上下文包括用戶問題、Assistant 的思維鏈輸出、工具觀察結(jié)果。這些數(shù)據(jù)在失敗后處理起來比較微妙如果全都清掉重試時 LLM 就失憶了會重新犯錯如果全保留失敗時的誤區(qū)會被帶進(jìn)重試?yán)風(fēng)LM 可能執(zhí)著于錯誤的思路。后面我會給一個更具體的處理策略。第二層是執(zhí)行軌跡。包括已完成的工具調(diào)用記錄、每輪 LLM 返回的中間 Action、耗時指標(biāo)、 token 消耗。這個狀態(tài)主要用于可觀測性和審計失敗后不一定要清但要標(biāo)記該軌跡對應(yīng)的任務(wù)已失敗避免它出現(xiàn)在后續(xù)查詢中。第三層是業(yè)務(wù)副作用狀態(tài)。這是最容易出問題的。Agent 已經(jīng)寫進(jìn)數(shù)據(jù)庫的記錄、已經(jīng)創(chuàng)建的文件、已經(jīng)發(fā)送的通知這些都不在 Agent 進(jìn)程內(nèi)但它們是 Agent 行為的結(jié)果。失敗后這些副作用需要被識別并根據(jù)業(yè)務(wù)規(guī)則決定是回滾、補(bǔ)償還是標(biāo)記為“孤兒數(shù)據(jù)”。第四層是運(yùn)行期上下文。包括當(dāng)前執(zhí)行到的步驟編號、臨時變量、工具參數(shù)緩存、重試計數(shù)等。這一層最簡單進(jìn)程內(nèi)對象請求結(jié)束自然銷毀但如果用了異步任務(wù)架構(gòu)要小心上下文對象在 worker 里滯留導(dǎo)致后續(xù)任務(wù)讀到上一輪的殘留數(shù)據(jù)。3.3 什么狀態(tài)該清什么不能清狀態(tài)清理最容易踩的坑就是“一刀切”。我們一開始的策略很簡單失敗就清空所有狀態(tài)重新開始效果差到令人崩潰。因為清理掉狀態(tài)的同時也清理掉了重試的意義LLM 沒有任何記憶的情況下重試和第一次執(zhí)行沒有任何區(qū)別該失敗的還是會失敗。后來我總結(jié)出的原則是分情況處理。如果重試策略是“從當(dāng)前 Agent 循環(huán)上下文繼續(xù)”那么消息歷史要保留但只保留到出問題步驟之前的部分出錯的那一步觀察結(jié)果要替換為新重試的結(jié)果。如果重試策略是“重置 Agent 內(nèi)部循環(huán)但保留用戶目標(biāo)”那么消息歷史要壓縮成一個整體摘要再把用戶原始需求拼接進(jìn)去。這樣 LLM 有背景知識又不會被上一次的具體錯誤路徑綁架。真正的死坑在第三層業(yè)務(wù)副作用狀態(tài)。這一層不存在統(tǒng)一的“清理”方案完全取決于業(yè)務(wù)語義。有些副作用是可回滾的比如數(shù)據(jù)庫操作fail 之后在 catch 塊里執(zhí)行反向操作就行。有些是不可回滾的比如發(fā)送郵件這類只能靠事后補(bǔ)償或人工介入。還有一種很隱蔽的情況Agent 在一次執(zhí)行中先調(diào)用了“創(chuàng)建工單”工具又調(diào)用了“分配負(fù)責(zé)人”工具最后因為超時失敗了。工單已創(chuàng)建但負(fù)責(zé)人分配沒完成這時如果簡單回滾掉“創(chuàng)建工單”會把一個本來部分有效的結(jié)果直接變成無用工單但如果不回滾用戶重試時又要重新創(chuàng)建一張工單就出現(xiàn)了重復(fù)數(shù)據(jù)。處理這種半完成業(yè)務(wù)狀態(tài)我最終采用的是快照與恢復(fù)模式。Agent 啟動時先建立業(yè)務(wù)快照記錄當(dāng)前業(yè)務(wù)狀態(tài)指紋執(zhí)行期間每次工具調(diào)用前先提交一次“檢查點”內(nèi)容是當(dāng)前商業(yè)操作的描述和參數(shù)失敗時根據(jù)檢查點生成一條懸空任務(wù)記錄讓用戶在 UI 里自行決定是繼續(xù)執(zhí)行、回滾還是作廢。這個模式避免了自動清理的粗暴性讓“清不清、怎么清”變成一個用戶可決策的流程。注意自動回滾業(yè)務(wù)副作用是一個高風(fēng)險的默認(rèn)行為。如果無法 100% 確認(rèn)工具操作的語義寧可保留現(xiàn)場并標(biāo)記為異常也不要在業(yè)務(wù)數(shù)據(jù)上執(zhí)行有創(chuàng)造性的“清理動作”。4. 實操中的實現(xiàn)細(xì)節(jié)4.1 帶狀態(tài)清理的 Agent 執(zhí)行環(huán)境設(shè)計具體到代碼層面我重構(gòu)了 Agent 的核心執(zhí)行器。整個設(shè)計可以用一句話概括所有手動“清狀態(tài)”的地方都被顯式化成一個 checkpoint 字段失敗后用統(tǒng)一的事件機(jī)制處理而不是散落在各種 except 塊里。下面給出核心執(zhí)行器的結(jié)構(gòu)代碼這是一個基于偽代碼的簡化示例體現(xiàn)的是分層思路實際工程實現(xiàn)會比這復(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: # 這里實際會取業(yè)務(wù)側(cè)關(guān)鍵數(shù)據(jù)表的CRC或版本號 return hash(str(self.tool_results)) dataclass class AgentAction: tool_name: str tool_args: dict thought: str我用 dataclass 定義了執(zhí)行上下文核心是 checkpoints 列表。列表里存放每一步工具調(diào)用的檢查點檢查點是后續(xù)做狀態(tài)清理的原始依據(jù)。然后定義超時參數(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 {} # 由外部注冊 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 獲取下一步動作 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í)行工具帶動作級超時和重試 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í)行成功記錄檢查點這里的檢查點就是“狀態(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)記錯誤并記錄到觀察結(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 # 會話超時 context.status TaskStatus.TIMEOUT return context def _call_tool_with_policy(self, context, action, deadline): # 執(zhí)行工具調(diào)用帶動作級超時 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 # 超時是最典型的瞬時錯誤做退避重試 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ù)校驗、權(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í)行工具且最近一步失敗時評估是否進(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)有過成功的業(yè)務(wù)操作后續(xù)再失敗就不直接標(biāo)記 FAILED而是標(biāo)記 SUSPENDED。SUSPENDED 狀態(tài)下系統(tǒng)會保留所有 checkpoint 和上下文用戶可以查看到達(dá)了哪一步、最后成功的是什么、失敗的是什么然后手動選擇繼續(xù)或回滾。4.2 狀態(tài)清理策略的實現(xiàn)上面的執(zhí)行器只負(fù)責(zé)把狀態(tài)遺漏保留下來真正的清理動作是在收到“確認(rèn)失敗”事件后觸發(fā)的。我實現(xiàn)了一個StateCleanupManager它的職責(zé)是根據(jù)業(yè)務(wù)配置決定某個工具調(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)清理策略注冊中心。 每個工具在注冊時必須同時聲明自己的清理策略。 這是解決狀態(tài)清理問題的最關(guān)鍵設(shè)計——清理不是萬能的后置處理 而是每個工具的固有屬性必須在工具研發(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生成清理計劃。 返回的每個計劃項都標(biāo)注了類型供上層做自動化或人工處理。 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, }這個設(shè)計的核心是一個原則工具的清理策略必須和工具本身同時注冊。我見過太多項目先開發(fā)幾十個工具最后才想起來做狀態(tài)清理結(jié)果每一個工具都需要靠逆向讀代碼來猜能不能安全回滾那個成本高到你想哭。如果從一開始每個工具定義時就順手聲明一下清理策略這個事基本是無縫集成的。清理策略也要有類型區(qū)分。我給每個清理動作打標(biāo)可自動回滾的比如修改類操作、可自動回退但破壞性的比如刪除數(shù)據(jù)、必須人工介入的比如發(fā)送消息。這三類在系統(tǒng)里走完全不同的流程自動回滾直接執(zhí)行破壞性回滾需要二次確認(rèn)人工介入則掛起并通知用戶。這樣既保證了自動化程度又避免了自動執(zhí)行的二次傷害。4.3 冪等設(shè)計的落地重試伴隨的另一個核心問題就是冪等。如果我的工具調(diào)用了“寫入數(shù)據(jù)庫”操作第一次寫成功了但沒有返回重試時會不會寫兩遍答案是會除非工具本身做了冪等控制。冪等控制的通用方案是在工具參數(shù)里強(qiáng)制要求一個request_id這個 ID 在 Agent 生成動作時就注入。數(shù)據(jù)庫側(cè)用這個 ID 作為唯一鍵重復(fù)寫入時只返回首次結(jié)果不再插入新記錄。這個設(shè)計對重試的收益是巨大的因為絕大多數(shù)重試場景都是“結(jié)果未知但請求可能已成功”冪等可以完美消除重復(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): # 另一個實例正在執(zhí)行同一請求等待結(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ù)”問題基本絕跡。這里說一句實在話冪等設(shè)計比狀態(tài)清理的任何后置方案都更根本如果工具本身可冪等大量的重試擾動都能被自動吸收。狀態(tài)清理是用來處理不可冪等場景的兜底兩者配合才是完整方案。提示給工具調(diào)用加冪等守衛(wèi)的成本遠(yuǎn)低于復(fù)盤事故成本。建議在 Agent 工具開發(fā)規(guī)范里把“必須支持 request_id 冪等”定為默認(rèn)要求而不是可有可無的加分項。5. 超時重試與狀態(tài)清理的完整聯(lián)動5.1 失敗后的完整處理流程當(dāng)一次 Agent 執(zhí)行進(jìn)入失敗路徑后完整流程是這樣的。先看錯誤類型。如果是網(wǎng)絡(luò)層瞬時錯誤且尚未產(chǎn)生任何業(yè)務(wù)副作用最簡單的處理是走自動重試重試上限內(nèi)大概率能成功這個場景不走狀態(tài)清理邏輯。如果是 LLM 層的邏輯死循環(huán)比如反復(fù)調(diào)用相同工具且參數(shù)完全一致此時啟動循環(huán)檢測截斷對話歷史到最近三次提示 LLM “檢測到重復(fù)動作請嘗試完全不同的方法”。這一步很有效因為很多死循環(huán)其實是 LLM 對上下文理解偏差導(dǎo)致的只要打斷它的慣性即可。如果已經(jīng)產(chǎn)生了業(yè)務(wù)副作用再分類處理。副作用是可以自動回滾的比如工具里有對應(yīng)的 rollback 函數(shù)就自動執(zhí)行回滾并記錄審計日志。副作用不可回滾那就把任務(wù)標(biāo)記為 SUSPENDED在 UI 上給用戶展示“當(dāng)前任務(wù)已暫?！钡目ㄆㄆ淹瓿傻牟襟E清單和剩余可選操作繼續(xù)執(zhí)行、整單作廢、標(biāo)記人工處理。這一步?jīng)]有任何自動化魔法是產(chǎn)品層面必須接受的復(fù)雜度。到這里整個流程就算閉環(huán)了超時保護(hù)住最壞情況下的資源占用重試覆蓋掉瞬時失敗和可恢復(fù)故障狀態(tài)清理處理掉失敗后的副作用殘留。三者組合才敢說這個 Agent 真正具備了一點工程化的穩(wěn)定性基礎(chǔ)。5.2 可觀測性要怎么配合狀態(tài)清理做得好不好很大程度上取決于能不能看到狀態(tài)。我強(qiáng)烈建議在 Agent 執(zhí)行上下文中把每一步的 checkpoint、狀態(tài)指紋、清理計劃全部暴露到日志系統(tǒng)里。我用過的最佳組合是結(jié)構(gòu)化日志JSON 格式進(jìn) ELK執(zhí)行軌跡進(jìn) Jaeger業(yè)務(wù)懸空任務(wù)單獨存一張表。三者配合排查一次線上事故的效率能提高一個數(shù)量級。這里有一個細(xì)節(jié)值得注意checkpoint 里的business_state_fingerprint每一次工具調(diào)用成功后我會對整個業(yè)務(wù)關(guān)鍵數(shù)據(jù)做一個輕量哈希記錄在檢查點里。后續(xù)如果要做回滾可以直接反查這個指紋驗證現(xiàn)狀是否與當(dāng)時執(zhí)行時一致。如果用戶已經(jīng)基于當(dāng)時的執(zhí)行結(jié)果做了額外操作指紋不匹配這時“無腦回滾”就是危險動作。這個設(shè)計把自動回滾的安全邊界劃得很清楚只允許回滾未被后續(xù)操作污染的數(shù)據(jù)。5.3 什么時候需要引入異步補(bǔ)償機(jī)制如果 Agent 任務(wù)本身是長時間運(yùn)行的比如分鐘級甚至小時級的編排任務(wù)同步執(zhí)行的狀態(tài)清理方案會變得不適用。因為一個任務(wù)的失敗和清理可能相隔幾十分鐘中間用戶可能已經(jīng)查詢過數(shù)據(jù)、其他任務(wù)可能已經(jīng)依賴了它產(chǎn)生的狀態(tài)。這種場景下必須把狀態(tài)清理做成異步補(bǔ)償機(jī)制通常走一條獨立的補(bǔ)償隊列。失敗事件進(jìn)入隊列之后由專門的補(bǔ)償 worker 根據(jù)檢查點執(zhí)行回滾或者用戶決策。這本質(zhì)上就是 Saga 模式在 Agent 場景的實現(xiàn)。如果用上了我上面講的StateCleanupManager和IdempotencyGuard遷移到異步隊列的改動成本很小因為清理邏輯本來就已經(jīng)和 Agent 執(zhí)行器解耦了。6. 常見問題與排查技巧實錄6.1 重試后重復(fù)副作用表現(xiàn)Agent 第一次調(diào)用“創(chuàng)建設(shè)備記錄”工具超時用戶重試出現(xiàn)了兩條相同的設(shè)備記錄。排查思路先看工具是否有冪等控制。如果是歷史代碼沒有冪等層優(yōu)先補(bǔ)request_id冪等方案。如果只是偶爾出現(xiàn)且不是高頻路徑可以退而求其次用清理策略做檢測——在每個檢查點里記錄創(chuàng)建記錄的唯一業(yè)務(wù)鍵失敗后反查該鍵是否已存在存在則視為“已副作用完成”不再回滾只返回首次結(jié)果。這個方案雖然不如真正的冪等優(yōu)雅但能解決兼容老工具的問題。6.2 會話超時真的觸發(fā)了但沒有用表現(xiàn)配置了 120 秒會話超時但請求實際運(yùn)行了 300 秒才返回。排查思路大概率是 LLM 流式調(diào)用阻塞了主線程而超時檢查在 while 循環(huán)里只在下一次迭代才生效。如果你調(diào)用 LLM 是阻塞式的且它一直不返回循環(huán)根本走不到超時檢查。解決方法是把 LLM 調(diào)用也放進(jìn)_run_tool_with_action_timeout的線程池執(zhí)行讓 Action 級和 Session 級的超時真正生效。這是我踩得最痛的坑之一agent 不是所有調(diào)用都會主動感知外層超時標(biāo)志的。6.3 狀態(tài)清理誤殺了正常數(shù)據(jù)表現(xiàn)任務(wù)失敗后自動回滾回滾后另一個正常任務(wù)的數(shù)據(jù)也消失了。排查思路典型的回滾邊界定義錯誤。某個工具的 rollback 函數(shù)寫得過于寬泛比如 DELETE 語句只按類型過濾而沒帶 request_id。修正方法就是升級清理策略的精度rollback 操作必須攜帶原調(diào)用的完整參數(shù)和 request_id且回滾范圍必須只能作用于該次調(diào)用產(chǎn)生的最小粒度數(shù)據(jù)。另外這也能用指紋校驗擋掉回滾前比對業(yè)務(wù)指紋不一致就不執(zhí)行自動回滾轉(zhuǎn)人工。6.4 清理策略速查表失敗類型是否重試典型工具示例狀態(tài)清理策略是否走人工網(wǎng)絡(luò)瞬斷是指數(shù)退避HTTP調(diào)用、LLM調(diào)用無需清理否LLM死循環(huán)否打斷重置推理步驟截斷上下文否非冪等寫入超時是需冪等數(shù)據(jù)庫INSERT冪等去重否已執(zhí)行的通知類工具否發(fā)郵件、推送通知不可回滾記錄懸空是已創(chuàng)建的工單未分配負(fù)責(zé)人否內(nèi)部業(yè)務(wù)API部分回滾/掛起是臨時文件、分布式鎖是緩存、文件、鎖自動清理否這張表是我們在實踐中沉淀下來的分類標(biāo)準(zhǔn)基本上所有工具都能歸入其中某一類。每次新增工具時第一件事就是想清楚它落在哪一行然后在代碼注冊時把清理策略配置好。這會成為 Agent 穩(wěn)定性的一道重要防線。7. 一點個人體會這次改造讓我對 Agent 工程化有了完全不同的理解。一開始我以為超時和重試是嚴(yán)謹(jǐn)性的體現(xiàn)做完了才發(fā)現(xiàn)它們頂多算 App 的入口保護(hù)真正的工程難點全在業(yè)務(wù)副作用的管理上。超時和重試本質(zhì)上是處理時間和頻率的問題而狀態(tài)清理處理的是真實世界被 Agent 改變了之后怎么辦的問題。后者復(fù)雜得多因為你不能理論化地“撤銷”一封已發(fā)出的郵件不能憑空“復(fù)原”一次外部系統(tǒng)的狀態(tài)修改。根據(jù)我的個人經(jīng)驗做 Agent 穩(wěn)定性的項目建議不要一開始就追求全自動清理而是先把“檢查點 懸空任務(wù) 人工決策”這套半自動方案跑通讓團(tuán)隊的認(rèn)知上線再逐步把高頻且安全的路徑自動化。如果倒過來一上來就寫一堆自動回滾代碼大概率會在生產(chǎn)環(huán)境的某個深夜搞出一場事故。最后再分享一個我實際用得很順手的小技巧給 Agent 上下文的每個 checkpoint 都加一個created_at字段配合任務(wù)級created_at可以算出“某個工具執(zhí)行后到底存活了多久”。當(dāng)一個任務(wù)失敗但用戶遲遲不處理懸空任務(wù)時超出 24 小時的懸空任務(wù)會被自動升級為特殊狀態(tài)并通知負(fù)責(zé)人。這個功能救了我好幾次因為在真實的業(yè)務(wù)里懸而不決的狀態(tài)往往比失敗本身更危險。希望這篇內(nèi)容對你的 Agent 工程化之路有一點點幫助如果你也在思考類似的問題歡迎拿這個故事去檢驗自己的設(shè)計很多時候只有在出過一次事之后才會真正理解狀態(tài)清理的分量。 AI Agent超時重試與狀態(tài)清理的工程實踐與踩坑指南
給 Agent 加超時重試本身不難難的是每次執(zhí)行失敗之后怎么把現(xiàn)場干干凈凈地收掉。這個教訓(xùn)我是在線上被真實流量教育過之后才徹底想明白的。今天把整個設(shè)計和踩坑過程整理出來希望能幫你少走一段彎路。1. 項目背景一個 AI Agent 的穩(wěn)定性改造先交代一下背景。我負(fù)責(zé)的項目是一個基于 LangChain FastAPI 的 AI Agent 服務(wù)核心能力是讓大模型自主完成多步任務(wù)比如查詢數(shù)據(jù)庫、調(diào)用內(nèi)部 API、生成報表、發(fā)通知甚至跨系統(tǒng)編排動作。外層接口是同步 HTTP內(nèi)部走的是循環(huán)式 Agent 執(zhí)行LLM 推理 → 工具調(diào)用 → 結(jié)果回填 → 再次推理直到模型判斷任務(wù)結(jié)束。項目上線初期穩(wěn)定運(yùn)行但流量一上來問題就暴露了。典型癥狀包括用戶請求偶爾 504、LLM 服務(wù)偶發(fā)超時導(dǎo)致整個請求卡死、工具調(diào)用失敗后 Agent 像“失憶”一樣重復(fù)做無用功。于是我們啟動了一輪穩(wěn)定性改造核心目標(biāo)就兩個字超時和重試。聽起來非常常規(guī)做起來也確實能解決一部分問題但改完之后真正折磨我的是第三個字狀態(tài)清理。這里也順便說清楚這篇文章適合誰看。如果你正在做 AI Agent 相關(guān)的工程化落地尤其是涉及多輪工具調(diào)用、狀態(tài)持久化、甚至異步任務(wù)編排的這篇文章提到的方案和坑會很有參考價值。如果你只是調(diào)通了一個基于 LangChain 的 demo對超時重試的認(rèn)知還停留在 requests 庫加個 timeout 參數(shù)那更建議從頭讀一遍因為你遲早會踩到同樣的坑。2. 超時與重試的落地實現(xiàn)2.1 三層超時設(shè)計先說說超時。這是最容易做錯的地方很多人給 Agent 加超時就是在 HTTP 客戶端上設(shè)置一下或者給 FastAPI 接口加個超時中間件但實際上 Agent 的超時體系是分層的至少拆成三層才夠用。第一層是網(wǎng)絡(luò)請求超時。這是最底層的針對 Agent 依賴的所有外部服務(wù)LLM 供應(yīng)商 API、數(shù)據(jù)庫連接、內(nèi)部工具服務(wù)的 HTTP 接口。這里我統(tǒng)一走的是 httpx 或 aiohttp 的 timeout 配置連接超時一般給 5 秒讀超時根據(jù)服務(wù)特點差異化配置。LLM 調(diào)用這類“慢服務(wù)”讀超時給 60 秒因為大模型流式輸出本來就慢內(nèi)部 API 讀超時給 10 秒超過這個閾值基本就是服務(wù)有問題了。第二層是動作級超時。這一層很多人會忽略。Agent 的“一個動作”不只是調(diào)用一次外部 API它包含一次完整的工具調(diào)用周期LLM 生成工具參數(shù) → 執(zhí)行工具 → 返回結(jié)果給 LLM。這個周期里任何一個環(huán)節(jié)都可能卡住尤其是工具本身內(nèi)部有重試邏輯的時候整個動作可能被拖到幾分鐘。所以我會給單個工具動作設(shè)置獨立超時稱為 action_timeout默認(rèn) 30 秒。第三層是會話級超時。這是最高層的兜底對應(yīng)整個 Agent 任務(wù)的總執(zhí)行時長?，F(xiàn)實場景里 LLM 可能陷入死循環(huán)不斷地調(diào)用同一個工具但參數(shù)永遠(yuǎn)不對如果只有動作超時沒有會話超時整個請求就會耗盡資源。會話超時我一般根據(jù)任務(wù)復(fù)雜度配 120 秒到 300 秒超過直接中斷并返回超時錯誤。三層超時的關(guān)系用數(shù)字表達(dá)就是網(wǎng)絡(luò)超時 動作超時 會話超時每一層都是上一層的兜底防線。這個金字塔結(jié)構(gòu)能保證任意一層出現(xiàn)問題都不會讓整個任務(wù)無限期掛起。2.2 重試策略不能一刀切重試比超時更微妙。一開始我們天真地對所有失敗做統(tǒng)一重試結(jié)果發(fā)現(xiàn)效果很差甚至產(chǎn)生了雙倍故障。原因很簡單重試是有前置條件的不是所有失敗都值得重試。我后來把所有 Agent 執(zhí)行過程中的失敗分成了三類。第一類是瞬時失敗典型代表是網(wǎng)絡(luò)抖動、連接池已滿、第三方服務(wù) 503 或 429。這類失敗是值得重試的但要遵循退避策略。我用的方案是初始等待 500ms每次重試等待時間翻倍最多重試 3 次同時加上一個小的隨機(jī)抖動來避免同時重試造成的驚群效應(yīng)。第二類是永久性失敗比如參數(shù)格式錯誤、API key 無效、業(yè)務(wù)規(guī)則不滿足。這類失敗重試多少次都沒用必須立刻失敗并把錯誤信息原樣返回上層讓 Agent 有機(jī)會重新規(guī)劃。注意這里有個關(guān)鍵點工具的永久性失敗要作為“觀察結(jié)果”反饋給 LLM而不是簡單地終結(jié)整個 Agent 循環(huán)。因為 LLM 可能根據(jù)錯誤信息調(diào)整參數(shù)重來這在 Agent 里是一次“更高級別的重試”。第三類是冪等性失敗也就是請求已經(jīng)成功執(zhí)行但響應(yīng)沒有收到典型例子是網(wǎng)絡(luò)超時后服務(wù)端其實已經(jīng)把數(shù)據(jù)庫記錄寫進(jìn)去了。這種情況下盲目重試會造成重復(fù)寫入。后面我也會說這類問題光靠重試參數(shù)解決不了必須在工具層做冪等設(shè)計。實現(xiàn)重試時我強(qiáng)烈建議不要自己手寫 while 循環(huán)直接用 tenacity 這類庫它支持指數(shù)退避、重試條件函數(shù)、最大重試次數(shù)等配置代碼極其簡潔。提示重試策略必須區(qū)分“值得重試的錯誤”和“不值得重試的錯誤”。統(tǒng)一重試是新手最容易犯的錯誤它在瞬時失敗場景下有效但在 LLM 工具調(diào)用場景下會放大副作用。3. 真正的坑狀態(tài)清理3.1 狀態(tài)清理是什么問題加完超時和重試后系統(tǒng)表面上是穩(wěn)定了但沒過多久就出現(xiàn)了更隱蔽的癥狀。比如用戶發(fā)起一個任務(wù)第一次執(zhí)行超時了用戶重試之后發(fā)現(xiàn)系統(tǒng)里出現(xiàn)了兩份數(shù)據(jù)再比如 Agent 中途失敗但已經(jīng)調(diào)用過的工具副效應(yīng)留在了系統(tǒng)里下次跑同樣的任務(wù)時舊數(shù)據(jù)還在導(dǎo)致結(jié)果錯亂。這就是狀態(tài)清理的范疇。超時和重試解決的是“任務(wù)如何結(jié)束”而狀態(tài)清理解決的是“任務(wù)結(jié)束后系統(tǒng)里不應(yīng)該留下任何實驗痕跡”。Agent 和普通接口最大的區(qū)別在于它有工具調(diào)用副作用。普通接口超時了客戶端不干了就行最多回滾一個數(shù)據(jù)庫事務(wù)但 Agent 超時時可能已經(jīng)調(diào)用了三個工具每個工具都有外部副作用有些還是不可回滾的——比如發(fā)了郵件、推送了通知、寫了日志。這些副效應(yīng)不會隨著超時自動消失它們就是狀態(tài)殘留。狀態(tài)殘留本質(zhì)上會讓系統(tǒng)從“每個請求獨立”退化成“請求之間有隱含依賴”。第一次請求失敗留下的殘留數(shù)據(jù)會污染第二次請求的判定邏輯重試觸發(fā)的同一工具調(diào)用會因為前一次的殘留數(shù)據(jù)而產(chǎn)生重復(fù)副作用。這些問題都比“一個接口超時”嚴(yán)重得多因為它們讓整個系統(tǒng)變得不可預(yù)期。3.2 Agent 狀態(tài)到底包含哪些東西要清理狀態(tài)得先定義狀態(tài)。Agent 執(zhí)行過程中的狀態(tài)不是單一的我拆出了四層。第一層是消息歷史。也就是 LLM 對話上下文包括用戶問題、Assistant 的思維鏈輸出、工具觀察結(jié)果。這些數(shù)據(jù)在失敗后處理起來比較微妙如果全都清掉重試時 LLM 就失憶了會重新犯錯如果全保留失敗時的誤區(qū)會被帶進(jìn)重試?yán)風(fēng)LM 可能執(zhí)著于錯誤的思路。后面我會給一個更具體的處理策略。第二層是執(zhí)行軌跡。包括已完成的工具調(diào)用記錄、每輪 LLM 返回的中間 Action、耗時指標(biāo)、 token 消耗。這個狀態(tài)主要用于可觀測性和審計失敗后不一定要清但要標(biāo)記該軌跡對應(yīng)的任務(wù)已失敗避免它出現(xiàn)在后續(xù)查詢中。第三層是業(yè)務(wù)副作用狀態(tài)。這是最容易出問題的。Agent 已經(jīng)寫進(jìn)數(shù)據(jù)庫的記錄、已經(jīng)創(chuàng)建的文件、已經(jīng)發(fā)送的通知這些都不在 Agent 進(jìn)程內(nèi)但它們是 Agent 行為的結(jié)果。失敗后這些副作用需要被識別并根據(jù)業(yè)務(wù)規(guī)則決定是回滾、補(bǔ)償還是標(biāo)記為“孤兒數(shù)據(jù)”。第四層是運(yùn)行期上下文。包括當(dāng)前執(zhí)行到的步驟編號、臨時變量、工具參數(shù)緩存、重試計數(shù)等。這一層最簡單進(jìn)程內(nèi)對象請求結(jié)束自然銷毀但如果用了異步任務(wù)架構(gòu)要小心上下文對象在 worker 里滯留導(dǎo)致后續(xù)任務(wù)讀到上一輪的殘留數(shù)據(jù)。3.3 什么狀態(tài)該清什么不能清狀態(tài)清理最容易踩的坑就是“一刀切”。我們一開始的策略很簡單失敗就清空所有狀態(tài)重新開始效果差到令人崩潰。因為清理掉狀態(tài)的同時也清理掉了重試的意義LLM 沒有任何記憶的情況下重試和第一次執(zhí)行沒有任何區(qū)別該失敗的還是會失敗。后來我總結(jié)出的原則是分情況處理。如果重試策略是“從當(dāng)前 Agent 循環(huán)上下文繼續(xù)”那么消息歷史要保留但只保留到出問題步驟之前的部分出錯的那一步觀察結(jié)果要替換為新重試的結(jié)果。如果重試策略是“重置 Agent 內(nèi)部循環(huán)但保留用戶目標(biāo)”那么消息歷史要壓縮成一個整體摘要再把用戶原始需求拼接進(jìn)去。這樣 LLM 有背景知識又不會被上一次的具體錯誤路徑綁架。真正的死坑在第三層業(yè)務(wù)副作用狀態(tài)。這一層不存在統(tǒng)一的“清理”方案完全取決于業(yè)務(wù)語義。有些副作用是可回滾的比如數(shù)據(jù)庫操作fail 之后在 catch 塊里執(zhí)行反向操作就行。有些是不可回滾的比如發(fā)送郵件這類只能靠事后補(bǔ)償或人工介入。還有一種很隱蔽的情況Agent 在一次執(zhí)行中先調(diào)用了“創(chuàng)建工單”工具又調(diào)用了“分配負(fù)責(zé)人”工具最后因為超時失敗了。工單已創(chuàng)建但負(fù)責(zé)人分配沒完成這時如果簡單回滾掉“創(chuàng)建工單”會把一個本來部分有效的結(jié)果直接變成無用工單但如果不回滾用戶重試時又要重新創(chuàng)建一張工單就出現(xiàn)了重復(fù)數(shù)據(jù)。處理這種半完成業(yè)務(wù)狀態(tài)我最終采用的是快照與恢復(fù)模式。Agent 啟動時先建立業(yè)務(wù)快照記錄當(dāng)前業(yè)務(wù)狀態(tài)指紋執(zhí)行期間每次工具調(diào)用前先提交一次“檢查點”內(nèi)容是當(dāng)前商業(yè)操作的描述和參數(shù)失敗時根據(jù)檢查點生成一條懸空任務(wù)記錄讓用戶在 UI 里自行決定是繼續(xù)執(zhí)行、回滾還是作廢。這個模式避免了自動清理的粗暴性讓“清不清、怎么清”變成一個用戶可決策的流程。注意自動回滾業(yè)務(wù)副作用是一個高風(fēng)險的默認(rèn)行為。如果無法 100% 確認(rèn)工具操作的語義寧可保留現(xiàn)場并標(biāo)記為異常也不要在業(yè)務(wù)數(shù)據(jù)上執(zhí)行有創(chuàng)造性的“清理動作”。4. 實操中的實現(xiàn)細(xì)節(jié)4.1 帶狀態(tài)清理的 Agent 執(zhí)行環(huán)境設(shè)計具體到代碼層面我重構(gòu)了 Agent 的核心執(zhí)行器。整個設(shè)計可以用一句話概括所有手動“清狀態(tài)”的地方都被顯式化成一個 checkpoint 字段失敗后用統(tǒng)一的事件機(jī)制處理而不是散落在各種 except 塊里。下面給出核心執(zhí)行器的結(jié)構(gòu)代碼這是一個基于偽代碼的簡化示例體現(xiàn)的是分層思路實際工程實現(xiàn)會比這復(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: # 這里實際會取業(yè)務(wù)側(cè)關(guān)鍵數(shù)據(jù)表的CRC或版本號 return hash(str(self.tool_results)) dataclass class AgentAction: tool_name: str tool_args: dict thought: str我用 dataclass 定義了執(zhí)行上下文核心是 checkpoints 列表。列表里存放每一步工具調(diào)用的檢查點檢查點是后續(xù)做狀態(tài)清理的原始依據(jù)。然后定義超時參數(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 {} # 由外部注冊 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 獲取下一步動作 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í)行工具帶動作級超時和重試 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í)行成功記錄檢查點這里的檢查點就是“狀態(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)記錯誤并記錄到觀察結(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 # 會話超時 context.status TaskStatus.TIMEOUT return context def _call_tool_with_policy(self, context, action, deadline): # 執(zhí)行工具調(diào)用帶動作級超時 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 # 超時是最典型的瞬時錯誤做退避重試 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ù)校驗、權(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í)行工具且最近一步失敗時評估是否進(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)有過成功的業(yè)務(wù)操作后續(xù)再失敗就不直接標(biāo)記 FAILED而是標(biāo)記 SUSPENDED。SUSPENDED 狀態(tài)下系統(tǒng)會保留所有 checkpoint 和上下文用戶可以查看到達(dá)了哪一步、最后成功的是什么、失敗的是什么然后手動選擇繼續(xù)或回滾。4.2 狀態(tài)清理策略的實現(xiàn)上面的執(zhí)行器只負(fù)責(zé)把狀態(tài)遺漏保留下來真正的清理動作是在收到“確認(rèn)失敗”事件后觸發(fā)的。我實現(xiàn)了一個StateCleanupManager它的職責(zé)是根據(jù)業(yè)務(wù)配置決定某個工具調(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)清理策略注冊中心。 每個工具在注冊時必須同時聲明自己的清理策略。 這是解決狀態(tài)清理問題的最關(guān)鍵設(shè)計——清理不是萬能的后置處理 而是每個工具的固有屬性必須在工具研發(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生成清理計劃。 返回的每個計劃項都標(biāo)注了類型供上層做自動化或人工處理。 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, }這個設(shè)計的核心是一個原則工具的清理策略必須和工具本身同時注冊。我見過太多項目先開發(fā)幾十個工具最后才想起來做狀態(tài)清理結(jié)果每一個工具都需要靠逆向讀代碼來猜能不能安全回滾那個成本高到你想哭。如果從一開始每個工具定義時就順手聲明一下清理策略這個事基本是無縫集成的。清理策略也要有類型區(qū)分。我給每個清理動作打標(biāo)可自動回滾的比如修改類操作、可自動回退但破壞性的比如刪除數(shù)據(jù)、必須人工介入的比如發(fā)送消息。這三類在系統(tǒng)里走完全不同的流程自動回滾直接執(zhí)行破壞性回滾需要二次確認(rèn)人工介入則掛起并通知用戶。這樣既保證了自動化程度又避免了自動執(zhí)行的二次傷害。4.3 冪等設(shè)計的落地重試伴隨的另一個核心問題就是冪等。如果我的工具調(diào)用了“寫入數(shù)據(jù)庫”操作第一次寫成功了但沒有返回重試時會不會寫兩遍答案是會除非工具本身做了冪等控制。冪等控制的通用方案是在工具參數(shù)里強(qiáng)制要求一個request_id這個 ID 在 Agent 生成動作時就注入。數(shù)據(jù)庫側(cè)用這個 ID 作為唯一鍵重復(fù)寫入時只返回首次結(jié)果不再插入新記錄。這個設(shè)計對重試的收益是巨大的因為絕大多數(shù)重試場景都是“結(jié)果未知但請求可能已成功”冪等可以完美消除重復(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): # 另一個實例正在執(zhí)行同一請求等待結(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ù)”問題基本絕跡。這里說一句實在話冪等設(shè)計比狀態(tài)清理的任何后置方案都更根本如果工具本身可冪等大量的重試擾動都能被自動吸收。狀態(tài)清理是用來處理不可冪等場景的兜底兩者配合才是完整方案。提示給工具調(diào)用加冪等守衛(wèi)的成本遠(yuǎn)低于復(fù)盤事故成本。建議在 Agent 工具開發(fā)規(guī)范里把“必須支持 request_id 冪等”定為默認(rèn)要求而不是可有可無的加分項。5. 超時重試與狀態(tài)清理的完整聯(lián)動5.1 失敗后的完整處理流程當(dāng)一次 Agent 執(zhí)行進(jìn)入失敗路徑后完整流程是這樣的。先看錯誤類型。如果是網(wǎng)絡(luò)層瞬時錯誤且尚未產(chǎn)生任何業(yè)務(wù)副作用最簡單的處理是走自動重試重試上限內(nèi)大概率能成功這個場景不走狀態(tài)清理邏輯。如果是 LLM 層的邏輯死循環(huán)比如反復(fù)調(diào)用相同工具且參數(shù)完全一致此時啟動循環(huán)檢測截斷對話歷史到最近三次提示 LLM “檢測到重復(fù)動作請嘗試完全不同的方法”。這一步很有效因為很多死循環(huán)其實是 LLM 對上下文理解偏差導(dǎo)致的只要打斷它的慣性即可。如果已經(jīng)產(chǎn)生了業(yè)務(wù)副作用再分類處理。副作用是可以自動回滾的比如工具里有對應(yīng)的 rollback 函數(shù)就自動執(zhí)行回滾并記錄審計日志。副作用不可回滾那就把任務(wù)標(biāo)記為 SUSPENDED在 UI 上給用戶展示“當(dāng)前任務(wù)已暫?！钡目ㄆㄆ淹瓿傻牟襟E清單和剩余可選操作繼續(xù)執(zhí)行、整單作廢、標(biāo)記人工處理。這一步?jīng)]有任何自動化魔法是產(chǎn)品層面必須接受的復(fù)雜度。到這里整個流程就算閉環(huán)了超時保護(hù)住最壞情況下的資源占用重試覆蓋掉瞬時失敗和可恢復(fù)故障狀態(tài)清理處理掉失敗后的副作用殘留。三者組合才敢說這個 Agent 真正具備了一點工程化的穩(wěn)定性基礎(chǔ)。5.2 可觀測性要怎么配合狀態(tài)清理做得好不好很大程度上取決于能不能看到狀態(tài)。我強(qiáng)烈建議在 Agent 執(zhí)行上下文中把每一步的 checkpoint、狀態(tài)指紋、清理計劃全部暴露到日志系統(tǒng)里。我用過的最佳組合是結(jié)構(gòu)化日志JSON 格式進(jìn) ELK執(zhí)行軌跡進(jìn) Jaeger業(yè)務(wù)懸空任務(wù)單獨存一張表。三者配合排查一次線上事故的效率能提高一個數(shù)量級。這里有一個細(xì)節(jié)值得注意checkpoint 里的business_state_fingerprint每一次工具調(diào)用成功后我會對整個業(yè)務(wù)關(guān)鍵數(shù)據(jù)做一個輕量哈希記錄在檢查點里。后續(xù)如果要做回滾可以直接反查這個指紋驗證現(xiàn)狀是否與當(dāng)時執(zhí)行時一致。如果用戶已經(jīng)基于當(dāng)時的執(zhí)行結(jié)果做了額外操作指紋不匹配這時“無腦回滾”就是危險動作。這個設(shè)計把自動回滾的安全邊界劃得很清楚只允許回滾未被后續(xù)操作污染的數(shù)據(jù)。5.3 什么時候需要引入異步補(bǔ)償機(jī)制如果 Agent 任務(wù)本身是長時間運(yùn)行的比如分鐘級甚至小時級的編排任務(wù)同步執(zhí)行的狀態(tài)清理方案會變得不適用。因為一個任務(wù)的失敗和清理可能相隔幾十分鐘中間用戶可能已經(jīng)查詢過數(shù)據(jù)、其他任務(wù)可能已經(jīng)依賴了它產(chǎn)生的狀態(tài)。這種場景下必須把狀態(tài)清理做成異步補(bǔ)償機(jī)制通常走一條獨立的補(bǔ)償隊列。失敗事件進(jìn)入隊列之后由專門的補(bǔ)償 worker 根據(jù)檢查點執(zhí)行回滾或者用戶決策。這本質(zhì)上就是 Saga 模式在 Agent 場景的實現(xiàn)。如果用上了我上面講的StateCleanupManager和IdempotencyGuard遷移到異步隊列的改動成本很小因為清理邏輯本來就已經(jīng)和 Agent 執(zhí)行器解耦了。6. 常見問題與排查技巧實錄6.1 重試后重復(fù)副作用表現(xiàn)Agent 第一次調(diào)用“創(chuàng)建設(shè)備記錄”工具超時用戶重試出現(xiàn)了兩條相同的設(shè)備記錄。排查思路先看工具是否有冪等控制。如果是歷史代碼沒有冪等層優(yōu)先補(bǔ)request_id冪等方案。如果只是偶爾出現(xiàn)且不是高頻路徑可以退而求其次用清理策略做檢測——在每個檢查點里記錄創(chuàng)建記錄的唯一業(yè)務(wù)鍵失敗后反查該鍵是否已存在存在則視為“已副作用完成”不再回滾只返回首次結(jié)果。這個方案雖然不如真正的冪等優(yōu)雅但能解決兼容老工具的問題。6.2 會話超時真的觸發(fā)了但沒有用表現(xiàn)配置了 120 秒會話超時但請求實際運(yùn)行了 300 秒才返回。排查思路大概率是 LLM 流式調(diào)用阻塞了主線程而超時檢查在 while 循環(huán)里只在下一次迭代才生效。如果你調(diào)用 LLM 是阻塞式的且它一直不返回循環(huán)根本走不到超時檢查。解決方法是把 LLM 調(diào)用也放進(jìn)_run_tool_with_action_timeout的線程池執(zhí)行讓 Action 級和 Session 級的超時真正生效。這是我踩得最痛的坑之一agent 不是所有調(diào)用都會主動感知外層超時標(biāo)志的。6.3 狀態(tài)清理誤殺了正常數(shù)據(jù)表現(xiàn)任務(wù)失敗后自動回滾回滾后另一個正常任務(wù)的數(shù)據(jù)也消失了。排查思路典型的回滾邊界定義錯誤。某個工具的 rollback 函數(shù)寫得過于寬泛比如 DELETE 語句只按類型過濾而沒帶 request_id。修正方法就是升級清理策略的精度rollback 操作必須攜帶原調(diào)用的完整參數(shù)和 request_id且回滾范圍必須只能作用于該次調(diào)用產(chǎn)生的最小粒度數(shù)據(jù)。另外這也能用指紋校驗擋掉回滾前比對業(yè)務(wù)指紋不一致就不執(zhí)行自動回滾轉(zhuǎn)人工。6.4 清理策略速查表失敗類型是否重試典型工具示例狀態(tài)清理策略是否走人工網(wǎng)絡(luò)瞬斷是指數(shù)退避HTTP調(diào)用、LLM調(diào)用無需清理否LLM死循環(huán)否打斷重置推理步驟截斷上下文否非冪等寫入超時是需冪等數(shù)據(jù)庫INSERT冪等去重否已執(zhí)行的通知類工具否發(fā)郵件、推送通知不可回滾記錄懸空是已創(chuàng)建的工單未分配負(fù)責(zé)人否內(nèi)部業(yè)務(wù)API部分回滾/掛起是臨時文件、分布式鎖是緩存、文件、鎖自動清理否這張表是我們在實踐中沉淀下來的分類標(biāo)準(zhǔn)基本上所有工具都能歸入其中某一類。每次新增工具時第一件事就是想清楚它落在哪一行然后在代碼注冊時把清理策略配置好。這會成為 Agent 穩(wěn)定性的一道重要防線。7. 一點個人體會這次改造讓我對 Agent 工程化有了完全不同的理解。一開始我以為超時和重試是嚴(yán)謹(jǐn)性的體現(xiàn)做完了才發(fā)現(xiàn)它們頂多算 App 的入口保護(hù)真正的工程難點全在業(yè)務(wù)副作用的管理上。超時和重試本質(zhì)上是處理時間和頻率的問題而狀態(tài)清理處理的是真實世界被 Agent 改變了之后怎么辦的問題。后者復(fù)雜得多因為你不能理論化地“撤銷”一封已發(fā)出的郵件不能憑空“復(fù)原”一次外部系統(tǒng)的狀態(tài)修改。根據(jù)我的個人經(jīng)驗做 Agent 穩(wěn)定性的項目建議不要一開始就追求全自動清理而是先把“檢查點 懸空任務(wù) 人工決策”這套半自動方案跑通讓團(tuán)隊的認(rèn)知上線再逐步把高頻且安全的路徑自動化。如果倒過來一上來就寫一堆自動回滾代碼大概率會在生產(chǎn)環(huán)境的某個深夜搞出一場事故。最后再分享一個我實際用得很順手的小技巧給 Agent 上下文的每個 checkpoint 都加一個created_at字段配合任務(wù)級created_at可以算出“某個工具執(zhí)行后到底存活了多久”。當(dāng)一個任務(wù)失敗但用戶遲遲不處理懸空任務(wù)時超出 24 小時的懸空任務(wù)會被自動升級為特殊狀態(tài)并通知負(fù)責(zé)人。這個功能救了我好幾次因為在真實的業(yè)務(wù)里懸而不決的狀態(tài)往往比失敗本身更危險。希望這篇內(nèi)容對你的 Agent 工程化之路有一點點幫助如果你也在思考類似的問題歡迎拿這個故事去檢驗自己的設(shè)計很多時候只有在出過一次事之后才會真正理解狀態(tài)清理的分量。