为什么 AI 应用需要特殊的测试策略
普通代码的单元测试逻辑很清晰：给定输入，断言输出。但 AI 应用面对三个麻烦：
- 非确定性：同样的 prompt 每次输出不一样
- 慢：调一次 GPT-4o 要 2-5 秒
- 贵：CI 跑 100 次测试可能花几美元 token 费用
Vercel AI SDK 内置了 mock 工具解决这三个问题——让测试快、便宜、可重复。
测试工具安装
# ai/test 是 ai 包的子路径，不需要额外安装
# 确保 ai 版本 >= 6.0
npm install ai
用 MockLanguageModelV4 测试 generateText
MockLanguageModelV4 让你控制模型的输出，不发任何真实 API 请求：
import { generateText } from 'ai';
import { MockLanguageModelV4 } from 'ai/test';
// 测试：验证函数能正确处理 AI 返回的内容
it('summarize() 应该返回摘要文本', async () => {
const model = new MockLanguageModelV4({
doGenerate: async () => ({
content: [{ type: 'text', text: '这是一段自动生成的摘要内容。' }],
finishReason: { unified: 'stop', raw: undefined },
usage: {
inputTokens: { total: 50, noCache: 50, cacheRead: undefined, cacheWrite: undefined },
outputTokens: { total: 20, text: 20, reasoning: undefined },
},
warnings: [],
}),
});
const { text } = await generateText({
model,
prompt: '请总结这篇文章：...',
});
expect(text).toBe('这是一段自动生成的摘要内容。');
});
doGenerate 完全控制返回内容，测试执行时间从 3 秒降到几毫秒。
用 simulateReadableStream 测试流式输出
流式响应需要模拟分块传输，simulateReadableStream 帮你构造：
import { streamText, simulateReadableStream } from 'ai';
import { MockLanguageModelV4 } from 'ai/test';
it('streamText 应该逐块输出文本', async () => {
const model = new MockLanguageModelV4({
doStream: async () => ({
stream: simulateReadableStream({
chunkDelayInMs: 0, // 测试时不需要延迟
chunks: [
{ type: 'text-start', id: 'text-1' },
{ type: 'text-delta', id: 'text-1', delta: '你好' },
{ type: 'text-delta', id: 'text-1', delta: '，世界！' },
{ type: 'text-end', id: 'text-1' },
{
type: 'finish',
finishReason: { unified: 'stop', raw: undefined },
logprobs: undefined,
usage: {
inputTokens: { total: 10, noCache: 10, cacheRead: undefined, cacheWrite: undefined },
outputTokens: { total: 5, text: 5, reasoning: undefined },
},
},
],
}),
}),
});
const result = streamText({ model, prompt: '打招呼' });
const text = await result.text;
expect(text).toBe('你好，世界！');
});
测试结构化输出
import { generateText, Output } from 'ai';
import { MockLanguageModelV4 } from 'ai/test';
import { z } from 'zod';
const RecipeSchema = z.object({
name: z.string(),
cookTime: z.number(),
ingredients: z.array(z.string()),
});
it('结构化输出应该正确解析 JSON', async () => {
const model = new MockLanguageModelV4({
doGenerate: async () => ({
content: [{
type: 'text',
text: JSON.stringify({
name: '番茄炒蛋',
cookTime: 10,
ingredients: ['番茄', '鸡蛋', '盐'],
}),
}],
finishReason: { unified: 'stop', raw: undefined },
usage: {
inputTokens: { total: 30, noCache: 30, cacheRead: undefined, cacheWrite: undefined },
outputTokens: { total: 40, text: 40, reasoning: undefined },
},
warnings: [],
}),
});
const { output } = await generateText({
model,
output: Output.object({ schema: RecipeSchema }),
prompt: '给我一个番茄炒蛋的食谱',
});
expect(output.name).toBe('番茄炒蛋');
expect(output.cookTime).toBe(10);
expect(output.ingredients).toHaveLength(3);
});
可观测性：用 OpenTelemetry 监控生产环境
测试保证代码正确，可观测性保证生产环境不出意外。AI SDK 内置了 OpenTelemetry 支持，接入 Langfuse、LangSmith 等平台只需几行代码。
基础配置
npm install @ai-sdk/otel @opentelemetry/sdk-node
在应用入口（instrumentation.ts for Next.js，或 index.ts）注册：
import { registerTelemetry } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';
// Next.js：放在 instrumentation.ts 的 register() 函数里
export async function register() {
registerTelemetry(new OpenTelemetry());
}
给每次 AI 调用打上标签
const result = await generateText({
model: openai('gpt-4o'),
prompt: '...',
telemetry: {
functionId: 'user-onboarding-summary', // 标识这是哪个功能的调用
},
});
// streamText 同样支持
const stream = streamText({
model: openai('gpt-4o'),
messages,
telemetry: {
functionId: 'chat-session',
},
});
functionId 会出现在 Langfuse / LangSmith 的每条 trace 里，让你按功能筛选、比较不同版本的 prompt 效果。
接入 Langfuse（推荐）
Langfuse 是目前最流行的 LLM 可观测性平台，有免费自托管版：
npm install langfuse-vercel
import { registerTelemetry } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';
import { LangfuseExporter } from 'langfuse-vercel';
registerTelemetry(
new OpenTelemetry({
exporter: new LangfuseExporter({
publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
secretKey: process.env.LANGFUSE_SECRET_KEY!,
baseUrl: 'https://cloud.langfuse.com',
}),
})
);
配置好之后，每次 generateText / streamText 调用会自动上报：
- 完整的 messages（输入/输出）
- Token 用量和成本估算
- 延迟（到第一个 token、总时长）
- 工具调用详情
- 错误信息
在 Langfuse 控制台就能看到每个用户、每个功能的 AI 调用明细，排查问题很方便。
自定义 Telemetry（轻量版，无需 OpenTelemetry）
如果你只想简单打日志，不想装 OTel，可以实现 Telemetry 接口：
import type { Telemetry } from 'ai';
import { registerTelemetry } from 'ai';
class ConsoleLogger implements Telemetry {
async onStart(event: any) {
console.log(`[AI] 开始调用: ${event.modelId}`);
}
async onStepEnd(event: any) {
const { inputTokens, outputTokens } = event.usage;
console.log(`[AI] 第 ${event.stepNumber} 步完成: in=${inputTokens.total} out=${outputTokens.total}`);
}
async onEnd(event: any) {
const totalTokens = event.usage.inputTokens.total + event.usage.outputTokens.total;
console.log(`[AI] 调用结束，总 tokens: ${totalTokens}`);
}
}
registerTelemetry(new ConsoleLogger());
实际项目建议
测试分层：
单元测试（MockLanguageModelV4）
→ 验证业务逻辑：prompt 构造正确吗？返回值处理正确吗？
→ 运行时间：毫秒级，CI 全量跑
集成测试（真实 API，但用便宜模型）
→ 验证 prompt 能引导模型给出正确格式的回答
→ 每个核心 prompt 跑一次，用 Gemini Flash 省钱
生产监控（Langfuse / OpenTelemetry）
→ 追踪实际成本、延迟、错误率
→ 发现 prompt 退化（同一个输入，某次之后质量变差）
关键指标要盯：
- P95 首 token 延迟（用户体感响应速度）
- 每次调用平均 token 数（成本控制）
- 工具调用失败率（Agent 健壮性）
finish_reason分布（length太多 = context 不够）
把这两章（MCP 集成 + 测试可观测性）加进你的项目，AI 功能才算真正生产就绪。