Agent应用在执行任务时，会连续调用大模型多次，每次调用都包含输入Token和输出Token。与单次聊天接口不同，Agent的Token消耗分布在推理、工具选择、结果汇总等多个阶段，如果只统计最终回复，往往会漏掉大量隐藏成本。建立一套清晰的Token消耗监控机制，是控制Agent运行成本、发现异常调用和优化提示词的前提。
一、Agent Token消耗为什么容易失控
普通LLM调用是单次请求响应，模型返回文本即可。Agent却需要循环：接收任务、生成下一步动作、调用工具、把工具结果再次交给模型。每次进入模型的上下文都会包含系统提示词、历史消息、工具定义、工具结果等。这些内容全部按Token计费，循环次数越多，重复计费的上下文越长。
隐藏Token还包括工具定义的JSON schema、function calling的语法占位、不同模型对消息角色转换的处理。例如某些平台会把工具定义转换成特殊标记，占用额外Token；多轮累积还会让同一个系统提示词在每一轮都被重复计费。因此不能用简单的一次对话的Token数来估算Agent成本，必须在每次模型调用点记录usage字段。
另一个容易被忽视的问题是工具返回内容。搜索结果、数据库查询结果或API响应会作为工具消息重新进入模型，如果工具返回了冗余字段或完整文档，后续每一步都会背上这些Token。这意味着即使模型本身的调用次数不多，单次调用的输入长度也可能持续膨胀，最终让成本成倍增加。
二、通过回调机制自动采集每次调用的Token
LangChain等框架提供回调接口，可以在LLM调用结束时读取response.llm_output中的token_usage信息。下面是一个使用LangChain回调处理器记录Token的示例。
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
import logging
logger = logging.getLogger("token_monitor")
class TokenUsageCallback(BaseCallbackHandler):
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
usage = response.llm_output.get("token_usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
logger.info(
"token_usage prompt=%s completion=%s total=%s",
prompt_tokens,
completion_tokens,
total_tokens,
)
回调方式的好处是非侵入式，不需要修改Agent核心逻辑。只要在构建执行器时把回调处理器加入callbacks列表，就能统一记录所有语言模型调用。对于流式输出，部分框架不会在回调中提供完整usage，这时需要在流结束后通过底层API的usage字段补记，或者根据实际返回的文本长度估算，但估算会引入误差。
另一个常见问题是重试机制。网络失败或限流导致自动重试时，前一次失败请求可能已经消耗了部分Token。监控系统需要记录每次尝试的请求ID和开始时间，避免只统计成功响应而低估成本。当模型返回usage为0或缺少字段时，应写入告警日志，说明该厂商接口暂不支持完整Token统计。
如果使用自定义请求封装而非框架回调，可以在模型客户端外层增加装饰器或中间件，在每次请求完成后统一提取usage。这样做的好处是不依赖特定框架，切换LangChain、LlamaIndex或自研执行器时都能复用同一套统计逻辑。
三、Token日志与成本归因
只记录总量还不够。Agent通常包含多个子任务，例如检索、计算、代码生成、总结。要定位高成本环节，需要给每次LLM调用加上标签：任务ID、会话ID、用户ID、节点名称、模型名称。结构化日志建议采用JSON格式，方便后续查询和聚合。
import json
import time
def log_token_usage(meta: dict, usage: dict):
record = {
"timestamp": time.time(),
"session_id": meta.get("session_id"),
"task_id": meta.get("task_id"),
"agent_step": meta.get("agent_step"),
"model": meta.get("model"),
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
}
print(json.dumps(record, ensure_ascii=False))
将日志接入Elasticsearch、ClickHouse或云监控后，可以按小时、按任务、按用户聚合Token消耗。例如某类Agent任务平均消耗9000 Token，而另一类只消耗1800 Token，差异往往来自冗长的工具描述或不必要的多轮反思。成本归因的关键是把Token映射到具体步骤，而不是把所有成本都记在Agent整体头上。
计费方面，不同模型的输入Token和输出Token价格不同，需要维护一张价格表。记录时最好同时保存单价快照，因为价格可能调整，事后计算时如果使用新价格会扭曲历史成本。计算公式很简单：成本等于输入Token数乘以输入单价，再加上输出Token数乘以输出单价。注意部分平台对缓存命中的输入Token有折扣，有些Agent框架会启用上下文缓存，这类Token需要单独统计。
归因粒度越细，越容易发现异常。比如某用户单次会话消耗了平时十倍以上的Token，可能是因为工具返回了超大结果，也可能是因为提示词诱导模型反复调用同一工具。通过结构化日志中的session_id和task_id可以快速定位原始请求，帮助排查问题。
四、预算告警与成本优化
拥有Token日志后，可以设置两级告警：单次任务Token超过阈值时发出警告，单日总Token接近预算上限时通知负责人。阈值需要根据任务类型区分，例如简单问答Agent的阈值设置在2000 Token，复杂数据分析Agent可能允许15000 Token。告警条件不应只看Token总量，还要结合错误率、循环次数等指标，避免单纯限制Token导致Agent提前终止。
成本优化可以从几个方向入手。第一是压缩系统提示词，删除重复的角色描述和示例，将固定知识外置到检索工具中。第二是控制上下文长度，对早轮对话做摘要或滑动窗口截断，减少重复计费。第三是选择合适的模型，简单意图分类和工具选择没必要使用超大模型，小模型能显著降低单位成本。第四是启用缓存和批处理，对于相同的工具定义和系统提示词，部分服务商支持前缀缓存，可以降低输入Token成本。
如果观察到某类工具调用的Token消耗异常高，可以检查工具返回内容是否过长。工具返回结果会被完整放入下一次请求，如果搜索结果返回了整篇文章而不是摘要，后续每个步骤都会背上这些Token。优化工具输出，只保留必要字段，往往能带来明显下降。最后建议每周查看Token消耗报表，结合用户反馈判断是否有模型被滥用或提示词失效。
把Token监控纳入Agent上线前的标准流程，而不是成本超标后再补救，能够帮助团队在功能迭代和成本控制之间找到平衡。持续记录、定期归因、定向优化，这三步构成了Agent成本追踪的基本闭环。