Token用量统计

要点

  • 大模型按 Token 计费，响应里的 usage 字段就是账单数据，不记录等于把钱扔掉了
  • Token 统计同时服务于计费、限流、成本分析和异常检测
  • KV 适合存时间窗口内的预聚合计数，关键是 key 的设计和 TTL 策略
  • 统计模块只负责提供数据，计费规则不应该写在统计模块里
  • waitUntil 让统计写入不阻塞主流程
  • 月度和长期数据通过 Cron Trigger 聚合后存入 D1，KV 只管近期高频读写

1. Token 用量统计在解决什么问题

前面的文章讲了怎么调用大模型、怎么管理会话、怎么做上下文截断。这些环节都有一个共同的输入：Token 数量。

大模型按 Token 收费。一次请求消耗的 Token = 输入 Token + 输出 Token，不同模型单价不同。调用 100 次 gpt-4o 和 100 次 gpt-4o-mini，成本差几十倍。如果你的服务对外提供 API，不统计每个用户的 Token 用量，就没办法回答：

  1. 这个用户这个月花了多少钱
  2. 哪些模型消耗了大部分预算
  3. 某个用户的用量突然翻倍，是正常增长还是接口被刷了

这些问题在日活十几个人的时候可以忽略，日活过千之后就变成运营硬约束。

2. 从模型响应中提取 usage

大模型 API 的响应自带用量数据：

// response.json
{
  "model": "gpt-4o",
  "choices": [{ "message": { "role": "assistant", "content": "..." } }],
  "usage": { "prompt_tokens": 156, "completion_tokens": 89, "total_tokens": 245 }
}

提取逻辑做一层防御性处理，模型 API 偶尔返回不完整的响应：

// src/stats/token-stats.ts
type TokenUsage = { promptTokens: number; completionTokens: number; totalTokens: number }
 
function extractUsage(response: unknown): TokenUsage | null {
  const usage = (response as any)?.usage
  if (!usage || typeof usage.prompt_tokens !== 'number') return null
  return {
    promptTokens: usage.prompt_tokens,
    completionTokens: usage.completion_tokens ?? 0,
    totalTokens: usage.total_tokens ?? 0
  }
}

流式响应的情况不同：SSE 的最后一个 chunk 才带 usage，中间 chunk 没有。如果需要在流式场景下统计，可以在流结束后异步更新，不阻塞主推送。

3. 按什么维度存储

先想清楚要回答哪些查询：

  • 用户 A 今天用了多少 Token — 按用户 + 天
  • 模型 gpt-4o 这周被调用了多少次 — 按模型 + 周
  • 用户 B 过去一小时的用量是否异常 — 按用户 + 小时

核心维度是三个：用户 ID、模型名称、时间窗口。

方案一：预聚合计数器

每次请求直接累加到时间窗口计数器。KV 存储：

key: stats:{userId}:{model}:{timeWindow}
value: { promptTokens, completionTokens, totalTokens, requestCount }

优点：写入快，一次 KV 读-改-写完成。读取快，一个 key 就是一个汇总值。

缺点：KV 读-改-写不是原子操作，并发可能丢计数。统计场景下这个误差通常可以接受。

方案二：写原始记录，异步聚合

每次写一条原始记录，定时任务扫描聚合。数据完整但写入量大，KV 做明细存储不太合适。

对 Cloudflare Workers + KV 的组合，方案一更实用。统计不需要绝对精确，差几条请求不影响决策。

4. KV 存储实现

时间窗口生成

// src/stats/time-window.ts
export type Granularity = 'hour' | 'day' | 'month'
 
export function getTimeWindow(timestamp: number, granularity: Granularity): string {
  const d = new Date(timestamp)
  const y = d.getUTCFullYear()
  const m = String(d.getUTCMonth() + 1).padStart(2, '0')
  const day = String(d.getUTCDate()).padStart(2, '0')
  const h = String(d.getUTCHours()).padStart(2, '0')
  if (granularity === 'hour') return `${y}${m}${day}${h}`
  if (granularity === 'day') return `${y}${m}${day}`
  return `${y}${m}`
}

写入统计

每次请求完成后，同时写入小时和天两个粒度：

// src/stats/token-stats.ts
type StatsRecord = { promptTokens: number; completionTokens: number; totalTokens: number; requestCount: number }
 
async function incrementStats(kv: KVNamespace, userId: string, model: string, usage: TokenUsage) {
  const now = Date.now()
  await incrementWindow(kv, userId, model, getTimeWindow(now, 'hour'), usage, 48 * 3600)
  await incrementWindow(kv, userId, model, getTimeWindow(now, 'day'), usage, 90 * 86400)
}
 
async function incrementWindow(kv: KVNamespace, userId: string, model: string, window: string, usage: TokenUsage, ttl: number) {
  const key = `stats:${userId}:${model}:${window}`
  const existing = await kv.get(key, 'json') as StatsRecord | null
  const record: StatsRecord = {
    promptTokens: (existing?.promptTokens ?? 0) + usage.promptTokens,
    completionTokens: (existing?.completionTokens ?? 0) + usage.completionTokens,
    totalTokens: (existing?.totalTokens ?? 0) + usage.totalTokens,
    requestCount: (existing?.requestCount ?? 0) + 1
  }
  await kv.put(key, JSON.stringify(record), { expirationTtl: ttl })
}

TTL 设置：小时粒度保留 48 小时（够用近 24h 趋势），天粒度保留 90 天（够季度报表）。月数据不在 KV 里存，而是从日数据聚合查询。

调用时机

在聊天接口里，模型调用成功后用 waitUntil 异步写入，不阻塞响应：

// src/routes/chat.ts
const result = await callModelAPI({ model: body.model, messages: body.messages })
const usage = extractUsage(result)
if (usage) {
  c.executionCtx.waitUntil(incrementStats(c.env.KV, userId, body.model, usage))
}
return c.json(result)

5. 统计查询接口

查询某用户某天的用量

// src/routes/stats.ts
statsApp.get('/usage/daily', async (c) => {
  const userId = c.get('user').id
  const dateStr = c.req.query('date') || getTimeWindow(Date.now(), 'day')
  const prefix = `stats:${userId}:`
  const suffix = `:${dateStr}`
 
  const { keys } = await c.env.KV.list({ prefix })
  const matched = keys.filter(k => k.name.endsWith(suffix))
 
  const results = await Promise.all(matched.map(async (k) => {
    const record = await c.env.KV.get(k.name, 'json') as StatsRecord | null
    const model = k.name.split(':')[2]
    return { model, ...record }
  }))
  return c.json({ date: dateStr, models: results })
})

KV 的 list() 只支持前缀匹配，所以 key 设计要把查询维度放在前缀能覆盖的位置。stats:{userId}: 做前缀列出某用户的所有记录，再用 endsWith 过滤时间窗口。

查询趋势数据

逐时间窗口查询，组装成数组：

// src/routes/stats.ts
statsApp.get('/usage/hourly-trend', async (c) => {
  const userId = c.get('user').id
  const hours = parseInt(c.req.query('hours') || '24')
  const now = Date.now()
  const points: { window: string; totalTokens: number; requestCount: number }[] = []
 
  for (let i = hours - 1; i >= 0; i--) {
    const window = getTimeWindow(now - i * 3600 * 1000, 'hour')
    // 对该窗口内所有模型求和，代码同 daily 查询，此处省略
    points.push({ window, totalTokens: 0, requestCount: 0 }) // 实际有值
  }
  return c.json({ userId, hours, points })
})

24 小时就是 24 次 KV list。如果查 7 天趋势，换成天粒度减少调用次数。

6. 长期聚合与 TTL 策略

整理各粒度的生命周期：

粒度TTL用途
小时48 小时实时监控、近 24h 趋势
天90 天日报表、月度聚合数据源
月从日数据聚合季度/年度报表

月数据在查询时从日粒度实时聚合。如果需要长期保存，用 Cron Trigger 定时聚合写入 D1：

// src/cron/daily-aggregate.ts
export async function dailyAggregate(env: Env) {
  const yesterday = getTimeWindow(Date.now() - 86400000, 'day')
  const { keys } = await env.KV.list({ prefix: 'stats:' })
  const userMap = new Map<string, StatsRecord>()
 
  for (const k of keys) {
    if (!k.name.endsWith(`:${yesterday}`)) continue
    const record = await env.KV.get(k.name, 'json') as StatsRecord | null
    if (!record) continue
    const userId = k.name.split(':')[1]
    const existing = userMap.get(userId) ?? { promptTokens: 0, completionTokens: 0, totalTokens: 0, requestCount: 0 }
    userMap.set(userId, {
      promptTokens: existing.promptTokens + record.promptTokens,
      completionTokens: existing.completionTokens + record.completionTokens,
      totalTokens: existing.totalTokens + record.totalTokens,
      requestCount: existing.requestCount + record.requestCount
    })
  }
 
  for (const [userId, record] of userMap) {
    await env.DB.prepare(`
      INSERT INTO monthly_stats (user_id, month, prompt_tokens, completion_tokens, total_tokens, request_count)
      VALUES (?, ?, ?, ?, ?, ?)
      ON CONFLICT(user_id, month) DO UPDATE SET
        prompt_tokens = prompt_tokens + excluded.prompt_tokens,
        completion_tokens = completion_tokens + excluded.completion_tokens,
        total_tokens = total_tokens + excluded.total_tokens,
        request_count = request_count + excluded.request_count
    `).bind(userId, yesterday.slice(0, 6), record.promptTokens, record.completionTokens, record.totalTokens, record.requestCount).run()
  }
}

wrangler.jsonc 配置 Cron Trigger：

// wrangler.jsonc
{ "triggers": { "crons": ["0 2 * * *"] } }  // 每天 UTC 2:00

7. 与计费系统对接

统计模块的职责到此为止：记录 Token 消耗，提供多维度查询。计费系统从统计模块拿数据，按自己的规则算钱。

统计模块不应该知道计费规则。 把单价、阶梯计费、免费额度写在统计模块里，改价格要动统计代码，不同客户的计费方案会让统计模块被 if-else 淹没。

给计费系统的数据接口，按模型和日期返回明细：

// src/routes/stats.ts
statsApp.get('/usage/billing', async (c) => {
  const userId = c.req.query('userId')
  const from = c.req.query('from') // YYYYMMDD
  const to = c.req.query('to')
  // ... 过滤并聚合，按模型分组
  return c.json({ userId, from, to, byModel: {
    "gpt-4o": [{ day: "20260601", promptTokens: 12000, completionTokens: 5400, requestCount: 45 }],
    "gpt-4o-mini": [{ day: "20260601", promptTokens: 45000, completionTokens: 22000, requestCount: 120 }]
  }})
})

计费模块拿到数据后按自己维护的价格表算钱。价格变了只改价格表，不动统计接口。

统计模块还有一个直接用途——实时限流。在中间件里查用户当前小时的 Token 消耗，超过阈值就拒绝：

// src/middleware/token-limit.ts
export const tokenLimit = async (c: any, next: () => Promise<void>) => {
  const userId = c.get('user').id
  const hourWindow = getTimeWindow(Date.now(), 'hour')
  // 读取该用户当前小时所有模型的总消耗
  let hourlyTotal = 0
  // ... 遍历 keys 求和，代码同 daily 查询
  const limit = c.get('user').tokenLimit ?? 100000
  if (hourlyTotal >= limit) {
    return c.json({ error: { code: 'TOKEN_LIMIT_EXCEEDED', message: '本小时 Token 用量已达上限' } }, 429)
  }
  c.set('hourlyTokenUsage', hourlyTotal)
  await next()
}

总结

回顾一下这篇的要点：

  • 大模型响应自带 usage 字段，统计模块的职责是记录、汇总、提供查询
  • 预聚合计数器适合 KV，读-改-写的竞态在统计场景下可以接受
  • 时间窗口按小时（TTL 48h）和天（TTL 90 天）两个粒度存储
  • 月度数据从日粒度聚合查询，长期归档通过 Cron Trigger 写入 D1
  • 统计模块只管数据不管计费规则，边界划清楚，两边独立演进
  • waitUntil 让统计写入不阻塞主流程

下一篇讨论怎么把统计数据展示在管理后台。