21.05-Token消耗日志

要点

Token 是 AI 调用的核心成本单位，需要精确追踪每次调用的消耗
区分 prompt_tokens（输入）和 completion_tokens（输出），两者价格不同
使用 Analytics Engine 或数据库记录 token 消耗，支持后续计费
设置配额和告警，防止异常消耗导致成本失控

内容

1. Token 消耗的三个维度

每次 AI 调用都会消耗 token，需要从三个维度追踪：

维度	含义	用途
单次调用	本次请求消耗了多少 token	计费和配额检查
用户累计	某个用户在一段时间内的总消耗	账单和套餐管理
全局统计	所有用户在过去 N 小时的总消耗	成本监控和预算告警

1.1 Token 的价格差异

不同模型、不同类型的 token 价格不同：

模型	Input Token 价格	Output Token 价格
gpt-4-turbo	$0.01 / 1K tokens	$0.03 / 1K tokens
gpt-3.5-turbo	$0.0005 / 1K tokens	$0.0015 / 1K tokens
claude-opus-4-6	$0.015 / 1K tokens	$0.075 / 1K tokens
claude-sonnet-4-6	$0.003 / 1K tokens	$0.015 / 1K tokens

注意：

Output token 通常比 input token 贵 2-5 倍
不同模型的价格差异很大（gpt-3.5 比 gpt-4 便宜 20 倍）
需要分别记录 prompt_tokens 和 completion_tokens，不能只看 total_tokens

2. 基础 Token 消耗记录

// src/lib/token-tracker.ts
export interface TokenUsage {
  userId: string
  model: string
  promptTokens: number
  completionTokens: number
  totalTokens: number
  estimatedCost: number
  timestamp: number
}
 
export async function recordTokenUsage(
  env: Env,
  usage: TokenUsage
) {
  // 记录到 Analytics Engine
  env.ANALYTICS.writeDataPoint({
    blobs: [usage.userId, usage.model],
    doubles: [
      usage.promptTokens,
      usage.completionTokens,
      usage.totalTokens,
      usage.estimatedCost,
    ],
    indexes: [usage.userId],
  })
 
  // 如果需要精确计费，还要记录到数据库
  await env.DB.prepare(`
    INSERT INTO token_usage (
      user_id, model, prompt_tokens, completion_tokens,
      total_tokens, estimated_cost, created_at
    ) VALUES (?, ?, ?, ?, ?, ?, ?)
  `).bind(
    usage.userId,
    usage.model,
    usage.promptTokens,
    usage.completionTokens,
    usage.totalTokens,
    usage.estimatedCost,
    usage.timestamp
  ).run()
}

// src/routes/chat.ts
import { recordTokenUsage } from '../lib/token-tracker'
import { calculateCost } from '../lib/pricing'
 
app.post('/v1/chat/completions', async (c) => {
  const userId = c.get('userId')
  const body = await c.req.json()
 
  const result = await callLLM(body, c.env)
 
  // 记录 token 消耗
  const usage: TokenUsage = {
    userId,
    model: result.model,
    promptTokens: result.usage.prompt_tokens,
    completionTokens: result.usage.completion_tokens,
    totalTokens: result.usage.total_tokens,
    estimatedCost: calculateCost(result.model, result.usage),
    timestamp: Date.now(),
  }
 
  await recordTokenUsage(c.env, usage)
 
  return c.json(result)
})

3. 成本计算

// src/lib/pricing.ts
export const PRICING = {
  'gpt-4-turbo': { input: 0.01, output: 0.03 },
  'gpt-4': { input: 0.03, output: 0.06 },
  'gpt-3.5-turbo': { input: 0.0005, output: 0.0015 },
  'claude-opus-4-6': { input: 0.015, output: 0.075 },
  'claude-sonnet-4-6': { input: 0.003, output: 0.015 },
  'claude-haiku-3-5': { input: 0.00025, output: 0.00125 },
}
 
export function calculateCost(
  model: string,
  usage: { prompt_tokens: number; completion_tokens: number }
): number {
  const pricing = PRICING[model]
  if (!pricing) {
    console.warn(`Unknown model pricing: ${model}`)
    return 0
  }
 
  const inputCost = (usage.prompt_tokens / 1000) * pricing.input
  const outputCost = (usage.completion_tokens / 1000) * pricing.output
 
  return inputCost + outputCost
}

注意：

价格会变化，建议从配置文件或环境变量读取
缓存命中的请求（prompt cache）可能享受折扣
批量 API（batch API）通常有 50% 折扣

4. 用户配额管理

在调用 AI 之前检查用户是否还有配额：

// src/middleware/quota-check.ts
import { Context, Next } from 'hono'
 
export async function quotaCheck(c: Context, next: Next) {
  const userId = c.get('userId')
  const env = c.env
 
  // 查询用户当前周期的 token 消耗
  const usage = await env.DB.prepare(`
    SELECT SUM(total_tokens) as total_tokens,
           SUM(estimated_cost) as total_cost
    FROM token_usage
    WHERE user_id = ?
      AND created_at > ?
  `).bind(
    userId,
    Date.now() - 30 * 24 * 60 * 60 * 1000  // 最近 30 天
  ).first()
 
  const totalTokens = usage?.total_tokens || 0
  const totalCost = usage?.total_cost || 0
 
  // 检查配额
  const quota = await env.DB.prepare(`
    SELECT token_limit, cost_limit
    FROM user_quotas
    WHERE user_id = ?
  `).bind(userId).first()
 
  if (quota) {
    if (quota.token_limit && totalTokens >= quota.token_limit) {
      return c.json({
        error: 'Token quota exceeded',
        used: totalTokens,
        limit: quota.token_limit,
      }, 429)
    }
 
    if (quota.cost_limit && totalCost >= quota.cost_limit) {
      return c.json({
        error: 'Cost quota exceeded',
        used: totalCost,
        limit: quota.cost_limit,
      }, 429)
    }
  }
 
  // 存储配额信息到上下文，后续可以记录日志
  c.set('quota', {
    used: totalTokens,
    cost: totalCost,
    limit: quota?.token_limit,
  })
 
  await next()
}

// src/index.ts
app.use('/v1/chat/completions', quotaCheck)

5. 实时配额更新

每次 AI 调用后，更新用户的实时配额计数器：

// src/lib/token-tracker.ts
export async function recordTokenUsage(
  env: Env,
  usage: TokenUsage
) {
  // 1. 记录详细日志（Analytics Engine + 数据库）
  env.ANALYTICS.writeDataPoint({
    blobs: [usage.userId, usage.model],
    doubles: [
      usage.promptTokens,
      usage.completionTokens,
      usage.totalTokens,
      usage.estimatedCost,
    ],
    indexes: [usage.userId],
  })
 
  await env.DB.prepare(`
    INSERT INTO token_usage (
      user_id, model, prompt_tokens, completion_tokens,
      total_tokens, estimated_cost, created_at
    ) VALUES (?, ?, ?, ?, ?, ?, ?)
  `).bind(
    usage.userId,
    usage.model,
    usage.promptTokens,
    usage.completionTokens,
    usage.totalTokens,
    usage.estimatedCost,
    usage.timestamp
  ).run()
 
  // 2. 更新实时计数器（KV）
  const currentPeriod = new Date().toISOString().slice(0, 7)  // 2026-06
  const kvKey = `quota:${usage.userId}:${currentPeriod}`
 
  const current = await env.QUOTA_KV.get(kvKey, 'json') || {
    tokens: 0,
    cost: 0,
  }
 
  await env.QUOTA_KV.put(
    kvKey,
    JSON.stringify({
      tokens: current.tokens + usage.totalTokens,
      cost: current.cost + usage.estimatedCost,
      lastUpdated: Date.now(),
    })
  )
}

这样可以在不查询数据库的情况下快速检查用户配额：

export async function quotaCheck(c: Context, next: Next) {
  const userId = c.get('userId')
  const env = c.env
 
  // 快速检查 KV 计数器
  const currentPeriod = new Date().toISOString().slice(0, 7)
  const kvKey = `quota:${userId}:${currentPeriod}`
  const current = await env.QUOTA_KV.get(kvKey, 'json') || {
    tokens: 0,
    cost: 0,
  }
 
  // 获取用户配额限制
  const quota = await env.DB.prepare(`
    SELECT token_limit, cost_limit
    FROM user_quotas
    WHERE user_id = ?
  `).bind(userId).first()
 
  if (quota?.token_limit && current.tokens >= quota.token_limit) {
    return c.json({
      error: 'Token quota exceeded',
      used: current.tokens,
      limit: quota.token_limit,
    }, 429)
  }
 
  if (quota?.cost_limit && current.cost >= quota.cost_limit) {
    return c.json({
      error: 'Cost quota exceeded',
      used: current.cost,
      limit: quota.cost_limit,
    }, 429)
  }
 
  c.set('quota', current)
  await next()
}

6. Token 消耗告警

当用户或全局的 token 消耗异常时，触发告警：

// src/lib/token-tracker.ts
export async function recordTokenUsage(
  env: Env,
  usage: TokenUsage
) {
  // ... 记录日志和更新计数器
 
  // 检查是否需要告警
  await checkTokenAlerts(env, usage)
}
 
async function checkTokenAlerts(env: Env, usage: TokenUsage) {
  // 1. 单次调用消耗异常（可能是 prompt 注入或死循环）
  if (usage.totalTokens > 100000) {
    await sendAlert(env, {
      type: 'high_single_call',
      userId: usage.userId,
      model: usage.model,
      tokens: usage.totalTokens,
      cost: usage.estimatedCost,
    })
  }
 
  // 2. 用户小时消耗异常（可能是账号被盗）
  const hourlyUsage = await env.DB.prepare(`
    SELECT SUM(total_tokens) as total_tokens
    FROM token_usage
    WHERE user_id = ?
      AND created_at > ?
  `).bind(
    usage.userId,
    Date.now() - 60 * 60 * 1000
  ).first()
 
  if (hourlyUsage?.total_tokens > 1000000) {
    await sendAlert(env, {
      type: 'high_hourly_usage',
      userId: usage.userId,
      tokens: hourlyUsage.total_tokens,
    })
  }
 
  // 3. 全局小时消耗异常（可能是攻击或 bug）
  const globalHourly = await env.DB.prepare(`
    SELECT SUM(total_tokens) as total_tokens
    FROM token_usage
    WHERE created_at > ?
  `).bind(Date.now() - 60 * 60 * 1000).first()
 
  if (globalHourly?.total_tokens > 10000000) {
    await sendAlert(env, {
      type: 'high_global_usage',
      tokens: globalHourly.total_tokens,
      estimatedCost: globalHourly.total_tokens * 0.00001,  // 粗略估算
    })
  }
}
 
async function sendAlert(env: Env, alert: any) {
  // 发送到 Slack、邮件或告警平台
  await env.ALERT_WEBHOOK.send(JSON.stringify({
    ...alert,
    timestamp: Date.now(),
  }))
}

7. 账单生成

月末根据 token_usage 表生成用户账单：

// src/cron/generate-bills.ts
export default {
  async scheduled(event, env, ctx) {
    // 每月 1 号 00:00 执行
    const lastMonth = new Date()
    lastMonth.setMonth(lastMonth.getMonth() - 1)
    const period = lastMonth.toISOString().slice(0, 7)  // 2026-05
 
    // 查询所有有消耗的用户
    const users = await env.DB.prepare(`
      SELECT DISTINCT user_id
      FROM token_usage
      WHERE created_at >= ? AND created_at < ?
    `).bind(
      new Date(period + '-01').getTime(),
      new Date().getTime()
    ).all()
 
    for (const user of users.results) {
      const usage = await env.DB.prepare(`
        SELECT
          model,
          SUM(prompt_tokens) as prompt_tokens,
          SUM(completion_tokens) as completion_tokens,
          SUM(total_tokens) as total_tokens,
          SUM(estimated_cost) as total_cost
        FROM token_usage
        WHERE user_id = ?
          AND created_at >= ? AND created_at < ?
        GROUP BY model
      `).bind(
        user.user_id,
        new Date(period + '-01').getTime(),
        new Date().getTime()
      ).all()
 
      // 生成账单
      await env.DB.prepare(`
        INSERT INTO bills (
          user_id, period, total_tokens, total_cost,
          breakdown, created_at
        ) VALUES (?, ?, ?, ?, ?, ?)
      `).bind(
        user.user_id,
        period,
        usage.results.reduce((sum, u) => sum + u.total_tokens, 0),
        usage.results.reduce((sum, u) => sum + u.total_cost, 0),
        JSON.stringify(usage.results),
        Date.now()
      ).run()
    }
  },
}

8. 实战：完整的 Token 追踪流程

// src/routes/chat.ts
import { recordTokenUsage } from '../lib/token-tracker'
import { calculateCost } from '../lib/pricing'
import { quotaCheck } from '../middleware/quota-check'
import { log } from '../lib/logger'
 
const app = new Hono()
 
// 1. 配额检查
app.use('/v1/chat/completions', quotaCheck)
 
// 2. AI 调用
app.post('/v1/chat/completions', async (c) => {
  const requestId = c.get('requestId')
  const userId = c.get('userId')
  const quota = c.get('quota')
 
  const body = await c.req.json()
  const startedAt = Date.now()
 
  try {
    const result = await callLLM(body, c.env)
 
    // 3. 计算成本
    const cost = calculateCost(result.model, result.usage)
 
    // 4. 记录 token 消耗
    await recordTokenUsage(c.env, {
      userId,
      model: result.model,
      promptTokens: result.usage.prompt_tokens,
      completionTokens: result.usage.completion_tokens,
      totalTokens: result.usage.total_tokens,
      estimatedCost: cost,
      timestamp: Date.now(),
    })
 
    // 5. 日志
    log.info('llm.call.success', {
      requestId,
      userId,
      model: result.model,
      tokens: result.usage.total_tokens,
      cost,
      duration: Date.now() - startedAt,
      quotaRemaining: quota.limit ? quota.limit - quota.used : null,
    })
 
    return c.json(result)
  } catch (err) {
    log.error('llm.call.failed', {
      requestId,
      userId,
      model: body.model,
      error: err.message,
    })
 
    throw err
  }
})

9. 小结

Token 消耗日志的关键点：

精确记录：区分 prompt_tokens 和 completion_tokens，分别计费
实时配额：用 KV 存储实时计数器，避免频繁查询数据库
成本计算：根据模型和 token 类型计算费用
异常告警：单次/小时/全局消耗异常时触发告警
账单生成：定期汇总生成用户账单

Token 是 AI 产品的核心成本，需要精确追踪和控制。下一节讲用户行为日志，看看怎么记录用户在系统中的操作。