企业官网建设流程全解析

做 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 上，再通过价格表换算成真实费用。

这里有几个基础概念需要先区分清楚：

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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 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 字段映射表，格式可以参考下面这样：

字段没对齐之前，不要急着建全局数据管道，否则后面做成本换算时会出现大量口径不一致的脏数据。

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_id=None, session_id=None): 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_id="robot-support", feature_id="order-status", user_id="u_1001") def get_order_status(client: OpenAI, order_no: str): prompt = build_prompt(order_no) return client.chat.completions.create( model="gpt-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(maxsize=10000) 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(timeout=2) 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(target=batch_worker, daemon=True).start()

队列满时直接丢弃记录是一个权衡方案。生产环境建议把丢弃行为替换成“降级到文件日志”，由独立采集进程扫描补写，避免成本数据丢失。学习环境只需要保证同步入库基本可用即可。

4. 成本换算与分摊：Token 数量不等于真实费用

4.1 按模型维度拆分定价

Token 数是原始数据，但要变成真正的成本，必须乘以单价。不同模型的价格差异很大，同一个模型的输入和输出价格也可能差数倍，加上缓存命中的 Token 通常更便宜，所以不能用一套固定比例去换算。

正规做法是维护一张按模型、按生效时间区间的价格表。每次写入 llm_usage 时，根据当时的模型和调用时间查询生效价格，计算出 cost_amount 一起落库。这样即使后续价格调整，历史记录的金额也不会被覆盖。

4.2 一次调用的成本计算示例

假设某个模型的价格配置如下（示例价格，实际落地时以官方计价文档为准）：

一次调用中，prompt_tokens 为 1200，其中 cached_tokens 为 600，completion_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 功能团队：

指标定义要在第一周就固定下来，不要频繁修改口径。口径一变，历史报表全部失去可比性。

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