大模型成本监控：记录每次推理的 Token 与费用

cost = (input_tokens / 1e6) * input_price + (output_tokens / 1e6) * output_price

因此，完整的成本监控必须记录每一次推理的**输入 Token 数、输出 Token 数、模型名称、推理时间戳**，并动态地将其转换为费用。

### 输入 Token 与输出 Token 的差异
- **输入 Token**：包括系统提示、用户消息、上下文历史、函数定义等  
- **输出 Token**：模型生成的文本或函数调用参数  
- 通常输出 Token 的价格是输入 Token 的 2~5 倍，长回复可能比短回复贵 10 倍以上

### 不同模型定价对比示例（2025 年初）

| 模型 | 输入价格/百万 Token | 输出价格/百万 Token |
|------|---------------------|---------------------|
| GPT-4o | $2.50 | $10.00 |
| GPT-4o-mini | $0.15 | $0.60 |
| Claude 3.5 Sonnet | $3.00 | $15.00 |
| Gemini 1.5 Pro | $1.25 | $5.00 |

一次典型的 GPT-4o 调用，输入约 2000 Token、输出约 500 Token，费用为：

(2000/1e6)*2.50 + (500/1e6)*10.00 = $0.005 + $0.005 = $0.01

看似微小，但如果每天有 10 万次调用，日成本就达到 $1000。

## 从头实现成本监控：日志结构与价格计算

### 标准日志字段

无论采用什么记录方式，每条推理日志至少应包含：

```json
{
  "timestamp": "2025-03-20T14:32:05.123Z",
  "model": "gpt-4o",
  "input_tokens": 1847,
  "output_tokens": 312,
  "input_cost": 0.0046175,
  "output_cost": 0.00312,
  "total_cost": 0.0077375,
  "api_endpoint": "/v1/chat/completions",
  "user_id": "u_123",
  "session_id": "s_abc"
}

• input_cost 和 output_cost 按照当时定价实时计算并写入日志
• 存储时可以考虑类型（如 Decimal 或毫分制整数），避免浮点误差积累

价格动态获取

不同模型、不同供应商的价格会变动，绝不能硬编码。应当维护一个价格配置表，并定期更新。示例 Python 实现：

PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
}

def calculate_cost(model, input_tokens, output_tokens):
    pricing = PRICING.get(model)
    if not pricing:
        raise ValueError(f"Unknown model: {model}")
    input_cost = (input_tokens / 1e6) * pricing["input"]
    output_cost = (output_tokens / 1e6) * pricing["output"]
    return round(input_cost + output_cost, 8)

日志存储方案

即便使用 LangChain 等框架，你仍需要显式地从回调中提取 Token 用量，并写入自己的监控系统，以防平台只统计部分数据或产生延迟。

维度化监控：不要只看总成本

仅仅记录总费用，不足以做出优化决策。必须为每条记录打上业务标签，才能回答关键问题。

必加标签维度

• 模型：区分不同模型版本的用量与成本
• 用户/组织：按租户分摊成本，用于计费或额度限制
• 功能/场景：例如“客服摘要”、“报告生成”、“代码审查”
• 来源：API、Web 应用、内部工具、自动化测试
• 会话 ID：追踪单次对话的完整 Token 消耗

示例高阶 SQL 查询：

SELECT 
  function_tag,
  SUM(total_cost) as cost_7d
FROM llm_inference_log
WHERE timestamp > now() - INTERVAL '7 days'
GROUP BY function_tag
ORDER BY cost_7d DESC;

成本异常检测

应设置按小时的成本告警，例如：

• 单小时成本超过过去 7 天同时间均值的 2 倍
• 单个用户 1 小时内消耗超过 $10
• 平均单次调用成本超过预设阈值（提示词可能突然变长）

这些规则可以通过定时任务扫描日志实现，或接入 Prometheus/Grafana 体系。

常见陷阱与调优实践

1. 忽略计费精度与舍入误差

• 每百万 Token 定价的小数点后通常有 2~4 位
• 累积百万次调用后，浮点误差可能导致账单偏差
• 推荐使用 Decimal 类型或以千分之一美分为单位的整数存储

2. 只记录成功调用

失败的推理也会消耗 Token（部分模型在返回错误前已产生输入 Token 费用）。务必记录包含错误码的日志，并统计失败请求的花费。

3. 使用流式输出时 Token 统计不准确

流式响应需要累积所有分片的 usage 字段。某些 SDK 的流式调用不会自动返回最终 Token 数量，必须显式读取最后一块数据。

4. 多模型组合场景的分配

如果一次用户请求内部调用了多个模型（如先用廉价模型分类，再用强模型回答），需要在日志中区分每个子调用，并通过跟踪 ID 关联。

5. 忽略上下文管理带来的隐性增长

随着对话变长，每次请求都会携带历史消息，输入 Token 数指数级上升。监控时应关注 平均输入 Token 数趋势，并设置截断或总结策略。

低成本快速启动方案

如果你现在就需要一个可用的监控方案，可以按以下步骤 30 分钟内搭建：

1. 创建一个 JSON Lines 文件作为日志 cost.log
2. 封装一个函数 log_usage(model, input_tokens, output_tokens, tags)，计算成本并追加写入
3. 在你的大模型调用代码返回后立即调用该函数
4. 用简单的 Python 脚本每日汇总：

import json
from collections import defaultdict

daily_cost = defaultdict(float)
with open('cost.log') as f:
    for line in f:
        record = json.loads(line)
        date = record['timestamp'][:10]
        daily_cost[date] += record['total_cost']

for date, cost in sorted(daily_cost.items()):
    print(f"{date}: ${cost:.2f}")

1. 当调用量增长时，将日志迁入 SQLite 或 PostgreSQL，并构建 Grafana 面板。

进阶：构建实时成本看板

当调用量达到每分钟数千次时，你需要实时可观测性。

数据管道设计

API 调用 → 结构化日志 → 消息队列（Kafka/Redis Stream）→ 流处理（计算分钟聚合）→ 时序数据库 → Grafana