2026 年,AI 应用开发已经从「调 API 拿结果」演进到了「构建沉浸式交互体验」。用户不再接受等待 10 秒才看到完整回答——流式输出(Streaming) 已成为标配。但自己实现一套可靠的流式 AI 前后端架构,涉及 SSE 协议处理、Token 解析、错误重试、多模型适配等大量工程细节。Vercel AI SDK(ai 包)正是为解决这些痛点而生,目前 npm 周下载量已超过 200 万,成为 TypeScript 生态中最流行的 AI 应用开发框架。
🚀 一、Vercel AI SDK 核心架构与设计理念
1.1 为什么不用 OpenAI SDK 直接调?
很多开发者的第一反应是:我直接用 openai 官方 SDK 不就行了?对于最简单的场景确实够用,但一旦涉及以下需求,你会发现手写代码量急剧膨胀:
| 需求场景 | 直接用 OpenAI SDK | 用 Vercel AI SDK |
|---|---|---|
| 流式文本输出 | 手动处理 ReadableStream + SSE 解析 | streamText() 一行搞定 |
| 前端流式渲染 | 自己写 EventSource + 状态管理 | useChat() Hook 自动处理 |
| Tool Calling | 手动解析 tool_calls、管理循环 | 内置自动工具调用循环 |
| 切换模型供应商 | 每家 SDK 不同,代码要重写 | 统一接口,改一行 provider |
| 结构化输出 | 手动 JSON Schema + 校验 | generateObject() + Zod |
| 多轮对话 | 手动维护 messages 数组 | Hook 自动管理状态 |
⚠️ 警告: 如果你的项目只用一次简单的文本生成,直接用官方 SDK 更轻量。Vercel AI SDK 的价值在于复杂 AI 应用的工程化,不是替代所有场景。
1.2 架构分层
Vercel AI SDK 分为三个层次:
- AI SDK Core(
ai包):服务端核心,提供generateText、streamText、generateObject、streamObject四大函数 - AI SDK UI(
@ai-sdk/react等):前端 Hooks,提供useChat、useCompletion、useObject - AI SDK Providers(
@ai-sdk/openai、@ai-sdk/anthropic等):各模型供应商适配层
// 安装核心依赖 — 以 OpenAI 为例
// npm install ai @ai-sdk/openai
import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
// 服务端:一行代码发起流式请求
const result = streamText({
model: openai('gpt-4o'),
prompt: '用 TypeScript 实现一个 LRU 缓存',
})
这种分层设计让你可以只用 Core 层(纯后端场景),也可以只用 UI 层配合自定义后端,灵活度很高。
🔧 二、四大核心 API 深度实战
2.1 streamText — 流式文本生成
streamText 是使用频率最高的 API。它返回一个 StreamTextResult 对象,支持多种消费方式:
// 服务端 API 路由(Next.js App Router 示例)
// app/api/chat/route.ts
import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
export async function POST(req: Request) {
const { messages } = await req.json()
const result = streamText({
model: openai('gpt-4o'),
system: '你是一个资深 TypeScript 开发者,回答要简洁、有代码示例。',
messages,
// 控制生成行为的关键参数
maxTokens: 2048,
temperature: 0.7,
// 自动重试配置 — 生产环境必备
maxRetries: 3,
})
// 直接返回流式响应,SDK 自动处理 SSE 协议
return result.toDataStreamResponse()
}
💡 提示:
toDataStreamResponse()使用的是 AI SDK 自定义的流协议(基于 Data Stream Protocol),比原生 SSE 更高效——它支持传输文本、工具调用、错误信息等多种消息类型,且自动处理了客户端重连逻辑。
streamText 还支持 onFinish 回调,非常适合做日志记录和用量统计:
const result = streamText({
model: openai('gpt-4o'),
messages,
onFinish({ text, usage, finishReason }) {
// 记录 Token 用量 — 用于成本核算
console.log(`输入: ${usage.promptTokens}, 输出: ${usage.completionTokens}`)
console.log(`费用估算: $${(usage.promptTokens * 0.005 / 1000 + usage.completionTokens * 0.015 / 1000).toFixed(4)}`)
// 存入数据库用于计费
db.usage.create({
promptTokens: usage.promptTokens,
completionTokens: usage.completionTokens,
model: 'gpt-4o',
})
},
})
2.2 generateObject — 结构化数据生成
让 LLM 输出 JSON 是最常见的需求之一,但也是坑最多的。直接让模型输出 JSON,你大概率会遇到:
- ❌ 输出被 Markdown 代码块包裹(
```json ... ```) - ❌ 末尾多了逗号(trailing comma)
- ❌ 字段类型不对(number 变成了 string)
- ❌ 缺少必填字段
generateObject 通过 Zod Schema 约束 + JSON Mode 彻底解决了这些问题:
// 用 Zod 定义严格的输出 Schema
import { generateObject } from 'ai'
import { openai } from '@ai-sdk/openai'
import { z } from 'zod'
const codeReviewSchema = z.object({
score: z.number().min(0).max(100).describe('代码质量评分'),
issues: z.array(z.object({
severity: z.enum(['error', 'warning', 'info']),
line: z.number(),
message: z.string(),
suggestion: z.string(),
})),
summary: z.string().describe('总体评价,一句话总结'),
autoFixable: z.boolean().describe('是否可以自动修复'),
})
const { object } = await generateObject({
model: openai('gpt-4o'),
schema: codeReviewSchema,
prompt: `审查以下 TypeScript 代码并打分:
\`\`\`typescript
function fetchData(url: string) {
const res = fetch(url)
return res.json()
}
\`\`\``,
})
// object 的类型自动推断为 z.infer<typeof codeReviewSchema>
// TypeScript 完全类型安全,无需手动 as
console.log(object.score) // number
console.log(object.issues[0].severity) // 'error' | 'warning' | 'info'
📌 记住:
generateObject内部使用了结构化输出(Structured Outputs)功能,OpenAI 和 Anthropic 的模型都支持。它不是靠 prompt 约束输出格式,而是在 API 层面强制 JSON Schema 验证,准确率接近 100%。
2.3 useChat — 前端对话 Hook
useChat 是整个 SDK 中设计最精妙的部分。它把流式 AI 对话的所有状态管理都封装好了:
<!-- Nuxt 3 / Vue 3 示例 — 配合 @ai-sdk/vue 使用 -->
<script setup lang="ts">
import { useChat } from '@ai-sdk/vue'
const { messages, input, handleSubmit, isLoading, error, stop, reload } = useChat({
api: '/api/chat',
// 可选:自定义请求头(如认证 token)
headers: { Authorization: `Bearer ${getToken()}` },
// 可选:错误处理
onError: (err) => {
console.error('聊天出错:', err.message)
},
// 可选:消息发送完成回调
onFinish: (message) => {
console.log('AI 回复完成:', message.content.slice(0, 50))
},
})
</script>
<template>
<div class="chat-container">
<div v-for="msg in messages" :key="msg.id" :class="msg.role">
<strong>{{ msg.role === 'user' ? '你' : 'AI' }}:</strong>
<p>{{ msg.content }}</p>
</div>
<form @submit="handleSubmit">
<input v-model="input" :disabled="isLoading" placeholder="输入消息..." />
<button type="submit" :disabled="isLoading">发送</button>
<button v-if="isLoading" type="button" @click="stop">停止</button>
</form>
</div>
</template>
useChat 自动帮你处理了:消息列表管理、流式文本追加、loading 状态、错误状态、停止生成、重新生成、自动滚动等。如果手写这些逻辑,至少需要 200+ 行代码。
2.4 多模型切换 — 统一接口的威力
Vercel AI SDK 最大的架构优势在于模型无关。切换模型只需更换 provider,不需要改任何业务逻辑:
// 切换模型只需改这一行
import { openai } from '@ai-sdk/openai'
import { anthropic } from '@ai-sdk/anthropic'
import { google } from '@ai-sdk/google'
// 方案一:直接硬编码
const model = openai('gpt-4o')
// 方案二:通过环境变量动态切换(推荐生产环境使用)
const providers = {
openai: () => openai(process.env.OPENAI_MODEL || 'gpt-4o'),
anthropic: () => anthropic(process.env.ANTHROPIC_MODEL || 'claude-sonnet-4-20250514'),
google: () => google(process.env.GOOGLE_MODEL || 'gemini-2.0-flash'),
}
const provider = process.env.AI_PROVIDER || 'openai'
const model = providers[provider]()
// 下面的代码完全不变
const result = streamText({ model, messages })
这对成本优化和容灾降级非常有价值。比如你可以设定:简单任务用 Gemini Flash(便宜快速),复杂推理用 Claude,代码生成用 GPT-4o。甚至可以在一个请求失败时自动 fallback 到另一个模型。
⚡ 三、生产环境避坑指南
3.1 流式响应的内存泄漏问题
这是最常见的生产 Bug。如果你的 onFinish 回调中有异步操作(如写数据库),但没有 await,在高并发下会导致连接池耗尽:
// ❌ 错误写法 — 异步操作未 await
const result = streamText({
model: openai('gpt-4o'),
messages,
onFinish({ text, usage }) {
// 这个 Promise 没有 await,会静默失败
db.usage.create({ data: { tokens: usage.completionTokens } })
},
})
// ✅ 正确写法 — 确保异步操作完成
const result = streamText({
model: openai('gpt-4o'),
messages,
async onFinish({ text, usage }) {
await db.usage.create({ data: { tokens: usage.completionTokens } })
await cache.set(`last-response:${userId}`, text, { ex: 3600 })
},
})
// 最稳妥的方式:使用 experimental_onFinish 并捕获错误
const result = streamText({
model: openai('gpt-4o'),
messages,
async onFinish({ text, usage }) {
try {
await db.usage.create({ data: { tokens: usage.completionTokens } })
} catch (err) {
// 记录日志但不影响用户响应
console.error('记录用量失败:', err)
}
},
})
3.2 Token 限制与上下文窗口管理
很多开发者只设置了 maxTokens(输出限制),却忽略了输入 Token 超限的问题。当对话轮次变多,messages 数组会超出模型的上下文窗口:
| 模型 | 上下文窗口 | 输入 Token 价格(每百万) | 输出 Token 价格(每百万) |
|---|---|---|---|
| GPT-4o | 128K | $2.50 | $10.00 |
| Claude Sonnet | 200K | $3.00 | $15.00 |
| Gemini 2.0 Flash | 1M | $0.10 | $0.40 |
| DeepSeek V3 | 128K | $0.27 | $1.10 |
⚡ 关键结论: 如果你的对话场景会积累大量历史消息,务必做消息裁剪。Gemini 的 1M 上下文虽然大,但价格也会线性增长。
import { streamText, type CoreMessage } from 'ai'
// 消息裁剪策略 — 保留 system prompt + 最近 N 条
function trimMessages(messages: CoreMessage[], maxMessages = 20): CoreMessage[] {
// 保留第一条 system message
const systemMsg = messages.find(m => m.role === 'system')
const nonSystemMsgs = messages.filter(m => m.role !== 'system')
// 只保留最近的对话
const recentMsgs = nonSystemMsgs.slice(-maxMessages)
return systemMsg ? [systemMsg, ...recentMsgs] : recentMsgs
}
export async function POST(req: Request) {
const { messages } = await req.json()
const trimmed = trimMessages(messages, 20)
const result = streamText({
model: openai('gpt-4o'),
messages: trimmed,
maxTokens: 4096,
})
return result.toDataStreamResponse()
}
3.3 错误处理与重试策略
AI API 的错误率比普通 REST API 高得多——限流(429)、服务过载(503)、超时都是家常便饭。SDK 内置了指数退避重试,但你需要正确配置:
import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
const result = streamText({
model: openai('gpt-4o'),
messages,
// 生产环境推荐配置
maxRetries: 3,
// 自定义 abort signal — 防止请求挂起太久
abortSignal: AbortSignal.timeout(30_000), // 30 秒超时
})
// 前端也需要处理流式中断的情况
// useChat 的 error 状态会自动捕获
⚠️ 警告: 永远不要在前端直接调用 AI API——你的 API Key 会暴露。必须通过后端 API 路由中转。即使使用
dangerouslyAllowBrowser: true也只适合原型开发。
3.4 成本控制实战
AI 应用的账单很容易失控。以下是经过验证的成本控制策略:
- ✅ 对简单任务用小模型(Gemini Flash、GPT-4o-mini),复杂任务用大模型
- ✅ 设置合理的
maxTokens,防止模型「话痨」 - ✅ 使用
onFinish记录每次调用的 Token 用量,设定预算告警 - ✅ 对用户输入做长度限制,防止恶意输入消耗大量输入 Token
- ❌ 不要对所有请求都用最高级模型,80% 的请求用小模型就够了
- ❌ 不要把 system prompt 写得过长,每轮对话都要重复计费
💡 四、与替代方案的深度对比
| 维度 | Vercel AI SDK | LangChain.js | OpenAI SDK | 自己封装 |
|---|---|---|---|---|
| 包大小 | ~50KB(核心) | ~500KB+ | ~100KB | 0 |
| 学习曲线 | 低 | 高 | 低 | 中 |
| 流式支持 | 开箱即用 | 需手动配置 | 部分支持 | 全手写 |
| 多模型切换 | 一行代码 | 通过 Runnable | 需换 SDK | 重写代码 |
| 前端 Hooks | useChat 等 | 无 | 无 | 全手写 |
| Tool Calling | 自动循环 | 需自己编排 | 基础支持 | 全手写 |
| TypeScript 类型 | 优秀 | 一般 | 良好 | 看水平 |
| 适用场景 | 全栈 AI 应用 | 复杂 Agent 编排 | 简单 API 调用 | 极端定制 |
⚡ 关键结论: 如果你在用 Next.js / Nuxt 构建 AI 应用,Vercel AI SDK 是当前最优解。如果你需要复杂的 Agent 编排(多步骤推理、条件分支、循环),LangChain.js 的 Runnable 更灵活。两者也可以结合使用——用 LangChain 做编排,用 AI SDK 做 UI 层。
✅ 五、最佳实践清单
经过多个生产项目的验证,以下是最关键的实践建议:
- 始终配置
maxRetries和超时 — AI API 不稳定是常态,不是异常 - 使用 Zod Schema 约束输出 — 不要靠 prompt 约束 JSON 格式,
generateObject远比手动解析可靠 - 做好消息裁剪 — 对话轮次超过 20 轮就要考虑裁剪或摘要压缩
- 分离 system prompt — 把 system prompt 放在环境变量或配置文件中,方便调优而不改代码
- 记录每次调用的 Token 用量 — 没有监控就没有优化,账单来了才后悔就晚了
- 前端用
useChat而非手写 — 它处理了你想不到的边界情况(网络断开、流中断、并发请求等) - 生产环境不用
streamText的默认 temperature — 显式设定temperature和maxTokens
🎯 总结
Vercel AI SDK 的核心价值不是「少写几行代码」,而是提供了一套经过验证的 AI 应用工程范式。它把流式传输、错误重试、类型安全、多模型适配这些容易出错的工程细节标准化了,让你可以把精力集中在产品逻辑上。
如果你正在用 TypeScript 构建 AI 应用,我的建议是:从 AI SDK 开始,在遇到复杂 Agent 编排需求时再引入 LangChain。不要一上来就用重量级框架,也不要从零手写流式处理逻辑。
相关资源:
- 🔗 Vercel AI SDK 官方文档
- 🔗 AI SDK GitHub 仓库
- 🔗 AI SDK Providers 列表
- 🔗 本文配套工具:JSON 格式化 | TypeScript 格式化 | Base64 编解码