1. 项目缘起当AI Agent开始“烧钱”最近在折腾一个AI Agent项目团队里几个小伙伴跑得挺欢各种功能测试、长对话、文件上传玩得不亦乐乎。直到月初负责财务的同事拿着云服务商的账单来找我眉头紧锁“这个月的AI API调用费用比上个月翻了快三倍你们到底在测什么” 我一时语塞因为我们的Agent系统接入了多个大模型服务比如OpenAI GPT、Claude、国内的一些大模型等每次对话、每次工具调用都在消耗Token但具体是哪个功能、哪个用户、甚至哪段代码消耗了多少我们完全是一笔糊涂账。这就像给家里装了个智能水龙头水流哗哗的月底一看水费账单傻眼了却不知道是洗澡用多了还是花园浇水管漏了。对于AI Agent这种“吞金兽”来说Token就是它的“水”是成本的核心构成。没有用量统计就谈不上成本控制、资源优化和商业化定价。于是给Agent加上一套清晰、准确的Token用量统计系统就成了一个必须立刻解决的工程问题。这不仅仅是看个账单更是从“玩具”走向“产品”的关键一步。2. 理解TokenAgent成本核算的“基本单位”在动手之前我们得先搞清楚我们要统计的到底是什么。Token对于大语言模型LLM而言就如同汽油对于汽车。它不是简单的“字数”而是模型处理文本时切分的最小语义单元。在英文中一个Token大约对应0.75个单词在中文里由于汉字和词语的复杂性一个汉字或一个常用词往往就是1-2个Token。为什么Token统计如此重要直接成本挂钩几乎所有主流云AI服务如OpenAI, Anthropic, 国内各大厂都按Token消耗量计费输入Prompt和输出Completion分开算。输入通常便宜输出昂贵。一次复杂的Agent推理可能包含多轮对话、长上下文和多次工具调用Token消耗是叠加的。性能瓶颈预警模型的上下文窗口Context Window有Token上限如128K。Agent在长时间运行中如果历史对话或检索到的信息不断累积很容易触及上限导致最前面的信息被“遗忘”或直接调用失败。实时统计Token有助于实现动态的上下文修剪Context Pruning策略。优化决策依据通过分析不同功能、不同提示词Prompt模板的Token消耗我们可以优化提示工程用更少的Token激发模型更好的表现或者对高消耗功能进行重构。在我们的Agent架构里Token消耗发生在多个环节用户输入用户的问题本身。系统提示词定义Agent角色、能力、规则的固定部分。历史对话为了让Agent有记忆我们需要把之前的对话内容也喂给它。工具描述当Agent需要调用外部函数如查天气、搜数据库时我们需要把这些函数的描述名称、参数说明也放入上下文。模型输出Agent的回复内容。中间过程在一些复杂架构中Agent可能会有“思考链”Chain-of-Thought这些内部推理步骤如果也调用模型同样会产生Token消耗。因此我们的统计系统必须能穿透这些层次进行细粒度的计量。3. 设计统计方案从粗放到精细的四个层级一开始我们想得很简单在每次调用模型API后把返回的usage字段通常包含prompt_tokens,completion_tokens,total_tokens累加起来不就行了但实际操作中这远远不够。我们设计了四个统计层级以满足不同场景的需求。3.1 层级一会话级统计Session-Level这是最基础的维度回答“这次对话总共花了多少钱”。统计对象一次完整的用户与Agent的交互会话可能包含多轮对话Multi-turn。实现方式在会话开始时创建一个计数器每次调用模型API无论是主模型还是子模型后累加其usage。会话结束时将总数据持久化到数据库。数据结构示例{ session_id: sess_abc123, user_id: user_001, start_time: 2024-05-27T10:00:00Z, end_time: 2024-05-27T10:05:30Z, total_prompt_tokens: 4500, total_completion_tokens: 1200, total_tokens: 5700, estimated_cost: 0.0114 // 根据模型单价估算 }价值快速评估单次服务成本用于用户级账单或内部成本分摊。3.2 层级二请求级统计Request-Level粒度更细回答“用户说的每一句话Agent的每一次回复各自花了多少Token”。统计对象单次模型API调用。实现方式在封装模型调用Client的代码层进行拦截。每次调用后不仅累加到会话计数器还将本次调用的明细单独记录。数据结构示例{ request_id: req_def456, session_id: sess_abc123, model: gpt-4-turbo, prompt_tokens: 850, completion_tokens: 300, total_tokens: 1150, timestamp: 2024-05-27T10:01:15Z, purpose: generate_response // 或 tool_calling, coarse_thinking }价值定位高消耗的对话轮次分析是用户问题复杂还是Agent回复冗长。3.3 层级三组件级统计Component-Level这是深度优化关键。回答“是系统提示词太啰嗦还是工具描述太占地方”统计对象构成一次模型请求的各个部分即Prompt的组成片段。实现方式在组装最终Prompt给模型之前对各个组件分别进行Token化计算。这需要集成一个独立的Tokenizer例如OpenAI的tiktoken库或Hugging Face的transformers库中的Tokenizer。计算过程将系统提示词、对话历史格式化后、工具描述列表、当前用户问题等分别保存为字符串。使用对应模型的Tokenizer对每个字符串进行编码并获取编码后的长度即Token数。将这些数字与请求记录关联存储。数据结构示例作为请求记录的扩展{ request_id: req_def456, component_breakdown: { system_prompt: 150, chat_history: 600, tool_descriptions: 90, user_query: 10, total_prompt: 850 // 应与request记录的prompt_tokens一致用于校验 } }价值这是成本优化的“显微镜”。我们曾发现一段精心设计但过于详细的系统提示词占了单次请求近30%的Token。通过精简和优化直接降低了近三分之一的提示成本。3.4 层级四业务级统计Business-Level这是面向产品和运营的维度回答“我们的‘旅行规划’功能比‘邮件润色’功能成本高多少”或“用户A是不是我们的高价值客户高消耗也可能意味着高活跃度”统计对象根据业务逻辑聚合的Token消耗。实现方式在会话或请求记录中增加业务标签business_unit,feature_flag,tenant_id。在后端通过ETL提取、转换、加载过程或实时分析引擎如Druid, ClickHouse进行聚合分析。分析视角按功能模块对比不同Agent Skill如数据分析、创意写作、代码生成的成本。按用户/租户识别高消耗用户为SaaS定价或资源配额提供依据。按时间周期观察成本趋势预测未来账单。注意组件级统计Tokenization是计算密集型操作尤其是在高并发下。绝对不要在每次请求的同步路径中进行实时Tokenize来计算预估成本这会极大增加响应延迟。正确的做法是1在异步任务或离线分析中执行2对固定内容如系统提示词、工具描述进行预计算并缓存结果。4. 核心实现拦截、计算与回调Callback机制理论讲完来看看代码怎么落地。核心思想是“非侵入式拦截”和“回调函数Callback注入”。我们不想重写每一处模型调用代码而是通过框架提供的扩展点来插入我们的统计逻辑。以使用LangChain框架为例它的Callback机制非常适合做这件事。我们创建一个自定义的BaseCallbackHandler。4.1 实现自定义Callback Handlerimport json from typing import Any, Dict, List from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import LLMResult from transformers import AutoTokenizer # 用于组件级统计 class TokenUsageStatsHandler(BaseCallbackHandler): 用于统计Token用量的回调处理器 def __init__(self, session_id: str, model_name: str gpt-3.5-turbo): super().__init__() self.session_id session_id self.model_name model_name self.tokenizer None # 懒加载 self.current_request_components {} # 存储当前请求的组件文本 # 初始化统计存储这里用内存示例生产环境应存DB self.session_stats { session_id: session_id, total_prompt_tokens: 0, total_completion_tokens: 0, requests: [] } def _get_tokenizer(self): 懒加载Tokenizer不同模型需不同处理 if self.tokenizer is None: # 示例使用transformers库的tokenizer需与目标模型匹配 # 对于OpenAI模型更推荐使用tiktoken这里仅为演示多方案 if gpt in self.model_name: # 实际使用import tiktoken; enc tiktoken.encoding_for_model(model_name) # 这里简化模拟 class MockTiktoken: def encode(self, text): return list(range(len(text.split()))) # 模拟 self.tokenizer MockTiktoken() else: # 假设是开源模型如Llama self.tokenizer AutoTokenizer.from_pretrained(meta-llama/Llama-2-7b-chat-hf) return self.tokenizer def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs): 在LLM开始运行时触发此时prompts已组装好 # 记录当前请求的完整prompt用于后续可能的组件拆分这里简化处理 self.current_prompt prompts[0] if prompts else # 在实际项目中你可以通过kwargs或序列化信息判断当前prompt的构成部分 def on_llm_end(self, response: LLMResult, **kwargs): 在LLM结束时触发这里包含usage信息 llm_output response.llm_output if not llm_output or token_usage not in llm_output: print(Warning: No token usage info in response.) return usage llm_output[token_usage] prompt_tokens usage.get(prompt_tokens, 0) completion_tokens usage.get(completion_tokens, 0) # 1. 累加会话总统计 self.session_stats[total_prompt_tokens] prompt_tokens self.session_stats[total_completion_tokens] completion_tokens # 2. 记录请求级明细 request_record { request_id: freq_{id(response)}, prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, total_tokens: prompt_tokens completion_tokens, timestamp: datetime.now().isoformat(), model: self.model_name, # 可以在这里尝试进行组件级分析异步或采样进行 # component_breakdown: self._analyze_components(self.current_prompt) } self.session_stats[requests].append(request_record) # 3. 可选实时打印或发送到监控系统 print(f[Token Stats] Session {self.session_id[:8]}... | Prompt: {prompt_tokens}, Completion: {completion_tokens}) def _analyze_components(self, full_prompt: str) - Dict: 分析Prompt各组件的Token消耗示例性逻辑需自定义 # 这是一个复杂部分需要你根据组装Prompt的逻辑来解析。 # 例如如果你的prompt模板是 # f\\\System: {system_prompt}\nHistory: {history}\nTools: {tools}\nUser: {query}\\\ # 你需要在这里将full_prompt按规则拆解回各个部分。 # 更推荐的做法是在组装prompt时就同步记录各部分的文本和预计算的token数。 breakdown {} try: tokenizer self._get_tokenizer() # 假设我们能通过分隔符解析实际情况更复杂 parts full_prompt.split(\n\n) # 简陋的示例分隔 for i, part in enumerate(parts): # 使用tokenizer计算注意不同库方法不同 # 对于tiktoken: tokens tokenizer.encode(part); count len(tokens) # 这里用模拟 simulated_count len(part.split()) // 0.75 # 非常粗略的模拟 breakdown[fcomponent_{i}] simulated_count except Exception as e: print(fComponent analysis failed: {e}) return breakdown def get_summary(self) - Dict: 获取本次会话的统计摘要 total self.session_stats[total_prompt_tokens] self.session_stats[total_completion_tokens] self.session_stats[total_tokens] total return self.session_stats4.2 在Agent运行时注入Handlerfrom langchain.agents import initialize_agent, AgentType from langchain.llms import OpenAI # 1. 创建带有统计功能的LLM llm OpenAI( temperature0, model_namegpt-3.5-turbo, callbacks[TokenUsageStatsHandler(session_idsess_123)] # 注入回调 ) # 2. 初始化Agent假设已有tools tools [...] # 你的工具列表 agent initialize_agent( tools, llm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue # verbose模式也会触发回调 ) # 3. 运行Agent try: result agent.run(请查询北京明天的天气并建议我是否要带伞。) finally: # 4. 获取统计结果 stats_handler llm.callbacks[0] summary stats_handler.get_summary() print(json.dumps(summary, indent2, ensure_asciiFalse))关键点解析回调的绑定我们将自定义的TokenUsageStatsHandler实例作为callbacks参数传递给LLM。这样LangChain框架在执行过程中会在关键节点on_llm_start,on_llm_end等自动调用我们handler里的方法。信息的获取on_llm_end方法中的response.llm_output包含了模型API返回的原始信息其中就有我们需要的token_usage。这是最准确的数据来源。异步与性能统计操作本身是轻量的累加和记录。但如果你需要进行实时的组件分析_analyze_components或远程写入数据库务必将其放入后台线程或异步任务队列避免阻塞主请求线程。5. 数据持久化、可视化与成本估算统计数据留在内存里没用我们需要存下来、展示出来、并算清楚多少钱。5.1 数据持久化方案选型时序数据库推荐Token消耗数据是典型的时间序列数据适合用InfluxDB、TimescaleDB或Prometheus存储。它们擅长高吞吐写入和按时间范围的聚合查询。关系型数据库如果数据量不大或需要与业务数据用户表、订单表做复杂关联PostgreSQL或MySQL也是可行的。可以为sessions和requests建立两张表。日志ELK将每次请求的Token使用情况作为结构化日志JSON格式输出然后用Logstash收集存入Elasticsearch最后用Kibana做可视化。这套方案运维复杂但查询灵活。数据仓库对于需要深度商业智能BI分析的情况可以定期将数据同步到Snowflake、BigQuery或ClickHouse中进行跨主题、历史趋势的分析。我们的选择是轻量级项目用PostgreSQL上了规模后迁移到时序数据库同时将聚合后的业务数据同步到数据仓库供BI使用。5.2 成本估算把Token数变成钱统计Token的最终目的是控制成本。我们需要一个成本估算模块。# cost_estimator.py MODEL_PRICING { # 示例价格美元/每千Token请以官方最新价格为准 gpt-3.5-turbo: {input: 0.0005, output: 0.0015}, gpt-4-turbo: {input: 0.01, output: 0.03}, claude-3-haiku: {input: 0.00025, output: 0.00125}, # 添加更多模型... } def estimate_cost(model_name: str, prompt_tokens: int, completion_tokens: int) - float: 估算单次请求的成本 if model_name not in MODEL_PRICING: raise ValueError(fPricing for model {model_name} not configured.) price MODEL_PRICING[model_name] cost (prompt_tokens / 1000) * price[input] (completion_tokens / 1000) * price[output] return round(cost, 6) def estimate_session_cost(session_stats: Dict) - float: 估算整个会话的成本可能包含多次请求、多种模型 total_cost 0.0 for req in session_stats.get(requests, []): model req.get(model, unknown) # 如果单次请求未区分模型则使用会话默认模型 if model unknown: model session_stats.get(default_model, gpt-3.5-turbo) cost estimate_cost(model, req[prompt_tokens], req[completion_tokens]) total_cost cost return round(total_cost, 4)重要提醒价格动态更新模型价格会变动尤其是促销或版本更新时。最好将价格配置放在数据库或配置中心支持热更新。汇率与税费如果是跨国服务需要考虑汇率换算和当地税费这部分逻辑可以放在成本估算的最终阶段。预留缓冲实际账单可能因为网络重试、配额外的额外Token如图片理解中的Vision Tokens而略有出入估算成本时应保留5-10%的缓冲空间。5.3 可视化看板让数据说话有了数据和成本我们需要一个仪表盘。对于初创团队用Grafana连接时序数据库或直接使用Metabase、Redash连接你的业务数据库是快速搭建看板的不二之选。核心监控面板应包含全局概览今日/本月总Token消耗、总估算成本、请求次数、活跃会话数。消耗趋势图按小时/天展示Token消耗区分Prompt/Completion和成本曲线。模型分布饼图展示不同模型消耗的Token占比和成本占比。功能热度/成本排行按业务标签功能模块排序找出最“烧钱”的功能。用户消耗TOP榜识别高消耗用户可能是重点客户也可能是异常行为需要排查。异常警报设置阈值如单会话成本超过$10或每分钟请求量暴增触发企业微信、钉钉或邮件告警。6. 避坑指南实践中遇到的五个“坑”与填法在实施过程中我们踩了不少坑这里分享出来希望大家能绕道走。坑一Tokenizer不匹配导致的统计误差现象我们自己用tiktoken计算的Prompt Token数总是比OpenAI API返回的prompt_tokens少一点。根因我们只计算了文本部分的Token但API在发送请求时会将消息的role如system,user,assistant以及一些特殊的控制标记也进行编码计入总Token。不同模型、不同API版本的具体处理方式可能有细微差别。填法以API返回的usage为准将其作为黄金标准。自计算的Token数仅用于预估和预警例如“当前上下文长度已接近模型上限”不作为计费依据。可以在日志中对比两者差异长期观察并校准自己的预估公式。坑二流式响应Streaming下的Token统计现象当启用流式输出streamTrue以获得更快的首字响应时间时on_llm_end回调中拿到的response.llm_output可能为空或不包含完整的token_usage。根因流式响应下Token使用信息可能在最终的完成消息[DONE]中才返回或者通过单独的字段传递。填法查阅所用框架如LangChain对Streaming Callback的支持。通常会有专门的on_llm_stream_end或类似回调。或者更简单粗暴的做法是对于需要精确计费的场景暂时关闭流式响应。如果必须用流式可能需要降级为使用模型输出文本的长度进行估算不精确。坑三Agent复杂工作流中的重复计算现象一个Agent任务总Token消耗异常高检查发现同一个问题被反复处理。根因Agent的推理链ReAct, Plan-and-Execute等可能导致多次调用LLM。例如先“思考”一步再“执行”工具再基于结果“思考”下一步。如果设计不当可能会把完整的对话历史重复传入每次调用造成Token的指数级浪费。填法优化上下文管理策略。采用“摘要式记忆”或“向量检索记忆”用固定长度的摘要或只检索相关历史片段来替代传递全部历史。在Callback中记录每次请求的purpose如“reasoning”,“action”分析哪类调用最耗Token针对性优化。坑四多租户Multi-tenancy下的数据隔离与聚合现象为不同客户租户部署的Agent成本核算混乱无法按客户出具账单。根因统计系统在设计初期未考虑租户隔离所有数据混在一起。填法在统计记录的每一层会话、请求都强制加上tenant_id字段。在数据查询和可视化层面实现严格的权限过滤确保每个租户只能看到自己的数据。聚合分析时tenant_id应作为核心维度。坑五离线批量处理任务的统计遗漏现象夜间运行的批量数据处理Agent消耗了大量Token但未计入统计系统。根因统计Callback只附着在实时API服务上而离线任务可能使用不同的脚本或直接调用SDK绕过了监控。填法统一模型调用客户端。将所有对AI模型的调用无论是实时还是离线都收敛到一个统一的、内置了统计功能的Client封装里。确保这个Client是团队内访问模型的唯一入口。7. 从统计到优化基于数据的Agent调优实战统计不是终点而是优化的起点。有了细粒度的数据我们可以做很多事。优化案例一精简系统提示词通过组件级统计我们发现一个用于“代码审查”的Agent其系统提示词长达1200个Token大部分是冗长的规则枚举。我们利用GPT-4本身让它根据示例总结出一套更精炼、更具概括性的规则将提示词压缩到400个Token效果不变单次调用成本立降65%。优化案例二实现动态上下文窗口我们发现很多会话的历史对话部分Token占比超过50%但很多早期对话已不再相关。我们实现了一个简单的策略当会话总Token数接近模型上限如GPT-4的128K的80%时自动触发上下文“修剪”。修剪策略不是简单删除最老的而是用另一个小模型如GPT-3.5-Turbo对早期对话进行摘要然后用摘要替换原始长文本成功将多个长会话的Token消耗维持在安全线内。优化案例三建立成本预算与熔断机制为每个用户/租户设置每日/每月Token预算。在统计系统的核心增加预算检查。当用户消耗达到预算的80%时发出警告达到100%时自动将该用户的请求路由到一个“降级模型”如从GPT-4降到GPT-3.5-Turbo或者直接返回友好提示告知预算已用尽。这有效防止了因意外或恶意请求导致的成本失控。优化案例四A/B测试提示词与模型当我们有两个功能相似的提示词模板或考虑升级模型时不再凭感觉决策。我们进行A/B测试将流量随机导入不同版本然后对比相同任务下的平均Token消耗、任务成功率和成本。数据会清晰地告诉我们哪个版本在效果和成本上取得了最佳平衡。给Agent加上用量统计就像给一辆车装上了油耗表和行车电脑。它让你从“盲开”变成了“精打细算的驾驶”。这个过程一开始会有点繁琐需要改造代码、搭建管道、设计看板。但一旦系统跑起来你会发现之前看不见的成本黑洞一个个浮现优化方向变得前所未有的清晰。更重要的是当你能向老板或客户清晰地展示“价值与成本”时整个项目会获得更大的信任和更健康的成长空间。