1. Agent错误处理的核心挑战与设计思路做Agent开发的人都有一个共识Demo跑通只要一天但让它稳定跑在生产环境可能要花上几个月。我见过太多团队在Agent项目上踩坑模型调用超时、工具执行失败、上下文丢失、重复扣费……这些问题在单次测试时几乎不会暴露一旦上了并发或者长时间运行各种幺蛾子就全出来了。Agent系统的错误处理和传统后端服务有本质区别。传统服务的错误大多是确定性的——数据库连不上、参数校验失败、空指针异常这些错误的边界清晰处理方式也相对固定。但Agent系统不一样它的错误来源极其复杂大模型本身的不确定性输出、外部工具调用的网络抖动、多轮对话中上下文的累积偏差、以及编排层逻辑的时序问题。更麻烦的是很多错误不是非黑即白的模型可能返回了一个格式正确但语义完全跑偏的结果这种软错误比硬崩溃更难处理。这一篇主要聊的是Agent在生产环境中怎么做好错误处理和工程化落地。核心围绕几个关键词展开重试策略、幂等性设计、错误分类与降级、可观测性建设。适合已经写过基础Agent、准备往生产环境推进的开发者也适合正在做Agent平台架构设计的同学参考。我会尽量把每个设计决策背后的为什么讲清楚而不是只丢一堆代码让你抄。1.1 为什么Agent的错误处理不能照搬传统方案传统后端服务的错误处理三板斧是try-catch、重试、降级。这套逻辑放到Agent系统里直接套用会出大问题。第一个坑是重试的副作用。传统接口大多是幂等的查询操作重试几次无所谓。但Agent调用的工具可能是发送邮件、创建订单、写入数据库这类有副作用的操作。你重试一次用户可能收到两封邮件或者被扣两次款。这就是为什么幂等性设计在Agent系统里不是可选项而是必选项。第二个坑是错误传播的放大效应。Agent的编排通常是链式的规划→调用工具→解析结果→再规划→再调用。如果第一步的规划就偏了后面每一步都会在错误的基础上继续执行最终产出一个看起来像模像样但完全不可用的结果。这种级联错误在传统服务里很少见但在Agent里是常态。第三个坑是错误判定的模糊性。传统服务里HTTP 500就是错误200就是成功。但Agent调用大模型返回200不代表结果可用——模型可能输出了幻觉内容、可能只输出了思考过程没有正文、可能格式不符合预期。你需要一套额外的结果质量校验机制来判断这次调用是否真的成功。我在实际项目中总结下来Agent的错误处理需要分三层来做传输层错误网络超时、连接失败、执行层错误工具调用失败、参数错误、语义层错误结果不符合预期、幻觉、格式错误。三层错误的处理策略完全不同混在一起处理必然出问题。1.2 错误分类体系先搞清楚敌人是谁在动手写任何错误处理代码之前先把错误分类做清楚。我一般会按下面的维度来划分错误类型典型场景可重试性处理策略网络超时模型API请求超时可重试指数退避重试限流错误API返回429可重试延迟后重试降低并发认证失败Token过期不可直接重试刷新凭证后重试参数错误工具调用参数格式不对可修复重试让模型重新生成参数工具执行失败外部服务返回错误视情况判断幂等性后决定语义错误模型输出幻觉/格式错误可重试带反馈重新生成上下文溢出超出模型上下文窗口不可重试压缩上下文后重试编排死循环Agent反复调用同一工具不可重试熔断人工介入这张表是我踩了无数坑之后总结出来的每一行背后都有血泪教训。比如参数错误这一类早期我的做法是直接抛异常终止后来发现更好的方式是把错误信息回传给模型让它自己修正参数重新调用。这个改动让工具调用的成功率从70%多提升到了95%以上。再比如上下文溢出很多人第一反应是截断历史消息但粗暴截断会丢失关键信息。我现在的做法是先用一个轻量模型对历史对话做摘要压缩保留关键决策和事实再拼接到新的上下文中。这个方案虽然多了一次模型调用但效果比截断好太多。注意错误分类不是一次性的工作。随着Agent能力扩展和工具接入增多新的错误类型会不断出现。建议在代码里预留一个未知错误的兜底分类并定期review错误日志把高频的未知错误归入已有分类或新增分类。2. 重试机制的设计与落地细节重试是错误处理里最基础也最容易做错的一环。很多人写重试就是简单套一个for循环加sleep但在Agent场景下这种朴素重试会引发一堆问题。2.1 指数退避与抖动别让重试变成雪崩最基础的重试策略是指数退避Exponential Backoff。核心思想是每次重试的等待时间指数增长给下游服务足够的恢复时间。公式很简单delay base_delay * (2 ^ retry_count)比如base_delay设为1秒那么第1次重试等1秒第2次等2秒第3次等4秒第4次等8秒。这个策略能有效避免在服务刚出问题时疯狂重试导致压力叠加。但光有指数退避还不够还需要加抖动Jitter。为什么假设你有100个Agent实例同时调用同一个模型APIAPI因为负载过高开始返回错误。如果没有抖动这100个实例会在完全相同的时刻发起重试形成重试风暴把刚恢复的服务再次打垮。抖动的做法是在计算出的delay上叠加一个随机值import random import time def retry_with_backoff(func, max_retries5, base_delay1.0, max_delay60.0): for attempt in range(max_retries): try: return func() except RetryableError as e: if attempt max_retries - 1: raise delay min(base_delay * (2 ** attempt), max_delay) # 全抖动在0到delay之间随机 jittered_delay random.uniform(0, delay) time.sleep(jittered_delay) raise MaxRetriesExceeded()抖动策略有几种变体全抖动Full Jitter是在0到delay之间随机等抖动Equal Jitter是delay的一半加上0到一半的随机去相关抖动Decorrelated Jitter是每次在上次delay的1到3倍之间随机。实测下来全抖动在Agent场景下表现最稳因为它能把重试请求最均匀地分散开。2.2 重试预算与熔断给重试设一个天花板无限重试是灾难。我见过一个Agent因为下游服务持续不可用重试了上千次把整个线程池占满导致其他正常请求也无法处理。重试预算Retry Budget的思路是给每个时间窗口内的重试次数设一个上限。比如每分钟最多重试100次超过这个数就不再重试直接走降级逻辑。这个预算可以按服务维度设也可以按Agent实例维度设。熔断器Circuit Breaker是另一个必备组件。它的逻辑是当某个下游服务的错误率超过阈值时直接熔断后续请求不再尝试调用而是快速失败。经过一段冷却时间后再放少量请求试探如果成功则恢复失败则继续熔断。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_count 0 self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.state CLOSED # CLOSED / OPEN / HALF_OPEN self.last_failure_time None def call(self, func): if self.state OPEN: if time.time() - self.last_failure_time self.recovery_timeout: self.state HALF_OPEN else: raise CircuitOpenError(服务已熔断) try: result func() if self.state HALF_OPEN: self.state CLOSED self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state OPEN raise熔断器在Agent系统里特别重要因为Agent经常会调用多个外部工具任何一个工具持续不可用都可能拖垮整个编排流程。有了熔断器至少能保证Agent快速失败并给出有意义的错误提示而不是卡死在那里。2.3 语义级重试让模型自己修正错误前面说的都是传输层和执行层的重试Agent系统里还有一类特殊的重试——语义级重试。当模型输出的结果不符合预期时比如格式错误、缺少必要字段、明显幻觉不是简单重发请求而是把错误信息作为反馈让模型重新生成。这个方案的关键在于错误反馈的构造。你不能只说你错了重新来而要具体指出哪里错了。比如你上一次的输出存在以下问题 1. JSON格式不合法第3行缺少闭合括号 2. 字段user_id缺失这是必填字段 3. amount字段的值是字符串类型应该是数字类型 请修正以上问题后重新输出。实测下来带具体反馈的重试成功率比盲目重试高出很多。我做过一个对比测试同样是格式错误的重试盲目重试3次成功率约60%带反馈重试3次成功率能到92%。但要注意语义级重试不能无限做。一般设置2-3次上限超过就说明这个任务可能超出了模型能力范围应该走降级或者转人工。3. 幂等性设计Agent工程化的生命线幂等性这个词在传统后端里不新鲜但在Agent系统里它的重要性被放大了十倍。原因很简单Agent会重试而重试意味着同一个操作可能被执行多次。如果这个操作有副作用扣款、发消息、写数据重复执行就是事故。3.1 幂等键的设计原则幂等性的核心是幂等键Idempotency Key。每次操作携带一个唯一标识服务端记录已处理的键遇到重复键直接返回上次的结果不重复执行。幂等键的设计有几个原则全局唯一不能有碰撞一般用UUID或者业务ID时间戳随机数的组合可追溯从键能反推出是哪个Agent、哪次任务、哪一步操作稳定同一个逻辑操作在重试时必须用同一个键不能每次重试都生成新键最后一点特别容易踩坑。我见过有团队在重试逻辑里每次都重新生成幂等键结果幂等性完全失效——因为服务端看到的是不同的键以为是不同的请求。正确的做法是在任务开始时生成幂等键整个任务生命周期内复用。比如一个Agent任务要调用支付工具那么幂等键应该在规划阶段就确定后续无论重试多少次用的都是同一个键。class AgentTask: def __init__(self, task_id): self.task_id task_id self.idempotency_keys {} # 按操作类型存储幂等键 def get_idempotency_key(self, operation_type, params): # 同一个操作类型参数组合返回同一个键 key f{self.task_id}:{operation_type}:{hash_params(params)} if key not in self.idempotency_keys: self.idempotency_keys[key] str(uuid.uuid4()) return self.idempotency_keys[key]3.2 幂等性检查用DB还是Redis这是热词里出现的一个高频问题我结合实际经验说一下。Redis方案用SETNX命令设置键和过期时间。优点是快、简单、天然支持TTL。缺点是Redis如果挂了或者数据丢失幂等性就没了。而且Redis的内存成本比磁盘高如果幂等键需要保留很长时间比如7天成本会比较高。数据库方案建一张幂等记录表用唯一索引保证不重复。优点是持久化可靠、支持复杂查询、可以记录更多元信息比如执行结果、时间戳。缺点是性能比Redis差高并发下可能成为瓶颈。我的建议是两者结合用Redis做第一层快速拦截挡住绝大部分重复请求用数据库做最终一致性保障Redis没挡住或者Redis不可用时数据库的唯一索引兜底。具体流程是请求进来先查Redis有没有这个幂等键有直接返回缓存的结果没有尝试在数据库插入幂等记录利用唯一索引插入成功说明是首次请求正常执行插入失败唯一键冲突说明是重复请求查询数据库返回已有结果执行完成后把结果写入Redis并设置TTL这个方案在性能和可靠性之间取得了比较好的平衡。实测在每秒几千次请求的场景下Redis层能挡住99%以上的重复请求数据库压力很小。注意幂等键的TTL设置要合理。太短了重试间隔长的请求可能穿透太长了占用存储。一般建议TTL至少覆盖任务的最大可能执行时间比如任务最长跑1小时TTL设24小时比较稳妥。3.3 工具层面的幂等性适配不是所有工具都天然支持幂等。有些第三方API没有幂等键机制这时候需要在Agent层面做适配。常见的适配方式有几种查询前置在执行操作前先查询当前状态如果已经是目标状态就跳过。比如要创建订单先查一下这个订单号是否已存在。状态机控制给每个操作定义状态流转只有特定状态才能执行特定操作。比如订单只有待支付状态才能执行支付已经已支付的订单再次收到支付请求直接拒绝。补偿事务对于无法做到幂等的操作记录操作日志出错时通过补偿逻辑回滚。这个方案复杂度最高一般只在金融级场景使用。我在实际项目里最常用的是查询前置状态机的组合。虽然多了一次查询开销但能覆盖绝大多数场景实现也简单。4. 工程化实践可观测性与降级策略错误处理做得好不好很大程度上取决于你能不能看到错误。如果错误发生了你都不知道那所有的重试、幂等、降级都是空谈。4.1 Agent的可观测性建设Agent的可观测性和传统服务有相似之处也有特殊的地方。相似的是都需要日志、指标、链路追踪三件套。特殊的是Agent需要额外关注决策过程的可观测性。我一般会在Agent的每个关键节点打点输入输出每次模型调用的prompt和response注意脱敏决策路径Agent选择了哪个工具、为什么选这个工具重试记录每次重试的原因、等待时间、最终结果幂等命中哪些请求命中了幂等缓存耗时分布每个阶段的耗时找出瓶颈这些数据汇总起来能回答很多关键问题Agent的成功率是多少失败主要集中在哪些环节重试的有效率如何有没有异常的调用模式链路追踪在Agent场景下尤其重要因为一个任务可能涉及多次模型调用和工具调用跨越多个服务。用Trace ID把这些调用串起来排查问题时能快速定位到具体是哪一步出了问题。import logging import time from contextlib import contextmanager logger logging.getLogger(agent) contextmanager def trace_step(step_name, trace_id, **extra): start time.time() logger.info(f[{trace_id}] step{step_name} statusstart extra{extra}) try: yield duration time.time() - start logger.info(f[{trace_id}] step{step_name} statussuccess duration{duration:.3f}s) except Exception as e: duration time.time() - start logger.error(f[{trace_id}] step{step_name} statusfailed duration{duration:.3f}s error{e}) raise4.2 降级策略优雅地失败降级的核心思想是当主要逻辑不可用时提供一个虽然不完美但可用的替代方案。Agent系统的降级策略一般分几个层次模型降级主模型不可用时切换到备用模型。比如GPT-4超时了降级到GPT-3.5。虽然效果差一些但至少能返回结果。工具降级某个工具不可用时用替代工具或者返回缓存结果。比如实时汇率接口挂了用最近一次缓存的汇率。流程降级复杂的多步编排降级为简单的单步处理。比如原本要调用5个工具现在只调用最关键的2个。兜底回复所有方案都失败时返回一个友好的错误提示而不是让用户看到一堆堆栈信息。降级策略的关键是提前定义好降级路径而不是等出问题了临时想。我一般会在Agent配置里为每个关键环节定义降级方案运行时根据错误类型自动选择。4.3 常见问题速查表最后整理一份我在实际项目中遇到的典型问题和解决方案方便大家快速对照排查问题现象可能原因排查方向解决方案Agent反复调用同一工具工具返回结果不符合模型预期检查工具返回格式增加结果校验超限熔断重试后产生重复副作用幂等键未复用检查幂等键生成逻辑任务级复用幂等键模型只输出思考过程无正文输出预算不足检查max_tokens配置逐级提升输出预算重试上下文溢出报错历史消息累积过多统计token数量摘要压缩后重试并发下成功率骤降下游限流查看429错误比例降低并发指数退避任务卡死不返回编排死循环检查循环检测逻辑设置最大步数熔断结果格式不稳定模型输出随机性检查temperature配置降低温度格式校验重试这份表是我从实际故障中一条条攒出来的每一条都对应过真实的事故。建议大家在项目初期就把这些检查项纳入监控告警能省下大量排查时间。关于Agent错误处理和工程化我的核心体会是不要试图消灭错误而是要学会和错误共处。Agent系统的不确定性是它的本质特征与其追求100%不出错不如把精力放在出错后能快速恢复、不产生副作用、能定位问题上。重试、幂等、降级、可观测性这四件事做好了Agent的稳定性就能上一个台阶。至于具体的参数调优比如重试几次、退避多久、熔断阈值设多少这些没有标准答案需要根据你的业务特点和下游服务的实际情况反复调整。我一般会先在测试环境跑一轮压测观察错误分布再据此设定初始参数上线后根据监控数据持续优化。