1. 项目概述为什么你的Agent需要一个“电表”最近在折腾AI Agent开发的朋友估计都遇到过同一个灵魂拷问这玩意儿到底“吃”了我多少Token尤其是在对接像OpenAI、DeepSeek这类按Token计费的模型服务时每次调用都像在烧钱心里没个谱儿可不行。这就好比给家里装了个大功率电器你总得看看电表走了多少字才能知道这个月电费会不会爆表。“Token账单怎么看”这个需求说白了就是给你的Agent加装一个“用量统计”模块。这不仅仅是简单的计数它涉及到从请求发出到响应返回的全链路追踪、不同模型和接口的差异化处理、以及最终清晰直观的数据呈现。我最近在重构自己的Agent框架时就花了大力气把这套统计体系给搭了起来过程踩了不少坑也总结了一些实用的方案。今天就来聊聊如何从零开始给你的Agent加上一套靠谱的用量统计系统让你对每一次AI交互的成本都了如指掌。2. 核心需求拆解我们要统计什么在动手写代码之前我们必须先明确统计的边界和维度。一个完整的用量统计系统远不止在调用API后读取返回的usage字段那么简单。2.1 统计维度的精细化划分首先我们需要定义清楚要统计哪些数据。至少应该包含以下几个核心维度按请求粒度统计这是最基础的。每一次向大模型服务发起的请求包括聊天补全、图像生成、函数调用等都需要记录其消耗的Token数。这通常包括Prompt Tokens你发送给模型的提示词所消耗的Token数。Completion Tokens模型生成的回复所消耗的Token数。Total Tokens本次请求的总Token数通常是前两者之和。按会话/任务粒度聚合一个复杂的Agent任务可能由多次模型调用组成例如规划、执行、反思等多个步骤。我们需要能够将这些分散的调用归属到同一个“会话”或“任务ID”下进行聚合统计从而评估完成一个完整用户目标的总成本。按模型和接口区分不同的模型如gpt-4o、gpt-4-turbo、claude-3-opus单价不同甚至同一个服务商的不同接口如Chat Completions vs. Embeddings计费方式也完全不同。统计时必须带上模型名称和接口类型后续才能做准确的成本核算。按用户/租户隔离如果你的Agent服务是多用户场景那么用量统计必须能够按用户ID或API Key进行隔离和汇总这是实现用量配额、计费扣费的基础。时间序列数据记录每次请求的时间戳这样我们就能分析用量随时间的变化趋势比如高峰时段、每日/每周用量等。2.2 非功能性需求性能与可靠性除了统计什么我们还得考虑怎么统计才靠谱。低侵入性与透明化理想的统计模块应该对业务代码的侵入性尽可能小。最好是通过中间件Middleware、装饰器Decorator或拦截器Interceptor的方式无缝集成业务开发人员无需关心统计逻辑。高性能与低延迟统计操作本身不能成为系统的性能瓶颈。记录用量信息应该是异步的、非阻塞的操作绝不能因为等待统计日志写入而拖慢模型响应的速度。数据可靠性用量数据直接关系到计费绝不能丢失。虽然允许一定的最终一致性比如先响应请求再异步记录日志但必须有一套可靠的机制确保数据最终落地如写入数据库、发送到消息队列。可扩展性统计模块应该易于扩展以便未来增加新的统计维度如计算延迟、记录错误类型或对接新的存储后端如从文件切换到数据库再切换到数据仓库。3. 技术方案选型与架构设计明确了需求接下来就是技术选型和架构设计。这里没有银弹需要根据你的团队规模、数据量和运维能力来权衡。3.1 核心架构模式拦截器 异步处理经过实践我认为最优雅和高效的架构是“拦截器 异步处理队列”模式。拦截器层在Agent框架的HTTP客户端或SDK调用层植入拦截器。当发起一个模型API调用时拦截器能捕获到请求的元数据如模型名、请求体。当收到响应时拦截器能解析响应头或响应体中的usage字段。这个过程是同步的但只做最低限度的数据提取和封装。事件发布拦截器将封装好的用量事件包含所有统计维度信息作为一个消息事件发布出去。这里切忌在拦截器内直接进行数据库写入等耗时操作。异步处理由一个独立的消费者服务或后台线程来订阅这些用量事件并将其持久化到存储系统中。这样模型请求的响应速度就不会受到统计操作的影响。3.2 存储方案对比用量数据的特点是写多读少偶尔需要聚合查询。以下是几种常见的存储方案存储方案优点缺点适用场景关系型数据库 (如 PostgreSQL, MySQL)事务支持好查询灵活易于做复杂聚合如按用户、按模型分组统计。高并发写入可能成为瓶颈。需要管理表结构。数据量不大日请求量百万以下需要实时复杂查询的场景。时序数据库 (如 InfluxDB, TimescaleDB)为时间序列数据优化写入性能高压缩比好内置时间窗口聚合函数。查询灵活性可能不如关系型数据库生态相对小众。主要用于监控和趋势分析需要高效存储和查询时间序列指标。日志文件 ELK Stack实现简单易于集成。Elasticsearch 搜索分析能力强。实时性稍差数据可靠性依赖日志采集链路成本可能较高。初期快速验证或已有成熟的日志监控体系。消息队列 (如 Kafka) 流处理解耦彻底吞吐量极高为后续实时计算和多个下游消费提供可能。架构复杂运维成本高。超大规模、需要实时风控或计费的场景。实操心得对于绝大多数中小型Agent项目我推荐从PostgreSQL开始。它足够通用用一张结构简单的表就能满足初期的所有统计和查询需求。当数据量增长到一定规模后可以考虑将历史冷数据归档或者引入时序数据库专门处理指标聚合。一开始就上Kafka这类重型架构属于过度设计会带来不必要的复杂度。3.3 关键工具与库开发语言根据你的Agent主体语言选择。Python生态有openai库可以方便地自定义HTTP客户端或使用回调函数Node.js生态有langchain提供了丰富的callback机制。异步处理Python的asyncioaiopg异步PostgreSQL驱动或celeryNode.js的bull或agenda任务队列。监控与可视化将聚合后的用量数据如每分钟请求数、Token消耗速率导出到Prometheus用Grafana制作实时监控看板效果非常直观。4. 核心实现细节与代码实战理论说再多不如看代码。下面我以Python openai官方库为例展示一个最核心的拦截器实现。4.1 构建一个可插拔的用量统计客户端我们不直接修改openai库的代码而是通过包装Wrapper的方式创建一个自定义客户端。import openai from openai import OpenAI import asyncio import json from datetime import datetime from typing import Dict, Any, Optional import threading from queue import Queue import logging # 初始化日志和事件队列 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class TokenUsageEvent: 用量事件数据类 def __init__(self, event_id: str, model: str, prompt_tokens: int, completion_tokens: int, total_tokens: int, user_id: Optional[str], session_id: Optional[str], api_type: str, timestamp: datetime): self.event_id event_id self.model model self.prompt_tokens prompt_tokens self.completion_tokens completion_tokens self.total_tokens total_tokens self.user_id user_id self.session_id session_id self.api_type api_type # 如 chat, completion, embedding self.timestamp timestamp def to_dict(self): return { event_id: self.event_id, model: self.model, prompt_tokens: self.prompt_tokens, completion_tokens: self.completion_tokens, total_tokens: self.total_tokens, user_id: self.user_id, session_id: self.session_id, api_type: self.api_type, timestamp: self.timestamp.isoformat() } class UsageStatisticsClient: 带用量统计功能的OpenAI客户端包装器 def __init__(self, api_key: str, base_url: Optional[str] None, default_user_id: Optional[str] None): # 初始化原生客户端 self._client OpenAI(api_keyapi_key, base_urlbase_url) # 用于关联用户和会话的上下文可通过线程局部存储实现更复杂场景 self._context {user_id: default_user_id, session_id: None} # 用量事件队列用于异步处理 self._event_queue Queue() # 启动后台消费者线程 self._consumer_thread threading.Thread(targetself._consume_events, daemonTrue) self._consumer_thread.start() logger.info(UsageStatisticsClient initialized with background consumer.) def set_context(self, user_id: Optional[str] None, session_id: Optional[str] None): 设置当前上下文用户、会话 if user_id is not None: self._context[user_id] user_id if session_id is not None: self._context[session_id] session_id def chat_completion_create(self, **kwargs): 包装聊天补全接口自动统计用量 # 1. 发起请求 response self._client.chat.completions.create(**kwargs) # 2. 提取用量信息 if response.usage: event TokenUsageEvent( event_idfchat_{datetime.utcnow().strftime(%Y%m%d%H%M%S%f)}, modelresponse.model, prompt_tokensresponse.usage.prompt_tokens, completion_tokensresponse.usage.completion_tokens, total_tokensresponse.usage.total_tokens, user_idself._context.get(user_id), session_idself._context.get(session_id), api_typechat, timestampdatetime.utcnow() ) # 3. 异步放入队列不阻塞主流程 self._event_queue.put(event) logger.debug(fUsage event queued: {event.model} - {event.total_tokens} tokens) # 4. 返回原始响应 return response def _consume_events(self): 后台消费者线程从队列取出事件并持久化 while True: try: event self._event_queue.get() if event is None: # 收到停止信号 break self._persist_event(event) except Exception as e: logger.error(fError consuming usage event: {e}, exc_infoTrue) def _persist_event(self, event: TokenUsageEvent): 持久化事件到存储这里以打印和写入文件为例可替换为DB操作 # 这里实现你的持久化逻辑例如 # 1. 打印到日志 logger.info(fPersisting usage: {event.to_dict()}) # 2. 写入本地文件示例 with open(token_usage.log, a) as f: f.write(json.dumps(event.to_dict()) \n) # 3. 实际项目中应替换为写入数据库如PostgreSQL # self._save_to_postgres(event) def shutdown(self): 优雅关闭发送停止信号给消费者线程 self._event_queue.put(None) self._consumer_thread.join() # 使用示例 if __name__ __main__: # 初始化带统计的客户端 client UsageStatisticsClient(api_keyyour-api-key, default_user_iduser_123) # 设置当前会话 client.set_context(session_idsession_abc) try: # 像使用原生客户端一样调用 response client.chat_completion_create( modelgpt-3.5-turbo, messages[{role: user, content: Hello, how are you?}], max_tokens50 ) print(response.choices[0].message.content) finally: # 程序退出前关闭客户端 client.shutdown()这段代码的核心在于UsageStatisticsClient类。它包装了原生的OpenAI客户端在chat_completion_create方法中先执行请求然后从响应中提取usage信息封装成事件放入一个队列。一个独立的后台线程会不断从队列中取出事件并进行持久化示例中是写日志文件。这样统计操作就不会影响API调用的返回速度。4.2 数据库表结构设计如果选择用PostgreSQL持久化表结构可以这样设计CREATE TABLE token_usage ( id BIGSERIAL PRIMARY KEY, event_id VARCHAR(64) NOT NULL UNIQUE, -- 事件唯一ID防重放 model VARCHAR(128) NOT NULL, -- 模型名称 prompt_tokens INTEGER NOT NULL, completion_tokens INTEGER NOT NULL, total_tokens INTEGER NOT NULL, user_id VARCHAR(128), -- 用户标识 session_id VARCHAR(128), -- 会话标识 api_type VARCHAR(50) NOT NULL, -- 接口类型 cost_estimate DECIMAL(12, 6), -- 估算成本可根据模型单价计算 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- 请求时间 recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -- 记录时间 ); -- 创建常用查询索引 CREATE INDEX idx_token_usage_user_id ON token_usage(user_id); CREATE INDEX idx_token_usage_session_id ON token_usage(session_id); CREATE INDEX idx_token_usage_created_at ON token_usage(created_at); CREATE INDEX idx_token_usage_model ON token_usage(model);cost_estimate字段是一个预计算字段你可以根据model和total_tokens结合事先配置好的模型单价表在写入时或之后通过定时任务计算并更新方便直接查询费用。4.3 处理Edge Cases与难点流式响应Streaming的统计对于流式响应OpenAI的API会在最后返回一个包含usage的独立data: [DONE]消息。我们的拦截器需要能够识别并捕获这个消息。这需要更精细地处理SSEServer-Sent Events数据流。非OpenAI格式的API响应如果你使用的是第三方中转API或自部署模型其返回的用量字段格式可能不同。拦截器需要具备可配置的解析器或者通过适配器模式来兼容不同后端的响应格式。错误请求的统计即使API请求因额度不足、网络超时等原因失败了如果已经消耗了Token比如请求已到达服务端并被处理理论上也应该统计。但这部分信息通常无法从错误响应中直接获取需要依赖服务商提供的账单API进行对账或者更保守地只在成功响应时统计。上下文关联的准确性在多线程或异步环境下如FastAPI服务器如何确保每个请求都能正确关联到发起它的用户和会话上下文这需要使用类似contextvarsPython或AsyncLocalStorageNode.js的线程/异步上下文管理工具而不是简单的全局变量。5. 数据展示与成本分析数据存好了最终目的是为了“看”和“用”。一个简单的命令行汇总当然可以但可视化的仪表盘更能直观地发现问题。5.1 常用分析SQL示例你可以直接连接数据库进行即席查询-- 查询今日总消耗 SELECT SUM(total_tokens) as today_total_tokens, SUM(cost_estimate) as today_total_cost FROM token_usage WHERE DATE(created_at) CURRENT_DATE; -- 按用户查看今日消耗TOP 10 SELECT user_id, SUM(total_tokens) as total_tokens, SUM(cost_estimate) as total_cost, COUNT(*) as request_count FROM token_usage WHERE DATE(created_at) CURRENT_DATE GROUP BY user_id ORDER BY total_cost DESC LIMIT 10; -- 按模型分析过去7天的用量趋势 SELECT DATE(created_at) as date, model, SUM(total_tokens) as daily_tokens FROM token_usage WHERE created_at NOW() - INTERVAL 7 days GROUP BY DATE(created_at), model ORDER BY date DESC, model;5.2 集成可视化监控Grafana更专业的做法是将聚合数据导入Grafana。你可以写一个简单的脚本定期如每分钟运行上述聚合查询将结果写入Prometheus可以抓取的格式如直接使用Prometheus的Python客户端库推数据或者在Grafana中直接配置PostgreSQL数据源。在Grafana中你可以创建这样的看板总览面板显示当前小时/今日/本月的总Token消耗和估算费用曲线。用户排行面板用条形图展示消耗最高的Top N用户。模型分布面板饼图展示不同模型的Token消耗占比。异常报警设置规则当某个用户的单位时间用量超过阈值或总费用接近预算时触发告警通过邮件、Slack等。6. 避坑指南与进阶思考在实施过程中我遇到了不少坑这里分享几个关键的注意事项。注意事项一Token计算的一致性不同模型、甚至同一模型的不同版本其Tokenizer分词器可能略有差异。服务端返回的usage是最权威的计费依据。如果你在客户端也尝试估算Prompt Token例如用tiktoken库请仅将其作为参考和预警最终成本以服务端返回为准。切勿用客户端的估算值作为计费唯一依据否则会产生偏差。注意事项二异步队列的可靠性示例中使用了内存队列(queue.Queue)这在单机部署且进程正常退出的情况下是可行的。但在生产环境如果进程意外崩溃内存队列中的数据会丢失。对于计费数据建议至少使用一个持久化的消息队列如Redis的List结构或者更正式的RabbitMQ、Kafka确保事件不丢失。注意事项三数据膨胀与归档用量日志表会快速增长。需要提前规划数据生命周期。例如保留最近3个月的明细数据供实时查询将3个月前的数据按月聚合后保留总和、平均值等转移到历史汇总表并从明细表中删除。这能有效控制主表大小保证查询性能。进阶思考从统计到管控有了精准的用量统计我们就可以做更多事预算与配额为每个用户或项目设置每日/每月Token预算或费用上限。在拦截器层可以在请求前查询当前已用量如果即将超限则直接拒绝请求或降级到更便宜的模型。成本优化分析分析哪些Prompt模板或会话模式消耗Token最多从而优化提示工程降低成本。性能监控同时记录每次请求的延迟将高延迟、高消耗的请求关联起来分析定位性能瓶颈。给Agent加上用量统计就像给汽车装上了油耗表和里程表。它不能直接让车跑得更快但能让你清楚地知道每一次“踩油门”的成本从而更聪明地规划路线避免不必要的浪费。这套系统看似是基础设施实则是Agent项目走向成熟和可运营的关键一步。从简单的日志记录开始逐步迭代到完整的监控、告警和成本管控体系你会发现对资源的掌控力越强你的Agent应用才会跑得越稳、越远。