做 AI 应用落地的团队十有八九会遇到同一个尴尬模型能力很强功能也上线了但老板问“这个月 AI 花了多少钱换回了什么”答案往往是模糊的。TokenSpend 这个项目标题直指问题核心——Token 是当前大模型应用最基础的计量单位Spend 是花费TokenSpend 要解决的不是简单的记账而是把每一笔 Token 消耗与业务收益对齐形成一套可复用的 AI ROI 分析闭环。下面直接进入工程实现角度。我会把 TokenSpend 当成一个真实要落地的系统来拆解覆盖数据模型、埋点采集、成本换算、指标分析、可视化看板以及 Agent 场景下的成本治理。无论你是写 AI 应用的工程师、搭平台的开发还是盯预算的负责人都能从里面拿到一个可以直接参考的实现思路而不是一句“要注意控制成本”的空话。先说明一个前提这篇文章写的是围绕“Token 维度追踪 AI 成本与收益”这一目标工程上可行、可以直接改造到现有项目里的方案。里面的表结构、代码片段和排查清单需要结合自己的技术栈、依赖版本和业务命名调整后再使用。1. 先算清楚 AI 投入的账TokenSpend 到底解决什么问题1.1 从 Credits 到 Token统一计量单位为什么重要大部分模型平台在计费时用的不是同一个词。有些平台直接按 Token 计费有些平台用 Credits点数计费还有些平台按请求次数、按时长、按并发数计费。对业务研发来说Credits 是一个很容易理解但又容易误用的概念它更像平台定义的点数不是模型真正消耗的 Token。同一个操作在不同模型、不同上下文长度下扣掉的 Credits 可能完全不同。Token 是模型处理文本时的最基本单位它才是一切成本计算的起点。一个英文单词可能拆成 1 到 2 个 Token一个中文字符可能对应 1 到 2 个 Token模型输入和输出分别计量。如果要搭建一套 ROI 分析系统第一步就是统一计量口径把所有平台的消耗都落到 Token 上再通过价格表换算成真实费用。这里有几个基础概念需要先区分清楚概念含义是否直接决定成本Credit平台定义的计费点数由平台规则决定不是原始计量Token模型处理文本的最小单位是Prompt Token输入到模型的 Token 数是Completion Token模型生成的 Token 数是Cached Token命中上下文缓存的输入 Token是通常价格更低1.2 完整的 ROI 链路记录、分摊、归因、对比ROI 如果只写成“收入减成本除以成本”这个公式在 AI 场景里基本没法落地。问题在于 AI 功能的收益很难从整体收入里剥离出来一次客服会话解决了一个订单问题一个 Agent 自动完成了一次报表生成一条 AI 生成的内容被用户大量围观这些收益形态差异很大无法简单用同一个收入字段表示。所以 TokenSpend 的思路是把 ROI 拆成四个环节记录每次 LLM 调用都记录模型、Token 数组、时间、调用方、业务维度。分摊把一次调用或一个会话的成本归属到具体功能、团队、客户。归因把订单、转化、任务完成、人工时间节约等业务结果关联到 token 消耗。对比按时间、功能、模型、用户群体对比成本和结果变化。只有这条链路完整才能回答“哪个功能亏了、哪个功能赚了、哪个模型适合继续用、哪个流程应该优化”。1.3 适用场景与读者这套方案适合以下场景企业内部部署了多个 AI 助手需要判断哪些部门、哪些流程在真正获益。C 端产品里嵌入了 AI 功能需要按功能模块核算成本决定是否收费、如何定价。以 Agent 为核心的自动化流程需要给每个任务设置预算和止损线。AI 编程工具、AI 客服、AI 内容生成工具需要评估提效价值是不是大于模型花费。目标读者包括负责功能的后端研发、做平台工程的开发、负责成本预算的产品经理以及需要给团队制定“AI 使用规范”的技术负责人。一个常见误区是等账单异常后再去查实际上到了账单层只能看到总金额看不到是哪个功能、哪个会话、哪次提示词调整引起的增长。成本数据只有在调用发生时采集事后几乎无法补全。2. TokenSpend 数据模型设计一张事实表加四张维度表2.1 llm_usage 事实表记录每一次模型调用的完整账目成本分析的第一步是设计一张能够承载“单次模型调用明细”的表。这张表是后续所有分析的核心它必须是明细表而不是汇总表。汇总表只能回答“花了多少”回答不了“为什么花这么多”。CREATE TABLE llm_usage ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, request_id VARCHAR(64) NOT NULL COMMENT 业务侧或网关侧请求ID, trace_id VARCHAR(64) DEFAULT NULL, model VARCHAR(128) NOT NULL COMMENT 模型名称如 gpt-4o-mini, provider VARCHAR(64) NOT NULL DEFAULT openai, endpoint VARCHAR(128) NOT NULL COMMENT 实际调用场景或API路径, user_id VARCHAR(64) DEFAULT NULL, project_id VARCHAR(64) DEFAULT NULL, feature_id VARCHAR(64) DEFAULT NULL COMMENT 最小业务功能单元, session_id VARCHAR(64) DEFAULT NULL COMMENT 一次业务会话的唯一标识, prompt_tokens INT UNSIGNED NOT NULL DEFAULT 0, completion_tokens INT UNSIGNED NOT NULL DEFAULT 0, total_tokens INT UNSIGNED NOT NULL DEFAULT 0, cached_tokens INT UNSIGNED NOT NULL DEFAULT 0, prompt_price DECIMAL(10,8) NOT NULL DEFAULT 0 COMMENT 每千输入Token单价, completion_price DECIMAL(10,8) NOT NULL DEFAULT 0, cost_amount DECIMAL(14,6) NOT NULL DEFAULT 0 COMMENT 本次调用成本, currency VARCHAR(8) NOT NULL DEFAULT USD, latency_ms INT UNSIGNED DEFAULT NULL, finish_reason VARCHAR(32) DEFAULT NULL, is_success TINYINT(1) NOT NULL DEFAULT 1, status_code SMALLINT UNSIGNED DEFAULT NULL, error_code VARCHAR(64) DEFAULT NULL, tags JSON DEFAULT NULL COMMENT 额外标签如请求来源、模型通道, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_created_at (created_at), KEY idx_model (model), KEY idx_project_feature (project_id, feature_id), KEY idx_session (session_id), KEY idx_request_id (request_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT大模型调用明细表;这张表里有几个设计点需要解释第一既存 Token 数也存单价和金额。这样做的原因是模型价格会调整。如果只存 Token 数供应商降价后回看历史成本就没办法对应当时的账单如果只存金额后续想统计“某类模型的总 Token 消耗”就没有原始数据。两者并存历史口碑才能稳定。第二request_id 单独建索引。这个字段用于和网关日志、模型平台日志对账排查“某次调用为什么没记录”时会非常有用。第三is_success 和 error_code 也要记录。调用失败的请求同样消耗了成本而且可能是没有完成业务价值的浪费它和成功调用一样需要进入成本分析。2.2 维度表把成本归属到业务单元llm_usage 里的 project_id、feature_id、user_id、session_id 都是维度字段。严格来说可以不做成单独的关联表直接在公司内部统一维护一套编码即可。但在实际工程中建议至少维护 project 和 feature 两张维度表目的是保证每个业务单元有责任人、负责人和成本目标。CREATE TABLE project_dim ( project_id VARCHAR(64) PRIMARY KEY, project_name VARCHAR(128) NOT NULL, owner_team VARCHAR(64) NOT NULL, budget_quota DECIMAL(14,6) DEFAULT NULL, status TINYINT(1) NOT NULL DEFAULT 1, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE feature_dim ( feature_id VARCHAR(64) PRIMARY KEY, project_id VARCHAR(64) NOT NULL, feature_name VARCHAR(128) NOT NULL, owner VARCHAR(64) DEFAULT NULL, enable_roi TINYINT(1) NOT NULL DEFAULT 0, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP );维度设计最重要的原则是维度的值必须在调用埋点时就已经存在而不是等数据进数仓后再靠规则去猜。比如“这个调用属于哪个功能”在业务代码里应该是显式传入的。常见的反模式是只记录模型名称和 Token 数等做报表时再根据消息内容倒推功能这种做法在调用量上来以后根本不可行。2.3 价格表与收益事件表价格表用于把 Token 数组换算成金额。它需要支持按模型、按生效区间维护这样才能在供应商调整价格后仍然正确回算历史成本。CREATE TABLE model_price ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, provider VARCHAR(64) NOT NULL, model VARCHAR(128) NOT NULL, input_price_per_1k DECIMAL(10,8) NOT NULL, output_price_per_1k DECIMAL(10,8) NOT NULL, cached_input_price_per_1k DECIMAL(10,8) NOT NULL DEFAULT 0, currency VARCHAR(8) NOT NULL DEFAULT USD, effective_from DATETIME NOT NULL, effective_to DATETIME DEFAULT NULL, remark VARCHAR(255) DEFAULT NULL, UNIQUE KEY uk_model_version (model, effective_from) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT模型价目表;收益事件表用于记录业务侧发生的结果事件例如下单、转化、任务完成、客服工单关闭。CREATE TABLE revenue_event ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, project_id VARCHAR(64) NOT NULL, feature_id VARCHAR(64) NOT NULL, session_id VARCHAR(64) DEFAULT NULL, user_id VARCHAR(64) DEFAULT NULL, event_type VARCHAR(64) NOT NULL COMMENT order/conversion/task_done/ticket_closed, revenue_amount DECIMAL(14,6) NOT NULL DEFAULT 0, currency VARCHAR(8) NOT NULL DEFAULT USD, related_usage_ids JSON DEFAULT NULL COMMENT 关联的调用记录ID列表, occurred_at DATETIME NOT NULL, tags JSON DEFAULT NULL, KEY idx_occurred_at (occurred_at), KEY idx_feature (feature_id, event_type) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT收益事件表;related_usage_ids 字段是可选的。对于简单场景直接通过 session_id 把收益事件和调用记录关联起来就够了不一定要保存具体的 usage id 列表。保存列表的好处是归因精确坏处是数据量很大写入成本高。学习阶段可以先忽略 tags 和 related_usage_ids只保留最简单字段先把记录和分析跑通再把复杂维度加上去。3. 采集层实现把 token 消耗从响应体变成可查数据3.1 usage 字段是采集的起点无论用哪个模型提供商的 SDK响应体里基本都会包含 usage 信息。以最常见的 OpenAI 响应结构为例{ model: gpt-4o-mini, usage: { prompt_tokens: 1200, completion_tokens: 350, total_tokens: 1550, prompt_tokens_details: { cached_tokens: 800 } } }这些字段就是成本采集的原始数据。要注意的是不同厂商的字段名不完全一样有的叫 input_tokens、output_tokens有点则把缓存命中单独命名为 cache_read_input_tokens。实现采集层之前先整理一份自己常用模型 API 的 usage 字段映射表格式可以参考下面这样厂商输入 Token 字段输出 Token 字段总 Token 字段缓存 Token 字段OpenAIprompt_tokenscompletion_tokenstotal_tokensprompt_tokens_details.cached_tokensAnthropicinput_tokensoutput_tokensusage 聚合值cache_read_input_tokens其他平台以官方文档为准以官方文档为准以官方文档为准以官方文档为准字段没对齐之前不要急着建全局数据管道否则后面做成本换算时会出现大量口径不一致的脏数据。3.2 用 Python 装饰器包装 OpenAI SDK最直接的埋点方式是在业务调用 LLM SDK 的位置统一增加一个跟踪装饰器。下面是一个基于 OpenAI Python SDK 的最小示例import functools import time import uuid from openai import OpenAI from app.tracking import write_usage def track_llm_call(project_id, feature_id, user_idNone, session_idNone): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): begin time.time() try: response func(*args, **kwargs) latency_ms int((time.time() - begin) * 1000) usage getattr(response, usage, None) if usage is None: return response prompt_tokens getattr(usage, prompt_tokens, 0) or 0 completion_tokens getattr(usage, completion_tokens, 0) or 0 total_tokens getattr(usage, total_tokens, 0) or 0 details getattr(usage, prompt_tokens_details, None) cached_tokens 0 if details is not None: cached_tokens getattr(details, cached_tokens, 0) or 0 write_usage({ request_id: str(uuid.uuid4()), model: getattr(response, model, None), project_id: project_id, feature_id: feature_id, user_id: user_id, session_id: session_id, prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, total_tokens: total_tokens, cached_tokens: cached_tokens, latency_ms: latency_ms, is_success: True, }) return response except Exception as exc: write_usage({ request_id: str(uuid.uuid4()), model: None, project_id: project_id, feature_id: feature_id, user_id: user_id, session_id: session_id, prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, cached_tokens: 0, is_success: False, error_code: type(exc).__name__, }) raise return wrapper return decorator track_llm_call(project_idrobot-support, feature_idorder-status, user_idu_1001) def get_order_status(client: OpenAI, order_no: str): prompt build_prompt(order_no) return client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], )这个示例有几个关键点。失败调用也要写记录否则成功率的指标会虚高。异常分支里拿不到 usage所以 Token 数组填 0但 error_code 要保留用来排查“是不是某种提示词导致模型频繁拒绝调用”。不要在业务线程里同步等待数据库写入。上面示例中是调用了 write_usage 函数它内部应该只做入队操作真正写库放到后台批量任务里。采集逻辑越轻对业务链路的影响越小。3.3 Spring AI 场景下的采集方式Java 技术栈使用 Spring AI 时不建议在每一个业务方法里手动埋点。更合理的做法是通过 AOP 切面统一拦截 ChatClient 的调用从 ChatResponse 中读取 usage 信息。Aspect Component public class LlmUsageAspect { private final TokenUsageRepository repository; public LlmUsageAspect(TokenUsageRepository repository) { this.repository repository; } Around(execution(* org.springframework.ai.chat.client.ChatClient.CallResponseSpec.call(..))) public Object recordTokenUsage(ProceedingJoinPoint pjp) throws Throwable { long begin System.currentTimeMillis(); try { Object result pjp.proceed(); long latencyMs System.currentTimeMillis() - begin; if (result instanceof ChatResponse response response.getMetadata() ! null response.getMetadata().getUsage() ! null) { ChatUsage usage response.getMetadata().getUsage(); repository.save(toRecord(usage, latencyMs, true)); } return result; } catch (Throwable t) { repository.save(toRecord(null, 0, false)); throw t; } } private LlmUsageRecord toRecord(ChatUsage usage, long latencyMs, boolean success) { // 将 usage 字段映射到自己的持久化对象 return new LlmUsageRecord(); } }Spring AI 的 API 在不同版本里差异比较大ChatResponse、ChatUsage 这些类的包名和字段可能发生变化。落地前要找到当前项目使用的版本文档确认 usage 的读取方式。切面表达式的匹配范围也建议只拦截自己项目里真正使用的 Client Bean避免误拦截。3.4 异步上报和批量写入采集层最需要考虑的是性能。每调用一次模型就同步执行一次 INSERT在高并发场景下会拖垮接口响应。推荐的做法是内存队列加批量写入。import logging import queue import threading import time logger logging.getLogger(tokenspend) usage_queue queue.Queue(maxsize10000) def write_usage(record: dict) - None: try: usage_queue.put_nowait(record) except queue.Full: logger.warning(usage queue is full, record dropped) def batch_worker() - None: batch [] while True: try: item usage_queue.get(timeout2) batch.append(item) except queue.Empty: if batch: flush_batch(batch) batch [] else: if len(batch) 100: flush_batch(batch) batch [] def flush_batch(batch) - None: # 按项目实际情况批量插入数据库 pass # 应用启动时开启后台线程 threading.Thread(targetbatch_worker, daemonTrue).start()队列满时直接丢弃记录是一个权衡方案。生产环境建议把丢弃行为替换成“降级到文件日志”由独立采集进程扫描补写避免成本数据丢失。学习环境只需要保证同步入库基本可用即可。4. 成本换算与分摊Token 数量不等于真实费用4.1 按模型维度拆分定价Token 数是原始数据但要变成真正的成本必须乘以单价。不同模型的价格差异很大同一个模型的输入和输出价格也可能差数倍加上缓存命中的 Token 通常更便宜所以不能用一套固定比例去换算。正规做法是维护一张按模型、按生效时间区间的价格表。每次写入 llm_usage 时根据当时的模型和调用时间查询生效价格计算出 cost_amount 一起落库。这样即使后续价格调整历史记录的金额也不会被覆盖。4.2 一次调用的成本计算示例假设某个模型的价格配置如下示例价格实际落地时以官方计价文档为准项单价每千 Token普通输入 Token0.150缓存命中输入 Token0.075输出 Token0.600一次调用中prompt_tokens 为 1200其中 cached_tokens 为 600completion_tokens 为 300。那么成本计算方式为1200 - 600 / 1000 * 0.150 600 / 1000 * 0.075 300 / 1000 * 0.600 0.090 0.045 0.180 0.315 美元用代码实现就是def calculate_cost(model, prompt_tokens, completion_tokens, cached_tokens, price_config): price price_config[model] normal_prompt prompt_tokens - cached_tokens normal_cost normal_prompt / 1000 * price[input_per_1k] cached_cost cached_tokens / 1000 * price[cached_input_per_1k] completion_cost completion_tokens / 1000 * price[output_per_1k] return round(normal_cost cached_cost completion_cost, 6)这里最容易出错的是 cached_tokens 大于 prompt_tokens 的情况。某些 SDK 返回的 cached_tokens 可能已经包含在 prompt_tokens 中也可能返回的是独立的 cache 创建费用含义不同计算前先确认厂商字段语义。4.3 多业务归属的分摊逻辑一个会话可能同时服务多个业务目标。比如一条用户消息既做了意图识别又生成了最终回复这两个动作算在哪个功能头上会直接影响 ROI 报表结论。实际项目里常用三种分摊方式分摊方式适用场景实现成本主功能全额承担调用有明确主目的低按功能比例拆分几个业务方共用一次调用中按会话整体归因一个会话对应一个业务结果低建议从“主功能全额承担”开始只有当同一个请求确实被多个部门复用时才增加比例拆分逻辑。分摊规则要写入文档避免一次调用同时出现在两个功能的报表里造成总成本翻倍。4.4 价格变更后的重算策略供应商调价后历史数据怎么处理主要看团队想回答什么问题如果只关心“当前模型组合下老业务的成本大概是多少”可以用新价格重算历史数据。如果关心“过去某个月实际花了多少钱”必须保留当时的单价不能用新价格覆盖。所以采集时保存单价快照是关键一步。价格表里使用 effective_from 和 effective_to 区间查询时用调用时间匹配当时的生效价格既不影响历史报表也能按需重算。5. 分析层让成本和收益进入同一个坐标系5.1 定义核心指标成本记录只解决了“花了多少”ROI 分析还需要定义一系列可执行的指标。下面这组指标适合大多数 AI 功能团队指标定义用途总成本指定周期内所有 LLM 调用的成本总量控制与预算对比每流程成本单个业务流程的平均 Token 和金额判断流程是否依然划算调用成功率成功调用数 / 总调用数排查无效消耗平均每千 Token 成本总成本 / 总 Token * 1000监控模型组合变化收益金额与 AI 功能直接相关的订单、转化或节约收益侧数据ROI收益 - 成本 / 成本判断该功能是否值得继续建设指标定义要在第一周就固定下来不要频繁修改口径。口径一变历史报表全部失去可比性。5.2 用 SQL 完成成本和调用量聚合明细表建好以后第一张有价值的报表是“按功能维度的成本排行”。下面是按 feature 和 model 分组的近 7 天聚合SELECT feature_id, model, COUNT(*) AS call_count, SUM(prompt_tokens) AS prompt_tokens, SUM(completion_tokens) AS completion_tokens, SUM(total_tokens) AS total_tokens, SUM(cost_amount) AS total_cost, ROUND( SUM(cost_amount) * 1000 / NULLIF(SUM(total TokenSpend：从Token计量到AI ROI分析的落地方案
做 AI 应用落地的团队十有八九会遇到同一个尴尬模型能力很强功能也上线了但老板问“这个月 AI 花了多少钱换回了什么”答案往往是模糊的。TokenSpend 这个项目标题直指问题核心——Token 是当前大模型应用最基础的计量单位Spend 是花费TokenSpend 要解决的不是简单的记账而是把每一笔 Token 消耗与业务收益对齐形成一套可复用的 AI ROI 分析闭环。下面直接进入工程实现角度。我会把 TokenSpend 当成一个真实要落地的系统来拆解覆盖数据模型、埋点采集、成本换算、指标分析、可视化看板以及 Agent 场景下的成本治理。无论你是写 AI 应用的工程师、搭平台的开发还是盯预算的负责人都能从里面拿到一个可以直接参考的实现思路而不是一句“要注意控制成本”的空话。先说明一个前提这篇文章写的是围绕“Token 维度追踪 AI 成本与收益”这一目标工程上可行、可以直接改造到现有项目里的方案。里面的表结构、代码片段和排查清单需要结合自己的技术栈、依赖版本和业务命名调整后再使用。1. 先算清楚 AI 投入的账TokenSpend 到底解决什么问题1.1 从 Credits 到 Token统一计量单位为什么重要大部分模型平台在计费时用的不是同一个词。有些平台直接按 Token 计费有些平台用 Credits点数计费还有些平台按请求次数、按时长、按并发数计费。对业务研发来说Credits 是一个很容易理解但又容易误用的概念它更像平台定义的点数不是模型真正消耗的 Token。同一个操作在不同模型、不同上下文长度下扣掉的 Credits 可能完全不同。Token 是模型处理文本时的最基本单位它才是一切成本计算的起点。一个英文单词可能拆成 1 到 2 个 Token一个中文字符可能对应 1 到 2 个 Token模型输入和输出分别计量。如果要搭建一套 ROI 分析系统第一步就是统一计量口径把所有平台的消耗都落到 Token 上再通过价格表换算成真实费用。这里有几个基础概念需要先区分清楚概念含义是否直接决定成本Credit平台定义的计费点数由平台规则决定不是原始计量Token模型处理文本的最小单位是Prompt Token输入到模型的 Token 数是Completion Token模型生成的 Token 数是Cached Token命中上下文缓存的输入 Token是通常价格更低1.2 完整的 ROI 链路记录、分摊、归因、对比ROI 如果只写成“收入减成本除以成本”这个公式在 AI 场景里基本没法落地。问题在于 AI 功能的收益很难从整体收入里剥离出来一次客服会话解决了一个订单问题一个 Agent 自动完成了一次报表生成一条 AI 生成的内容被用户大量围观这些收益形态差异很大无法简单用同一个收入字段表示。所以 TokenSpend 的思路是把 ROI 拆成四个环节记录每次 LLM 调用都记录模型、Token 数组、时间、调用方、业务维度。分摊把一次调用或一个会话的成本归属到具体功能、团队、客户。归因把订单、转化、任务完成、人工时间节约等业务结果关联到 token 消耗。对比按时间、功能、模型、用户群体对比成本和结果变化。只有这条链路完整才能回答“哪个功能亏了、哪个功能赚了、哪个模型适合继续用、哪个流程应该优化”。1.3 适用场景与读者这套方案适合以下场景企业内部部署了多个 AI 助手需要判断哪些部门、哪些流程在真正获益。C 端产品里嵌入了 AI 功能需要按功能模块核算成本决定是否收费、如何定价。以 Agent 为核心的自动化流程需要给每个任务设置预算和止损线。AI 编程工具、AI 客服、AI 内容生成工具需要评估提效价值是不是大于模型花费。目标读者包括负责功能的后端研发、做平台工程的开发、负责成本预算的产品经理以及需要给团队制定“AI 使用规范”的技术负责人。一个常见误区是等账单异常后再去查实际上到了账单层只能看到总金额看不到是哪个功能、哪个会话、哪次提示词调整引起的增长。成本数据只有在调用发生时采集事后几乎无法补全。2. TokenSpend 数据模型设计一张事实表加四张维度表2.1 llm_usage 事实表记录每一次模型调用的完整账目成本分析的第一步是设计一张能够承载“单次模型调用明细”的表。这张表是后续所有分析的核心它必须是明细表而不是汇总表。汇总表只能回答“花了多少”回答不了“为什么花这么多”。CREATE TABLE llm_usage ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, request_id VARCHAR(64) NOT NULL COMMENT 业务侧或网关侧请求ID, trace_id VARCHAR(64) DEFAULT NULL, model VARCHAR(128) NOT NULL COMMENT 模型名称如 gpt-4o-mini, provider VARCHAR(64) NOT NULL DEFAULT openai, endpoint VARCHAR(128) NOT NULL COMMENT 实际调用场景或API路径, user_id VARCHAR(64) DEFAULT NULL, project_id VARCHAR(64) DEFAULT NULL, feature_id VARCHAR(64) DEFAULT NULL COMMENT 最小业务功能单元, session_id VARCHAR(64) DEFAULT NULL COMMENT 一次业务会话的唯一标识, prompt_tokens INT UNSIGNED NOT NULL DEFAULT 0, completion_tokens INT UNSIGNED NOT NULL DEFAULT 0, total_tokens INT UNSIGNED NOT NULL DEFAULT 0, cached_tokens INT UNSIGNED NOT NULL DEFAULT 0, prompt_price DECIMAL(10,8) NOT NULL DEFAULT 0 COMMENT 每千输入Token单价, completion_price DECIMAL(10,8) NOT NULL DEFAULT 0, cost_amount DECIMAL(14,6) NOT NULL DEFAULT 0 COMMENT 本次调用成本, currency VARCHAR(8) NOT NULL DEFAULT USD, latency_ms INT UNSIGNED DEFAULT NULL, finish_reason VARCHAR(32) DEFAULT NULL, is_success TINYINT(1) NOT NULL DEFAULT 1, status_code SMALLINT UNSIGNED DEFAULT NULL, error_code VARCHAR(64) DEFAULT NULL, tags JSON DEFAULT NULL COMMENT 额外标签如请求来源、模型通道, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_created_at (created_at), KEY idx_model (model), KEY idx_project_feature (project_id, feature_id), KEY idx_session (session_id), KEY idx_request_id (request_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT大模型调用明细表;这张表里有几个设计点需要解释第一既存 Token 数也存单价和金额。这样做的原因是模型价格会调整。如果只存 Token 数供应商降价后回看历史成本就没办法对应当时的账单如果只存金额后续想统计“某类模型的总 Token 消耗”就没有原始数据。两者并存历史口碑才能稳定。第二request_id 单独建索引。这个字段用于和网关日志、模型平台日志对账排查“某次调用为什么没记录”时会非常有用。第三is_success 和 error_code 也要记录。调用失败的请求同样消耗了成本而且可能是没有完成业务价值的浪费它和成功调用一样需要进入成本分析。2.2 维度表把成本归属到业务单元llm_usage 里的 project_id、feature_id、user_id、session_id 都是维度字段。严格来说可以不做成单独的关联表直接在公司内部统一维护一套编码即可。但在实际工程中建议至少维护 project 和 feature 两张维度表目的是保证每个业务单元有责任人、负责人和成本目标。CREATE TABLE project_dim ( project_id VARCHAR(64) PRIMARY KEY, project_name VARCHAR(128) NOT NULL, owner_team VARCHAR(64) NOT NULL, budget_quota DECIMAL(14,6) DEFAULT NULL, status TINYINT(1) NOT NULL DEFAULT 1, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE feature_dim ( feature_id VARCHAR(64) PRIMARY KEY, project_id VARCHAR(64) NOT NULL, feature_name VARCHAR(128) NOT NULL, owner VARCHAR(64) DEFAULT NULL, enable_roi TINYINT(1) NOT NULL DEFAULT 0, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP );维度设计最重要的原则是维度的值必须在调用埋点时就已经存在而不是等数据进数仓后再靠规则去猜。比如“这个调用属于哪个功能”在业务代码里应该是显式传入的。常见的反模式是只记录模型名称和 Token 数等做报表时再根据消息内容倒推功能这种做法在调用量上来以后根本不可行。2.3 价格表与收益事件表价格表用于把 Token 数组换算成金额。它需要支持按模型、按生效区间维护这样才能在供应商调整价格后仍然正确回算历史成本。CREATE TABLE model_price ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, provider VARCHAR(64) NOT NULL, model VARCHAR(128) NOT NULL, input_price_per_1k DECIMAL(10,8) NOT NULL, output_price_per_1k DECIMAL(10,8) NOT NULL, cached_input_price_per_1k DECIMAL(10,8) NOT NULL DEFAULT 0, currency VARCHAR(8) NOT NULL DEFAULT USD, effective_from DATETIME NOT NULL, effective_to DATETIME DEFAULT NULL, remark VARCHAR(255) DEFAULT NULL, UNIQUE KEY uk_model_version (model, effective_from) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT模型价目表;收益事件表用于记录业务侧发生的结果事件例如下单、转化、任务完成、客服工单关闭。CREATE TABLE revenue_event ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, project_id VARCHAR(64) NOT NULL, feature_id VARCHAR(64) NOT NULL, session_id VARCHAR(64) DEFAULT NULL, user_id VARCHAR(64) DEFAULT NULL, event_type VARCHAR(64) NOT NULL COMMENT order/conversion/task_done/ticket_closed, revenue_amount DECIMAL(14,6) NOT NULL DEFAULT 0, currency VARCHAR(8) NOT NULL DEFAULT USD, related_usage_ids JSON DEFAULT NULL COMMENT 关联的调用记录ID列表, occurred_at DATETIME NOT NULL, tags JSON DEFAULT NULL, KEY idx_occurred_at (occurred_at), KEY idx_feature (feature_id, event_type) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT收益事件表;related_usage_ids 字段是可选的。对于简单场景直接通过 session_id 把收益事件和调用记录关联起来就够了不一定要保存具体的 usage id 列表。保存列表的好处是归因精确坏处是数据量很大写入成本高。学习阶段可以先忽略 tags 和 related_usage_ids只保留最简单字段先把记录和分析跑通再把复杂维度加上去。3. 采集层实现把 token 消耗从响应体变成可查数据3.1 usage 字段是采集的起点无论用哪个模型提供商的 SDK响应体里基本都会包含 usage 信息。以最常见的 OpenAI 响应结构为例{ model: gpt-4o-mini, usage: { prompt_tokens: 1200, completion_tokens: 350, total_tokens: 1550, prompt_tokens_details: { cached_tokens: 800 } } }这些字段就是成本采集的原始数据。要注意的是不同厂商的字段名不完全一样有的叫 input_tokens、output_tokens有点则把缓存命中单独命名为 cache_read_input_tokens。实现采集层之前先整理一份自己常用模型 API 的 usage 字段映射表格式可以参考下面这样厂商输入 Token 字段输出 Token 字段总 Token 字段缓存 Token 字段OpenAIprompt_tokenscompletion_tokenstotal_tokensprompt_tokens_details.cached_tokensAnthropicinput_tokensoutput_tokensusage 聚合值cache_read_input_tokens其他平台以官方文档为准以官方文档为准以官方文档为准以官方文档为准字段没对齐之前不要急着建全局数据管道否则后面做成本换算时会出现大量口径不一致的脏数据。3.2 用 Python 装饰器包装 OpenAI SDK最直接的埋点方式是在业务调用 LLM SDK 的位置统一增加一个跟踪装饰器。下面是一个基于 OpenAI Python SDK 的最小示例import functools import time import uuid from openai import OpenAI from app.tracking import write_usage def track_llm_call(project_id, feature_id, user_idNone, session_idNone): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): begin time.time() try: response func(*args, **kwargs) latency_ms int((time.time() - begin) * 1000) usage getattr(response, usage, None) if usage is None: return response prompt_tokens getattr(usage, prompt_tokens, 0) or 0 completion_tokens getattr(usage, completion_tokens, 0) or 0 total_tokens getattr(usage, total_tokens, 0) or 0 details getattr(usage, prompt_tokens_details, None) cached_tokens 0 if details is not None: cached_tokens getattr(details, cached_tokens, 0) or 0 write_usage({ request_id: str(uuid.uuid4()), model: getattr(response, model, None), project_id: project_id, feature_id: feature_id, user_id: user_id, session_id: session_id, prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, total_tokens: total_tokens, cached_tokens: cached_tokens, latency_ms: latency_ms, is_success: True, }) return response except Exception as exc: write_usage({ request_id: str(uuid.uuid4()), model: None, project_id: project_id, feature_id: feature_id, user_id: user_id, session_id: session_id, prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, cached_tokens: 0, is_success: False, error_code: type(exc).__name__, }) raise return wrapper return decorator track_llm_call(project_idrobot-support, feature_idorder-status, user_idu_1001) def get_order_status(client: OpenAI, order_no: str): prompt build_prompt(order_no) return client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], )这个示例有几个关键点。失败调用也要写记录否则成功率的指标会虚高。异常分支里拿不到 usage所以 Token 数组填 0但 error_code 要保留用来排查“是不是某种提示词导致模型频繁拒绝调用”。不要在业务线程里同步等待数据库写入。上面示例中是调用了 write_usage 函数它内部应该只做入队操作真正写库放到后台批量任务里。采集逻辑越轻对业务链路的影响越小。3.3 Spring AI 场景下的采集方式Java 技术栈使用 Spring AI 时不建议在每一个业务方法里手动埋点。更合理的做法是通过 AOP 切面统一拦截 ChatClient 的调用从 ChatResponse 中读取 usage 信息。Aspect Component public class LlmUsageAspect { private final TokenUsageRepository repository; public LlmUsageAspect(TokenUsageRepository repository) { this.repository repository; } Around(execution(* org.springframework.ai.chat.client.ChatClient.CallResponseSpec.call(..))) public Object recordTokenUsage(ProceedingJoinPoint pjp) throws Throwable { long begin System.currentTimeMillis(); try { Object result pjp.proceed(); long latencyMs System.currentTimeMillis() - begin; if (result instanceof ChatResponse response response.getMetadata() ! null response.getMetadata().getUsage() ! null) { ChatUsage usage response.getMetadata().getUsage(); repository.save(toRecord(usage, latencyMs, true)); } return result; } catch (Throwable t) { repository.save(toRecord(null, 0, false)); throw t; } } private LlmUsageRecord toRecord(ChatUsage usage, long latencyMs, boolean success) { // 将 usage 字段映射到自己的持久化对象 return new LlmUsageRecord(); } }Spring AI 的 API 在不同版本里差异比较大ChatResponse、ChatUsage 这些类的包名和字段可能发生变化。落地前要找到当前项目使用的版本文档确认 usage 的读取方式。切面表达式的匹配范围也建议只拦截自己项目里真正使用的 Client Bean避免误拦截。3.4 异步上报和批量写入采集层最需要考虑的是性能。每调用一次模型就同步执行一次 INSERT在高并发场景下会拖垮接口响应。推荐的做法是内存队列加批量写入。import logging import queue import threading import time logger logging.getLogger(tokenspend) usage_queue queue.Queue(maxsize10000) def write_usage(record: dict) - None: try: usage_queue.put_nowait(record) except queue.Full: logger.warning(usage queue is full, record dropped) def batch_worker() - None: batch [] while True: try: item usage_queue.get(timeout2) batch.append(item) except queue.Empty: if batch: flush_batch(batch) batch [] else: if len(batch) 100: flush_batch(batch) batch [] def flush_batch(batch) - None: # 按项目实际情况批量插入数据库 pass # 应用启动时开启后台线程 threading.Thread(targetbatch_worker, daemonTrue).start()队列满时直接丢弃记录是一个权衡方案。生产环境建议把丢弃行为替换成“降级到文件日志”由独立采集进程扫描补写避免成本数据丢失。学习环境只需要保证同步入库基本可用即可。4. 成本换算与分摊Token 数量不等于真实费用4.1 按模型维度拆分定价Token 数是原始数据但要变成真正的成本必须乘以单价。不同模型的价格差异很大同一个模型的输入和输出价格也可能差数倍加上缓存命中的 Token 通常更便宜所以不能用一套固定比例去换算。正规做法是维护一张按模型、按生效时间区间的价格表。每次写入 llm_usage 时根据当时的模型和调用时间查询生效价格计算出 cost_amount 一起落库。这样即使后续价格调整历史记录的金额也不会被覆盖。4.2 一次调用的成本计算示例假设某个模型的价格配置如下示例价格实际落地时以官方计价文档为准项单价每千 Token普通输入 Token0.150缓存命中输入 Token0.075输出 Token0.600一次调用中prompt_tokens 为 1200其中 cached_tokens 为 600completion_tokens 为 300。那么成本计算方式为1200 - 600 / 1000 * 0.150 600 / 1000 * 0.075 300 / 1000 * 0.600 0.090 0.045 0.180 0.315 美元用代码实现就是def calculate_cost(model, prompt_tokens, completion_tokens, cached_tokens, price_config): price price_config[model] normal_prompt prompt_tokens - cached_tokens normal_cost normal_prompt / 1000 * price[input_per_1k] cached_cost cached_tokens / 1000 * price[cached_input_per_1k] completion_cost completion_tokens / 1000 * price[output_per_1k] return round(normal_cost cached_cost completion_cost, 6)这里最容易出错的是 cached_tokens 大于 prompt_tokens 的情况。某些 SDK 返回的 cached_tokens 可能已经包含在 prompt_tokens 中也可能返回的是独立的 cache 创建费用含义不同计算前先确认厂商字段语义。4.3 多业务归属的分摊逻辑一个会话可能同时服务多个业务目标。比如一条用户消息既做了意图识别又生成了最终回复这两个动作算在哪个功能头上会直接影响 ROI 报表结论。实际项目里常用三种分摊方式分摊方式适用场景实现成本主功能全额承担调用有明确主目的低按功能比例拆分几个业务方共用一次调用中按会话整体归因一个会话对应一个业务结果低建议从“主功能全额承担”开始只有当同一个请求确实被多个部门复用时才增加比例拆分逻辑。分摊规则要写入文档避免一次调用同时出现在两个功能的报表里造成总成本翻倍。4.4 价格变更后的重算策略供应商调价后历史数据怎么处理主要看团队想回答什么问题如果只关心“当前模型组合下老业务的成本大概是多少”可以用新价格重算历史数据。如果关心“过去某个月实际花了多少钱”必须保留当时的单价不能用新价格覆盖。所以采集时保存单价快照是关键一步。价格表里使用 effective_from 和 effective_to 区间查询时用调用时间匹配当时的生效价格既不影响历史报表也能按需重算。5. 分析层让成本和收益进入同一个坐标系5.1 定义核心指标成本记录只解决了“花了多少”ROI 分析还需要定义一系列可执行的指标。下面这组指标适合大多数 AI 功能团队指标定义用途总成本指定周期内所有 LLM 调用的成本总量控制与预算对比每流程成本单个业务流程的平均 Token 和金额判断流程是否依然划算调用成功率成功调用数 / 总调用数排查无效消耗平均每千 Token 成本总成本 / 总 Token * 1000监控模型组合变化收益金额与 AI 功能直接相关的订单、转化或节约收益侧数据ROI收益 - 成本 / 成本判断该功能是否值得继续建设指标定义要在第一周就固定下来不要频繁修改口径。口径一变历史报表全部失去可比性。5.2 用 SQL 完成成本和调用量聚合明细表建好以后第一张有价值的报表是“按功能维度的成本排行”。下面是按 feature 和 model 分组的近 7 天聚合SELECT feature_id, model, COUNT(*) AS call_count, SUM(prompt_tokens) AS prompt_tokens, SUM(completion_tokens) AS completion_tokens, SUM(total_tokens) AS total_tokens, SUM(cost_amount) AS total_cost, ROUND( SUM(cost_amount) * 1000 / NULLIF(SUM(total