AI Agent超时重试与状态清理的工程实践与踩坑指南

给 Agent 加超时重试本身不难难的是每次执行失败之后怎么把现场干干净净地收掉。这个教训我是在线上被真实流量教育过之后才彻底想明白的。今天把整个设计和踩坑过程整理出来希望能帮你少走一段弯路。1. 项目背景一个 AI Agent 的稳定性改造先交代一下背景。我负责的项目是一个基于 LangChain FastAPI 的 AI Agent 服务核心能力是让大模型自主完成多步任务比如查询数据库、调用内部 API、生成报表、发通知甚至跨系统编排动作。外层接口是同步 HTTP内部走的是循环式 Agent 执行LLM 推理 → 工具调用 → 结果回填 → 再次推理直到模型判断任务结束。项目上线初期稳定运行但流量一上来问题就暴露了。典型症状包括用户请求偶尔 504、LLM 服务偶发超时导致整个请求卡死、工具调用失败后 Agent 像“失忆”一样重复做无用功。于是我们启动了一轮稳定性改造核心目标就两个字超时和重试。听起来非常常规做起来也确实能解决一部分问题但改完之后真正折磨我的是第三个字状态清理。这里也顺便说清楚这篇文章适合谁看。如果你正在做 AI Agent 相关的工程化落地尤其是涉及多轮工具调用、状态持久化、甚至异步任务编排的这篇文章提到的方案和坑会很有参考价值。如果你只是调通了一个基于 LangChain 的 demo对超时重试的认知还停留在 requests 库加个 timeout 参数那更建议从头读一遍因为你迟早会踩到同样的坑。2. 超时与重试的落地实现2.1 三层超时设计先说说超时。这是最容易做错的地方很多人给 Agent 加超时就是在 HTTP 客户端上设置一下或者给 FastAPI 接口加个超时中间件但实际上 Agent 的超时体系是分层的至少拆成三层才够用。第一层是网络请求超时。这是最底层的针对 Agent 依赖的所有外部服务LLM 供应商 API、数据库连接、内部工具服务的 HTTP 接口。这里我统一走的是 httpx 或 aiohttp 的 timeout 配置连接超时一般给 5 秒读超时根据服务特点差异化配置。LLM 调用这类“慢服务”读超时给 60 秒因为大模型流式输出本来就慢内部 API 读超时给 10 秒超过这个阈值基本就是服务有问题了。第二层是动作级超时。这一层很多人会忽略。Agent 的“一个动作”不只是调用一次外部 API它包含一次完整的工具调用周期LLM 生成工具参数 → 执行工具 → 返回结果给 LLM。这个周期里任何一个环节都可能卡住尤其是工具本身内部有重试逻辑的时候整个动作可能被拖到几分钟。所以我会给单个工具动作设置独立超时称为 action_timeout默认 30 秒。第三层是会话级超时。这是最高层的兜底对应整个 Agent 任务的总执行时长。现实场景里 LLM 可能陷入死循环不断地调用同一个工具但参数永远不对如果只有动作超时没有会话超时整个请求就会耗尽资源。会话超时我一般根据任务复杂度配 120 秒到 300 秒超过直接中断并返回超时错误。三层超时的关系用数字表达就是网络超时 动作超时 会话超时每一层都是上一层的兜底防线。这个金字塔结构能保证任意一层出现问题都不会让整个任务无限期挂起。2.2 重试策略不能一刀切重试比超时更微妙。一开始我们天真地对所有失败做统一重试结果发现效果很差甚至产生了双倍故障。原因很简单重试是有前置条件的不是所有失败都值得重试。我后来把所有 Agent 执行过程中的失败分成了三类。第一类是瞬时失败典型代表是网络抖动、连接池已满、第三方服务 503 或 429。这类失败是值得重试的但要遵循退避策略。我用的方案是初始等待 500ms每次重试等待时间翻倍最多重试 3 次同时加上一个小的随机抖动来避免同时重试造成的惊群效应。第二类是永久性失败比如参数格式错误、API key 无效、业务规则不满足。这类失败重试多少次都没用必须立刻失败并把错误信息原样返回上层让 Agent 有机会重新规划。注意这里有个关键点工具的永久性失败要作为“观察结果”反馈给 LLM而不是简单地终结整个 Agent 循环。因为 LLM 可能根据错误信息调整参数重来这在 Agent 里是一次“更高级别的重试”。第三类是幂等性失败也就是请求已经成功执行但响应没有收到典型例子是网络超时后服务端其实已经把数据库记录写进去了。这种情况下盲目重试会造成重复写入。后面我也会说这类问题光靠重试参数解决不了必须在工具层做幂等设计。实现重试时我强烈建议不要自己手写 while 循环直接用 tenacity 这类库它支持指数退避、重试条件函数、最大重试次数等配置代码极其简洁。提示重试策略必须区分“值得重试的错误”和“不值得重试的错误”。统一重试是新手最容易犯的错误它在瞬时失败场景下有效但在 LLM 工具调用场景下会放大副作用。3. 真正的坑状态清理3.1 状态清理是什么问题加完超时和重试后系统表面上是稳定了但没过多久就出现了更隐蔽的症状。比如用户发起一个任务第一次执行超时了用户重试之后发现系统里出现了两份数据再比如 Agent 中途失败但已经调用过的工具副效应留在了系统里下次跑同样的任务时旧数据还在导致结果错乱。这就是状态清理的范畴。超时和重试解决的是“任务如何结束”而状态清理解决的是“任务结束后系统里不应该留下任何实验痕迹”。Agent 和普通接口最大的区别在于它有工具调用副作用。普通接口超时了客户端不干了就行最多回滚一个数据库事务但 Agent 超时时可能已经调用了三个工具每个工具都有外部副作用有些还是不可回滚的——比如发了邮件、推送了通知、写了日志。这些副效应不会随着超时自动消失它们就是状态残留。状态残留本质上会让系统从“每个请求独立”退化成“请求之间有隐含依赖”。第一次请求失败留下的残留数据会污染第二次请求的判定逻辑重试触发的同一工具调用会因为前一次的残留数据而产生重复副作用。这些问题都比“一个接口超时”严重得多因为它们让整个系统变得不可预期。3.2 Agent 状态到底包含哪些东西要清理状态得先定义状态。Agent 执行过程中的状态不是单一的我拆出了四层。第一层是消息历史。也就是 LLM 对话上下文包括用户问题、Assistant 的思维链输出、工具观察结果。这些数据在失败后处理起来比较微妙如果全都清掉重试时 LLM 就失忆了会重新犯错如果全保留失败时的误区会被带进重试里LLM 可能执着于错误的思路。后面我会给一个更具体的处理策略。第二层是执行轨迹。包括已完成的工具调用记录、每轮 LLM 返回的中间 Action、耗时指标、 token 消耗。这个状态主要用于可观测性和审计失败后不一定要清但要标记该轨迹对应的任务已失败避免它出现在后续查询中。第三层是业务副作用状态。这是最容易出问题的。Agent 已经写进数据库的记录、已经创建的文件、已经发送的通知这些都不在 Agent 进程内但它们是 Agent 行为的结果。失败后这些副作用需要被识别并根据业务规则决定是回滚、补偿还是标记为“孤儿数据”。第四层是运行期上下文。包括当前执行到的步骤编号、临时变量、工具参数缓存、重试计数等。这一层最简单进程内对象请求结束自然销毁但如果用了异步任务架构要小心上下文对象在 worker 里滞留导致后续任务读到上一轮的残留数据。3.3 什么状态该清什么不能清状态清理最容易踩的坑就是“一刀切”。我们一开始的策略很简单失败就清空所有状态重新开始效果差到令人崩溃。因为清理掉状态的同时也清理掉了重试的意义LLM 没有任何记忆的情况下重试和第一次执行没有任何区别该失败的还是会失败。后来我总结出的原则是分情况处理。如果重试策略是“从当前 Agent 循环上下文继续”那么消息历史要保留但只保留到出问题步骤之前的部分出错的那一步观察结果要替换为新重试的结果。如果重试策略是“重置 Agent 内部循环但保留用户目标”那么消息历史要压缩成一个整体摘要再把用户原始需求拼接进去。这样 LLM 有背景知识又不会被上一次的具体错误路径绑架。真正的死坑在第三层业务副作用状态。这一层不存在统一的“清理”方案完全取决于业务语义。有些副作用是可回滚的比如数据库操作fail 之后在 catch 块里执行反向操作就行。有些是不可回滚的比如发送邮件这类只能靠事后补偿或人工介入。还有一种很隐蔽的情况Agent 在一次执行中先调用了“创建工单”工具又调用了“分配负责人”工具最后因为超时失败了。工单已创建但负责人分配没完成这时如果简单回滚掉“创建工单”会把一个本来部分有效的结果直接变成无用工单但如果不回滚用户重试时又要重新创建一张工单就出现了重复数据。处理这种半完成业务状态我最终采用的是快照与恢复模式。Agent 启动时先建立业务快照记录当前业务状态指纹执行期间每次工具调用前先提交一次“检查点”内容是当前商业操作的描述和参数失败时根据检查点生成一条悬空任务记录让用户在 UI 里自行决定是继续执行、回滚还是作废。这个模式避免了自动清理的粗暴性让“清不清、怎么清”变成一个用户可决策的流程。注意自动回滚业务副作用是一个高风险的默认行为。如果无法 100% 确认工具操作的语义宁可保留现场并标记为异常也不要在业务数据上执行有创造性的“清理动作”。4. 实操中的实现细节4.1 带状态清理的 Agent 执行环境设计具体到代码层面我重构了 Agent 的核心执行器。整个设计可以用一句话概括所有手动“清状态”的地方都被显式化成一个 checkpoint 字段失败后用统一的事件机制处理而不是散落在各种 except 块里。下面给出核心执行器的结构代码这是一个基于伪代码的简化示例体现的是分层思路实际工程实现会比这复杂一些但骨架是有效的。# 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 # 半完成状态需要人工决策 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: # 这里实际会取业务侧关键数据表的CRC或版本号 return hash(str(self.tool_results)) dataclass class AgentAction: tool_name: str tool_args: dict thought: str我用 dataclass 定义了执行上下文核心是 checkpoints 列表。列表里存放每一步工具调用的检查点检查点是后续做状态清理的原始依据。然后定义超时参数和执行器主体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. 调用 LLM 获取下一步动作 action self._invoke_llm_with_timeout(context, deadline) if action is None: return self._mark_failed(context, llm response none after retries) # LLM 判断任务结束 if action.tool_name FINISH: context.status TaskStatus.SUCCEEDED return context # 2. 执行工具带动作级超时和重试 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]: # 执行成功记录检查点这里的检查点就是“状态清理”的基础 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: # 失败处理标记错误并记录到观察结果让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]}} ) # 每步结束后检查是否需要提前挂起半完成状态 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): # 执行工具调用带动作级超时 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: # 区分可重试异常和永久异常 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): # 实际判断会看异常类型连接错误、超时、限流可重试 # 参数校验、权限、业务拒绝不可重试。 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): # 当存在已执行工具且最近一步失败时评估是否进入挂起 if context.run_state.get(last_error) is not None: # 如果已经产生了不可忽略的业务用户流程如创建了工单、发了通知就挂起 return len(context.checkpoints) 0 return False这里_needs_suspend的逻辑是核心只要之前已经有过成功的业务操作后续再失败就不直接标记 FAILED而是标记 SUSPENDED。SUSPENDED 状态下系统会保留所有 checkpoint 和上下文用户可以查看到达了哪一步、最后成功的是什么、失败的是什么然后手动选择继续或回滚。4.2 状态清理策略的实现上面的执行器只负责把状态遗漏保留下来真正的清理动作是在收到“确认失败”事件后触发的。我实现了一个StateCleanupManager它的职责是根据业务配置决定某个工具调用结果要不要回滚、怎么回滚。# 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 # 是否是破坏性回滚如删除数据 requires_manual: bool # 是否需要人工决策 class StateCleanupManager: 状态清理策略注册中心。 每个工具在注册时必须同时声明自己的清理策略。 这是解决状态清理问题的最关键设计——清理不是万能的后置处理 而是每个工具的固有属性必须在工具研发阶段就定义。 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]: 根据上下文里的 checkpoints生成清理计划。 返回的每个计划项都标注了类型供上层做自动化或人工处理。 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, }这个设计的核心是一个原则工具的清理策略必须和工具本身同时注册。我见过太多项目先开发几十个工具最后才想起来做状态清理结果每一个工具都需要靠逆向读代码来猜能不能安全回滚那个成本高到你想哭。如果从一开始每个工具定义时就顺手声明一下清理策略这个事基本是无缝集成的。清理策略也要有类型区分。我给每个清理动作打标可自动回滚的比如修改类操作、可自动回退但破坏性的比如删除数据、必须人工介入的比如发送消息。这三类在系统里走完全不同的流程自动回滚直接执行破坏性回滚需要二次确认人工介入则挂起并通知用户。这样既保证了自动化程度又避免了自动执行的二次伤害。4.3 幂等设计的落地重试伴随的另一个核心问题就是幂等。如果我的工具调用了“写入数据库”操作第一次写成功了但没有返回重试时会不会写两遍答案是会除非工具本身做了幂等控制。幂等控制的通用方案是在工具参数里强制要求一个request_id这个 ID 在 Agent 生成动作时就注入。数据库侧用这个 ID 作为唯一键重复写入时只返回首次结果不再插入新记录。这个设计对重试的收益是巨大的因为绝大多数重试场景都是“结果未知但请求可能已成功”幂等可以完美消除重复副作用。# idempotent_tool.py import hashlib import json from typing import Callable class IdempotencyGuard: 幂等守卫给任意工具调用包一层幂等控制。 核心思路用 request_id 工具名 参数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() # 尝试获取已有结果 if self.store.exists(key): return self.store.get(key) # 执行前先做“预定”标记防止并发下重复执行 if not self.store.acquire_lock(key): # 另一个实例正在执行同一请求等待结果 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)这样改造之后之前线上出现的“用户重试导致双份数据”问题基本绝迹。这里说一句实在话幂等设计比状态清理的任何后置方案都更根本如果工具本身可幂等大量的重试扰动都能被自动吸收。状态清理是用来处理不可幂等场景的兜底两者配合才是完整方案。提示给工具调用加幂等守卫的成本远低于复盘事故成本。建议在 Agent 工具开发规范里把“必须支持 request_id 幂等”定为默认要求而不是可有可无的加分项。5. 超时重试与状态清理的完整联动5.1 失败后的完整处理流程当一次 Agent 执行进入失败路径后完整流程是这样的。先看错误类型。如果是网络层瞬时错误且尚未产生任何业务副作用最简单的处理是走自动重试重试上限内大概率能成功这个场景不走状态清理逻辑。如果是 LLM 层的逻辑死循环比如反复调用相同工具且参数完全一致此时启动循环检测截断对话历史到最近三次提示 LLM “检测到重复动作请尝试完全不同的方法”。这一步很有效因为很多死循环其实是 LLM 对上下文理解偏差导致的只要打断它的惯性即可。如果已经产生了业务副作用再分类处理。副作用是可以自动回滚的比如工具里有对应的 rollback 函数就自动执行回滚并记录审计日志。副作用不可回滚那就把任务标记为 SUSPENDED在 UI 上给用户展示“当前任务已暂停”的卡片卡片包含已完成的步骤清单和剩余可选操作继续执行、整单作废、标记人工处理。这一步没有任何自动化魔法是产品层面必须接受的复杂度。到这里整个流程就算闭环了超时保护住最坏情况下的资源占用重试覆盖掉瞬时失败和可恢复故障状态清理处理掉失败后的副作用残留。三者组合才敢说这个 Agent 真正具备了一点工程化的稳定性基础。5.2 可观测性要怎么配合状态清理做得好不好很大程度上取决于能不能看到状态。我强烈建议在 Agent 执行上下文中把每一步的 checkpoint、状态指纹、清理计划全部暴露到日志系统里。我用过的最佳组合是结构化日志JSON 格式进 ELK执行轨迹进 Jaeger业务悬空任务单独存一张表。三者配合排查一次线上事故的效率能提高一个数量级。这里有一个细节值得注意checkpoint 里的business_state_fingerprint每一次工具调用成功后我会对整个业务关键数据做一个轻量哈希记录在检查点里。后续如果要做回滚可以直接反查这个指纹验证现状是否与当时执行时一致。如果用户已经基于当时的执行结果做了额外操作指纹不匹配这时“无脑回滚”就是危险动作。这个设计把自动回滚的安全边界划得很清楚只允许回滚未被后续操作污染的数据。5.3 什么时候需要引入异步补偿机制如果 Agent 任务本身是长时间运行的比如分钟级甚至小时级的编排任务同步执行的状态清理方案会变得不适用。因为一个任务的失败和清理可能相隔几十分钟中间用户可能已经查询过数据、其他任务可能已经依赖了它产生的状态。这种场景下必须把状态清理做成异步补偿机制通常走一条独立的补偿队列。失败事件进入队列之后由专门的补偿 worker 根据检查点执行回滚或者用户决策。这本质上就是 Saga 模式在 Agent 场景的实现。如果用上了我上面讲的StateCleanupManager和IdempotencyGuard迁移到异步队列的改动成本很小因为清理逻辑本来就已经和 Agent 执行器解耦了。6. 常见问题与排查技巧实录6.1 重试后重复副作用表现Agent 第一次调用“创建设备记录”工具超时用户重试出现了两条相同的设备记录。排查思路先看工具是否有幂等控制。如果是历史代码没有幂等层优先补request_id幂等方案。如果只是偶尔出现且不是高频路径可以退而求其次用清理策略做检测——在每个检查点里记录创建记录的唯一业务键失败后反查该键是否已存在存在则视为“已副作用完成”不再回滚只返回首次结果。这个方案虽然不如真正的幂等优雅但能解决兼容老工具的问题。6.2 会话超时真的触发了但没有用表现配置了 120 秒会话超时但请求实际运行了 300 秒才返回。排查思路大概率是 LLM 流式调用阻塞了主线程而超时检查在 while 循环里只在下一次迭代才生效。如果你调用 LLM 是阻塞式的且它一直不返回循环根本走不到超时检查。解决方法是把 LLM 调用也放进_run_tool_with_action_timeout的线程池执行让 Action 级和 Session 级的超时真正生效。这是我踩得最痛的坑之一agent 不是所有调用都会主动感知外层超时标志的。6.3 状态清理误杀了正常数据表现任务失败后自动回滚回滚后另一个正常任务的数据也消失了。排查思路典型的回滚边界定义错误。某个工具的 rollback 函数写得过于宽泛比如 DELETE 语句只按类型过滤而没带 request_id。修正方法就是升级清理策略的精度rollback 操作必须携带原调用的完整参数和 request_id且回滚范围必须只能作用于该次调用产生的最小粒度数据。另外这也能用指纹校验挡掉回滚前比对业务指纹不一致就不执行自动回滚转人工。6.4 清理策略速查表失败类型是否重试典型工具示例状态清理策略是否走人工网络瞬断是指数退避HTTP调用、LLM调用无需清理否LLM死循环否打断重置推理步骤截断上下文否非幂等写入超时是需幂等数据库INSERT幂等去重否已执行的通知类工具否发邮件、推送通知不可回滚记录悬空是已创建的工单未分配负责人否内部业务API部分回滚/挂起是临时文件、分布式锁是缓存、文件、锁自动清理否这张表是我们在实践中沉淀下来的分类标准基本上所有工具都能归入其中某一类。每次新增工具时第一件事就是想清楚它落在哪一行然后在代码注册时把清理策略配置好。这会成为 Agent 稳定性的一道重要防线。7. 一点个人体会这次改造让我对 Agent 工程化有了完全不同的理解。一开始我以为超时和重试是严谨性的体现做完了才发现它们顶多算 App 的入口保护真正的工程难点全在业务副作用的管理上。超时和重试本质上是处理时间和频率的问题而状态清理处理的是真实世界被 Agent 改变了之后怎么办的问题。后者复杂得多因为你不能理论化地“撤销”一封已发出的邮件不能凭空“复原”一次外部系统的状态修改。根据我的个人经验做 Agent 稳定性的项目建议不要一开始就追求全自动清理而是先把“检查点 悬空任务 人工决策”这套半自动方案跑通让团队的认知上线再逐步把高频且安全的路径自动化。如果倒过来一上来就写一堆自动回滚代码大概率会在生产环境的某个深夜搞出一场事故。最后再分享一个我实际用得很顺手的小技巧给 Agent 上下文的每个 checkpoint 都加一个created_at字段配合任务级created_at可以算出“某个工具执行后到底存活了多久”。当一个任务失败但用户迟迟不处理悬空任务时超出 24 小时的悬空任务会被自动升级为特殊状态并通知负责人。这个功能救了我好几次因为在真实的业务里悬而不决的状态往往比失败本身更危险。希望这篇内容对你的 Agent 工程化之路有一点点帮助如果你也在思考类似的问题欢迎拿这个故事去检验自己的设计很多时候只有在出过一次事之后才会真正理解状态清理的分量。