课 2 · 上线前的工程准备

本课目标

完成 Token 成本控制、缓存策略、日志/Tracing/监控告警等上线准备。课后你会有一份上线 checklist。

这一课先掌握“首版上线最少还要补什么”，不要求第一次就把语义缓存、Tracing、监控告警全部做到生产级。

质量控制解决的是"模型输出可靠"，工程准备解决的是"系统整体可运维"——成本可控、问题可追踪、异常可告警。

最低配上线版 vs 进阶生产版

把模块 8 两课合起来看，建议按“先保命，再优化”的顺序理解：

• 最低配上线版必修：上一课的超时 / 重试 / 基础 fallback，加上这一课的日志、成本感知、基础 checklist
• 进阶生产版扩展：语义缓存、Tracing、监控告警、更细的预算控制

也就是说，这一课不是要求你第一次上线就把整套生产体系都补齐，而是让你知道哪些是“现在就该做”，哪些是“业务跑起来后再继续补”。

Token 成本计算与控制

这一部分先建立成本感知就够了：知道调用为什么会花钱、哪些地方最容易失控、如何先做粗粒度预算。

成本结构

大模型 API 按 Token 计费，分输入和输出：

以 Qwen3.6-Plus 为例，输入和输出都会按 Token 分开计费。具体单价会变动，以上线时控制台或官方计费页为准。

成本估算公式

单次请求成本 = 输入 Token × 输入单价 + 输出 Token × 输出单价
日成本 = 单次成本 × 日请求量
月成本 = 日成本 × 30

控制策略

1. 选对模型

import { generateText, Output } from 'ai'
import { z } from 'zod'

// 简单任务也先统一走 Qwen3.6-Plus
const classifyResult = await generateText({
  model,
  output: Output.object({
    schema: z.object({ category: z.enum(['tech', 'business', 'other']) }),
  }),
  prompt: `分类: ${text}`,
})

// 复杂任务同样可以先用同一模型打通链路
const analysisResult = await generateText({
  model,
  prompt: `详细分析: ${text}`,
})

2. 控制输入长度

// 限制上下文长度
function trimContext(messages: ModelMessage[], maxTokens: number) {
  let totalTokens = 0
  const trimmed: ModelMessage[] = []

  // 从最新消息开始保留
  for (let i = messages.length - 1; i >= 0; i--) {
    const msgTokens = estimateTokens(messages[i])
    if (totalTokens + msgTokens > maxTokens) break
    trimmed.unshift(messages[i])
    totalTokens += msgTokens
  }
  return trimmed
}

// 粗略估算 token 数（1 个中文字 ≈ 1.5 token，1 个英文词 ≈ 1 token）
function estimateTokens(message: ModelMessage): number {
  const text = typeof message.content === 'string' ? message.content : ''
  return Math.ceil(text.length * 1.5)
}

3. 限制输出长度

const result = await generateText({
  model,
  maxTokens: 500,  // 限制输出最多 500 token
  prompt: '简要说明什么是 RAG',
})

4. 设置预算告警

let totalCost = 0
const DAILY_BUDGET = 10 // $10/天

function trackCost(inputTokens: number, outputTokens: number) {
  const cost = (inputTokens * 3 + outputTokens * 15) / 1_000_000  // 示例单价，按实际模型价格替换
  totalCost += cost

  if (totalCost > DAILY_BUDGET * 0.8) {
    console.warn(`⚠️ 日预算已用 ${((totalCost / DAILY_BUDGET) * 100).toFixed(0)}%`)
  }
  if (totalCost > DAILY_BUDGET) {
    throw new Error('已超出日预算上限')
  }
}

语义缓存

相似的问题不需要每次都调 API。语义缓存通过向量相似度判断问题是否已经被回答过。

这部分属于典型的进阶优化项。首版如果还没有稳定流量，先用基础缓存 / 重复问题缓存就够了；等你确认调用量和成本确实开始上涨，再考虑把语义缓存补上。

这里要区分两层口径：

• 这节课会讲语义缓存的生产设计思路，因为它确实是上线后常见的降本方案
• 但当前基础项目为了保持本地 REPL 简洁，只展示基础缓存/重复问题缓存，不强制落地 embedding 相似度缓存

工作原理

用户提问: "什么是 RAG？"
  ↓
Embedding → 查缓存 → 相似度 > 0.95?
  ├── 是 → 返回缓存结果（不调 API）
  └── 否 → 调 API → 存入缓存 → 返回结果

实现

interface CacheEntry {
  question: string
  embedding: number[]
  answer: string
  timestamp: number
}

const cache: CacheEntry[] = []
const SIMILARITY_THRESHOLD = 0.95
const CACHE_TTL = 3600_000 // 1小时过期

async function cachedGenerate(question: string): Promise<string> {
  const now = Date.now()

  // 计算问题的 embedding
  // embeddingModel 可沿用前文初始化的 provider.embedding('text-embedding-v4')
  const { embedding } = await embed({
    model: embeddingModel,
    value: question,
  })

  // 在缓存中找相似问题
  for (const entry of cache) {
    if (now - entry.timestamp > CACHE_TTL) continue // 过期
    const similarity = cosineSimilarity(embedding, entry.embedding)
    if (similarity > SIMILARITY_THRESHOLD) {
      console.log(`📦 缓存命中 (相似度: ${similarity.toFixed(3)})`)
      return entry.answer
    }
  }

  // 未命中，调 API
  const { text } = await generateText({
    model,
    prompt: question,
  })

  // 存入缓存
  cache.push({ question, embedding, answer: text, timestamp: now })
  return text
}

效果

• 重复问题（如 FAQ）命中率可达 60-80%
• 直接降低 API 调用成本
• 响应速度从秒级降到毫秒级

可观测性：日志与 Tracing

如果按优先级排，建议先把“日志”做到位，再看 Tracing。日志是最低配上线版就该有的，Tracing 则是在多步骤链路开始变长之后更有价值。

结构化日志

interface LLMLog {
  requestId: string
  timestamp: string
  model: string
  inputTokens: number
  outputTokens: number
  latencyMs: number
  status: 'success' | 'error'
  error?: string
  toolCalls?: string[]
}

function logLLMCall(log: LLMLog) {
  // 结构化输出，方便日志系统采集
  console.log(JSON.stringify(log))
}

Tracing

Multi-step Agent 调用涉及多次 LLM 交互，需要 Tracing 把整个链路串起来。

这一步先理解“为什么需要把多次调用串成一条链”即可，不要求现在就接入完整观测平台。

// 简单的 Trace 实现
class Trace {
  id: string
  spans: Array<{
    name: string
    startTime: number
    endTime?: number
    metadata?: Record<string, any>
  }> = []

  constructor() {
    this.id = crypto.randomUUID()
  }

  startSpan(name: string) {
    const span = { name, startTime: Date.now() }
    this.spans.push(span)
    return {
      end: (metadata?: Record<string, any>) => {
        span.endTime = Date.now()
        span.metadata = metadata
      },
    }
  }

  summary() {
    return {
      traceId: this.id,
      totalMs: Math.max(...this.spans.map(s => s.endTime || 0)) -
               Math.min(...this.spans.map(s => s.startTime)),
      spans: this.spans.map(s => ({
        name: s.name,
        durationMs: (s.endTime || Date.now()) - s.startTime,
        ...s.metadata,
      })),
    }
  }
}

// 使用
const trace = new Trace()

const embedSpan = trace.startSpan('embedding')
const embedding = await embed({ model: openaiEmbed, value: query })
embedSpan.end({ tokens: query.length })

const searchSpan = trace.startSpan('vector-search')
const docs = await vectorSearch(embedding)
searchSpan.end({ results: docs.length })

const genSpan = trace.startSpan('generation')
const answer = await generateText({ model, prompt: `...` })
genSpan.end({ tokens: answer.usage })

console.log(trace.summary())

监控告警

监控告警同样属于从“能跑”走向“稳定运维”的进阶能力。第一次上线时，你至少要知道该盯哪些指标；至于是否马上接完整告警系统，可以按业务压力决定。

核心监控指标

指标	含义	告警阈值
失败率	API 调用失败比例	> 5%
P99 延迟	99% 请求的响应时间	> 30s
Token 用量	日 Token 消耗	超预算 80%
缓存命中率	基础缓存或语义缓存的命中比例	< 30%（可能缓存失效）

简单监控实现

class Monitor {
  private metrics = {
    totalCalls: 0,
    failedCalls: 0,
    totalLatencyMs: 0,
    totalTokens: 0,
  }

  record(success: boolean, latencyMs: number, tokens: number) {
    this.metrics.totalCalls++
    if (!success) this.metrics.failedCalls++
    this.metrics.totalLatencyMs += latencyMs
    this.metrics.totalTokens += tokens
  }

  report() {
    const { totalCalls, failedCalls, totalLatencyMs, totalTokens } = this.metrics
    return {
      failureRate: totalCalls > 0 ? (failedCalls / totalCalls * 100).toFixed(1) + '%' : '0%',
      avgLatencyMs: totalCalls > 0 ? Math.round(totalLatencyMs / totalCalls) : 0,
      totalTokens,
      totalCalls,
    }
  }

  checkAlerts() {
    const { totalCalls, failedCalls } = this.metrics
    if (totalCalls > 10 && failedCalls / totalCalls > 0.05) {
      console.error('🚨 告警：失败率超过 5%')
    }
  }
}

权限与安全边界

AI 应用上线前必须检查的安全事项：

1. API Key 管理：环境变量存储，不提交到代码仓库
2. 工具权限：Agent 能调用哪些工具、能访问哪些数据，必须显式定义
3. 输入过滤：Prompt Injection 防护（Module 2 已讲）
4. 输出过滤：敏感信息检测（PII、密码、密钥）
5. 速率限制：单用户请求频率限制，防止滥用

// 输出敏感信息过滤
function filterSensitiveOutput(text: string): string {
  return text
    .replace(/\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b/g, '[邮箱已隐藏]')
    .replace(/\b1[3-9]\d{9}\b/g, '[手机号已隐藏]')
    .replace(/sk-[a-zA-Z0-9]{20,}/g, '[API Key 已隐藏]')
}

上线 Checklist

首版必过项

□ 模型选择：任务匹配模型能力，不过度使用贵模型
□ 重试策略：指数退避 + 可重试错误判断
□ Fallback：模型级和功能级降级方案
□ 超时控制：单次请求超时 + 整体超时
□ 输入校验：长度限制 + Prompt Injection 防护
□ 输出校验：Zod Schema + 二次验证
□ 成本控制：至少有 Token 预算意识和基础阈值
□ 日志系统：至少有结构化日志
□ 安全检查：API Key 管理、权限边界、敏感信息过滤

有余力再补项

□ 缓存策略：本地基础缓存，生产可扩展为语义缓存 + TTL
□ Tracing：把多步调用串成完整链路
□ 监控告警：失败率、延迟、Token 用量
□ 压力测试：并发请求、大输入、异常输入
□ 更细的预算控制：日预算、月预算、阈值告警

本课产物

• ✅ Token 成本计算和控制策略
• ✅ 缓存策略与语义缓存设计思路
• ✅ 请求监控与 Tracing 实现思路
• ✅ 监控告警核心指标
• ✅ 上线 Checklist

完整代码在 basic/examples/08-production/02-engineering/index.ts。

试试看

cd daqi-ai-agent
pnpm exec tsx basic/examples/08-production/02-engineering/index.ts

1. 连续提几个问题后输入 cost，查看总调用次数、Token 用量和成本估算
2. 输入 monitor，查看成功率、平均延迟、P95/P99 延迟以及告警是否生效
3. 输入 checklist，对照当前程序输出核对上线前检查项

面试追问

Q：大模型应用的成本控制有哪些方法？

四个方向：1. 先统一模型打通链路，再按成本和效果做分层。2. 控制输入——裁剪对话历史、压缩上下文。3. 限制输出——设置 maxTokens。4. 语义缓存——相似问题命中缓存直接返回，不调 API。实际项目中最有效的往往是第一条和第三条。

Q：怎么评测大模型应用的效果？

三个层面：1. 离线评测——用标注数据集跑 RAGAS 指标（Faithfulness、Relevancy、Precision、Recall），得到量化分数。2. 在线监控——跟踪失败率、延迟、用户反馈。3. A/B 测试——新旧版本分流对比。离线评测发现系统性问题，在线监控发现运行时问题。

Q：如何实现 AI 应用的可观测性？

三个支柱：1. 日志——结构化日志记录每次 LLM 调用的模型、Token、延迟、状态。2. Tracing——多步调用用 traceId 串联，追踪端到端耗时。3. 指标——失败率、P99 延迟、Token 用量、缓存命中率，配合告警阈值。