🎯 课程主题

ToolRetryMiddleware 基于指数退避 + 随机抖动算法，在工具调用失败时自动重试，避免瞬时并发对服务造成二次冲击。

📝 核心知识点

1. 指数退避（Exponential Backoff）

  • 概念说明：操作失败后不立即重试，而是逐步增加重试间隔（如 1s → 2s → 4s → 8s → ...），让服务端有时间恢复。
  • 关键细节：
    • 公式：delay = initial_delay × (backoff_factor ^ retry_count)
    • backoff_factor 默认为 2（以 2 为底的指数增长）。
    • 设置 max_delay 上限防止延迟无限增长。
    • 对比"马上重试"的弊端：大量客户端同时重试等同于对已瘫痪的服务器发起 DDoS 攻击。

2. 随机抖动（Jitter）

  • 概念说明：在指数退避的基础上加入随机偏移，让同一批次失败请求的重试时间点错峰分布，避免"惊群效应"。
  • 关键细节：
    • 例如理论间隔 2s，加入抖动后可能是 1.8s 或 2.1s。
    • 关闭抖动（jitter=False）则退化为严格固定的指数退避，每次间隔更可预测。
    • 生产环境强烈建议开启抖动。

3. 核心参数

参数类型说明
max_retriesint最大重试次数（不含首次调用，设 6 则共执行 7 次）
backoff_factorint退避因子，每次延迟乘以该值（默认 2）
initial_delayfloat初始延迟秒数（默认 1）
max_delayfloat最大延迟上限秒数（默认 60）
jitterbool是否启用随机抖动（默认 True）
retry_ontuple触发重试的异常类型元组
on_failurestr达到最大重试后的行为："continue" 继续执行 或 "error" 抛异常

🏗️ 架构与工作流

Agent 选择工具 → 工具调用
                     ↓ (失败)
                 等待 delay 秒（含抖动）
                     ↓
                 重试工具调用
                     ↓ (失败)
                 等待 delay × backoff_factor 秒
                     ↓
                 ... 重复至 max_retries 次
                     ↓ (仍失败)
             on_failure="continue" → Agent 继续执行
             on_failure="error"    → 抛出异常

💻 代码实战

示例 1：开启抖动（jitter=True）

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
import time

call_count = 0
last_call_time = None

def get_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    global call_count, last_call_time
    current_time = time.time()
    if last_call_time is None:
        last_call_time = current_time
    interval = current_time - last_call_time
    call_count += 1

    # 记录每次调用的间隔到文件
    with open("tool_call_log.txt", "a") as f:
        f.write(f"第{call_count}次调用, 间隔: {interval:.2f}秒\n")
    last_call_time = current_time

    # 模拟工具调用失败
    raise TimeoutError("工具调用超时")

tool_retry_middleware = ToolRetryMiddleware(
    max_retries=6,          # 最大重试次数 6（共执行 7 次）
    backoff_factor=2,       # 指数底数
    initial_delay=1.0,      # 初始延迟 1 秒
    max_delay=10.0,         # 最大延迟上限 10 秒
    jitter=True,            # 开启抖动
    retry_on=(TimeoutError,),  # 仅此类异常触发重试
    on_failure="continue"   # 重试耗尽后让 Agent 继续
)

agent = create_agent(
    model="deepseek-v4-pro",
    tools=[get_weather],
    middleware=[tool_retry_middleware]
)

result = agent.invoke({"messages": [{"role": "user", "content": "北京天气如何？"}]})
print(result["messages"][-1].content)
# 输出: 抱歉，我暂时无法获取北京的天气信息

# 查看 tool_call_log.txt，间隔大致为：0s, ~1s, ~2s, ~4s, ~8s, ~10s, ~10s（受抖动和上限影响）

示例 2：关闭抖动（jitter=False）

tool_retry_middleware = ToolRetryMiddleware(
    max_retries=6,
    backoff_factor=2,
    initial_delay=1.0,
    max_delay=10.0,
    jitter=False,           # 关闭抖动
    retry_on=(TimeoutError,),
    on_failure="continue"
)
# 此时间隔更接近严格的指数增长：0s, 1s, 2s, 4s, 8s, 10s, 10s

⚠️ 常见问题与避坑指南

  • max_retries 是重试次数而非总调用次数，设 6 则共调用 7 次（1 次初始 + 6 次重试）。
  • retry_on 必须是元组类型，即使只捕获一种异常也要写成 (TimeoutError,)。
  • 关闭抖动（jitter=False）后间隔更接近严格指数增长，生产环境建议开启以分散重试压力。

💡 个人总结与延伸

指数退避 + 抖动是分布式系统中经典的容错模式，在 AWS SDK、gRPC 等主流框架中均有实现。LangChain 将其封装为中间件，使得 Agent 工具调用天然具备健壮的重试能力。实际开发中可结合具体工具的超时特性自定义 backoff_factor 和 max_delay。

最近更新:
Contributors: yangwenguang