Agent自己管账：多智能体API连环扣费的三个断点

2026年上半年，一家中型电商公司把客服Agent接入订单系统后，单月API调用成本从预估的8000元飙到4.3万。查账发现，一个退货退款流程触发了7个Agent之间的23次工具调用，其中3次因为参数格式错误陷入无限重试，直接烧掉1200元。

这不是个例。据《2026年下半年AI代理账单危机》观察，Agent从演示走向企业流程后，真正拉开成本差距的，不是模型参数大小，而是屏幕理解、工具调用、上下文记忆和失败重试这些细节。多智能体协作场景下，这个问题被成倍放大。

账单失控，往往不是单价问题。很多团队盯着模型API单价优化，却忽略了多Agent架构里的三个成本黑洞：上下文膨胀、工具重试风暴和后台空转。Agent A把整个对话历史传给Agent B，Agent B再把结果加上自己调用的工具输出传回Agent A。每轮协作，token数指数级增长。一个简单的订单查询，如果经过三个Agent接力，上下文可能从500 token膨胀到8000 token。Agent调用外部API失败时，默认的重试逻辑常常没有上限。如果下游服务返回了一个"格式错误"的响应，Agent可能连续重试数十次，每次都在消耗token。Agent等待另一个Agent响应时，维持会话状态的轮询请求持续计费。据Airwallex在2026年7月发布的《AI金融不止于结账》报告指出，智能体商业缺失的金融执行层，正是指这类"执行过程中的隐性消耗"——Agent能推理、能决策，但对自身行动的成本完全没有感知。

解法不是让Agent变聪明，而是让它变"穷"——给它一个硬性预算，花完就停。我们设计的防超额计费中间件，核心是三个组件：令牌桶限流器、死循环监测器和硬性熔断开关。这三个组件部署在Agent与LLM API之间，对所有进出请求进行计数、计时和拦截。

令牌桶限流：按预算花，不按需花

令牌桶算法的核心逻辑很简单：桶里预先放入固定数量的令牌，每次API调用消耗对应token数的令牌。令牌耗尽时，请求直接被拒绝，而不是排队等待。

import time
import threading
from collections import defaultdict

class TokenBucketLimiter:
    def __init__(self, max_tokens: int, refill_rate: float = 0):
        self.max_tokens = max_tokens
        self.current_tokens = max_tokens
        self.refill_rate = refill_rate
        self.last_refill = time.time()
        self.lock = threading.Lock()
        self.user_budgets = defaultdict(lambda: max_tokens)
    
    def consume(self, user_id: str, tokens: int) -> bool:
        with self.lock:
            if self.refill_rate > 0:
                now = time.time()
                elapsed = now - self.last_refill
                refill_amount = elapsed * self.refill_rate
                self.current_tokens = min(
                    self.max_tokens, 
                    self.current_tokens + refill_amount
                )
                self.last_refill = now
            
            budget = self.user_budgets[user_id]
            if budget >= tokens:
                self.user_budgets[user_id] -= tokens
                return True
            return False
    
    def remaining(self, user_id: str) -> int:
        return self.user_budgets[user_id]

实际部署时，max_tokens按项目预算设定。假设你给一个Agent协作流程的日预算是50万token（GPT-4o模式下约合人民币200-300元），就把max_tokens设为500000，refill_rate设为0。预算花完，当天所有调用直接熔断。

这里有个关键设计决策：按user_id而非session_id做预算隔离。多Agent协作时，一个用户请求可能触发多个session，如果按session限流，Agent B的调用可能绕过Agent A的预算限制。按user_id聚合，确保整个协作链条共享同一个预算池。

死循环监测：不是看次数，是看"轨迹相似度"

单纯限制重试次数不够。Agent的"死循环"往往不是重复完全相同的请求，而是在几个相似的错误模式之间来回跳转。

我们的监测器不计数，而是对最近N次请求做向量化，计算轨迹相似度。当连续5次请求的语义相似度超过0.85时，判定为循环，强制中断。

import numpy as np
from collections import deque
from openai import OpenAI

class LoopDetector:
    def __init__(self, similarity_threshold: float = 0.85, window_size: int = 5):
        self.threshold = similarity_threshold
        self.window_size = window_size
        self.request_windows = defaultdict(
            lambda: deque(maxlen=window_size)
        )
        self.embedding_client = OpenAI()
    
    def _get_embedding(self, text: str) -> list:
        response = self.embedding_client.embeddings.create(
            model="text-embedding-3-small",
            input=text[:8000]
        )
        return response.data[0].embedding
    
    def check_and_record(self, chain_id: str, request_content: str) -> bool:
        window = self.request_windows[chain_id]
        
        if len(window) < self.window_size:
            window.append(request_content)
            return False
        
        current_emb = np.array(self._get_embedding(request_content))
        past_embs = [np.array(self._get_embedding(req)) for req in window]
        
        similarities = [
            np.dot(current_emb, past_emb) / 
            (np.linalg.norm(current_emb) * np.linalg.norm(past_emb))
            for past_emb in past_embs
        ]
        avg_similarity = np.mean(similarities)
        
        window.append(request_content)
        
        if avg_similarity > self.threshold:
            return True
        return False

注意成本：每次检测调用一次embedding API，text-embedding-3-small模型每1000 token约0.00002美元。对于一个日均10万次调用的系统，检测成本约2美元/天。相比它能拦截的死循环损失，这个开销几乎可以忽略。

chain_id是多Agent协作的关键标识。同一个用户请求触发的所有Agent调用，应该共享同一个chain_id，这样Agent A和Agent B之间的来回重试才能被正确识别为同一个循环。

硬性熔断：不要优雅降级，要立刻停

很多系统的熔断策略是"部分降级"——比如换用小模型、缩短上下文。但在成本控制的场景下，降级策略有个致命问题：它仍然在花钱，只是花得慢一点。

我们的熔断器只有两个状态：开和关。触发条件有三条：令牌桶耗尽（预算花完）、死循环检测命中、单次调用token数超过阈值（防止单次超大上下文）。

from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"

class HardCircuitBreaker:
    def __init__(self, 
                 max_single_call_tokens: int = 16000,
                 cooldown_minutes: int = 30):
        self.max_single_call_tokens = max_single_call_tokens
        self.cooldown = timedelta(minutes=cooldown_minutes)
        self.state = defaultdict(lambda: CircuitState.CLOSED)
        self.open_since = {}
        self.break_reasons = {}
    
    def check(self, user_id: str, tokens: int, 
              limiter: TokenBucketLimiter,
              loop_detected: bool) -> tuple[bool, str]:
        if self.state[user_id] == CircuitState.OPEN:
            if datetime.now() - self.open_since[user_id] < self.cooldown:
                return False, f"熔断冷却中，剩余{(self.cooldown - (datetime.now() - self.open_since[user_id])).seconds}秒"
            else:
                self.state[user_id] = CircuitState.CLOSED
        
        if tokens > self.max_single_call_tokens:
            self._trip(user_id, f"单次调用{tokens}token超过上限{self.max_single_call_tokens}")
            return False, self.break_reasons[user_id]
        
        if not limiter.consume(user_id, tokens):
            remaining = limiter.remaining(user_id)
            self._trip(user_id, f"预算耗尽，剩余配额{remaining}token不足")
            return False, self.break_reasons[user_id]
        
        if loop_detected:
            self._trip(user_id, "检测到死循环，强制熔断")
            return False, self.break_reasons[user_id]
        
        return True, ""
    
    def _trip(self, user_id: str, reason: str):
        self.state[user_id] = CircuitState.OPEN
        self.open_since[user_id] = datetime.now()
        self.break_reasons[user_id] = reason

冷却期设为30分钟，是经验值。太短（5分钟）可能导致反复熔断-恢复-熔断的抖动，太长（24小时）会让正常用户在误触发后等太久。30分钟足够让运维人员介入排查，又不至于严重影响业务。

组装：中间件部署代码

三个组件组装成一个FastAPI中间件，拦截所有指向LLM API的请求：

from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class AgentCostGuardMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, daily_budget_tokens: int = 500000):
        super().__init__(app)
        self.limiter = TokenBucketLimiter(
            max_tokens=daily_budget_tokens, 
            refill_rate=0
        )
        self.loop_detector = LoopDetector(
            similarity_threshold=0.85, 
            window_size=5
        )
        self.breaker = HardCircuitBreaker(
            max_single_call_tokens=16000,
            cooldown_minutes=30
        )
    
    async def dispatch(self, request: Request, call_next):
        if not request.url.path.startswith("/v1/chat/completions"):
            return await call_next(request)
        
        body = await request.json()
        user_id = request.headers.get("X-User-ID", "default")
        chain_id = request.headers.get("X-Agent-Chain-ID", user_id)
        
        messages = body.get("messages", [])
        estimated_tokens = sum(
            len(str(msg.get("content", ""))) // 4 
            for msg in messages
        )
        
        request_text = str(messages[-1].get("content", "")) if messages else ""
        loop_detected = self.loop_detector.check_and_record(
            chain_id, request_text
        )
        
        allowed, reason = self.breaker.check(
            user_id, estimated_tokens, self.limiter, loop_detected
        )
        
        if not allowed:
            raise HTTPException(
                status_code=429, 
                detail={
                    "error": "budget_exceeded",
                    "reason": reason,
                    "remaining_tokens": self.limiter.remaining(user_id)
                }
            )
        
        return await call_next(request)

部署后，Agent侧的代码需要做两件事：在每个API请求头里带上X-User-ID和X-Agent-Chain-ID，以及捕获429错误，向用户返回明确的预算提示，而不是让Agent继续尝试。

为什么不是"让Agent学会省钱"

有人会问：为什么不训练Agent自己感知成本？据《AI Agent智能体技术发展报告》指出，当前Agent的规划能力主要依赖ReAct、CoT等框架做任务分解，但成本感知不在这些框架的设计范围内。让LLM在推理时同时考虑"这个工具调用值不值得"，会显著增加推理开销，反而更贵。

华为发布的《金融智能体应用协同指南》也提到，多智能体协同的通信保障和安全可信，需要外部规范而非Agent自觉。成本控制同理——它是一个架构层的约束，不该交给Agent自行判断。

这套中间件的设计哲学很明确：Agent负责聪明，中间件负责抠门。在项目初期部署它，比事后查账单、改架构要便宜得多。

上线前最后一步：压测你的预算上限

部署后别直接上线。先用历史日志回放一周的真实调用数据，观察令牌桶耗尽的时间点（如果每天下午3点就耗尽，预算需要上调或优化Agent逻辑）、死循环检测的误报率（如果误报超过5%，调整similarity_threshold）、熔断后的业务降级方案（用户看到什么提示？是否有手动恢复通道？）。

据《Agent账单偷跑，先从这几处查起》一文的实践经验，参考AWS FinOps Agent案例，成本审计和止损措施需要在Agent上线前就内嵌到架构里，而不是等账单炸了再补救。

这套中间件已在GitHub开源（agent-cost-guard），适配OpenAI、Anthropic和国产模型API。30分钟冷却、85%相似度阈值、16000 token单次上限都是可配置项，根据你的业务场景调整。

锁死成本上限，不是让Agent变笨。是让它学会在预算内干活。