给 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_factory=list) tool_results: list[dict] = field(default_factory=list) checkpoints: list[Checkpoint] = field(default_factory=list) run_state: dict[str, Any] = field(default_factory=dict) 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"] = f"tool 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_id=str(uuid.uuid4()), step_index=step_index, tool_name=action.tool_name, tool_args=action.tool_args, business_state_fingerprint=ctx_fp, created_at=time.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": f"Error: {tool_result['error']}"} ) else: context.run_state["last_error"] = tool_result["error"] context.message_history.append( {"role": "tool", "tool_name": action.tool_name, "content": f"Fatal 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 = f"action 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_workers=2) as pool: future = pool.submit(tool_fn, **args) return future.result(timeout=timeout) @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_name=tool_name, rollback_fn=rollback_fn, is_destructive=is_destructive, requires_manual=requires_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_keys=True)}".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, timeout=10) 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 工程化之路有一点点帮助，如果你也在思考类似的问题，欢迎拿这个故事去检验自己的设计，很多时候只有在出过一次事之后，才会真正理解状态清理的分量。
