解决AI Agent重复扣费：一个轻量级的幂等性实战方案

分布式系统里有个经典坑叫“非幂等操作”，现在这坑被 AI Agent 继承了。很多开发者在给 Agent 配置 Tool Calling 时，如果网络波动导致超时，Agent 的客户端逻辑通常会触发 Retry（重试）。但问题在于，之前的请求可能已经在服务端运行且即将成功，此时重试会导致用户被扣两次款、发两封邮件或创建两个重复订单。

这种 Bug 最阴险的地方在于：Agent 的日志显示“最终请求成功”，但数据库里的账单却是双份的。

核心痛点：超时重试的陷阱

我们可以用一段简单的 Python 代码复现这个场景。假设支付接口响应需要 0.4s，但 Agent 的客户端超时设置是 0.2s：

import threading
import time

ledger = {} # 模拟账单数据库

def charge_card(order_id, amount):
    time.sleep(0.4) # 模拟支付 API 响应缓慢
    ledger[order_id] = ledger.get(order_id, 0) + 1
    return {"order_id": order_id, "status": "charged"}

def agent_charge_with_naive_retry(order_id, amount, max_retries=2):
    for attempt in range(max_retries):
        # 模拟 Agent 发起异步调用
        thread = threading.Thread(target=lambda: charge_card(order_id, amount))
        thread.start()
        thread.join(timeout=0.2) # Agent 客户端超时时间
        
        if thread.is_alive():
            # 客户端认为超时了，直接进入下一次重试
            continue 
        return # 收到响应，结束

在这个实操案例中，第一次请求在后台依然运行，并没有因为客户端的 timeout 而被取消。当第二次重试请求到达时，两次请求都会执行成功，导致 ledger 里的订单金额翻倍。

解决方案：引入 Idempotency Key（幂等键）

要解决这个问题，不能靠优化网络，而要靠在函数调用中引入一个唯一的 idempotency_key。

我为了方便在 Agent 工作流中使用，写了一个轻量级的库 latch。它的核心逻辑是通过装饰器拦截请求，如果同一个 Key 已经有执行记录且在有效期内，直接返回缓存结果，不再触发函数体。

实战配置示例

使用 @idempotent 装饰器后，代码变成了这样：

from latch import idempotent

@idempotent()
def create_order(order_id: str, amount: float) -> dict:
    # 这里调用真实的支付 API
    return payments_api.charge(order_id, amount)

# 第一次调用，正常执行并记录 key
create_order(order_id="A1", amount=42.0, idempotency_key="run-7-step-3")

# 第二次调用，使用相同的 key，直接命中缓存，不会产生第二次扣费
create_order(order_id="A1", amount=42.0, idempotency_key="run-7-step-3")

这里有一个设计细节： 为什么不直接对函数参数进行 Hash 来自动生成 Key？因为“什么算作重复操作”应该由调用者（Agent 的编排层）决定，而不是由工具函数猜测。如果参数微调但业务上仍视为同一次操作，手动传入 Key 是最稳妥的。

从单点幂等到 Agent 鲁棒性增强

在实际部署 AI Agent 的过程中，幂等性只是稳定性的一环。为了构建一个生产级别的 Agent 工作流，我还给这个库增加了几个关键的装饰器，建议在开发 Agent Tool 时一并考虑：

• 熔断机制 (@circuit_breaker)： 当下游 API 持续报错（比如 503）时，直接快速失败，避免 Agent 在死循环中不断冲击已经崩溃的服务。
• 硬超时控制 (@with_timeout)： 防止某个 Tool 调用挂起导致整个 Agent 的推理循环被阻塞。
• 预算护栏 (@budget_guardrail)： 针对 Token 消耗或 API 费用设置上限，防止 Agent 进入死循环导致账户被刷爆。
• Saga 模式 (Saga 类)： 处理多步联动操作（如：扣款 → 定机票 → 定酒店）。如果第三步失败，Saga 能触发前两步的补偿机制（退款/取消机票），防止数据不一致。

对于那些正在从简单的 Prompt 实验转向实际产品部署的开发者，建议把这套逻辑集成到你们的 tools.py 中。比起在 Prompt 里告诉 AI “请不要重复执行”，在代码层实现幂等性才是真正的保姆级解决方案。