先说结论超时和重试这两个东西加之前觉得是保命符加完之后才发现它们只是把问题往后挪了一步。真正让线上Agent事故从偶发变成必现的是状态没有清理干净。这篇文章就把我在这条路上踩过的坑、试过的方案、最后沉淀下来的做法完整讲一遍。不管是刚给Agent写第一个工作流还是已经在扛生产环境的调用量这篇文章都应该能帮你在动手之前少走几趟弯路。我负责的一个基于FastAPI LangGraph搭建的AI Agent服务核心职责是根据用户指令编排多个工具调用——查库存、创建工单、发送通知偶尔还会跑一些需要几十秒的批处理。最初上线时客户端经常遇到响应超时于是第一反应就是给所有外部调用包上超时控制再配上重试。改动上线当天确实稳了不少可第二天就出了大问题用户反馈同一个工单被创建了三次还有人收到了一模一样的通知消息。查日志时发现每次重试都带着上一轮执行留下的中间状态相当于把一个没打扫干净的厨房再次交给下一个厨师菜当然越做越乱。后来我把状态清理这四个字当成一等公民来做才算是真正把Agent从能用变成了敢用。下面按我实际的推进顺序把全过程拆开说。1. 超时与重试的设计逐层拆解1.1 为什么加上超时这件事本身就有这么多讲究很多人以为超时就是给HTTP请求设置一个timeout30就结束了但在AI Agent场景里这个参数远比想象中复杂。Agent的调用链通常是这样的用户请求进入编排层编排层调用大模型大模型返回工具调用参数编排层再去调业务API最后把结果汇总返回给用户。这条链路上每一环的超时要求都不一样不能一刀切。大模型调用属于长尾延迟服务正常情况下可能3秒就返回了但遇到模型排队或者上下文很长10秒、20秒都是家常便饭。如果统一用5秒超时那基本是逼着用户不停重试。反过来业务API比如查库存、提交工单必须要求快速响应这类接口通常有明确的SLO超过8秒大概率就是出问题了再等下去只会白白占用线程和连接池。我当时给大模型调用设置了30秒超时给工具调用设置了10秒超时整体链路预留了40秒的上限。之所以这样配是因为最坏情况下一次模型调用超时触发重试重试成功后又调工具工具调用的超时和重试需要单独计算。这里有个很容易被忽视的点超时时间不能简单相加还要考虑底层连接池的排队时间。如果连接池已经满了新请求可能在池子里排队就已经把超时时间耗掉了。实际操作里我在每个环节都额外保留了20%的buffer。比如工具调用SLO是8秒我超时就设成10秒。这样既不会因为网络抖动频繁触发重试也不会把用户卡在那里等一个注定失败的请求。1.2 重试策略不是失败了再试一次这么简单重试机制里最经典的三个参数是重试次数、退避算法和最大退避时间。很多人只设置了次数忽略退避直接导致雪崩一次第三方服务故障所有Agent请求在同一秒开始重试把已经出问题的服务彻底打垮。我当时用的是指数退避加抖动jitter。指数退避很容易理解第一次重试等2秒第二次等4秒第三次等8秒倍数增长。但纯粹的指数退避有个问题如果所有请求都在同一时刻失败它们重试的时机仍然会撞在一起。抖动就是给等待时间加上随机偏移让重试请求自然散开。代码上用tenacity库实现非常简单from tenacity import ( retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type, before_sleep_log ) retry( stopstop_after_attempt(3), waitwait_random_exponential(multiplier1, min2, max30), retryretry_if_exception_type((TimeoutError, ConnectionError, ApiError)), ) async def call_tool_with_retry(func, *args, **kwargs): return await func(*args, **kwargs)注意这里我没有对任何异常都重试。业务校验错误比如参数不合法、库存不足重试一万次也是同样的结果甚至可能因为重复提交产生脏数据。只有TimeoutError、ConnectionError这类可重试异常才进入重试逻辑。还有一个关键点重试必须有全局唯一请求ID贯穿始终。如果每次重试都生成新的请求ID那么下游系统根本不知道这是同一个操作的延续重复执行就无法识别了。我在请求入口生成一个trace_id同时把这个ID作为业务幂等键传给所有工具调用。1.3 幂等设计重试的前提条件没有它一切白搭如果说超时和重试是Agent的骨架幂等设计就是血肉。没有幂等重试不是保护机制而是事故放大器。以创建工单为例用户发了帮我创建一个售后工单Agent调用创建工单API服务端已经成功创建了工单但响应在网络上超时了。此时客户端以为失败触发重试第二个工单又被创建了。这就是典型的非幂等操作遇上重试后产生的重复副作用。幂等有两种做法。第一种是下游接口本身支持幂等键大部分云服务都支持在请求头上传一个幂等键Idempotency-Key服务端会记录这个键对应的处理结果重复请求直接返回第一次的结果。第二种是代理层自己做幂等在调用下游之前先查一张已执行操作表如果这个trace_id已经执行过某个操作就不重复调用了直接把之前的结果读出来返回。我在项目里两种都用了核心业务接口工单、支付、通知强制要求传入幂等键非核心接口比如查询类的在代理层做去重。这样双重保险之下重试才真正安全。2. 状态清理才是真正的分水岭2.1 状态管理Agent和普通接口的本质区别普通的Web接口是无状态的请求来了处理完就返回什么也不留下。Agent不一样一个完整任务往往要经过理解意图→规划步骤→调用工具→汇总结果好几个阶段每个阶段都可能产生中间状态。这些状态散落在内存变量、缓存、数据库、甚至外部系统里。状态大致分成三类。第一类是会话状态也就是对话历史、用户意图的中间表示、当前正在执行哪一步。第二类是外部副作用状态比如已经通过工具调用创建的资源工单ID、订单号、已发送的通知以及为此加过的分布式锁、占用的连接和临时文件。第三类是编排状态比如任务执行到第几步了、哪些步骤已经完成、哪些还在pending。很多人做Agent的时候只关注第一类会话状态。把消息列表存到Redis就算管理了。第二类副作用状态往往被忽略直到重试把同一笔操作执行了两遍才意识到它带来的问题比会话状态严重得多。2.2 超时和重试如何一步步把状态搞脏我复盘了那次工单被创建三次的事故完整的链条是这样的用户发起请求Agent开始第一步——调用大模型获取行动计划。模型正常返回Agent拿着计划去执行创建工单这个工具。工具调用发出后服务端成功创建了工单但响应没回来客户端这边触发了超时。此时LangGraph的执行节点其实已经把工具调用的参数、意图信息都放进了状态里。重试开始后新的节点执行时直接读取了旧状态里的参数而旧的执行记录、步骤标记还留在状态里。于是同一个工具被再次调用第二次创建工单成功。接着第三次……更隐蔽的问题是重试是哪个层级的重试如果是整个编排层的重试那不仅工具会重复执行连大模型的调用次数都会翻倍token消耗直接翻三倍。如果是工具层的重试那我上面说的情况就无法避免。正常做法是每次进入新节点之前把上一轮执行留下的执行痕迹清理干净把pending的工具调用标记为已处理或已撤回然后再执行当前节点。同时记录重试前已经成功执行的副作用清单以便在最终放弃时做补偿。2.3 状态清理的三种实用模式踩过坑之后我把状态清理的做法整理成三种模式根据场景选着用。第一种叫会话快照重构Snapshot Reconstruction。在执行关键步骤之前把当前Agent状态做一次快照存到独立的存储里。如果后续执行失败、需要重试就直接加载这个快照从一个干净的状态开始重新执行。这种方式最简单实现成本低适合执行链路短、外部副作用少的场景。缺点是如果第一步执行中已经产生了副作用快照也救不了你因为副作用无法被快照回滚。第二种叫副作用补偿Compensation。在Agent执行过程中维护一个副作用登记表记录每次成功调用的工具、参数、返回结果、外部资源ID。如果任务最终失败或需要回滚就按登记表的逆序逐一做补偿操作——创建了工单就关闭它发出去的通知就追加更正说明占用了锁就释放锁。这个模式的优点是能处理真实世界的外部状态缺点是补偿逻辑本身也要写得很严谨否则补偿动作本身也可能失败。第三种叫资源生命周期绑定Resource Lifecycle Binding。把Agent的执行上下文看成一个资源容器任务开始时申请资源锁、连接、临时存储任务结束时统一释放。关键在于每个任务上下文都有唯一的生命周期ID无论是超时、重试、还是正常结束都必须走同一个释放资源入口。我用FastAPI的依赖注入系统来做这件事每个请求进来时创建一个上下文管理器请求结束时会话自动清理重试时则先销毁旧上下文、再创建新上下文。实际上我在生产环境用的是后两种的组合副作用登记表负责外部补偿资源生命周期绑定负责内部资源释放。会话快照只用在最核心的编排路径上作为兜底。2.4 为什么状态清理比超时本身难十倍超时和重试本质上是时间维度的问题你只需要回答两个问题等多久算失败失败后试几次而状态清理是空间维度的问题你得回答这些哪些数据属于这次任务如果任务中断这些数据怎么回收重试时旧数据里哪些该保留、哪些该清掉如果多个任务并发它们的状态会不会互相污染这就是真正的坑所在。超时重试是确定性的技术方案而状态清理需要你对自己系统里的每一个数据流都了如指掌。很多Agent框架包括LangGraph默认会把状态保存在内存对象里重试时直接复用这个对象这会让问题变得异常隐蔽。你看到日志里重试成功但实际上成功的是带着脏状态的第二次执行而不是干净的第二次执行。更麻烦的是分布式环境下的状态。如果你的Agent服务是多个副本同时跑那么状态就不能只存在单机内存里必须放到Redis这类中心化存储中。可一旦放到中心化存储就要考虑状态过期、锁竞争、并发访问冲突。我在做并发压测时就遇到过同一个用户的连续两个Agent任务因为会话ID相同第二个任务读取到了第一个任务留下的残留状态导致行为完全错乱。3. 实操落地在LangGraph里做出干净的Agent重试3.1 项目架构和状态定义先说说我们这个项目的具体技术栈FastAPI提供API入口LangGraph做Agent编排Redis存会话和状态PostgreSQL存业务数据。LangGraph本身允许自定义StateSchema这是我实现状态清理的基础。定义一个干净的Agent状态第一原则是只放当前任务必需的数据不是什么东西都往里塞。一开始我图方便把大模型的完整响应、中间计算结果、甚至是局部日志都塞进状态里结果状态越来越臃肿重试时带着一堆无关数据排查问题都要翻半天。后面我重新设计了状态结构核心只有四个字段from typing import TypedDict, Optional from enum import Enum class TaskStatus(str, Enum): PENDING pending RUNNING running SUCCEEDED succeeded FAILED failed COMPENSATED compensated class AgentState(TypedDict): trace_id: str # 全局唯一请求ID重试时保持不变 task_status: TaskStatus conversation: list # 必要的对话历史用于模型上下文 pending_actions: list # 当前待执行的工具调用 executed_actions: list # 已经成功执行并产生副作用的操作登记表 retry_count: int # 当前重试次数3.2 执行流程中的状态清理触发点我给LangGraph的每个节点都定义了三种行为正常行为、异常行为、补偿行为。正常行为就是执行工具调用异常行为是捕获异常后把当前待执行的行动打回pending状态补偿行为是如果某个节点彻底失败就对这个节点之前成功执行的操作做撤销。关键代码逻辑如下简化自真实项目class ToolExecutionNode: async def __call__(self, state: AgentState) - AgentState: if state[task_status] TaskStatus.PENDING: state[pending_actions] [] state[executed_actions] [] action state[pending_actions].pop(0) if state[pending_actions] else None if action is None: return state try: result await call_tool_with_retry(action[tool_name], action[params]) state[executed_actions].append({ action: action, result: result, status: done }) state[task_status] TaskStatus.RUNNING except TimeoutError: # 超时后进入补偿路径 state[task_status] TaskStatus.FAILED state[pending_actions] [action, *state[pending_actions]] await self.compensate(state) return state你留意到没有每次进入节点时如果任务状态是PENDING就先把pending_actions和executed_actions清空。这就是状态清理的第一个触发点——一个任务从零开始或者重试重生时必须从空白状态出发绝不能带着上一轮的残留行动列表。3.3 副作用登记与补偿机制副作用登记表是这次改造里最关键的部分。每次工具调用成功我会把这次调用产生的外部资源信息和补偿方法一起登记到executed_actions里。后续如果需要清理就遍历这个列表逐个执行补偿。async def register_side_effect(action_name: str, params: dict, result: dict): if action_name create_work_order: return { compensate_method: close_work_order, compensate_params: {work_order_id: result[work_order_id]}, } elif action_name acquire_lock: return { compensate_method: release_lock, compensate_params: {lock_id: params[lock_id]}, } # 其他工具类似补偿执行时有一个非常容易忽视的细节补偿操作本身也可能失败所以补偿也要有幂等设计。比如释放一个已经不存在的锁成功或者返回锁不存在都应该视为成功因为最终状态已经达成了。我处理的方式是补偿方法只捕获异常但不抛出并且给补偿操作也标记一个独立的追踪ID方便查日志。3.4 资源释放让每次请求都有干净的出生和死亡在FastAPI层我用依赖注入为每个Agent请求创建独立的上下文管理器。这个上下文管理器集中管理三类资源Redis连接、临时文件句柄、分布式锁。from contextlib import asynccontextmanager asynccontextmanager async def agent_context(trace_id: str): # 进入上下文时创建锁、初始化存储、写入开始标记 lock await redis_client.lock(fagent:{trace_id}, timeout30) await state_store.initialize(trace_id) try: yield {trace_id: trace_id, lock: lock} finally: # 退出上下文时无论成功失败超时统一走清理入口 await state_store.destroy(trace_id) await lock.release()这个设计的核心就是把超时清理和正常完成统一到同一个出口。FastAPI的finally块保证不管Agent是正常跑完、中途抛异常、还是被asyncio超时中断资源释放逻辑都会执行。我之前犯过的一个错误是只在成功路径上写了清理失败路径漏了结果失败的任务占着锁不释放后续重试全部卡死。3.5 并发场景下的状态隔离状态清理的另一个重要战场是并发。同一个用户的多个Agent任务可能并发到达如果它们复用同一个会话ID状态就会互相覆盖。我的做法是每个任务都有自己的task_id而会话ID只用于保存跨任务的长期上下文比如用户的长期偏好绝不用于保存短期执行状态。短期执行状态统一以task_id为键存在Redis里并设置TTL。这样一来即使某个任务的清理逻辑因为极端情况没有执行Redis中的状态也会在几分钟后自动过期不会造成永久污染。这个TTL要设置得比正常任务耗时长一些——我当时设的是最大任务耗时的两倍确保任务还在执行时状态不会被提前回收。同时对外部副作用的状态隔离也不能忽略。比如两个并发任务都要调用同一个分布式锁那么锁的键必须包含task_id否则任务A锁住了资源任务B重试时会因为拿不到锁而失败。4. 常见问题与排查技巧实录4.1 事故档案三次典型状态污染故障这里记录几件我在实际运行中遇到、并且最终排查清楚的故障。这些案例基本覆盖了Agent状态清理最常见的坑。第一个是重试时使用了同一个会话ID导致上下文串味。用户连续发起了两个任务第一个任务超时了重试时直接复用了同一个会话对象。当时状态存储用的是Redis哈希键是会话ID字段是任务相关数据。两个任务写入同一个哈希后面的把前面的数据覆盖了导致第一个任务最终生成的结果混入了第二个任务的信息。排查时发现Redis里同一个会话ID下同时有两个task_status字段才确定问题根源。修复方式是短期状态键改为task_id会话ID只负责长期上下文。第二个是工具调用重复执行且没有幂等键。这个前面提到过工单被创建了三次。排查时发现executed_actions列表为空——因为工具层超时的时候成功响应已经返回给服务端了但客户端状态里根本没有记录这次调用。修复方式是检查所有工具调用是否都传了幂等键并把幂等键统一绑定trace_id。第三个是资源泄漏导致的重试死锁。有一个任务要调用外部数据库写数据超时后触发重试但旧事务占用的数据库连接没有释放新重试就拿不到连接一直超时直到达到最大重试次数。最后整个Agent不可用。修复方式就是把数据库连接改成了上面提到的上下文管理器模式在finally里强制关闭。4.2 排查方法如何快速定位状态污染排查状态污染的难点在于问题往往不是稳定复现的而是偶发的。我的经验是用日志贯穿整个状态生命周期。在状态初始化的入口打一条日志记录初始状态的完整快照在每次工具调用前打一条日志记录当前状态中的pending_actions和executed_actions在每次任务结束时打一条日志记录最终状态和清理动作。这样一旦出问题你可以把这个任务从开始到结束的所有状态变化拉出来对比预期状态和实际状态的差异。我排查那三次事故的时候都是靠这个日志追踪在几分钟内定位的。还有一个小技巧给每次工具调用打印性能日志时附带trace_id、retry_count、action_name三个字段。重试时同一个trace_id会对应多条日志把重试前后的日志放在一起看就能清楚看到状态是否被上一次执行污染了。4.3 状态清理避坑清单我把踩过的坑整理成一张速查表每次给Agent加新工具调用时都会过一遍这张表检查项常见误区正确做法工具调用是否幂等默认下游接口是幂等的所有可能产生副作用的工具必须显式支持幂等键状态变量是否最小化把中间结果全部塞进状态只放当前节点必需的数据其他用临时变量重试时是否清理旧状态重试直接复用旧状态对象进入新节点前重置pending和executed列表资源释放是否覆盖所有路径只在成功路径上释放用上下文管理器保证任何路径都走统一的释放逻辑状态过期时间是否合理TTL过短导致任务中途状态丢失TTL设为最大任务耗时的两倍并发任务状态是否隔离按会话ID存短期状态按task_id存短期状态会话ID只管长期上下文补偿操作是否幂等补偿失败导致状态不一致补偿操作也带幂等设计失败视为达成目标条件日志是否覆盖状态生命周期只在出错时才打印状态初始、调用前、结束时都必须有状态快照日志4.4 几个值得提前想清楚的取舍做状态清理方案时有几个地方是没有绝对正确只有适合自己的取舍这里聊聊我的选择。重试到底应该放在哪一层如果放在工具层粒度细不会重复调用大模型但状态污染的防护要靠自己实现。如果放在编排层整个Agent任务重跑实现简单但token成本成倍上升副作用重复的风险也更大。我最终选择工具层重试编排层兜底工具层负责应对瞬时故障如果工具层重试都失败了编排层会根据之前登记的副作用做补偿然后整个任务以失败收场不继续往下跑。这个方案在成本和可靠性之间比较均衡。状态清理是完成任务后一次性做还是每个节点间持续做我之前以为一次性做完就够了结果发现任务中途超时重试时旧状态在下一个节点就已经污染了执行。后来改成每个节点结束时都做一次轻量清理把已经执行完的操作从pending移到executed任务彻底结束时再做全量补偿和资源释放。代价是多一点点CPU开销但换来的是任何一步出错都不会带着脏状态进入下一步。状态是放内存还是放Redis单个Agent任务实例运行时内存访问快性能好但无法跨副本共享崩溃就丢。Redis可以跨副本、有TTL但每次读写都有网络开销。我的选择是会话状态放Redis因为要跨请求、跨副本步骤执行中的临时计算放内存因为生命周期极短出了作用域就销毁。只有真正需要跨任务保留的数据才值得放进Redis这个原则让状态管理简单了一大截。结尾的几个建议这些改动上线跑了两周之后我又专门做了一次压测同时模拟100个并发任务每个任务都在中途注入随机的网络超时。结果没有再出现一次重复创建工单、没有一次锁泄漏、也没有一次状态串味。这说明状态清理的方案是真正经受住了考验的。如果你现在正要给Agent加超时和重试我的建议是先把每种工具调用的副作用列清楚想好哪类操作重试是安全的哪类操作重试前必须做补偿然后设计好状态结构宁可少放也不要多放最后把资源释放统一收口到一个出口让每个任务都像一次干净的事务一样有明确的开始和结束。超时和重试只是给Agent穿上了防弹衣状态清理才是让它真正有了免疫系统。这两者配合好Agent才敢在真实业务里放手干活。后面我准备把LangGraph里状态清理的这套逻辑抽成一个独立模块封装成可复用的中间件让新接入的Agent项目不用从头踩一遍这些坑。到时候如果有效果再来分享。
