1. 为什么错误处理才是 Agent 工程化的分水岭做 Agent 开发的人大概都有过这种体验Demo 阶段一切丝滑工具调用、多轮推理、记忆读写全都跑得通可一旦放到真实环境里跑上几天各种稀奇古怪的报错就开始往外冒。模型返回的 JSON 少了个括号、工具接口超时、上下文被截断、并发一上来状态就串了——这些问题在演示视频里永远不会出现但在生产环境里几乎每天都会遇到。我做了几年大模型应用开发越来越确信一件事Agent 能不能上线拼的不是推理能力而是错误处理能力。一个能扛住各种异常、失败后能自愈、重复执行不出错的 Agent才算是真正工程化的 Agent。这也是Agent 系列走到 9.4 这个节点必须专门聊错误处理与工程化实践的原因——前面几篇把架构、记忆、工具调用都铺完了剩下的就是把这些零件组装成一个摔不坏的系统。这篇文章面向的是已经写过至少一个能跑通的 Agent、正准备把它推向真实业务场景的开发者。如果你还在纠结 Agent 是什么、框架怎么选那可以先去看前面的基础篇。但如果你已经踩过接口超时导致整条链路崩掉重试之后数据写了两遍这类坑那这篇就是写给你的。我会把错误分类、重试策略、幂等设计、并发控制、可观测性这几块拆开讲透每个点都给出可直接抄作业的方案和参数计算过程。2. Agent 错误分类与处理思路拆解2.1 先搞清楚错误从哪来四层错误模型很多人处理错误的方式是一把梭——不管什么错先 try-catch 包起来失败了就重试。这在简单脚本里能用但在 Agent 里会出大问题。因为 Agent 的错误来源是分层的不同层的错误处理逻辑完全不同混在一起处理只会让问题更难定位。我习惯把 Agent 的错误分成四层错误层级典型表现是否可重试处理策略模型层输出格式错误、内容截断、拒绝回答部分可重试重试格式修复降级工具层接口超时、返回异常、限流大多可重试退避重试熔断编排层状态丢失、步骤错乱、死循环谨慎重试状态快照回滚系统层内存溢出、进程崩溃、网络中断需人工介入告警恢复这个分层不是拍脑袋来的。模型层的错误本质是概率性输出同一个 prompt 多跑几次可能就对了所以重试有效工具层的错误大多是瞬时故障网络抖动、服务端限流退避重试能解决大部分编排层的错误往往是逻辑缺陷盲目重试只会让状态更乱系统层的错误则是资源问题重试没用得先修环境。提示判断一个错误该不该重试核心看它是不是幂等可恢复的。如果重试会让系统状态变得更糟那宁可失败也不要重试。2.2 重试不是万能药什么时候该放弃我见过太多项目把重试次数设成 10 次、间隔 1 秒结果一个坏请求把整个服务拖垮。重试的本质是用时间换成功率但它有成本占用连接、消耗配额、放大下游压力。所以重试必须配合退避策略和熔断机制。退避策略我一般用指数退避加抖动。基础公式是delay min(max_delay, base_delay * 2^attempt) random_jitter举个具体例子base_delay 设 0.5 秒max_delay 设 30 秒那么第 1 次重试等约 0.5 秒第 2 次约 1 秒第 3 次约 2 秒第 4 次约 4 秒……到第 6 次就接近 30 秒封顶。加抖动是为了避免惊群效应——如果一百个请求同时失败同时重试会把下游瞬间打爆加个随机偏移就能把压力摊开。熔断则是另一道保险。当某个工具连续失败超过阈值比如 10 次里失败 8 次就直接切断这个工具的调用快速失败等一段时间后再半开试探。这样能防止一个坏掉的下游把整个 Agent 拖死。2.3 幂等重试的前提条件重试最大的风险是重复执行。如果一次工具调用是扣款 100 元超时后重试结果扣了两次那就是事故。所以任何可能被重试的操作都必须是幂等的。幂等的定义是同一个请求执行一次和执行多次对系统状态的影响相同。实现幂等最常用的手段是幂等键idempotency key。客户端在发起请求时生成一个唯一 ID服务端记录这个 ID 的处理结果重复请求直接返回缓存结果不再真正执行。这里有个常见的选择题幂等性检查用数据库实现好还是用 Redis 好我的经验是分场景强一致场景涉及金额、库存用数据库唯一索引虽然慢但绝对可靠事务保证不会漏。高并发场景消息去重、状态标记用 Redis 的 SETNX性能高配合过期时间自动清理。两者结合Redis 做第一道快速拦截数据库做最终兜底兼顾性能和可靠性。3. 核心细节解析与实操要点3.1 模型输出格式错误的修复套路Agent 里最常见的错误之一就是模型返回的 JSON 不合法。少个引号、多个逗号、被 markdown 代码块包起来这些都会让解析直接崩掉。我的处理分三步走。第一步是预处理清洗。模型经常把 JSON 包在json里或者前面加一句好的这是结果。所以解析前先做正则提取把代码块标记和前后废话剥掉。这一步能解决大概六成的格式问题。第二步是容错解析。标准库的 json.loads 太严格我一般会用一个宽松解析器比如 Python 的json5或者自己写个修复函数处理尾逗号、单引号、未转义字符这些常见毛病。下面是我常用的一个修复函数import json import re def robust_json_parse(text): # 剥离 markdown 代码块 text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 修复常见问题尾逗号、单引号 text re.sub(r,\s*([}\]]), r\1, text) text text.replace(, ) try: return json.loads(text) except json.JSONDecodeError: return None第三步是重试加提示。如果清洗和容错都救不回来就把错误信息拼回 prompt让模型重新生成一次明确告诉它上次输出不是合法 JSON请只输出 JSON不要任何额外文字。实测下来加上这句提示后第二次成功率能到 95% 以上。注意重试次数别超过 2 次。如果模型连续两次都输出不了合法格式说明这个任务本身可能超出了它的能力继续重试只是浪费 token不如直接降级到兜底逻辑。3.2 工具调用的超时与降级设计工具调用是 Agent 和外部世界交互的接口也是最容易出问题的地方。我的原则是每个工具调用都必须有超时每个工具都必须有降级方案。超时时间怎么定不能拍脑袋。我的方法是先统计这个工具在正常情况下的 P99 响应时间然后把超时设成 P99 的 2 到 3 倍。比如一个搜索接口 P99 是 800 毫秒那超时设 2 秒比较合理。设太短会误杀正常请求设太长会让 Agent 卡死。降级方案则要提前想好。搜索工具挂了能不能用缓存结果数据库查询超时了能不能返回一个默认值这些都要在代码里写死而不是等出事了再临时想。我一般会给每个工具配一个 fallback 函数主逻辑失败时自动切换。def call_with_fallback(primary, fallback, timeout2.0): try: return primary(timeouttimeout) except (TimeoutError, ConnectionError) as e: log.warning(fprimary failed: {e}, using fallback) return fallback()3.3 状态快照让 Agent 能回到过去编排层的错误最麻烦因为 Agent 是有状态的。多轮对话、多步任务中间任何一步出错状态就可能乱掉。我的做法是在每个关键步骤前打快照。快照不需要存全部状态只存恢复所需的最小信息就够了。比如当前执行到第几步、已经收集了哪些参数、上一步的输出是什么。这样一旦出错可以回滚到上一个稳定状态重新执行而不是从头再来。快照的存储我一般用 Rediskey 是会话 ID 加步骤号value 是序列化后的状态设个合理的过期时间比如 1 小时。这样既不会占太多内存又能保证短时间内可以恢复。4. 实操过程与核心环节实现4.1 搭建一个带完整错误处理的 Agent 骨架光讲理论没意思我直接给一个可运行的骨架。这个骨架把前面讲的四层错误处理都串起来了你可以直接拿去改。import time import random import logging from functools import wraps logger logging.getLogger(__name__) class RetryConfig: def __init__(self, max_attempts3, base_delay0.5, max_delay30): self.max_attempts max_attempts self.base_delay base_delay self.max_delay max_delay def with_retry(config: RetryConfig, retryable(TimeoutError, ConnectionError)): def decorator(func): wraps(func) def wrapper(*args, **kwargs): last_exc None for attempt in range(config.max_attempts): try: return func(*args, **kwargs) except retryable as e: last_exc e if attempt config.max_attempts - 1: break delay min( config.max_delay, config.base_delay * (2 ** attempt) ) random.uniform(0, 0.5) logger.warning( f{func.__name__} attempt {attempt1} failed: {e}, fretry in {delay:.2f}s ) time.sleep(delay) raise last_exc return wrapper return decorator这个装饰器的关键点在于只对可重试的异常生效。像 ValueError 这种逻辑错误重试一百次也没用所以不在 retryable 里。退避公式里加了random.uniform(0, 0.5)的抖动避免并发场景下的同步重试。4.2 幂等键的生成与校验实现幂等键的设计有几个细节容易踩坑。第一键的生成必须在客户端不能服务端生成否则重试时生成新键就失去意义了。第二键要能唯一标识这一次业务操作通常用业务 ID 加操作类型拼出来。import hashlib import redis r redis.Redis() def make_idempotency_key(user_id, action, biz_id): raw f{user_id}:{action}:{biz_id} return hashlib.sha256(raw.encode()).hexdigest() def execute_idempotent(key, func, ttl3600): # SETNX 抢占成功说明是首次执行 if r.set(key, processing, nxTrue, exttl): try: result func() r.set(key, fdone:{result}, exttl) return result except Exception as e: r.delete(key) # 失败释放允许重试 raise else: # 已存在检查状态 status r.get(key) if status and status.startswith(bdone:): return status.decode().split(:, 1)[1] raise RuntimeError(duplicate request in progress)这段代码的逻辑是用 SETNX 抢占一个键抢到就执行执行完把结果写进去没抢到说明有重复请求如果前一个已经完成就直接返回缓存结果如果还在处理中就报错让客户端稍后重试。这里有个坑要注意——失败时必须删除键否则一次失败会让后续所有重试都被判定为重复请求永远执行不了。4.3 并发场景下的 Agent 状态隔离AI Agent 怎么扛并发是很多人关心的问题。核心就一句话状态必须隔离共享资源必须加锁。Agent 的状态分两种会话状态和全局状态。会话状态天然隔离每个会话一个 key 就行。全局状态比如共享的缓存、计数器才需要小心。我的做法是能用无状态就不用有状态实在需要共享的用 Redis 的原子操作而不是自己加锁。比如限流不要用读-判断-写这种三步操作直接用 Redis 的 INCR 加过期时间def rate_limit(key, limit, window60): current r.incr(key) if current 1: r.expire(key, window) return current limitINCR 是原子操作天然并发安全比加锁简单得多。这个思路可以推广到大部分并发场景——优先用原子操作其次用乐观锁最后才考虑悲观锁。5. 常见问题与排查技巧实录5.1 错误处理速查表我把实际项目里遇到的高频问题整理成了一张表出问题时可以对照排查现象可能原因排查方向解决手段Agent 卡住不返回工具调用无超时看日志最后停在哪一步加超时熔断数据重复写入重试未做幂等查是否有重复业务 ID加幂等键状态错乱并发共享状态看是否有全局变量状态隔离原子操作格式解析失败模型输出不规范打印原始输出清洗容错重试内存持续增长快照未清理看 Redis 内存设过期时间重试风暴无退避无抖动看重试时间分布指数退避抖动5.2 几个只有踩过才知道的坑第一个坑是重试放大了副作用。有次我们的 Agent 调用一个发短信的工具超时后重试结果用户收到了三条短信。后来加了幂等键才解决。教训是任何有副作用的操作上线前必须问一句重试会怎样。第二个坑是日志里没有上下文。出错时只看到timeout不知道是哪个会话、哪一步、什么参数。后来我在每个日志里都带上 trace_id 和 step排查效率直接翻倍。这个投入非常值得建议一开始就做。第三个坑是熔断阈值设得太敏感。有次把阈值设成连续 3 次失败就熔断结果下游只是短暂抖动Agent 就自己把自己熔断了白白损失了可用性。后来改成10 次里失败 8 次稳定多了。阈值这东西没有标准答案得根据自己的业务容忍度调。第四个坑是快照存了太多东西。一开始我把整个对话历史都塞进快照结果 Redis 内存暴涨。后来只存恢复必需的最小状态内存占用降了 90%。快照的原则是够用就好不是越多越安全。5.3 可观测性让错误无处可藏错误处理做得好不好很大程度上取决于你能不能看见错误。我的建议是至少埋三类指标成功率、延迟分布、错误分类计数。成功率按工具维度统计能快速定位是哪个工具在拖后腿。延迟分布看 P50、P95、P99能发现慢查询和性能退化。错误分类计数则按前面说的四层模型来分能看出问题是出在模型、工具还是编排上。这些指标不用搞得很复杂一个简单的计数器加定时上报就够了。关键是要有而不是追求完美。我见过太多项目连基本的错误日志都没有出了问题只能靠猜那才是真的痛苦。6. 工程化落地的几个经验之谈6.1 配置化别把参数写死在代码里重试次数、超时时间、熔断阈值这些参数千万别硬编码。不同环境、不同工具的最优值都不一样写死了改起来就得重新发版。我的做法是全部抽到配置文件里按工具维度配置tools: search: timeout: 2.0 max_attempts: 3 base_delay: 0.5 fallback: cache_search database: timeout: 5.0 max_attempts: 2 base_delay: 1.0 fallback: null这样调参不用改代码改配置重启就行。而且配置本身就是文档新人一看就知道每个工具的重试策略是什么。6.2 测试错误路径比正常路径更重要大部分人写测试只测正常流程但错误处理的测试才是真正值钱的。我一般会专门写一组故障注入测试模拟工具超时、模拟返回非法格式、模拟并发重复请求验证 Agent 能不能正确处理。故障注入不用搞得很复杂用 mock 把工具替换成必定抛异常或必定超时的版本就行。重点验证三件事重试有没有生效、幂等有没有守住、降级有没有触发。这三件事测过了上线心里就有底了。6.3 渐进式上线别一次性全量错误处理策略再完善也架不住真实环境的复杂性。我的建议是灰度上线先放 1% 的流量观察错误率和重试率没问题再逐步放大。这样即使策略有问题影响面也可控。灰度期间重点看两个指标重试率和降级率。重试率突然升高说明下游有问题降级率升高说明主逻辑不稳定。这两个指标是错误处理策略的体温计能提前发现隐患。6.4 关于 Agent 框架选择的补充顺便说一句框架的事。现在主流的 Agent 框架不少但错误处理这块框架能帮你做的其实有限。大部分框架提供了重试的钩子但幂等、熔断、状态快照这些还是得自己实现。所以选框架时别太看重它内置了多少错误处理而要看它容不容易让你插入自己的错误处理逻辑。钩子丰富、扩展点多的框架才是工程化的好选择。我在实际项目里的体会是错误处理这件事没有银弹它是一堆小决策的累积每个工具的超时设多少、每个操作要不要幂等、每个错误该不该重试。这些决策单独看都不难难的是系统性地想清楚并坚持执行。等你把这些都理顺了Agent 才算真正从能跑变成了能扛。最后分享一个小技巧给每个 Agent 会话生成一个 trace_id贯穿所有日志和工具调用。出问题时一个 trace_id 就能把整条链路串起来排查效率能提升好几倍。这个投入很小但回报极大强烈建议一开始就加上。