AI Agent 工具调用失败后的重试与幂等设计

AI Agent 的工具调用不能把“失败”统一理解为“可以再调一次”。连接超时可能发生在工具已经完成扣款之后，参数错误则重试多少次都不会成功。可靠方案是先分类失败，再为有副作用的调用生成稳定幂等键，由工具服务保存首次执行状态和结果；同一业务动作的后续尝试复用同一个键。

一、旧方案的问题：看到异常就重试

我在联调 Agent 工作流时遇到过最危险的一类状态：客户端超时，但工具服务其实已经完成写入。若编排器重新生成一个请求并再次执行，就会产生重复订单、重复通知或重复工单。此时“调用失败”只是客户端没有收到确认，不等于副作用没有发生。

二、新规则：一次业务动作只有一个幂等键

幂等键应绑定“业务动作”，而不是绑定某次 HTTP 尝试。Agent 第一次决定执行 create_ticket 时生成键，并把它写入持久化执行记录；网络重试、进程重启和工作流恢复都继续使用同一个键。若每次尝试都重新生成 UUID，去重机制就失效。

from dataclasses import dataclass
import hashlib

@dataclass(frozen=True)
class ToolIntent:
    workflow_id: str
    step_id: str
    tool_name: str

def idempotency_key(intent: ToolIntent) -> str:
    # 业务动作字段固定，重试时可得到同一个键。
    raw = f"{intent.workflow_id}:{intent.step_id}:{intent.tool_name}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

不要把当前时间、随机重试编号或易变化的模型文本放进键。工具参数可以另算摘要并保存，用于检测“同一键却提交了不同参数”的冲突。

三、工具服务端保存状态和首次结果

服务端收到请求后，应在执行副作用之前原子地登记幂等键。常用状态包括 PROCESSING、SUCCEEDED 和 FAILED_FINAL。重复请求遇到成功记录时直接返回首次结果；遇到处理中记录时返回可查询状态，而不是再执行一次。

def execute_once(store, key, args, tool):
    # 原子占位保证并发请求只有一个执行者。
    record, created = store.create_if_absent(key, args)
    if not created:
        if record.status == "SUCCEEDED":
            return record.result
        if record.status == "PROCESSING":
            raise RuntimeError("仍在处理中，请稍后查询")
        raise RuntimeError(record.error)

    try:
        result = tool(**args)
        # 先持久化结果，再向调用方返回。
        store.mark_succeeded(key, result)
        return result
    except Exception as exc:
        store.mark_failed(key, str(exc))
        raise

生产实现还要处理执行者崩溃后的租约过期、参数摘要冲突和结果保留期限。幂等不是简单加一张唯一索引表，而是要定义每个状态是否允许重放。

四、重试只放在一个层级

模型层、Agent 编排层、HTTP SDK 和工具服务如果都各重试三次，最坏会形成倍增请求。应指定一个负责重试的层级，其他层只上报结构化失败。可重试错误采用指数退避并加入随机抖动，同时设置最大次数和总耗时上限。

import random
import time

def retry_tool(call, retryable, max_attempts=4):
    for attempt in range(max_attempts):
        try:
            return call()
        except Exception as exc:
            # 参数或权限错误立即停止，不浪费调用额度。
            if not retryable(exc) or attempt == max_attempts - 1:
                raise
            # 指数退避加抖动，避免多个 Agent 同时重试。
            delay = min(8.0, 0.5 * (2 ** attempt))
            time.sleep(delay * random.uniform(0.7, 1.3))

五、Agent 需要看到结构化结果

工具返回值至少包含 status、retryable、operation_id 和安全的错误说明。Agent 不应根据自然语言错误猜测是否重试。对于结果未知，应返回查询操作标识；对于最终失败，应提供可修正字段或明确进入人工队列。

六、采用建议

读取类工具通常天然适合重试；创建、发送、扣款和状态迁移必须先确认幂等能力。若第三方工具不支持幂等键，可以在自己的网关建立动作记录，或改成“先创建草稿、再明确确认”的两阶段交互。无法判断是否执行成功时，宁可暂停并查询，也不要让模型盲目再调用。

七、常见问题

1. 幂等是否等于工具只执行一次？

不完全等同。底层可能收到多次请求，但相同业务动作只产生一次可见副作用，并返回一致结果。

2. 所有错误都应该自动重试吗？

不应该。参数、权限和业务规则错误需要修正；只有暂时性失败且操作可安全重放时才适合重试。

3. 幂等记录保存多久？

至少覆盖工作流可能重放和客户端可能重试的窗口。涉及长期业务凭证时，应按业务主键保留更久，而不是只依赖短 TTL 缓存。