智能体工具调用失败后怎样设计可控重试

我第一次给智能体接上“自动重试”时，规则很简单：工具报错就再调三次。测试环境里它看起来很可靠，到了真实任务才暴露问题——查询接口被重复轰炸，创建工单的请求在超时后又执行了一次，模型还会把同一个坏参数换种说法继续提交。

直接结论：可控重试不是“失败后重复 N 次”，而是先做错误分类，再同时约束可重试性、幂等性、重试预算、退避、熔断和观测。参数错误应让模型修正；超时、限流和部分 5xx 才进入自动重试；鉴权、权限、业务拒绝等永久错误要立即停止。有副作用的调用如果没有服务端幂等保障，就不应自动重放。

OpenAI Function Calling 官方文档：https://developers.openai.com/api/docs/guides/function-calling

先比较三种失败处理方案

工具调用失败后，常见选择其实只有三种：把结构化错误返回给模型，让它修改参数；由执行层原样重试；停止并升级给人工或上层流程。它们解决的是不同问题，不能用一个固定次数配置替代。

三种策略不是互相替代，而是由失败类型决定：参数错误交给模型修正，瞬时故障进入受控重试，永久故障立即停止。

我现在会先问四个问题：这次失败换参数能修好吗？同样请求稍后执行成功概率会明显变高吗？工具有没有副作用？上一次请求到底有没有成功？最后一个问题最容易被忽略：客户端看到超时，只说明没有及时收到响应，并不证明服务端没有完成创建、发送或扣减。

分类时不要只看异常名称

异常类型只是线索，真正的决策要结合工具语义。比如读取天气的超时与发送消息的超时都叫 TimeoutError，前者通常可以安全重试，后者如果没有幂等键就可能重复发送。建议把原始异常归一化为以下四类：

• ARGUMENT：工具已接收请求，但参数不满足 schema 或业务前置条件。返回结构化、可修正的错误给模型，不在执行层原样重放。
• TRANSIENT：网络中断、超时、限流或临时服务错误。满足幂等要求时，进入有限次数的退避重试。
• PERMANENT：无效凭据、权限不足、策略禁止、确定不存在的资源。立即停止，避免把确定失败变成流量放大器。
• UNKNOWN：无法确认类别或无法确认上一次操作结果。默认停止；只有只读且天然幂等的工具才考虑一次保守重试。

参数错误的返回也应可执行，而不是只写“调用失败”。更实用的结果会包含错误码、问题字段、允许值和是否可修正，例如 {"code":"INVALID_DATE","field":"start_date","retryable_by_model":true}。模型获得的是新的约束，而不是同一失败的自然语言复述。

把重试做成独立控制器

把“请重试”写在系统提示词里很方便，却很难统一统计次数，也挡不住 SDK、HTTP 客户端和网关各自再试一轮。我更倾向于把重试做成工具适配器旁边的独立控制器：智能体只决定要完成什么任务，控制器负责一次调用最多花多少时间、可以尝试几次、什么时候熔断以及每次决策如何留痕。

把重试从智能体提示词里抽离为独立控制器，才能统一管理预算、退避、幂等、熔断和审计。

共享预算比局部次数更重要

假设模型最多重新规划 3 次，工具适配器每次重试 3 次，底层 HTTP 客户端又默认重试 3 次，最坏情况下一个用户任务可能触发 27 次请求。解决办法不是逐个把次数改小，而是给整个任务分配共享预算，例如最多 5 次工具尝试、总等待不超过 8 秒、同一工具不超过 3 次。每一层都消耗同一个预算。

退避必须有上限和抖动

瞬时故障可以使用指数退避，但延迟要封顶，并加入少量随机抖动，避免大量智能体在同一时间重新冲击服务。可用的计算方式是 min(max_delay, base_delay × 2^(attempt-1)) + jitter。如果服务返回了可信的 Retry-After，应优先遵守它，但仍受任务总时限约束。

副作用工具必须有幂等策略

创建订单、发送通知、提交表单和删除资源都属于副作用工具。可靠做法是由调用方生成稳定的操作标识，将幂等键传给服务端，并在幂等账本里记录请求指纹与最终结果。同一任务重试时复用同一个键，而不是每次生成新键。若服务端不支持幂等，就应把“结果未知”视为需要查询或人工确认的状态，而不是再次执行。

一个最小的 Python 控制器

下面的示例故意不绑定具体智能体 SDK。工具适配器只要把异常映射为统一的 ToolFailure，就可以复用同一套策略。

from dataclasses import dataclass
from enum import Enum
import random
import time


class FailureKind(Enum):
    ARGUMENT = "argument"
    TRANSIENT = "transient"
    PERMANENT = "permanent"
    UNKNOWN = "unknown"


class ToolFailure(Exception):
    def __init__(self, kind: FailureKind, message: str):
        super().__init__(message)
        self.kind = kind


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 3
    base_delay: float = 0.4
    max_delay: float = 4.0
    max_elapsed: float = 8.0


def call_with_retry(tool, args, *, policy, side_effect=False,
                    idempotency_key=None):
    # 有副作用的工具缺少幂等键时禁止自动重试
    if side_effect and not idempotency_key:
        return {"ok": False, "decision": "manual_check",
                "error": "missing_idempotency_key"}

    started = time.monotonic()
    for attempt in range(1, policy.max_attempts + 1):
        try:
            # 同一任务的所有尝试必须复用同一个幂等键
            return tool(args, idempotency_key=idempotency_key)
        except ToolFailure as exc:
            # 参数错误交给模型修正，永久或未知错误立即停止
            if exc.kind == FailureKind.ARGUMENT:
                return {"ok": False, "decision": "model_fix",
                        "error": str(exc)}
            if exc.kind != FailureKind.TRANSIENT:
                return {"ok": False, "decision": "stop",
                        "error": str(exc)}

            elapsed = time.monotonic() - started
            if attempt == policy.max_attempts or elapsed >= policy.max_elapsed:
                return {"ok": False, "decision": "budget_exhausted",
                        "error": str(exc)}

            # 指数退避封顶，并加入随机抖动避免同时重试
            delay = min(policy.max_delay,
                        policy.base_delay * (2 ** (attempt - 1)))
            delay += random.uniform(0, delay * 0.2)
            if elapsed + delay > policy.max_elapsed:
                return {"ok": False, "decision": "deadline_exceeded",
                        "error": str(exc)}
            time.sleep(delay)

生产实现还应把睡眠替换为异步等待，并把任务级共享预算作为参数传入；示例中的重点是决策顺序：先检查副作用和幂等性，再分类失败，然后检查次数与总时限，最后才计算退避。顺序反过来，系统就可能先重放一次危险操作，再发现它其实不该重试。

熔断和审计决定系统能否长期运行

单次调用有预算，还不足以应对服务整体故障。当同一工具在短窗口内连续出现瞬时失败，应打开熔断器，暂停自动调用并走降级方案。半开状态只放少量探测请求；探测恢复后再关闭熔断。这样能阻止成百上千个智能体同时执行各自“合理”的三次重试。

我至少会记录这些字段：任务 ID、工具名、调用 ID、尝试序号、失败类别、决策、耗时、退避时长、幂等键摘要、熔断状态和最终结果。日志里不要保存原始密钥、完整令牌或不必要的敏感参数。审计目标是回答“为什么又调了一次”，而不是把所有上下文无差别落盘。

哪些情况不适合自动重试

• 支付、发信、建单、删除等副作用操作没有服务端幂等支持。
• 上一次执行结果未知，而且没有查询操作状态的接口。
• 错误属于权限、合规、内容策略或明确的业务拒绝。
• 工具成本很高，单次调用已经接近任务预算。
• 实时交互剩余时限不足，退避后的结果已经失去价值。

落地时可以直接使用的决策表

最终我会把“重试成功率”与“任务成功率”分开观察。高重试成功率不一定是好事，它也可能说明上游持续制造了大量本可避免的错误。真正健康的指标应该同时包含首次成功率、平均工具尝试次数、预算耗尽率、重复副作用事件数和熔断次数。只有这样，可控重试才是可靠性机制，而不是把故障藏得更深。