跳转至

Agent 调用外部工具时,如何区分 应该重试 和 不应该重试 的错误?

当 Agent 调用外部工具(搜索、计算、发送请求等)出错时,有些错误重试一下就能成功,有些重试反而雪上加霜。如何自动区分“该重试”和“不该重试”的错误?

这个问题本质上是:如何把异常从一堆 Exception 变成有语义的决策信号。


🧠 核心思路:让异常“说话”

重试策略不该靠“猜”,而应让每个异常明确回答两个问题:

  • 🔄 瞬态还是持久? 网络闪断是瞬态的,权限拒绝是持久的。

  • ♻️ 重复执行安全吗? 即接口是否幂等(例如查询天然可重试,创建订单重复提交就可能扣款两次)。

因此,我们需要在错误抛出时就打上标签,重试框架根据标签自动决策。


🏷️ 错误三分法:可重试 / 不可重试 / 需谨慎

类型 图标 典型场景 决策
瞬时性错误 🔄 网络超时、服务 503、连接拒绝、限流 429 ✅ 应重试(带退避)
明确业务失败 🚫 参数校验失败、权限不足、资源不存在 (404) ❌ 不应重试,直接向上抛
需判断的中间态 ⚠️ 远程服务返回 409 Conflict、408 Timeout、部分写入 需结合幂等性和状态检查决定

有了这个分类,我们把错误包装成不同异常类,重试器只看类型即可。


⚙️ 落地实现:异常分层 + 重试装饰器

1️⃣ 设计异常树

class ToolError(Exception):
    """所有工具异常的基类"""

class RetryableError(ToolError):
    """标记:此错误可重试"""
    pass

class NonRetryableError(ToolError):
    """标记:此错误不可重试"""
    pass

# 具体例子
class ExternalServiceTimeout(RetryableError):
    pass

class InvalidParameterError(NonRetryableError):
    pass

2️⃣ 工具调用时抛出对应异常

def call_search_api(query):
    resp = requests.get("...", timeout=5)
    if resp.status_code == 503:
        raise ExternalServiceTimeout("服务暂时不可用")
    if resp.status_code == 400:
        raise InvalidParameterError(f"参数错误: {query}")
    return resp.json()

3️⃣ 智能重试装饰器(识别异常类型)

import time
from functools import wraps

def tool_retry(max_retries=3, backoff=1.5):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, max_retries+1):
                try:
                    return func(*args, **kwargs)
                except RetryableError as e:
                    last_exc = e
                    print(f"🔄 可重试错误,第{attempt}次重试: {e}")
                    time.sleep(backoff * (attempt-1))  # 线性退避
                except NonRetryableError as e:
                    raise  # 不可重试,直接抛出
            raise last_exc  # 重试耗尽
        return wrapper
    return decorator

4️⃣ 使用效果

@tool_retry(max_retries=3)
def search_tool(query):
    return call_search_api(query)

# 如果是 ExternalServiceTimeout,会自动重试3次
# 如果是 InvalidParameterError,会立刻失败

现在 Agent 调用 search_tool 时,就透明地拥有了按异常类型决策的重试能力。


📊 一个更聪明的做法:让异常自己携带重试建议

有时候重试策略不是二元的,比如 HTTP 429 应该根据 Retry-After 头来退避。可以让异常承载这些信息:

class RateLimitedError(RetryableError):
    def __init__(self, retry_after_seconds):
        self.retry_after = retry_after_seconds
        super().__init__(f"限流,需等待{retry_after_seconds}s")

重试装饰器再根据 e.retry_after 动态计算休眠时间。这样错误本身就成了重试策略的“参数”。