Vercel AI SDK 实战:用 TypeScript 构建生产级流式 AI 应用

深入解析 Vercel AI SDK 核心架构与 API,对比 OpenAI SDK、LangChain.js 方案差异,涵盖流式对话、Tool Calling、结构化输出、多模型切换等实战场景,附完整可运行代码。

前端开发 2026-05-29 14 分钟

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 Coreai 包):服务端核心,提供 generateTextstreamTextgenerateObjectstreamObject 四大函数
  • AI SDK UI@ai-sdk/react 等):前端 Hooks,提供 useChatuseCompletionuseObject
  • 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 层。

✅ 五、最佳实践清单

经过多个生产项目的验证,以下是最关键的实践建议:

  1. 始终配置 maxRetries 和超时 — AI API 不稳定是常态,不是异常
  2. 使用 Zod Schema 约束输出 — 不要靠 prompt 约束 JSON 格式,generateObject 远比手动解析可靠
  3. 做好消息裁剪 — 对话轮次超过 20 轮就要考虑裁剪或摘要压缩
  4. 分离 system prompt — 把 system prompt 放在环境变量或配置文件中,方便调优而不改代码
  5. 记录每次调用的 Token 用量 — 没有监控就没有优化,账单来了才后悔就晚了
  6. 前端用 useChat 而非手写 — 它处理了你想不到的边界情况(网络断开、流中断、并发请求等)
  7. 生产环境不用 streamText 的默认 temperature — 显式设定 temperaturemaxTokens

🎯 总结

Vercel AI SDK 的核心价值不是「少写几行代码」,而是提供了一套经过验证的 AI 应用工程范式。它把流式传输、错误重试、类型安全、多模型适配这些容易出错的工程细节标准化了,让你可以把精力集中在产品逻辑上。

如果你正在用 TypeScript 构建 AI 应用,我的建议是:从 AI SDK 开始,在遇到复杂 Agent 编排需求时再引入 LangChain。不要一上来就用重量级框架,也不要从零手写流式处理逻辑。

相关资源:

📚 相关文章