LLM 应用可观测性实战：Token 追踪、延迟监控与成本告警全链路方案

2026 年，大语言模型（LLM）应用已经从 Demo 阶段全面进入生产环境。但一个被严重低估的问题是：超过 70% 的 LLM 应用上线后处于「黑盒」状态——你不知道每次调用消耗了多少 Token、延迟波动的原因是什么、模型输出是否出现了幻觉、成本是否在失控。根据 Langfuse 社区调查数据，没有可观测性的 LLM 应用，其生产事故率是有完善监控体系的 4.7 倍。

📌 **记住：**LLM 应用的可观测性 ≠ 传统 Web 应用的可观测性。LLM 有三个独特的监控维度：Token 经济学（成本）、语义质量（幻觉/相关性）、推理延迟（TTFT/TPS）。传统 APM 工具根本覆盖不了这些。

📊 一、为什么传统监控不够？LLM 可观测性的三层架构

1.1 传统 APM vs LLM 监控的核心差异

传统 Web 应用的可观测性关注请求延迟、错误率、吞吐量。但 LLM 应用引入了全新的监控维度：

⚡ **关键结论：**你需要一套专门针对 LLM 的可观测性体系，而不是简单复用现有的 APM 方案。

1.2 三层监控架构设计

我将 LLM 可观测性分为三层，每层解决不同的问题：

• 🔵 基础设施层 — API 可用性、错误率、网络延迟（传统 APM 覆盖）
• 🟡 调用链路层 — Token 用量、TTFT/TPS、模型路由、重试逻辑、成本追踪
• 🔴 语义质量层 — 幻觉检测、相关性评分、输出格式正确率

大多数团队只做了第一层，少数做到了第二层，几乎没有团队做到第三层。本文将带你完整搭建后两层监控体系。

1.3 成本对比：主流模型的真实开销

在搭建监控之前，你需要了解不同模型的成本结构。以下是 2026 年 5 月主流模型在典型场景下的月成本估算（按每天 10,000 次调用，平均输入 1,000 tokens、输出 500 tokens 计算）：

⚠️ **警告：**GPT-4o 和 Claude Sonnet 的输出价格是输入价格的 4-5 倍。如果你的应用输出 token 很多（如代码生成、长文写作），成本会迅速飙升。一定要设置单次调用成本上限。

⚡ **关键结论：**在不需要顶级推理能力的场景（分类、提取、简单问答），使用 GPT-4o-mini 或 DeepSeek 可以将成本降低 10-30 倍，而质量损失通常不超过 15%。

🔧 二、全链路 Trace 与成本监控的完整实现

2.1 构建 LLM Trace 中间件

核心思路是用一个包装器拦截所有 LLM 调用，自动采集关键指标。以下是一个基于 OpenAI SDK 的完整实现：

// llm-tracer.ts — LLM 调用全链路追踪器
import OpenAI from 'openai'
import { randomUUID } from 'crypto'

interface LLMTrace {
  traceId: string
  spanId: string
  model: string
  promptTokens: number
  completionTokens: number
  totalTokens: number
  ttft: number          // 首 Token 延迟 (ms)
  tps: number           // 每秒生成 Token 数
  totalLatency: number  // 总延迟 (ms)
  inputPreview: string  // 输入前 200 字符
  outputPreview: string // 输出前 200 字符
  timestamp: string
  status: 'success' | 'error' | 'timeout'
  errorMessage?: string
  cost: number          // 估算成本 (USD)
}

// 模型定价表 (2026 年 5 月最新价格)
const MODEL_PRICING: Record<string, { input: number; output: number }> = {
  'gpt-4o':          { input: 2.50 / 1_000_000, output: 10.00 / 1_000_000 },
  'gpt-4o-mini':     { input: 0.15 / 1_000_000, output: 0.60 / 1_000_000 },
  'claude-sonnet-4-20250514': { input: 3.00 / 1_000_000, output: 15.00 / 1_000_000 },
  'deepseek-chat':   { input: 0.14 / 1_000_000, output: 0.28 / 1_000_000 },
}

function calculateCost(model: string, promptTokens: number, completionTokens: number): number {
  const pricing = MODEL_PRICING[model]
  if (!pricing) return 0
  return promptTokens * pricing.input + completionTokens * pricing.output
}

class LLMTracer {
  private traces: LLMTrace[] = []
  private onTrace?: (trace: LLMTrace) => void

  constructor(options?: { onTrace?: (trace: LLMTrace) => void }) {
    this.onTrace = options?.onTrace
  }

  // 包装 OpenAI client，自动采集 trace
  wrapClient(client: OpenAI): OpenAI {
    const originalCreate = client.chat.completions.create.bind(client.chat.completions)
    const tracer = this

    client.chat.completions.create = async function (params: any) {
      const traceId = randomUUID()
      const spanId = randomUUID()
      const startTime = Date.now()
      let firstTokenTime = 0
      let tokenCount = 0

      try {
        // 流式调用
        if (params.stream) {
          const stream = await originalCreate({ ...params, stream: true })
          const wrappedStream = async function* () {
            for await (const chunk of stream) {
              if (firstTokenTime === 0 && chunk.choices[0]?.delta?.content) {
                firstTokenTime = Date.now()
              }
              if (chunk.choices[0]?.delta?.content) tokenCount++
              yield chunk
            }
            // 流结束，记录 trace
            const endTime = Date.now()
            const trace: LLMTrace = {
              traceId, spanId, model: params.model,
              promptTokens: 0, completionTokens: tokenCount, totalTokens: tokenCount,
              ttft: firstTokenTime > 0 ? firstTokenTime - startTime : -1,
              tps: tokenCount / ((endTime - (firstTokenTime || startTime)) / 1000),
              totalLatency: endTime - startTime,
              inputPreview: JSON.stringify(params.messages).slice(0, 200),
              outputPreview: '', timestamp: new Date(startTime).toISOString(),
              status: 'success', cost: 0,
            }
            tracer.record(trace)
          }
          return wrappedStream() as any
        }

        // 非流式调用
        const response = await originalCreate(params)
        const endTime = Date.now()
        const usage = response.usage
        const trace: LLMTrace = {
          traceId, spanId, model: params.model,
          promptTokens: usage?.prompt_tokens ?? 0,
          completionTokens: usage?.completion_tokens ?? 0,
          totalTokens: usage?.total_tokens ?? 0,
          ttft: endTime - startTime,
          tps: (usage?.completion_tokens ?? 0) / ((endTime - startTime) / 1000),
          totalLatency: endTime - startTime,
          inputPreview: JSON.stringify(params.messages).slice(0, 200),
          outputPreview: (response.choices[0]?.message?.content ?? '').slice(0, 200),
          timestamp: new Date(startTime).toISOString(),
          status: 'success',
          cost: calculateCost(params.model, usage?.prompt_tokens ?? 0, usage?.completion_tokens ?? 0),
        }
        tracer.record(trace)
        return response

      } catch (error: any) {
        const endTime = Date.now()
        tracer.record({
          traceId, spanId, model: params.model,
          promptTokens: 0, completionTokens: 0, totalTokens: 0,
          ttft: -1, tps: 0, totalLatency: endTime - startTime,
          inputPreview: JSON.stringify(params.messages).slice(0, 200),
          outputPreview: '', timestamp: new Date(startTime).toISOString(),
          status: 'error', errorMessage: error.message, cost: 0,
        })
        throw error
      }
    } as any
    return client
  }

  private record(trace: LLMTrace) {
    this.traces.push(trace)
    this.onTrace?.(trace)
    if (this.traces.length > 10000) this.traces = this.traces.slice(-5000)
  }

  getStats() {
    if (this.traces.length === 0) return null
    const recent = this.traces.slice(-100)
    return {
      totalCalls: this.traces.length,
      avgLatency: recent.reduce((s, t) => s + t.totalLatency, 0) / recent.length,
      avgTTFT: recent.filter(t => t.ttft > 0).reduce((s, t) => s + t.ttft, 0) / recent.filter(t => t.ttft > 0).length,
      totalCost: this.traces.reduce((s, t) => s + t.cost, 0),
      errorRate: this.traces.filter(t => t.status === 'error').length / this.traces.length,
    }
  }
}

export { LLMTracer, type LLMTrace }

2.2 实际使用与实时告警

// app.ts — 在应用中使用 Tracer
import OpenAI from 'openai'
import { LLMTracer } from './llm-tracer'

const tracer = new LLMTracer({
  onTrace: (trace) => {
    console.log(`[LLM] ${trace.model} | ${trace.totalLatency}ms | ${trace.totalTokens} tokens | $${trace.cost.toFixed(6)}`)
    // 超过 5 秒的慢调用告警
    if (trace.totalLatency > 5000) {
      console.warn(`⚠️ SLOW LLM CALL: ${trace.totalLatency}ms on ${trace.model}`)
    }
    // 成本超过 $0.01 的调用告警
    if (trace.cost > 0.01) {
      console.warn(`💰 EXPENSIVE CALL: $${trace.cost.toFixed(4)} on ${trace.model}`)
    }
  }
})

const client = tracer.wrapClient(new OpenAI({ apiKey: process.env.OPENAI_API_KEY }))

// 正常使用，所有调用自动被追踪
const response = await client.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: '解释 JavaScript 闭包' }],
  stream: true,
})

// 查看统计
console.log(tracer.getStats())

💡 **提示：**Tracer 包装器对原有代码零侵入——你只需要在初始化时 wrapClient 一次，之后所有调用自动被追踪，不需要修改任何业务代码。

2.3 成本追踪与 Prometheus 导出

// cost-tracker.ts — LLM 成本实时追踪、告警与 Prometheus 指标导出
interface CostRecord {
  timestamp: number
  model: string
  promptTokens: number
  completionTokens: number
  cost: number
  userId?: string
}

class CostTracker {
  private records: CostRecord[] = []
  private dailyBudget: number
  private singleCallLimit: number
  private onAlert: (alert: { type: string; message: string }) => void

  constructor(options: { dailyBudget?: number; singleCallLimit?: number; onAlert?: (a: any) => void } = {}) {
    this.dailyBudget = options.dailyBudget ?? 10.0
    this.singleCallLimit = options.singleCallLimit ?? 0.5
    this.onAlert = options.onAlert ?? ((a) => console.warn(`🚨 ${a.message}`))
  }

  record(data: Omit<CostRecord, 'timestamp'>) {
    const record = { ...data, timestamp: Date.now() }
    this.records.push(record)

    // 单次调用成本检查
    if (record.cost > this.singleCallLimit) {
      this.onAlert({
        type: 'single_call',
        message: `单次调用成本 $${record.cost.toFixed(4)} 超过限制 $${this.singleCallLimit}`,
      })
    }

    // 日预算检查
    const todayCost = this.getTodayCost()
    if (todayCost > this.dailyBudget * 0.8) {
      this.onAlert({
        type: 'daily_budget',
        message: `今日成本 $${todayCost.toFixed(2)} 已达预算的 ${((todayCost / this.dailyBudget) * 100).toFixed(0)}%`,
      })
    }
  }

  getTodayCost(): number {
    const todayStart = new Date(); todayStart.setHours(0, 0, 0, 0)
    return this.records.filter(r => r.timestamp >= todayStart.getTime()).reduce((s, r) => s + r.cost, 0)
  }

  getCostByModel(): Record<string, number> {
    const result: Record<string, number> = {}
    for (const r of this.records) result[r.model] = (result[r.model] ?? 0) + r.cost
    return result
  }

  // Prometheus 格式导出
  toPrometheus(): string {
    const lines: string[] = []
    lines.push('# HELP llm_cost_total Total LLM cost in USD')
    lines.push('# TYPE llm_cost_total counter')
    for (const [model, cost] of Object.entries(this.getCostByModel())) {
      lines.push(`llm_cost_total{model="${model}"} ${cost.toFixed(6)}`)
    }
    const totalTokens = this.records.reduce((s, r) => s + r.promptTokens + r.completionTokens, 0)
    lines.push('# HELP llm_tokens_total Total tokens consumed')
    lines.push(`llm_tokens_total ${totalTokens}`)
    return lines.join('\n')
  }
}

export { CostTracker }

2.4 Grafana 告警规则配置

将 LLM 指标推送到 Prometheus 后，在 Grafana 中配置告警规则：

# alert-rules.yml — LLM 应用告警规则
groups:
  - name: llm-alerts
    rules:
      # 成本告警：日成本超过预算 80%
      - alert: LLMDailyCostHigh
        expr: increase(llm_cost_total[24h]) > 8
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM 日成本超过 $8"

      # 延迟告警：P95 TTFT 超过 3 秒
      - alert: LLMHighLatency
        expr: histogram_quantile(0.95, rate(llm_ttft_seconds_bucket[5m])) > 3
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "LLM P95 TTFT 超过 3 秒"

      # 错误率告警：错误率超过 5%
      - alert: LLMHighErrorRate
        expr: rate(llm_calls_total{status="error"}[5m]) / rate(llm_calls_total[5m]) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "LLM 错误率超过 5%"

核心 PromQL 查询用于 Grafana 面板：

# 每分钟 Token 消耗量
rate(llm_tokens_total[5m]) * 60

# P95 首 Token 延迟
histogram_quantile(0.95, rate(llm_ttft_seconds_bucket[5m]))

# 每小时成本趋势
increase(llm_cost_total[1h])

# 各模型成本占比
sum by (model) (llm_cost_total) / sum(llm_cost_total)

🔍 三、语义质量监控：幻觉检测与输出校验

3.1 基于规则的输出校验器

LLM 输出的质量不能只靠「看起来对」来判断。以下是一个覆盖 5 个维度的输出校验器：

// output-validator.ts — LLM 输出质量校验器
interface ValidationResult {
  passed: boolean
  checks: { name: string; passed: boolean; detail: string }[]
}

class LLMOutputValidator {
  // 1. JSON 格式校验
  static validateJSON(output: string) {
    try { JSON.parse(output); return { passed: true, detail: 'JSON 格式正确' } }
    catch (e: any) { return { passed: false, detail: `JSON 解析失败: ${e.message}` } }
  }

  // 2. 输出完整性检测 (可能被截断)
  static validateCompleteness(output: string) {
    const trimmed = output.trim()
    if (/[，,、；;：:（(]$/.test(trimmed)) {
      return { passed: false, detail: '输出可能被截断：以标点符号中间结尾' }
    }
    const endsWithComplete = /[。！？.!?\n]$/.test(trimmed) || /```\s*$/.test(trimmed)
    if (!endsWithComplete && trimmed.length > 100) {
      return { passed: false, detail: '输出可能不完整：没有以正常标点结尾' }
    }
    return { passed: true, detail: '输出完整性检查通过' }
  }

  // 3. 重复内容检测
  static validateNoRepetition(output: string) {
    const sentences = output.split(/[。！？.!?\n]+/).filter(s => s.trim().length > 10)
    if (sentences.length < 3) return { passed: true, detail: '句子数太少，跳过重复检测' }
    let dupCount = 0
    for (let i = 0; i < sentences.length; i++) {
      for (let j = i + 1; j < sentences.length; j++) {
        if (this.similarity(sentences[i], sentences[j]) > 0.8) dupCount++
      }
    }
    if (dupCount > sentences.length * 0.3) {
      return { passed: false, detail: `检测到 ${dupCount} 组高度相似句子，可能存在重复生成` }
    }
    return { passed: true, detail: '无明显重复内容' }
  }

  // 4. 拒绝回答检测
  static validateNotRefused(output: string) {
    const patterns = [
      /作为.*AI.*我(不能|无法|不方便)/,
      /I('m| am) (sorry|afraid),?\s*(but )?I (can't|cannot)/i,
      /超出.*能力范围/,
    ]
    for (const p of patterns) {
      if (p.test(output)) return { passed: false, detail: `检测到拒绝回答: ${p.source}` }
    }
    return { passed: true, detail: '未检测到拒绝回答' }
  }

  // 5. PII 泄露检测
  static validateNoPII(output: string) {
    const pii = [
      { name: '手机号', pattern: /1[3-9]\d{9}/ },
      { name: '身份证号', pattern: /\d{17}[\dXx]/ },
      { name: '邮箱', pattern: /[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/ },
    ]
    for (const { name, pattern } of pii) {
      if (pattern.test(output)) return { passed: false, detail: `检测到可能的 ${name} 泄露` }
    }
    return { passed: true, detail: '未检测到 PII 泄露' }
  }

  // 执行全部校验
  static validateAll(output: string, options?: { expectJSON?: boolean }): ValidationResult {
    const checks = []
    if (options?.expectJSON) checks.push({ name: 'JSON 格式', ...this.validateJSON(output) })
    checks.push({ name: '输出完整性', ...this.validateCompleteness(output) })
    checks.push({ name: '重复检测', ...this.validateNoRepetition(output) })
    checks.push({ name: '拒绝回答', ...this.validateNotRefused(output) })
    checks.push({ name: 'PII 泄露', ...this.validateNoPII(output) })
    return { passed: checks.every(c => c.passed), checks }
  }

  private static similarity(a: string, b: string): number {
    const setA = new Set(a), setB = new Set(b)
    const intersection = new Set([...setA].filter(x => setB.has(x)))
    return intersection.size / Math.max(setA.size, setB.size)
  }
}

export { LLMOutputValidator }

💡 **提示：**规则校验只是第一道防线。对于真正的幻觉（事实性错误），你需要用另一个 LLM 做交叉验证，或者将输出与 RAG 检索结果做语义比对。

3.2 幻觉检测的实用策略

幻觉检测没有银弹，但以下策略组合使用可以覆盖 80%+ 的场景：

• ✅ 引用验证 — 要求模型在回答中标注引用来源，然后验证引用是否存在
• ✅ 交叉验证 — 用不同模型（或同模型不同 temperature）对同一问题生成多次回答，比较一致性
• ✅ 知识库比对 — 将模型输出的关键事实与 RAG 检索结果做语义相似度比对
• ✅ 置信度标记 — 在 System Prompt 中要求模型为每个事实声明标注置信度（高/中/低）

⚠️ **警告：**不要完全依赖模型自评置信度。研究表明，GPT-4 在「低置信度」声明上的准确率仍然有 72%，而在「高置信度」声明上也只有 89%。自评置信度有参考价值，但不能作为唯一依据。

3.3 采样策略与隐私合规

生产环境中不需要记录每一条 trace 的完整数据，建议分层采样：

• ✅ 错误请求 — 100% 记录（必须）
• ✅ 慢请求（> P95） — 100% 记录
• ✅ 高成本请求（> $0.05） — 100% 记录
• ✅ 正常请求 — 10% 采样记录

隐私合规注意事项：

• ❌ **避免：**将完整用户输入输出存入监控系统
• ✅ **推荐：**只存储前 200 字符的 preview + hash
• ✅ **推荐：**对 PII 字段做脱敏后再存储
• ⚠️ **注意：**医疗、金融等领域，监控数据的存储也需要符合相应法规（如 HIPAA、GDPR）

✅ 总结与工具推荐

LLM 应用的可观测性不是「有了更好」，而是生产级应用的必要条件。核心要点回顾：

• 🔵 调用链路层 — 用 LLM Tracer 自动采集 Token 用量、TTFT/TPS、成本数据，零侵入接入
• 🔴 语义质量层 — 用规则校验 + 交叉验证 + 知识库比对覆盖输出质量
• 💰 成本控制 — 用 CostTracker 实时监控日/月成本，设置多级告警阈值

⚡ **关键结论：**可观测性投入的 ROI 极高——一个完善的监控体系可以在第一周就帮你发现 20%+ 的 Token 浪费（比如不必要的长 Context、低效的 Prompt、重复调用），仅成本优化一项就能覆盖监控系统的搭建成本。

推荐工具