引言
在Agent系统中,失败是常态。LLM API超时、工具调用失败、网络波动——这些故障随时会发生。如何处理这些失败,决定了系统的可靠性。
重试是最直接的容错手段,但"盲目重试"——不分析原因、不调整策略地反复重试——不仅浪费资源,还可能加剧问题。2026年的Agent系统需要的是"智能重试"——根据故障类型、上下文和历史数据做出最优的重试决策。
一、何时应该重试
1.1 可重试错误 vs 不可重试错误
可重试错误:
- 网络超时
- 服务暂时不可用(503)
- 速率限制(429)
- 临时性数据冲突
不可重试错误:
- 参数错误(400)
- 认证失败(401/403)
- 资源不存在(404)
- 业务逻辑错误(如"余额不足")
1.2 判断框架
def should_retry(error, attempt_count, context):
# 不可重试错误直接返回False
if error.type in NON_RETRYABLE_ERRORS:
return False
# 超过最大重试次数
if attempt_count >= context.max_retries:
return False
# 速率限制:可以重试,但需要等待
if error.type == "rate_limit":
wait_time = error.retry_after or calculate_backoff(attempt_count)
return True, wait_time
# 网络超时:检查是否值得重试
if error.type == "timeout":
if context.estimated_remaining_time > context.deadline:
return False # 重试也无法在截止时间内完成
return True, calculate_backoff(attempt_count)
# 服务不可用:重试
if error.type == "service_unavailable":
return True, calculate_backoff(attempt_count)
# 未知错误:保守重试
return True, calculate_backoff(attempt_count)
二、退避算法
2.1 固定间隔
每次重试间隔相同时间。简单但可能造成"重试风暴"——所有失败请求同时重试。
2.2 指数退避
每次重试间隔按指数增长:
Retry 1: wait 1s
Retry 2: wait 2s
Retry 3: wait 4s
Retry 4: wait 8s
Retry 5: wait 16s
def exponential_backoff(attempt, base=1, factor=2, max_delay=60):
delay = min(base * (factor ** attempt), max_delay)
return delay
指数退避是最常用的策略,有效避免重试风暴。
2.3 抖动退避
在指数退避基础上添加随机抖动,进一步分散重试时间:
def jittered_backoff(attempt, base=1, factor=2, max_delay=60):
delay = min(base * (factor ** attempt), max_delay)
jitter = random.uniform(0, delay * 0.1) # 10%抖动
return delay + jitter
2026年的最佳实践是"全抖动"(Full Jitter):
def full_jitter_backoff(attempt, base=1, factor=2, max_delay=60):
delay = min(base * (factor ** attempt), max_delay)
return random.uniform(0, delay) # 在0和计算延迟间随机
2.4 适应性退避
根据历史重试成功率动态调整退避参数:
class AdaptiveBackoff:
def __init__(self):
self.success_rate = 1.0
self.base_delay = 1.0
def update(self, success):
if success:
self.success_rate = self.success_rate * 0.9 + 1.0 * 0.1
else:
self.success_rate = self.success_rate * 0.9 + 0.0 * 0.1
# 成功率低时增加延迟
if self.success_rate < 0.5:
self.base_delay = min(self.base_delay * 1.5, 10.0)
# 成功率高时减少延迟
elif self.success_rate > 0.9:
self.base_delay = max(self.base_delay * 0.9, 0.5)
def get_delay(self, attempt):
delay = min(self.base_delay * (2 ** attempt), 60)
return random.uniform(0, delay)
三、Agent特定的重试场景
3.1 LLM调用重试
LLM调用失败的原因多样,需要分别处理:
| 失败原因 | 重试策略 | 特殊处理 |
|---|---|---|
| 速率限制 | 指数退避 | 遵守Retry-After头 |
| 上下文超长 | 不重试 | 截断上下文后重新请求 |
| 内容被拒绝 | 不重试 | 修改prompt后重新请求 |
| 服务不可用 | 抖动退避 | 切换备用模型 |
| 超时 | 抖动退避 | 减少max_tokens |
3.2 工具调用重试
工具调用的重试需要考虑工具的幂等性:
幂等工具:可以安全重试。如查询数据、搜索信息。
非幂等工具:重试可能导致副作用。如发送邮件、创建订单。对于非幂等工具,重试前需要检查上一次操作是否实际执行了。
async def safe_retry_tool_call(tool, params, max_retries=3):
for attempt in range(max_retries):
try:
result = await tool.call(**params)
return result
except TimeoutError:
# 超时:可能执行了也可能没执行
if tool.is_idempotent:
continue # 幂等工具直接重试
else:
# 非幂等工具:检查是否已执行
existing = await tool.check_existing(params)
if existing:
return existing # 已执行,返回之前的结果
continue # 确认未执行,重试
except Exception as e:
if not is_retryable(e):
raise
await asyncio.sleep(calculate_backoff(attempt))
3.3 多步任务重试
Agent执行多步任务时,某一步失败后的重试策略:
从失败步骤重试:不需要重做之前的步骤。前提是步骤间无依赖。
从检查点重试:从最近的检查点恢复,可能需要重做部分步骤。
完全重试:从头开始。简单但浪费资源,适合短任务。
async def execute_with_retry(steps, checkpoint_manager, max_retries=3):
for attempt in range(max_retries):
# 从检查点恢复
start_step = checkpoint_manager.get_last_checkpoint()
for i, step in enumerate(steps[start_step:], start=start_step):
try:
result = await step.execute()
checkpoint_manager.save_checkpoint(i, result)
except RetryableError as e:
if attempt < max_retries - 1:
await asyncio.sleep(calculate_backoff(attempt))
break # 跳出内循环,进入下一次外循环重试
else:
raise
else:
return # 所有步骤成功完成
四、重试中的上下文处理
4.1 上下文保留
重试时保留原始上下文,但可能需要调整:
async def retry_with_context_adjustment(original_request, error, attempt):
adjusted_request = original_request.copy()
if error.type == "context_too_long":
# 截断上下文
adjusted_request["messages"] = truncate_messages(
original_request["messages"],
max_tokens=original_request.get("max_tokens", 4096)
)
if error.type == "content_filter":
# 修改prompt
adjusted_request["messages"] = sanitize_messages(
original_request["messages"]
)
if attempt > 2 and error.type == "timeout":
# 减少输出长度
adjusted_request["max_tokens"] = original_request.get("max_tokens", 4096) // 2
return adjusted_request
4.2 重试信息注入
在重试时注入失败信息,帮助LLM避免重复错误:
原始请求: "生成一个Python函数来读取CSV文件"
第一次失败: 生成的代码有语法错误
重试请求: "生成一个Python函数来读取CSV文件。
注意:上次尝试生成的代码有以下语法错误:[错误详情]。请避免这些问题。"
五、降级策略
当重试耗尽时,需要降级策略:
5.1 模型降级
GPT-4o → GPT-4o-mini → GPT-3.5 → 规则引擎
每次降级到更快但可能更弱的模型。
5.2 功能降级
完整分析 → 简化分析 → 返回原始数据
逐步减少功能,确保至少返回有用结果。
5.3 缓存降级
实时计算 → 使用缓存结果(标注可能过时)→ 返回默认值
5.4 人工降级
自动处理 → 请求人工处理 → 告知用户稍后处理
六、重试监控
6.1 关键指标
- 重试率:需要重试的请求比例。目标:<10%
- 重试成功率:重试后成功的比例。目标:>70%
- 平均重试次数:每个请求的平均重试次数。目标:<1.5
- 重试延迟:重试增加的额外延迟
- 降级率:降级处理的比例。目标:<1%
6.2 重试分析
定期分析重试数据,发现系统性问题:
- 哪些工具/LLM重试率最高?
- 重试失败后降级的主要原因是什么?
- 是否存在"重试风暴"(大量请求同时重试)?
- 重试成功率是否有下降趋势?
结语
重试策略看似简单——“失败了再来一次”——但做好重试策略需要深入理解系统特性、失败模式和业务需求。好的重试策略应该是"聪明的"——知道何时重试、如何重试、何时停止。
2026年的重试策略正在向"预测性重试"方向演进——系统根据历史数据预测某次请求的失败概率,在发送前就调整策略(如提前切换到备用模型)。这种预测性方法可以进一步减少不必要的失败和重试,提升系统整体可靠性。
但无论策略多么智能,重试的本质仍然是"面对失败的韧性"。一个可靠的Agent系统不是不失败的系统,而是能够优雅地从失败中恢复的系统。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞,认知差异创造无限可能。