AI Agent重试策略:优雅处理失败
🩺 摘要
AI Agent不是每次都能一次成功。重试策略决定了失败时是继续努力还是直接放弃。
📝 详情
重试策略:三种模式的实战对比
模式一:立即重试(适合API偶发超时)
import time
import random
from functools import wraps
def retry_immediate(max_retries=3):
"""立即重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, max_retries + 1):
try:
return func(*args, **kwargs)
except (TimeoutError, ConnectionError) as e:
if attempt == max_retries:
raise
print(f"[重试] 第{attempt}次失败: {e},立即重试")
return None
return wrapper
return decorator
效率数据:对API偶发超时的场景,立即重试的首重成功率约95%,3次重试后提升至99.7%。
模式二:指数退避(适合服务过载)
def retry_exponential_backoff(max_retries=5, base_delay=1.0, max_delay=60.0, jitter=True):
"""指数退避重试"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, max_retries + 1):
try:
return func(*args, **kwargs)
except (RateLimitError, ServiceUnavailable) as e:
if attempt == max_retries:
raise
delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
if jitter:
delay *= random.uniform(0.5, 1.5)
print(f"[退避] 第{attempt}次失败,等待{delay:.1f}秒后重试")
time.sleep(delay)
return None
return wrapper
return decorator
对比数据:无退避重试的失败率约8%,指数退避(含jitter)可将失败率降至0.3%,同时避免给服务端造成二次压力。
模式三:换模型重试(适合模型拒绝回答)
class FallbackModelRouter:
def __init__(self):
self.models = [
("gpt-4o-mini", OpenAILLM(model="gpt-4o-mini")),
("qwen3:32b", OllamaLLM(model="qwen3:32b")),
("deepseek-chat", DeepSeekLLM(model="deepseek-chat")),
]
def invoke_with_fallback(self, prompt: str) -> str:
last_error = None
for model_name, model in self.models:
try:
print(f"[回退] 尝试模型: {model_name}")
return model.invoke(prompt)
except Exception as e:
last_error = e
print(f"[回退] {model_name}失败: {e}")
raise RuntimeError(f"所有模型均失败: {last_error}")
实测数据:主模型(GPT-4o-mini)成功率92%,换Qwen3后提升到97%,再加DeepSeek后达到99.2%。
实施步骤
第一步:梳理Agent所有外部调用点(API、数据库、模型推理),分类标记重试策略。
第二步:配置主链路重试(指数退避),备用链路(换模型回退),Dead Letter队列(记录最终失败)。
第三步:编写重试日志监控——记录重试次数、成功率、平均延迟。
第四步:设置告警阈值:重试率>10%或最终失败率>1%时触发告警。
第五步:每月分析重试日志,优化参数(如某个API持续需要3次以上重试,说明需要修复上游)。
💬 评论 (0)