21.11-成本监控

要点

AI 服务的成本主要来自 token 消耗，需要实时监控和预算控制
区分不同维度的成本：用户级、功能级、全局级
设置预算告警，防止成本失控
成本优化：缓存、模型路由、prompt 压缩

内容

1. AI 服务的成本结构

AI 服务的成本主要来自 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 倍
不同模型的价格差异可达 20 倍
缓存命中的请求可能享受折扣（如 Anthropic 的 prompt caching）

2. 成本追踪的三个维度

2.1 用户级成本

追踪每个用户的 token 消耗和费用：

// src/lib/cost-tracker.ts
export interface CostRecord {
  userId: string
  model: string
  promptTokens: number
  completionTokens: number
  totalTokens: number
  estimatedCost: number
  timestamp: number
}
 
export async function recordCost(env: Env, record: CostRecord) {
  // 1. 记录到数据库
  await env.DB.prepare(`
    INSERT INTO cost_records (
      user_id, model, prompt_tokens, completion_tokens,
      total_tokens, estimated_cost, created_at
    ) VALUES (?, ?, ?, ?, ?, ?, ?)
  `).bind(
    record.userId,
    record.model,
    record.promptTokens,
    record.completionTokens,
    record.totalTokens,
    record.estimatedCost,
    record.timestamp
  ).run()
 
  // 2. 更新用户实时计数器
  const month = new Date().toISOString().slice(0, 7)  // 2026-06
  const kvKey = `cost:${record.userId}:${month}`
 
  const current = await env.COST_KV.get(kvKey, 'json') || {
    totalTokens: 0,
    totalCost: 0,
  }
 
  await env.COST_KV.put(
    kvKey,
    JSON.stringify({
      totalTokens: current.totalTokens + record.totalTokens,
      totalCost: current.totalCost + record.estimatedCost,
      lastUpdated: Date.now(),
    })
  )
}

查询用户成本：

// src/routes/admin/cost.ts
app.get('/admin/cost/user/:userId', async (c) => {
  const userId = c.req.param('userId')
  const month = c.req.query('month') || new Date().toISOString().slice(0, 7)
 
  const cost = await c.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 cost_records
    WHERE user_id = ?
      AND strftime('%Y-%m', datetime(created_at / 1000, 'unixepoch')) = ?
    GROUP BY model
  `).bind(userId, month).all()
 
  return c.json({
    userId,
    month,
    breakdown: cost.results,
    total: cost.results.reduce((sum, r) => sum + r.total_cost, 0),
  })
})

2.2 功能级成本

追踪不同功能的成本：

// src/routes/chat.ts
app.post('/v1/chat/completions', async (c) => {
  const body = await c.req.json()
  const feature = body.feature || 'default'  // chat, summary, translate, etc.
 
  const result = await callLLM(body, c.env)
 
  // 记录成本时带上 feature 标签
  await recordCost(c.env, {
    userId: c.get('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),
    feature,
    timestamp: Date.now(),
  })
})

查询各功能的成本：

SELECT
  feature,
  SUM(total_tokens) as total_tokens,
  SUM(estimated_cost) as total_cost
FROM cost_records
WHERE created_at > ?
GROUP BY feature
ORDER BY total_cost DESC

2.3 全局成本

追踪整个系统的总成本：

// src/cron/cost-summary.ts
export default {
  async scheduled(event, env, ctx) {
    // 每天 00:00 执行
 
    const today = new Date()
    today.setHours(0, 0, 0, 0)
 
    const yesterday = new Date(today)
    yesterday.setDate(yesterday.getDate() - 1)
 
    // 昨天的总成本
    const cost = await env.DB.prepare(`
      SELECT
        SUM(total_tokens) as total_tokens,
        SUM(estimated_cost) as total_cost,
        COUNT(DISTINCT user_id) as active_users
      FROM cost_records
      WHERE created_at >= ? AND created_at < ?
    `).bind(yesterday.getTime(), today.getTime()).first()
 
    // 记录到成本汇总表
    await env.DB.prepare(`
      INSERT INTO daily_cost_summary (
        date, total_tokens, total_cost, active_users
      ) VALUES (?, ?, ?, ?)
    `).bind(
      yesterday.toISOString().slice(0, 10),
      cost.total_tokens,
      cost.total_cost,
      cost.active_users
    ).run()
 
    // 检查是否超出预算
    const monthlyBudget = parseFloat(env.MONTHLY_BUDGET || '1000')
    const monthStart = new Date()
    monthStart.setDate(1)
    monthStart.setHours(0, 0, 0, 0)
 
    const monthlyCost = await env.DB.prepare(`
      SELECT SUM(estimated_cost) as total_cost
      FROM cost_records
      WHERE created_at >= ?
    `).bind(monthStart.getTime()).first()
 
    if (monthlyCost.total_cost > monthlyBudget) {
      await sendAlert(env, {
        level: 'P1',
        type: 'budget_exceeded',
        message: `月度成本 $${monthlyCost.total_cost.toFixed(2)} 超出预算 $${monthlyBudget}`,
      })
    }
  },
}

3. 成本预算控制

3.1 用户预算

给每个用户设置月度预算：

// src/middleware/budget-check.ts
export async function budgetCheck(c: Context, next: Next) {
  const userId = c.get('userId')
  const env = c.env
 
  // 查询用户预算
  const budget = await env.DB.prepare(`
    SELECT monthly_budget FROM user_budgets WHERE user_id = ?
  `).bind(userId).first()
 
  if (!budget?.monthly_budget) {
    await next()
    return
  }
 
  // 查询用户本月已消耗
  const monthStart = new Date()
  monthStart.setDate(1)
  monthStart.setHours(0, 0, 0, 0)
 
  const current = await env.COST_KV.get(
    `cost:${userId}:${new Date().toISOString().slice(0, 7)}`,
    'json'
  ) || { totalCost: 0 }
 
  // 检查是否超出预算
  if (current.totalCost >= budget.monthly_budget) {
    return c.json({
      error: 'Budget exceeded',
      used: current.totalCost,
      limit: budget.monthly_budget,
    }, 429)
  }
 
  // 存储预算信息到上下文
  c.set('budget', {
    used: current.totalCost,
    limit: budget.monthly_budget,
    remaining: budget.monthly_budget - current.totalCost,
  })
 
  await next()
}

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

3.2 预算告警

当用户成本接近预算时，提前告警：

// src/lib/cost-tracker.ts
export async function recordCost(env: Env, record: CostRecord) {
  // ... 记录成本
 
  // 检查预算告警
  const budget = await env.DB.prepare(`
    SELECT monthly_budget, alert_threshold FROM user_budgets WHERE user_id = ?
  `).bind(record.userId).first()
 
  if (budget?.alert_threshold) {
    const current = await env.COST_KV.get(
      `cost:${record.userId}:${new Date().toISOString().slice(0, 7)}`,
      'json'
    ) || { totalCost: 0 }
 
    const usagePercent = current.totalCost / budget.monthly_budget * 100
 
    if (usagePercent >= budget.alert_threshold && usagePercent < 100) {
      // 发送预算告警
      await sendAlert(env, {
        level: 'P2',
        type: 'budget_warning',
        userId: record.userId,
        message: `用户 ${record.userId} 成本已达预算的 ${usagePercent.toFixed(0)}%`,
      })
    }
  }
}

4. 成本优化策略

4.1 缓存

缓存 LLM 响应，避免重复调用：

// src/lib/cache.ts
export async function getWithCache<T>(
  env: Env,
  key: string,
  loader: () => Promise<T>,
  ttl: number = 3600
): Promise<{ data: T; fromCache: boolean }> {
  // 1. 尝试从缓存读取
  const cached = await env.CACHE.get(key)
 
  if (cached) {
    return {
      data: JSON.parse(cached),
      fromCache: true,
    }
  }
 
  // 2. 缓存未命中，调用 LLM
  const data = await loader()
 
  // 3. 写入缓存
  await env.CACHE.put(key, JSON.stringify(data), {
    expirationTtl: ttl,
  })
 
  return {
    data,
    fromCache: false,
  }
}

// src/routes/chat.ts
app.post('/v1/chat/completions', async (c) => {
  const body = await c.req.json()
  const cacheKey = generateCacheKey(body)
 
  const { data: result, fromCache } = await getWithCache(
    c.env,
    cacheKey,
    () => callLLM(body, c.env),
    3600
  )
 
  // 记录缓存命中率
  await recordMetric(c.env, fromCache ? 'cache.hit' : 'cache.miss', {
    userId: c.get('userId'),
  }, [1])
 
  return c.json(result)
})

4.2 模型路由

根据任务复杂度选择不同模型：

// src/lib/model-router.ts
export function selectModel(messages: ChatMessage[]): string {
  // 简单任务（如翻译、摘要）用便宜模型
  const isSimpleTask = messages.some(m =>
    m.content.includes('translate') ||
    m.content.includes('summarize') ||
    m.content.includes('翻译') ||
    m.content.includes('摘要')
  )
 
  if (isSimpleTask) {
    return 'gpt-3.5-turbo'  // 便宜 20 倍
  }
 
  // 复杂任务（如代码生成、推理）用贵模型
  return 'gpt-4-turbo'
}

// src/routes/chat.ts
app.post('/v1/chat/completions', async (c) => {
  const body = await c.req.json()
 
  // 自动选择模型（如果用户没有指定）
  const model = body.model || selectModel(body.messages)
 
  const result = await callLLM({ ...body, model }, c.env)
  return c.json(result)
})

4.3 Prompt 压缩

压缩 prompt 以减少 token 消耗：

// src/lib/prompt-compressor.ts
export function compressPrompt(prompt: string): string {
  // 1. 移除多余空格
  prompt = prompt.replace(/\s+/g, ' ')
 
  // 2. 移除无用的换行
  prompt = prompt.replace(/\n{2,}/g, '\n')
 
  // 3. 缩写常见词（可选）
  const abbreviations: Record<string, string> = {
    'please': 'plz',
    'thank you': 'thx',
    'because': 'bc',
  }
 
  for (const [full, short] of Object.entries(abbreviations)) {
    prompt = prompt.replace(new RegExp(`\\b${full}\\b`, 'gi'), short)
  }
 
  return prompt
}

注意：prompt 压缩可能影响模型质量，需要谨慎使用。

5. 成本报表

5.1 日报

// src/cron/cost-report.ts
export default {
  async scheduled(event, env, ctx) {
    const today = new Date()
    today.setDate(today.getDate() - 1)
    const dateStr = today.toISOString().slice(0, 10)
 
    // 昨天的成本数据
    const cost = 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,
        COUNT(*) as call_count
      FROM cost_records
      WHERE strftime('%Y-%m-%d', datetime(created_at / 1000, 'unixepoch')) = ?
      GROUP BY model
      ORDER BY total_cost DESC
    `).bind(dateStr).all()
 
    const totalCost = cost.results.reduce((sum, r) => sum + r.total_cost, 0)
 
    // 发送到 Slack
    await sendSlackAlert(env, {
      level: 'INFO',
      type: 'daily_cost_report',
      message: `📊 ${dateStr} 成本报告\n\n` +
        `总成本: $${totalCost.toFixed(2)}\n` +
        cost.results.map(r =>
          `${r.model}: $${r.total_cost.toFixed(2)} (${r.total_tokens} tokens)`
        ).join('\n'),
    })
  },
}

5.2 月报

// src/routes/admin/cost-report.ts
app.get('/admin/cost-report/monthly', async (c) => {
  const month = c.req.query('month') || new Date().toISOString().slice(0, 7)
 
  const report = await c.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,
      COUNT(DISTINCT user_id) as user_count,
      COUNT(*) as call_count
    FROM cost_records
    WHERE strftime('%Y-%m', datetime(created_at / 1000, 'unixepoch')) = ?
    GROUP BY model
    ORDER BY total_cost DESC
  `).bind(month).all()
 
  const totalCost = report.results.reduce((sum, r) => sum + r.total_cost, 0)
 
  return c.json({
    month,
    totalCost,
    breakdown: report.results,
  })
})

6. 成本监控 Dashboard

在 Cloudflare Dashboard 创建成本监控图表：

进入 Workers → 你的 Worker → Analytics
创建自定义 Dashboard
添加图表：
- 每日成本趋势（折线图）
- 各模型成本占比（饼图）
- Top 10 用户成本（柱状图）
- 缓存命中率（折线图）

7. 实战：完整的成本监控系统

// src/lib/cost-tracker.ts
import { calculateCost } from './pricing'
import { sendAlert } from './alert-router'
 
export async function recordCost(env: Env, record: CostRecord) {
  // 1. 记录到数据库
  await env.DB.prepare(`
    INSERT INTO cost_records (
      user_id, model, prompt_tokens, completion_tokens,
      total_tokens, estimated_cost, created_at
    ) VALUES (?, ?, ?, ?, ?, ?, ?)
  `).bind(
    record.userId,
    record.model,
    record.promptTokens,
    record.completionTokens,
    record.totalTokens,
    record.estimatedCost,
    record.timestamp
  ).run()
 
  // 2. 更新实时计数器
  const month = new Date().toISOString().slice(0, 7)
  const kvKey = `cost:${record.userId}:${month}`
  const current = await env.COST_KV.get(kvKey, 'json') || {
    totalCost: 0,
  }
 
  const newCost = current.totalCost + record.estimatedCost
  await env.COST_KV.put(kvKey, JSON.stringify({
    totalCost: newCost,
    lastUpdated: Date.now(),
  }))
 
  // 3. 检查预算告警
  const budget = await env.DB.prepare(`
    SELECT monthly_budget, alert_threshold FROM user_budgets WHERE user_id = ?
  `).bind(record.userId).first()
 
  if (budget?.monthly_budget && budget?.alert_threshold) {
    const usagePercent = newCost / budget.monthly_budget * 100
 
    if (usagePercent >= budget.alert_threshold && usagePercent < 100) {
      await sendAlert(env, {
        level: 'P2',
        type: 'budget_warning',
        message: `用户 ${record.userId} 成本已达预算的 ${usagePercent.toFixed(0)}%`,
      })
    }
  }
}

8. 小结

成本监控的关键点：

三个维度：用户级、功能级、全局级成本追踪
预算控制：给用户设置月度预算，超出时拒绝请求
成本告警：预算接近阈值时提前告警
成本优化：缓存、模型路由、prompt 压缩
成本报表：日报、月报，可视化成本趋势

AI 服务的成本可以快速增长，没有监控和预算控制，很容易出现意外账单。通过实时监控、预算控制和成本优化，可以有效管理 AI 服务的成本。