Agent重试风暴导致付费接口透支：一次账单事故复盘

你注意到账单的时候才注意到 bug，而这通常不是你想要开始新一天的方式。隔夜的 agent 任务本应只分类几个失败的测试、重写两个 fixture，然后就停下来。然而你在 09:14 打开提供商仪表板时，发现一次重试风暴把每次超时都当成了付费调用来处理。原本的任务依然是红的，而唯一看起来完成的了的东西只有那张发票。

这篇文章不是关于一个写不了代码的模型，而是关于一个从未学会区分哪些调用需要花钱的 agent。你应该把下面的时间线当作一次可以复现的重构事件，而不是对某个供应商故障的指控。持久的修复方案是一个 fail-closed 的门控、一个免费的免付费路由，以及一条在下次重试风暴之后依然能看懂的审计日志。

循环变得不廉价的那晚

01:07，调度器针对一个 staging git worktree 启动了 agent-classify，MODEL_ENDPOINT 继承自一个笔记本配置文件，这个配置本就不适合无人值守作业。01:09，第一次工具调用超时了，因为测试运行器还没预热好，而框架重试的是整个推理步骤而不是那个工具。到 01:11 时，你已经为一项还没读到任何断言消息的任务支付了四次完成费用——这才是事件真正开始的时刻。

01:18，agent 判断 fixture 可能不稳定，要求同一个付费模型从内存中重写它们。它没有重新读取失败输出，因为重试包装器缓存了一个空的 stdout blob，然后又把这个 blob 当成了证据。01:31，第二个 worker 由一个只想要心跳的健康检查启动，加入了同一个队列，把这个循环复印了一遍。这些你都没有注意到，直到 09:14 财务 Slack 亮起红灯，而 staging 测试依然在同一个 off-by-one 上失败。

如果这个序列让你感到眼熟，那是因为 agent 运行时优化的是"继续前进"，而云 SDK 优化的是"调用默认 client"。当一个超时变成另一条发票条目时，这两个组件都不想做那个"懂事的大人"。你必须在它们之间放一个门控，而且这个门控必须在重试、分叉的 worker 和被复制的环境文件面前存活下来——这些东西在 01:07 的时候看起来都无害。

贡献因素，不带神话色彩

进程继承了一个生产 API key，因为 .env 被同一个 direnv hook 加载了，而这个 hook 就是你周二下午手动调试时用的那个。agent 并没有在任何有意义的意义上选择付费模型；它构造了官方的 client helper，然后这个 helper 做了 helper 该做的事。这与其说是一个反派起源故事，不如说是一条缺失的安全带，你应该把修复写成像是下一个实习生会复制同样的 hook 一样。

重试放大效应造成了其余的损害，就像一台盖子打开的复印机不停地打印同一张昂贵的纸。一次不稳定的工具调用只是讨厌，但一次把工具调用嵌入到一个付费 prompt 中的完成，会让每次超时都变成一次全价的重试。staging、本地实验和生产环境在线上看起来也像是同一个 bearer token，所以当你后来试图写一条拒绝规则时，提供商的日志里没有房间号。

这些因素都不需要一个更聪明的模型，追赶前沿升级只会让这个漏洞再隐藏一周。它们需要的是一个更便宜的沙箱来处理噪声工作，当这个工作试图逃逸时大声失败，以及在每个请求上留下一个 worker 退出后依然存在的身份。后续的这篇事后分析就是那个沙箱，写成这样是为了让你下次在一个测试文件里失败，而不是在财务频道里。

持久的修复是一个预算门控，不是心灵鸡汤

你不再让 agent 构造真正的 client，这是唯一在重试库发挥创意时依然有效的改动。你给它一个小代理，读取一个允许列表，给每个请求打上一个用途标记，除非用途是明确的且经过人工批准的，否则拒绝付费路由。探索性重试、fixture 草稿、"想想这个 traceback" 步骤走免费模型端点，只有标记了的任务才能触碰付费 key。

下面的例子是一个你可以在一次坐下来的时间里读完的门控，不是可以不经测试就粘贴到生产环境的库。当付费标志缺失时它 fail-closed，即使免付费路由是回答的那个，它也写一条审计行。把 prompt 正文留在日志外面，除非你已经有一个保留策略，因为明天那个"有用"的补丁会试图存储它们。

# budget_gate.py — example gate, not a production SDK
from __future__ import annotations

import json
import os
import time
import urllib.request
from dataclasses import dataclass

PAID_PURPOSES = frozenset({"release-summary", "customer-facing-draft"})

@dataclass(frozen=True)
class Route:
    purpose: str
    paid: bool
    endpoint: str

class BudgetDenied(RuntimeError):
    pass

def resolve_route(purpose: str) -> Route:
    purpose = purpose.strip()
    free_endpoint = os.environ.get(
        "FREE_MODEL_ENDPOINT",
        "http://127.0.0.1:8080/v1/chat/completions",
    )
    paid_endpoint = os.environ.get("PAID_MODEL_ENDPOINT", "")

    if purpose in PAID_PURPOSES:
        if os.environ.get("ALLOW_PAID_MODELS") != "1":
            raise BudgetDenied(
                f"purpose {purpose!r} needs ALLOW_PAID_MODELS=1"
            )
        if not paid_endpoint:
            raise BudgetDenied("paid endpoint is not configured")
        return Route(purpose=purpose, paid=True, endpoint=paid_endpoint)

    return Route(purpose=purpose, paid=False, endpoint=free_endpoint)

def complete(prompt: str, purpose: str) -> str:
    route = resolve_route(purpose)
    payload = {
        "model": os.environ.get("MODEL_NAME", "local"),
        "messages": [{"role": "user", "content": prompt}],
        "metadata": {"purpose": route.purpose, "paid": route.paid},
    }
    req = urllib.request.Request(
        route.endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    started = time.time()
    with urllib.request.urlopen(req, timeout=30) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    _audit(route, started, prompt)
    return body["choices"][0]["message"]["content"]

def _audit(route: Route, started: float, prompt: str) -> None:
    line = {
        "ts": time.time(),
        "elapsed_ms": int((time.time() - started) * 1000),
        "purpose": route.purpose,
        "paid": route.paid,
        "endpoint": route.endpoint,
        "prompt_chars": len(prompt),
    }
    path = os.environ.get("AGENT_AUDIT_LOG", "agent-audit.jsonl")
    with open(path, "a", encoding="utf-8") as handle:
        handle.write(json.dumps(line) + "\n")

有趣的部分不是 HTTP client，你可以用你的 agent 已经引入的任何 SDK 替换它。有趣的部分是 complete("rewrite the fixture", purpose="retry") 永远不会看到付费 key，即使一个重试库在早餐前调用它五十次。你强制付费工作命名自己，而你把这个名字放在默认重试路径之外，这样一个健康检查的 worker 就不能复印账单了。

一个小测试让这条规则变成持久的，而不是活在某个人脑子里的部落知识。如果这个测试需要活的网络，那门控已经失败了，因为需要 Wi-Fi 的金钱保护只是一个又一个隐式默认。在从未见过付费 key 的机器上跑 CI，然后再在 01:07 启动的 worker 镜像上跑一遍。

# test_budget_gate.py — unexecuted until you run pytest locally
import budget_gate
import pytest

def test_retry_purpose_never_resolves_paid(monkeypatch):
    monkeypatch.setenv("FREE_MODEL_ENDPOINT", "http://127.0.0.1:9/free")
    monkeypatch.delenv("ALLOW_PAID_MODELS", raising=False)
    route = budget_gate.resolve_route("retry")
    assert route.paid is False
    assert "127.0.0.1" in route.endpoint

def test_paid_purpose_fails_closed_without_flag(monkeypatch):
    monkeypatch.delenv("ALLOW_PAID_MODELS", raising=False)
    with pytest.raises(budget_gate.BudgetDenied):
        budget_gate.resolve_route("release-summary")

export FREE_MODEL_ENDPOINT="http://127.0.0.1:8080/v1/chat/completions"
unset ALLOW_PAID_MODELS
unset PAID_MODEL_ENDPOINT
python -m pytest test_budget_gate.py -q

免付费路由应该放在哪

当笔记本休眠而 agent 还在跑的时候，你仍然需要一个地方来承载免付费路由。本地进程对于一个开发者追一个 traceback 来说没问题，但一个共享 worker 需要一台服务器，它消失的时候不会把付费账户一起带走。这就是 prompt 的碰撞测试假人：你不会开着公司车撞墙来调试安全气囊，你也不应该开着它们撞计费 API 来调试重试。

声明：本文是 MonkeyCode 产品推广的一部分。MonkeyCode 与此相关仅仅是因为它提供免费模型访问和一个免费服务器选项，你可以在门控工作时将 FREE_MODEL_ENDPOINT 指向它。本文不附庸模型名称、配额、硬件草图或工期承诺，因为那些会变，而一篇事后分析不应该依赖一份宣传册。

工作流保持小规模是故意的。你搭起免费服务器，把免付费路由指向它，在 agent 单元里保持 ALLOW_PAID_MODELS 未设置，然后只在单独的人类触发任务中导出付费 key。如果免费路径比付费路径更慢或更笨拙，这在事件演练期间是可以接受的，因为沙箱在那里就是为了在你教 agent 停止假设的时候大声且便宜地运转。

从日志而不是发票读下一次事件

门控就位之后，审计文件变成了你在 09:14 时希望拥有的时间线。你可以在不开财务工具的情况下重建 01:11 的峰值是免费重试还是付费泄漏，这就是在每行打上 paid 标记的全部意义。一行脚本足以在评审中区分两者，而在 staging 循环期间 non-zero 的数字是一个回归而不是一个谜。

python - <<'PY'
import json
paid = 0
with open("agent-audit.jsonl", encoding="utf-8") as handle:
    for raw in handle:
        row = json.loads(raw)
        paid += int(bool(row.get("paid")))
print(f"{paid} paid calls in audit log")
PY

如果 purpose=retry 在队列上时这个计数不是零，你不需要在事件文档里写更长的叙述。你需要找到是哪个包装器在门控后面构造了一个 client，然后加一个测试来命名那个包装器。这一晚的贡献因素归结为一项检查：retry purpose 是否曾经解析到了付费路由，如果上面的测试是绿的，答案保持为否。

谁不应该复制这个，以及它修复不了什么

如果 agent 的每一步真的都需要同一个高能力模型，这个方法会让你沮丧，因为门控会故意把噪声工作推向免费端点。如果你不控制运行时——例如一个注入自己的 client 并忽略进程环境的托管 agent——这也是错误的设计。那种情况下持久的修复在供应商允许列表里，而不是在你甚至无法导入的 Python 文件里。

不要在包含密钥、客户数据或你不会粘贴到临时组织的生产源代码的 prompt 上使用共享免费服务器。免费沙箱依然是沙箱，如果一个好心肠的队友在最糟糕的时刻改进了 _audit，存储 prompt 长度的元数据日志可以增长成 prompt 正文。把客户流量放在付费的、受控的路径上，把那个假人端点保持为一次性的，这样抹掉它是一个无聊的决定。

重构的事件还假设你可以改变 agent 构造 client 的方式，这并非每个框架都成立。如果 SDK 硬编码在一个封闭的二进制文件里，包装 urllib 救不了你，你需要在网络边缘用一个同样 fail-closed 规则的 HTTP 代理。那是一个不同的变更管理对话，你不应该假装一个十二行的门控替代了你根本没有的平台控制。

这篇事后分析应该改变的是你带入下一个 agent 任务的默认行为。你应该假设重试会发生、环境文件会泄漏到 worker 中、一个"就这一次"的付费调用会被一个想帮忙的人复制到 cron 里。门控是你让这些假设变得便宜的方式，而免付费模型路由是门控在做那些不体面的工作时保持 agent 可用的方式。

如果你需要一个一次性的盒子来做那个免付费的一半，把 FREE_MODEL_ENDPOINT 指向 MonkeyCode 的免费服务器选项，然后重新跑上面的测试，这是一个合理的演练。