Agent工具调用失败自动重试、回退降级策略
Agent在调用搜索、数据库、第三方API等外部工具时,常会遇到网络超时、限流、服务端5xx等瞬时故障。
如果不做任何处理,一次工具失败会导致整个任务中断;
如果简单粗暴地重试多次,又会加剧上游压力。
本文给出一个可直接落地的策略:先把错误分成可重试与不可重试,再用指数退避控制重试节奏,最后通过主备工具和人工兜底完成回退降级,让Agent任务在真实故障中仍能稳定运行。
先分清哪些失败值得重试
不是所有错误都适合重试。
可重试的错误通常具备"瞬时性":连接超时、读超时、429限流、500/502/503/504等。
不可重试的错误通常是"确定性"的:400参数错误、401未授权、404不存在、405方法不允许等。
重试前者能恢复,重试后者只会浪费时间和资源。
建议在调用工具前包一层错误分类函数:
def is_retryable(error):
if hasattr(error, "status_code"):
return error.status_code in (408, 429, 500, 502, 503, 504)
return isinstance(error, (TimeoutError, ConnectionError))
这样后续重试逻辑只对可重试错误生效。
实现带指数退避的自动重试
指数退避是重试的基础策略:每次失败后等待时间按指数增长,再加上随机抖动,避免多个请求同时重试造成"惊群"。
最大重试次数建议设置为3次,基础等待1秒,第1次失败等1秒,第2次等2秒,第3次等4秒,整体耗时约7秒,不会拖垮任务链路。
import time
import random
def call_with_retry(func, max_retries=3, base_delay=1):
for attempt in range(max_retries + 1):
try:
return func()
except Exception as err:
if attempt >= max_retries or not is_retryable(err):
raise
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
time.sleep(delay)
如果所有重试都失败,再进入降级逻辑。
回退降级:主工具失败就用备用方案
重试耗尽后,不能让Agent直接报错,而应按优先级切换备用方案。
典型的回退顺序是:主工具 → 备用工具 → 缓存数据或规则引擎 → 人工处理队列。
以股票信息查询为例,实时API失败后,可先降级到延迟行情接口;
再失败就使用本地缓存;
最后仍然不行,则返回"暂时无法获取,请稍后再试",并把任务写入待人工确认队列。
def call_with_fallback(*callables):
for callable_ in callables:
try:
return callable_()
except Exception:
continue
raise RuntimeError("所有工具均调用失败")
调用时按优先级传入函数列表,系统会按顺序自动切换。
用日志和指标验证策略是否生效
只写代码不观察,等于没做。
建议在调用入口输出结构化日志,至少包含:工具名称、错误类型、重试次数、是否触发降级、降级后的工具名。
如果你有监控系统,可以记录三个核心指标:tool_call_total(总调用次数)、tool_retry_count(重试次数)、tool_fallback_count(降级次数)。
验证方法很简单:用Mock接口模拟返回500一次、429两次,再模拟第三个接口正常返回,观察日志中重试次数是否递减、等待间隔是否接近1秒、2秒、4秒,降级顺序是否按预设执行。
避坑清单
- 幂等性:重试前必须确认工具调用是幂等的,否则可能重复扣费、重复提交订单。必要时给请求加唯一请求ID,让上游去重。
- 超时上限:总重试时间不能无限增长,建议设置整体超时,比如最重试加降级不超过15秒。
- 错误分类要准确:如果把403、404误判为可重试,会白白消耗大量重试次数,甚至触发上游风控。
- 降级不能无底洞:所有备用工具都失败时,要明确返回用户可理解的提示,并记录工单,而不是让任务卡在死循环里。
如果你正在处理Agent工具调用失败自动重试、回退降级策略,建议先按上述步骤实现重试和fallback链,再逐步接入监控;
遇到异常时优先回看避坑清单,定位是重试逻辑问题还是错误分类问题。