AI应用成本失控？TokenSpend架构设计与最小实现解析

最近一段时间在推进 AI 应用落地时有一个问题比模型选型更让人头疼每次调大模型 API 都是“看不见的支出”等到月底账单出来才发现成本已经分散到了各个项目、各种调试脚本和无数轮实验里。团队里如果有人问“这个 AI 功能到底花了多少钱、带来了多少收益”往往很难立刻给出准确答案。本文要聊的 TokenSpend正是一个面向 AI 应用成本的 ROI 解决方案。接下来我会把它拆开来看它的核心概念是什么解决什么问题技术上是如何设计的以及我们自己动手做一个最小可用版本需要哪些关键步骤。文章会包含完整代码示例和落地避坑经验适合正在做 AI 应用开发、负责 AI 平台建设、或者负责成本治理的开发者阅读。1. AI 成本管理为什么 Token 需要被“看见”1.1 从账单困境说起传统后端服务的成本很直观一台云服务器多少钱、一个数据库实例多少钱采购时基本就能估算出来。但大模型应用不一样成本是按 token 消耗实时产生的。一个用户对话可能消耗几百 token一个长文档总结可能消耗几万 token一次批量处理任务可能直接烧掉几十万 token。这种模式下成本与你写的代码逻辑、用户输入长度、模型版本、上下文窗口大小全都耦合在一起。更麻烦的是同一个项目可能同时接入 OpenAI、Anthropic、通义千问、DeepSeek 等多个模型每个模型的价格体系和计费单位还不一样。如果没有一个统一的计量体系财务看账单只能看到“某月 AI 费用 XX 万元”但完全说不清钱花在哪个业务线上。1.2 TokenSpend 是什么TokenSpend 的定位很清晰它是围绕 AI 调用成本设计的可观测与投资回报分析平台。简单来说它做的事情可以概括为四步采集每一次大模型调用的 token 消耗数据将费用归因到项目、团队、用户、业务场景实时统计成本并比对预算结合业务产出数据计算 ROI给出“这笔 AI 花销值不值”的结论。它的核心价值不是把账单做得更漂亮而是让 AI 成本从“事后震惊”变成“事前可控、事中可观测、事后可复盘”。1.3 使用 TokenSpend 的典型场景多项目成本分摊公司同时跑着客服问答、内容生成、代码助手、数据分析等多个 AI 应用需要把一份总账单拆到各个项目头上。预算实时管控某个测试环境的 key 被异常高频调用需要第一时间发现并熔断。模型成本对比同一业务用 GPT-4o 和 GPT-4o-mini 分别跑成本差异到底有多大效果差了多远。ROI 评估一个 AI 功能上线后节省了多少人工成本、提升了多少转化率需要和 token 成本放在同一张表里看。2. Token 成本归因要解决的核心难题2.1 Token 的两段式计费几乎所有主流大模型 API 都会把 token 消耗拆成两部分输入 tokenprompt tokens和输出 tokencompletion tokens。输入通常包含系统提示词、历史对话、工具定义、参考文档输出就是模型生成的文本内容。两部分的单价往往不同输出 token 通常更贵。这意味着成本统计不能只看一个总数。你在做埋点或者日志采集时必须把 prompt_tokens 和 completion_tokens 分开记录。如果只记总 token 数后续价格调整或者成本分摊都会出问题。2.2 归因维度怎么设计一次标准的模型调用除了 token 数量之外还需要记录以下信息维度示例价值项目智能客服、内容助手成本分摊到业务线模型gpt-4o、claude-3-5-sonnet对比模型性价比场景在线问答、离线批处理识别高成本场景用户/团队产品部、研发部内部账单核算请求来源Web、移动端、内部脚本定位异常消耗业务标识订单号、会话 ID关联业务产出这些维度需要在上游请求链路中提前注入。最简单的方式是在网关层统一封装从 header 或参数中读取 project、user、scene 等字段再透传到日志系统。2.3 单价管理的难点模型价格并不是一成不变的。OpenAI、Anthropic 等厂商会调整定价同一个模型在不同时期、不同账号类型下的价格可能不同。如果在代码里写死价格表后期维护成本很高。更合理的做法是把价格表独立成配置按时间版本管理。计算成本时根据调用发生的时间段选择对应版本的单价。这部分在自建方案时可以做成一张 price_versions 表记录 model、prompt_unit_price、completion_unit_price、effective_date 等字段。3. TokenSpend 技术架构与数据模型设计3.1 整体架构分层抛开具体的商业化实现一个完整的 AI 成本管理系统通常分成五层采集层通过 SDK 埋点、API 网关日志、云厂商账单导出等方式拿到原始调用记录。传输层使用消息队列如 Kafka、RabbitMQ或者日志管道如 Fluentd、Filebeat把数据集中起来。存储层明细数据存 OLTP 或列式数据库聚合数据存 OLAP 或时序数据库。计算层定时任务做小时级/天级聚合检测预算使用率计算趋势和 ROI。展示与告警层成本看板、预算进度条、异常告警、团队日报。对于中小团队链路可以简化应用服务直接异步上报调用记录到 ClickHouse 或 MySQL通过定时任务完成聚合。3.2 明细表结构设计明细表是成本分析的基础。以 MySQL 为例一张核心的调用明细表可以这样设计CREATE TABLE llm_call_records ( id BIGINT NOT NULL AUTO_INCREMENT, call_id VARCHAR(64) NOT NULL COMMENT 调用唯一 ID, project_id VARCHAR(64) NOT NULL COMMENT 项目 ID, user_id VARCHAR(64) DEFAULT COMMENT 用户标识, scene VARCHAR(64) DEFAULT COMMENT 业务场景, model_name VARCHAR(64) NOT NULL COMMENT 模型名称, prompt_tokens INT NOT NULL DEFAULT 0, completion_tokens INT NOT NULL DEFAULT 0, total_tokens INT NOT NULL DEFAULT 0, cost_usd DECIMAL(12, 6) NOT NULL DEFAULT 0 COMMENT 成本单位美元, created_at DATETIME NOT NULL COMMENT 请求时间, PRIMARY KEY (id), KEY idx_project_time (project_id, created_at), KEY idx_model_time (model_name, created_at) ) ENGINE InnoDB DEFAULT CHARSET utf8mb4;这里有几个需要注意的点cost_usd字段可以在写入明细时就算好方便后续查询。但如果价格发生变化历史数据是否需要重算需要事先约定。索引要结合真实查询习惯。大多数场景是“按项目看时间趋势”“按模型看占比”所以复合索引按照这个方向建。明细表数据量会很大建议按月份做分区保留近 6 个月的明细更早的数据转存归档表或对象存储。3.3 聚合表设计明细表不适合频繁做 group by 查询尤其是当数据量达到千万级以上时。更常用的方案是维护一张天级聚合表CREATE TABLE daily_call_stats ( stat_date DATE NOT NULL, project_id VARCHAR(64) NOT NULL, model_name VARCHAR(64) NOT NULL, scene VARCHAR(64) DEFAULT , call_count INT NOT NULL DEFAULT 0, prompt_tokens BIGINT NOT NULL DEFAULT 0, completion_tokens BIGINT NOT NULL DEFAULT 0, total_tokens BIGINT NOT NULL DEFAULT 0, cost_usd DECIMAL(12, 6) NOT NULL DEFAULT 0, PRIMARY KEY (stat_date, project_id, model_name, scene) );定时任务在每天凌晨跑一次聚合把前一天的数据从明细表汇总到这张表。日常报表、看板查询都走聚合表性能会好很多。4. 亲自实现一个最小版 TokenSpend接下来我们写一个可运行的最小版 AI 成本统计服务。技术栈使用 Python 标准库加少量第三方依赖重点是理解整个成本计算和 ROI 评估的逻辑闭环。4.1 项目结构token-spend-demo/ ├── main.py # 入口模拟采集 统计报表 ├── core.py # 成本计算、数据结构 ├── analyzer.py # 聚合分析与 ROI 预估 └── sample_data.py # 模拟数据生成4.2 定义基础数据结构和价格表# core.py from dataclasses import dataclass, asdict from datetime import datetime from typing import Optional dataclass class LLMCall: call_id: str project: str model: str prompt_tokens: int completion_tokens: int scene: str user: str created_at: datetime property def total_tokens(self) - int: return self.prompt_tokens self.completion_tokens # 价格表单位美元 / 1M tokens # 注意价格会根据厂商调整生产环境建议放配置中心 PRICE_TABLE { gpt-4o: { prompt: 2.50, completion: 10.00, }, gpt-4o-mini: { prompt: 0.15, completion: 0.60, }, claude-3-5-sonnet: { prompt: 3.00, completion: 15.00, }, } def calc_cost_usd(call: LLMCall) - float: 根据模型价格表计算单次调用成本美元 model_price PRICE_TABLE.get(call.model) if not model_price: return 0.0 prompt_cost call.prompt_tokens * model_price[prompt] / 1_000_000 completion_cost call.completion_tokens * model_price[completion] / 1_000_000 return prompt_cost completion_cost def enrich_call(call: LLMCall) - dict: 将调用记录转换为带成本字段的字典方便后续持久化 record asdict(call) record[total_tokens] call.total_tokens record[cost_usd] round(calc_cost_usd(call), 6) return record在这个函数中我们把 prompt_tokens 和 completion_tokens 分开乘以对应的百万 token 单价得到的就是一次调用的美元成本。这个计算逻辑是整个 TokenSpend 的基石后面所有看板、告警、ROI 分析都要依赖它。4.3 生成模拟数据为了让演示可以直接跑起来我写了一个简单的 mock 数据生成器模拟多个项目、多个模型在几天内的调用情况。# sample_data.py import random import uuid from datetime import datetime, timedelta from core import LLMCall PROJECTS [智能客服, 内容助手, 数据分析, 代码助手] MODELS [gpt-4o, gpt-4o-mini, claude-3-5-sonnet] SCENES [在线问答, 离线批处理, 摘要生成, 意图识别] USERS [alice, bob, carol, david, eve] def random_call(dt: datetime) - LLMCall: model random.choice(MODELS) prompt_tokens random.randint(200, 8000) completion_tokens random.randint(50, 4000) return LLMCall( call_idstr(uuid.uuid4()), projectrandom.choice(PROJECTS), modelmodel, prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, scenerandom.choice(SCENES), userrandom.choice(USERS), created_atdt, ) def generate_sample_calls(days: int 7, calls_per_day: int 200) - list[LLMCall]: calls [] now datetime.now() for day_offset in range(days, 0, -1): day now - timedelta(daysday_offset) for _ in range(calls_per_day): dt day timedelta( hoursrandom.randint(9, 20), minutesrandom.randint(0, 59), secondsrandom.randint(0, 59), ) calls.append(random_call(dt)) return calls这段数据生成器生产环境用不到但用来本地演示成本统计流程非常方便。实际项目中这里应该被替换为真实的 SDK 埋点数据或网关日志。4.4 聚合分析与成本报表接下来是我们最关心的部分按项目、按模型聚合成本并生成一张简单的成本报表。# analyzer.py from collections import defaultdict from datetime import date from core import LLMCall, enrich_call class CostAnalyzer: def __init__(self, calls: list[LLMCall]): self.records [enrich_call(call) for call in calls] def daily_report(self) - dict: 按日期统计调用量、token 量、成本 daily defaultdict(lambda: { call_count: 0, total_tokens: 0, cost_usd: 0.0, }) for record in self.records: key record[created_at].date().isoformat() daily[key][call_count] 1 daily[key][total_tokens] record[total_tokens] daily[key][cost_usd] record[cost_usd] return dict(sorted(daily.items())) def project_report(self) - list[dict]: 按项目聚合成本输出成本占比 project_cost defaultdict(float) for record in self.records: project_cost[record[project]] record[cost_usd] total_cost sum(project_cost.values()) report [] for project, cost in sorted(project_cost.items(), keylambda item: -item[1]): report.append({ project: project, cost_usd: round(cost, 4), ratio: f{cost / total_cost * 100:.2f}%, }) return report def model_report(self) - list[dict]: 按模型聚合成本用于对比不同模型的开销 model_cost defaultdict(float) model_calls defaultdict(int) for record in self.records: model_cost[record[model]] record[cost_usd] model_calls[record[model]] 1 report [] for model, cost in sorted(model_cost.items(), keylambda item: -item[1]): report.append({ model: model, call_count: model_calls[model], cost_usd: round(cost, 4), avg_cost_per_call: round(cost / model_calls[model], 6), }) return report这里我用三个方法分别从时间、项目、模型三个维度呈现成本。日常使用中最关注“各项目成本排序”它可以直接回答“钱主要花在哪”。4.5 主程序运行# main.py from sample_data import generate_sample_calls from analyzer import CostAnalyzer def main(): calls generate_sample_calls(days7, calls_per_day200) analyzer CostAnalyzer(calls) print( 每日成本趋势 ) for day, stats in analyzer.daily_report().items(): print(f{day}: 调用 {stats[call_count]} 次, token {stats[total_tokens]}, 成本 ${stats[cost_usd]:.4f}) print(\n 项目成本占比 ) for row in analyzer.project_report(): print(f{row[project]}: ${row[cost_usd]:.4f}占比 {row[ratio]}) print(\n 模型成本对比 ) for row in analyzer.model_report(): print(f{row[model]}: 调用 {row[call_count]} 次, 成本 ${row[cost_usd]:.4f}, f单次均摊 ${row[avg_cost_per_call]:.6f}) if __name__ __main__: main()直接运行python main.py可以看到类似下面的输出 每日成本趋势 2025-01-10: 调用 200 次, token 1183678, 成本 $1.8342 2025-01-11: 调用 200 次, token 1067842, 成本 $1.5210 ... 项目成本占比 内容助手: $3.2041占比 29.31% 智能客服: $2.8755占比 26.30% 数据分析: $2.4207占比 22.14% 代码助手: $2.4322占比 22.25% 模型成本对比 claude-3-5-sonnet: 调用 509 次, 成本 $4.1562, 单次均摊 $0.008167 gpt-4o: 调用 476 次, 成本 $3.9820, 单次均摊 $0.008366 gpt-4o-mini: 调用 415 次, 成本 $1.2342, 单次均摊 $0.002974可以看到分析结果很直观地展示了各项目和模型的成本分布。这就是一个简化版 TokenSpend 的核心统计能力。5. 预算控制与告警让成本不失控5.1 分级预算机制仅仅做成本统计还不够真正的 AI ROI 解决方案必须包含“成本控制”能力。常见做法是给项目设置月度预算并按照使用率分阶段告警。预算使用率 50%发送提示提醒项目负责人关注。预算使用率 80%发送预警建议检查是否有异常调用。预算使用率 100%触发强限制可以降级到低成本模型或者直接熔断非核心调用。预算模型很简单可以放在 Redis 或者 MySQL 中BUDGET_TABLE { 智能客服: 100.0, # 月预算单位美元 内容助手: 80.0, 数据分析: 60.0, 代码助手: 50.0, }检查逻辑# budget.py from analyzer import CostAnalyzer def check_budget(analyzer: CostAnalyzer, budget_table: dict) - list[dict]: project_cost {row[project]: row[cost_usd] for row in analyzer.project_report()} alerts [] for project, budget in budget_table.items(): used project_cost.get(project, 0.0) usage used / budget * 100 if budget else 0 if usage 100: level BLOCK elif usage 80: level WARNING elif usage 50: level NOTICE else: level OK alerts.append({ project: project, used: round(used, 4), budget: budget, usage: f{usage:.2f}%, level: level, }) return alerts实际项目中升级到告警动作时应该发送到钉钉、飞书、企业微信或者 Slack Webhook。这里注意熔断操作权限要严格控制避免自动化脚本误伤线上核心链路。5.2 异常调用检测AI 成本失控最常见的几个原因没有设置max_tokens模型无限输出。循环里没有判断上下文长度对话历史无限膨胀。测试脚本用了高配额 key并被反复执行。爬虫或攻击者拿到了未鉴权的 API key疯狂调用。针对这些场景可以在网关层做简单的调用频率和单次 token 量限制。单次 total_tokens 超过阈值的请求应该被记录并标记为异常。6. ROI 计算从“花了多少钱”到“值不值”6.1 简单 ROI 模型成本统计只是 TokenSpend 的一半另一半是 ROI 分析。ROI 的定义不复杂ROI (业务收益 - 成本) / 成本难点在于“业务收益”怎么量化。不同场景的收益指标完全不同客服场景节约了多少人力成本、提升了多少问题解决率。内容助手提升了多少内容生产速度、带来了多少流量增长。数据分析助手节省了多少数据分析师工时。代码助手提升了多少开发效率、减少了多少 bug 修复时间。我们可以把业务产出数据外挂到成本系统做一个简单的关联统计。在代码中可以维护一张业务产出表# roi.py # 业务产出数据项目 - 当月节省工时或收益金额 BUSINESS_VALUE { 智能客服: 200.0, # 节省人工成本 内容助手: 150.0, 数据分析: 120.0, 代码助手: 180.0, } def calc_roi(project_cost: dict, business_value: dict) - list[dict]: results [] for project, value in business_value.items(): cost project_cost.get(project, 0.0) if cost 0: continue roi (value - cost) / cost * 100 results.append({ project: project, cost: round(cost, 4), value: value, roi: f{roi:.2f}%, }) return sorted(results, keylambda row: -float(row[roi].strip(%)))在实际产品中业务收益数据不会像演示代码这样硬编码而是由业务系统通过 API 上报或者从数据仓库中拉取。核心思路是成本侧数据来自 TokenSpend收益侧数据来自业务系统两者在统一维度上做关联计算。6.2 成本效率指标除了 ROI还有几个值得关注的效率指标每千 token 成本cost / total_tokens * 1000可以横向对比不同模型。单次会话成本总成本 / 会话数用于评估对话场景的盈利能力。有效 token 比率completion_tokens / (prompt_tokens completion_tokens)这个值过高说明输出太长可能浪费过低说明上下文太长可能有大量无用信息灌入。这些指标不需要额外采集数据只要明细字段完整聚合时就能顺手算出来。7. 常见问题与排查思路在搭建 AI 成本管理系统时最常遇到的问题集中在数据采集、价格计算、预算告警三个方面。下面整理成一张排查表。问题现象常见原因解决思路统计成本与账单差异大prompt/completion 拆分错误或价格表过期统一按官方返回的 usage 字段记录价格表按时间版本管理看不到某个项目的成本调用链路未传递 project 字段在网关层强制读取并校验 project缺省值设为 unknown预算告警一直没触发告警检查任务没定时跑或使用率计算口径不对确认定时任务执行时间检查分母是月度预算还是实时预算单次调用 token 数异常高上下文无限增长或超长文档被重复塞入在应用层限制最大上下文为单次调用设置 token 上限明细表查询越来越慢数据量增长索引失效按月份分区定期归档历史数据报表走聚合表成本计算出 NaN 或负数价格表字段异常或存在除零对 token 和价格字段做非负校验异常数据跳过并告警在排查这类问题时建议先从明细表随机抽几条记录人工核对模型的 usage 字段和最终成本确认基础计算没有偏差再往上排查聚合和报表逻辑。8. 生产落地工程化建议8.1 埋点方式选择最小演示版是直接在调用完成后同步计算成本并上报。生产环境推荐异步化避免在请求链路中引入额外的网络延迟。如果大模型调用是通过公司内部 API 网关统一转发的可以在网关层做统一日志采集业务方只需要在请求 header 中带上项目、场景等元信息。这种方式接入成本最低也最容易形成规范。如果是 SDK 直连云厂商那么更推荐使用 OpenTelemetry 或类似框架在调用前后记录 token 用量和元数据统一导出到日志系统。8.2 数据安全与权限AI 调用数据里可能包含用户输入内容。虽然成本系统主要关心 token 数量和费用但日志采集时很可能会顺手把请求正文和响应正文一起记录下来。这里必须注意默认不持久化完整输入输出除非有明确的合规评估。需要保存样本时必须先脱敏并且设置访问权限。成本报表的查询权限要按角色控制财务、项目负责人、研发看到的数据范围应该不同。涉及生产环境变更时先在测试环境验证遵循最小权限原则。8.3 成本为重算机制模型厂商调价之后如何修正历史成本是一个容易被忽略的问题。我的建议是新产生的调用使用新价格。历史数据是否重算取决于业务需求。如果只需要看“未来成本趋势”不需要重算如果要出同比环比报告则需要按价格版本统一重算。做重算时更新聚合表而不是直接改明细表避免在百万级明细上全表 update。8.4 告警的动作设计告警不是发一条消息就结束了。一个合格的告警链路应该包括触发条件预算使用率超过阈值、单日成本环比突增、某个 key 调用频率异常。通知对象项目负责人、平台管理员。附加上下文趋势图、当前 TOP 调用方、异常检测结果。处置动作标记阻塞、降级模型、暂时禁用 key。这里特别说明任何自动熔断动作都必须有手动恢复通道并且要设置冷却时间防止系统在“熔断-恢复”之间反复抖动。9. 总结与下一步学习方向本文从 AI 应用成本失控这个痛点出发完整拆解了 TokenSpend 这类 AI ROI 解决方案的核心设计token 采集、成本计算、预算告警、ROI 评估。并且用 Python 实现了一个最小可运行版本包括调用记录模型、价格表、聚合统计、预算检查和 ROI 计算。整体看下来这套方案的实现难度并不高真正的挑战在于数据规范、价格版本管理和跨部门协同。如果你准备在自己的团队或项目中落地类似能力我的建议是分三步走先做“看得见”统一请求网关采集所有模型调用的 usage 和元数据。再做“管得住”接预算模型和分级告警把成本异常提前暴露出来。最后做“算得清”接入业务收益数据逐步形成项目级别的 ROI 报表。下一步可以继续深入学习的方向包括ClickHouse 在大规模日志存储上的应用、Apache Calcite 或 Presto 做多维 SQL 分析、OpenTelemetry 的模型调用链路追踪以及如何结合网关做更精细的 token 级限流。把这些串起来就是一套可以在生产环境长期运行的 AI 成本治理基础设施。