在大模型应用落地的过程中，Agent的调用频率往往随着业务自动化程度提升而快速增长。如果不对每次推理所产生的Token消耗做精确记录，到了月底结算时很容易发现成本远超预期。Token消耗监控的本质，是把每一次与模型交互的输入长度、输出长度以及隐含的系统提示词开销都转化为可统计的数值，再按项目、用户或会话维度聚合。预算管理则是在此基础上设定阈值，当累计消耗接近上限时触发限制或告警。
Agent成本追踪的底层原理与数据采集点
要真正看懂Token消耗在哪里，必须先理解一次Agent调用的完整链路。通常一个Agent会先加载系统提示词，再拼接用户问题、历史记忆、工具返回结果，最后送给模型生成回答。这些环节都会占用Token，但很多开发者只统计了用户消息和最终回复，忽略了系统提示词和工具上下文。系统提示词可能在每次请求中都重复发送，若长度为两千字，一天一万次调用就白白浪费两千万Token。
因此监控的第一个关键采集点应该放在请求发出前。我们可以在封装模型调用的客户端层，把即将发送的报文序列化并计算长度。由于不同模型分词器不同，精确Token数需要调用模型方提供的计数接口，但在内部做字符级估算也能满足趋势监控。第二个采集点是响应返回后，从API回包的usage字段提取准确输入输出Token。把这两处数据上报到统一日志服务，就形成了最基础的成本追踪流水。
除了单点采集，还要考虑Agent多轮工具调用的场景。一个任务可能循环调用五次模型，每次都带上了之前所有的工具输出。如果不做上下文裁剪，Token会指数增长。监控时需要把一次任务的所有子调用聚合为一个trace，用唯一任务ID串联，这样才能算出单个业务动作的真实成本，而不是被零散请求迷惑。
基于中间件拦截的Token监控代码实现
在网关或SDK中间层做统一拦截，是避免业务代码入侵的最佳方式。下面这段Python示例展示了一个简单的包装器，它在调用模型前后记录Token并推送到监控队列。注意这里把HTML特殊字符都做了转义，仅作为逻辑演示。
import time
import json
class TokenMonitor:
def __init__(self, upstream_call):
self.upstream_call = upstream_call
self.logs = []
def chat(self, prompt, session_id):
# 请求前本地估算输入长度
input_est = len(prompt) // 2
start = time.time()
# 实际调用模型，回包中包含usage
resp = self.upstream_call(prompt)
cost_ms = int((time.time() - start) * 1000)
usage = resp.get('usage', {})
record = {
'session_id': session_id,
'input_token': usage.get('prompt_tokens', input_est),
'output_token': usage.get('completion_tokens', 0),
'cost_ms': cost_ms,
'ts': int(time.time())
}
self.logs.append(record)
# 伪代码：推送到中心化监控
# push_to_kafka('agent_token_log', json.dumps(record))
return resp['content']
def fake_model(p):
return {'content': 'done', 'usage': {'prompt_tokens': 120, 'completion_tokens': 30}}
mon = TokenMonitor(fake_model)
print(mon.chat('帮我查一下订单', 'sess_001'))
上面的代码把监控逻辑从业务函数里抽离出来。实际生产中，upstream_call可能是对远程接口的封装，我们在其中解析返回的JSON获取usage字段。如果模型不支持返回用量，就需要本地集成分词器，比如用tiktoken库对输入输出文本编码后数长度。
这种中间件的优势在于可复用。无论是基于HTTP的OpenAI兼容接口，还是私有化部署的推理服务，只要进出参数结构一致，就能套用同一套监控。团队还可以给中间件加采样率，高频调用只采千分之一详单，既控成本又留痕迹。
预算管理中的配额划分与告警机制
采集到Token流水只是第一步，真正控制开支要靠预算。预算一般分三层：全局月预算、项目周预算、用户日预算。全局预算防止总账单爆炸；项目预算方便核算ROI；用户预算避免单个异常账号拖垮全局。在数据库里维护一张配额表，每次请求前做原子扣减，失败则拒绝服务，这是最常见方案。
告警不能只等超了才通知。建议设置梯度比例，例如达到预算的百分之五十发周报，百分之八十发预警，百分之百熔断。下面是一段伪代码展示扣减逻辑，其中使用了行内code标签说明函数名，标签名讨论时做了转义。
def consume_quota(user_id, token_count):
# 使用事务保证并发安全
with db.transaction():
row = db.query('SELECT used, limit FROM quota WHERE uid=%s', user_id)
if row.used + token_count > row.limit:
return False, 'quota_exceeded'
db.execute('UPDATE quota SET used=used+%s WHERE uid=%s', token_count, user_id)
return True, 'ok'
ok, msg = consume_quota('user_99', 150)
if not ok:
# 这里可返回友好错误，而不是继续调用模型
print('预算不足，请求被拦截')
在网页端展示配额时，可以用<progress>元素直观呈现消耗进度，但注意在正文中提及该标签名需转义。对于超预算的处置，除了硬阻断，也可降级到小模型或缩短历史上下文，用体验换成本。这种弹性策略比直接报错更易被业务接受。
最后要强调的是，预算管理不是一次性配置。应每周复盘实际消耗曲线，把突发增长归因到具体功能，再反过来优化Agent的提示词模板和工具调用次数。只有把监控和预算闭环，成本追踪才真正产生价值。