Vercel AI SDK 全景解析

从零手写流式 AI 接口需要处理 ReadableStream、TextEncoder、SSE 协议、错误恢复、客户端状态管理——大量样板代码。Vercel AI SDK 把这些全部封装好，让你用几行代码就能构建生产级 AI 应用。本章完整讲解 AI SDK 的三层架构、核心 API 和多模型适配策略。

1. AI SDK 架构总览

AI SDK 分为三层，各司其职。这种分层借鉴了经典的依赖反转原则——业务代码依赖抽象接口（Core 层），而非具体的模型提供商。当 OpenAI 出故障、Anthropic 发布更好的模型时，你只需要改 Provider 层的一行代码，Core 层和 UI 层完全不动。

┌─────────────────────────────────────────────┐
│  AI SDK UI（前端）                            │
│  useChat / useCompletion / useObject         │
│  自动管理消息状态、流式渲染、加载/错误状态      │
├─────────────────────────────────────────────┤
│  AI SDK Core（核心层）                        │
│  generateText / streamText / generateObject  │
│  统一的模型调用接口，与具体提供商无关           │
├─────────────────────────────────────────────┤
│  AI SDK Providers（模型适配层）                │
│  @ai-sdk/openai / @ai-sdk/anthropic / ...   │
│  每个提供商一个包，实现统一接口                 │
└─────────────────────────────────────────────┘

AI SDK Core（ai 包）：提供 streamText、generateText、generateObject 等函数，在服务端调用
AI SDK UI（ai/react）：提供 useChat、useCompletion 等 React Hook，在客户端使用
AI SDK Providers：每个 AI 提供商一个独立的包（@ai-sdk/openai、@ai-sdk/anthropic 等）

这种分层设计的好处：切换模型只需要改一行代码——业务逻辑和 UI 完全不变。

1.1 安装

# 核心包（必装）
pnpm add ai
 
# 模型提供商（按需安装）
pnpm add @ai-sdk/openai      # OpenAI / GPT-4o
pnpm add @ai-sdk/anthropic   # Anthropic / Claude
pnpm add @ai-sdk/google      # Google / Gemini
pnpm add @ai-sdk/mistral     # Mistral

1.2 环境变量

# .env.local
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_GENERATIVE_AI_API_KEY=...

AI SDK 的 Provider 会自动读取对应的环境变量——不需要手动传 API Key。

2. AI SDK Core：服务端 API

Core 层运行在服务端（Route Handler / Server Action），负责调用 AI 模型。

2.1 streamText — 流式文本生成

最常用的函数。调用大模型，返回流式文本响应。

为什么需要流式？ LLM 生成一段 500 字的回复可能需要 3-5 秒。如果用 generateText 等全部生成完再返回，用户会盯着空白屏幕 5 秒——体验很差。流式输出让第一个 Token 在 200-500ms 内就到达用户眼前（这个指标叫 TTFT, Time to First Token），用户看到文字逐字出现，感觉 AI "正在思考"，等待感大大降低。

流式的底层原理：LLM 是逐 Token 生成的（自回归模型），每生成一个 Token 就可以立即发送给客户端。streamText 利用 HTTP 的 Transfer-Encoding: chunked 或 SSE（Server-Sent Events），把每个 Token 实时推送到浏览器。

// app/api/chat/route.ts
import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
 
export async function POST(request: Request) {
  const { messages } = await request.json()
 
  const result = streamText({
    model: openai('gpt-4o'),
    messages,
  })
 
  return result.toDataStreamResponse()
}

三行核心代码做了所有事情：调用 OpenAI、创建流、构建 Response。toDataStreamResponse() 内部创建了一个 ReadableStream，把 AI 模型的逐 Token 输出转换成 AI SDK 的 Data Stream 协议格式，再包装成标准的 Response 对象。如果你手动实现同样的功能，需要处理 ReadableStream、TextEncoder、SSE 格式、错误恢复等至少 50 行代码。

streamText 的完整参数：

const result = streamText({
  model: openai('gpt-4o'),          // 模型
  messages,                          // 消息历史
  system: '你是一个专业的 AI 助手',    // 系统提示词
  temperature: 0.7,                  // 创造性（0-2）
  maxTokens: 4096,                   // 最大输出 Token
  topP: 0.9,                         // 核采样
  frequencyPenalty: 0.5,             // 频率惩罚
  presencePenalty: 0.5,              // 存在惩罚
  stopSequences: ['---'],            // 停止序列
  tools: { /* ... */ },              // 工具定义（第 35 章详讲）
  maxSteps: 5,                       // 工具调用最大步数
  onFinish: async ({ text, usage }) => {
    // 流完成后的回调：保存数据库、记录 Token 用量
    await saveMessage(text)
    await recordUsage(usage)
  },
})

2.2 generateText — 非流式文本生成

不需要流式输出时使用——比如在 Server Action 中生成摘要、在后台任务中批量处理。和 streamText 的区别：generateText 等 AI 把所有 Token 生成完毕后，一次性返回完整结果。适合后台任务（用户不在等待）和需要完整结果才能继续的场景（比如生成标题后要存数据库）。

import { generateText } from 'ai'
import { openai } from '@ai-sdk/openai'
 
export async function generateSummary(content: string) {
  const { text, usage } = await generateText({
    model: openai('gpt-4o-mini'),
    prompt: `请用 3 句话总结以下内容：\n\n${content}`,
  })
 
  return { summary: text, tokens: usage.totalTokens }
}

generateText 返回完整结果对象：

const result = await generateText({ ... })
 
result.text           // 生成的文本
result.usage          // { promptTokens, completionTokens, totalTokens }
result.finishReason   // 'stop' | 'length' | 'tool-calls' | 'error'
result.toolCalls      // 工具调用结果（如果有）
result.toolResults    // 工具执行结果

2.3 generateObject — 结构化输出

让 AI 返回结构化 JSON 而非自由文本。用 zod 定义 Schema，AI SDK 保证输出严格符合 Schema。

原理：传统做法是在 Prompt 里写 "请返回 JSON 格式"——但 AI 可能返回多余的文字、漏掉字段、或者格式不对。generateObject 的底层使用了 OpenAI 的 Structured Outputs（或其他模型的 JSON Mode + Function Calling）。它把 zod Schema 转换成 JSON Schema 发给模型，模型在生成时受到 Schema 的 约束解码（Constrained Decoding）——每一步只允许生成符合 Schema 的 Token，从根本上保证输出格式正确。这不是 "希望" AI 返回正确 JSON，而是强制它返回正确 JSON。

import { generateObject } from 'ai'
import { openai } from '@ai-sdk/openai'
import { z } from 'zod'
 
const sentimentSchema = z.object({
  sentiment: z.enum(['positive', 'negative', 'neutral']),
  confidence: z.number().min(0).max(1),
  keywords: z.array(z.string()).max(5),
  summary: z.string(),
})
 
export async function analyzeSentiment(text: string) {
  const { object } = await generateObject({
    model: openai('gpt-4o'),
    schema: sentimentSchema,
    prompt: `分析以下文本的情感倾向：\n\n${text}`,
  })
 
  // object 的类型是 { sentiment: 'positive' | ..., confidence: number, ... }
  return object
}

这比让 AI "请返回 JSON 格式" 再手动解析可靠得多——AI SDK 内部使用了 OpenAI 的 Structured Outputs（或其他模型的 JSON mode），保证格式正确。

2.4 streamObject — 流式结构化输出

实时展示结构化数据的生成过程（比如逐步填充表格）：

import { streamObject } from 'ai'
 
const result = streamObject({
  model: openai('gpt-4o'),
  schema: z.object({
    title: z.string(),
    sections: z.array(z.object({
      heading: z.string(),
      content: z.string(),
    })),
  }),
  prompt: '写一篇关于 Next.js 性能优化的文章大纲',
})
 
return result.toTextStreamResponse()

3. AI SDK UI：前端 Hook

3.1 useChat — 对话交互

useChat 是 AI SDK 最核心的前端 Hook——自动管理消息列表、发送请求、处理流式响应、维护加载状态：

'use client'
 
import { useChat } from 'ai/react'
 
export function ChatUI() {
  const {
    messages,        // Message[] — 完整的消息历史
    input,           // string — 当前输入框内容
    handleInputChange, // 输入框 onChange 处理
    handleSubmit,    // 表单 onSubmit 处理
    isLoading,       // boolean — AI 是否正在回复
    stop,            // () => void — 停止生成
    reload,          // () => void — 重新生成最后一条回复
    error,           // Error | undefined — 错误信息
    setMessages,     // 手动设置消息列表
    append,          // 编程式追加消息
  } = useChat({
    api: '/api/chat',           // 后端接口（默认 /api/chat）
    initialMessages: [],         // 初始消息（从数据库加载历史）
    body: { model: 'gpt-4o' },  // 额外参数，每次请求都携带
    onFinish: (message) => {
      // AI 回复完成后回调
      console.log('AI 回复完成：', message.content)
    },
    onError: (error) => {
      toast.error('AI 回复失败：' + error.message)
    },
  })
 
  return (
    <div className="flex flex-col h-full">
      {/* 消息列表 */}
      <div className="flex-1 overflow-y-auto p-4 space-y-4">
        {messages.map((msg) => (
          <div
            key={msg.id}
            className={msg.role === 'user' ? 'text-right' : 'text-left'}
          >
            <div className={`inline-block p-3 rounded-lg max-w-[80%] ${
              msg.role === 'user'
                ? 'bg-blue-500 text-white'
                : 'bg-gray-100 text-gray-900'
            }`}>
              {msg.content}
            </div>
          </div>
        ))}
      </div>
 
      {/* 输入框 */}
      <form onSubmit={handleSubmit} className="p-4 border-t flex gap-2">
        <input
          value={input}
          onChange={handleInputChange}
          placeholder="输入消息..."
          className="flex-1 border rounded-lg px-4 py-2"
          disabled={isLoading}
        />
        {isLoading ? (
          <button type="button" onClick={stop} className="px-4 py-2 bg-red-500 text-white rounded-lg">
            停止
          </button>
        ) : (
          <button type="submit" className="px-4 py-2 bg-blue-500 text-white rounded-lg">
            发送
          </button>
        )}
      </form>
    </div>
  )
}

useChat 内部做了大量工作：

用户点击发送 → 把消息追加到 messages
发起 POST 请求到 /api/chat，携带完整消息历史
接收 AI SDK 的 Data Stream 协议，逐 Token 更新 messages 中最后一条 AI 回复
流完成后触发 onFinish，设置 isLoading = false

3.2 useCompletion — 单次补全

比 useChat 更简单——输入一段文本，AI 返回补全结果。适合摘要生成、翻译、代码补全等场景：

'use client'
 
import { useCompletion } from 'ai/react'
 
export function SummaryGenerator() {
  const { completion, input, handleInputChange, handleSubmit, isLoading } = useCompletion({
    api: '/api/completion',
  })
 
  return (
    <div>
      <form onSubmit={handleSubmit}>
        <textarea
          value={input}
          onChange={handleInputChange}
          placeholder="粘贴需要总结的文章..."
          rows={10}
        />
        <button type="submit" disabled={isLoading}>
          {isLoading ? '生成中...' : '生成摘要'}
        </button>
      </form>
      {completion && (
        <div className="mt-4 p-4 bg-gray-50 rounded-lg">
          <h3>摘要：</h3>
          <p>{completion}</p>
        </div>
      )}
    </div>
  )
}

服务端只需要用 streamText 返回 toTextStreamResponse()：

// app/api/completion/route.ts
import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
 
export async function POST(request: Request) {
  const { prompt } = await request.json()
 
  const result = streamText({
    model: openai('gpt-4o-mini'),
    prompt: `请用中文生成以下内容的精炼摘要：\n\n${prompt}`,
  })
 
  return result.toTextStreamResponse()
}

3.3 useObject — 流式结构化数据

配合 streamObject 使用——实时渲染结构化数据的生成过程：

'use client'
 
import { useObject } from 'ai/react'
import { z } from 'zod'
 
const articleSchema = z.object({
  title: z.string(),
  outline: z.array(z.object({
    heading: z.string(),
    keyPoints: z.array(z.string()),
  })),
  estimatedReadTime: z.number(),
})
 
export function ArticleOutlineGenerator() {
  const { object, submit, isLoading } = useObject({
    api: '/api/outline',
    schema: articleSchema,
  })
 
  return (
    <div>
      <button onClick={() => submit({ topic: 'Next.js 性能优化' })} disabled={isLoading}>
        生成大纲
      </button>
 
      {/* object 在流式生成过程中逐步填充，字段可能是 undefined */}
      {object && (
        <div>
          <h2>{object.title ?? '生成中...'}</h2>
          {object.outline?.map((section, i) => (
            <div key={i}>
              <h3>{section.heading}</h3>
              <ul>
                {section.keyPoints?.map((point, j) => <li key={j}>{point}</li>)}
              </ul>
            </div>
          ))}
          {object.estimatedReadTime && <p>预计阅读 {object.estimatedReadTime} 分钟</p>}
        </div>
      )}
    </div>
  )
}

4. Data Stream 协议

4.1 为什么不用纯 SSE

标准 SSE（Server-Sent Events）只能传文本。但 AI 对话需要传输的信息远不止文本——还有工具调用请求、工具执行结果、Token 使用统计、完成原因等结构化元数据。如果用纯 SSE，你得自己定义协议格式并在客户端解析。

AI SDK 设计了 Data Stream 协议——在 SSE 基础上，用类型前缀区分不同数据类型。useChat 自动解析这个协议，把文本追加到消息、把工具调用触发回调、把完成信号用于更新状态。你通常不需要直接处理它，但理解它有助于调试。

4.2 协议格式

useChat 和 toDataStreamResponse() 之间用 Data Stream 协议 通信——它是一种扩展的 SSE 格式，支持传输文本、工具调用、元数据等多种类型：

0:"Hello"           → 文本 Token
0:" World"          → 文本 Token
9:{"toolCallId":"abc","toolName":"search","args":{"q":"test"}}  → 工具调用
a:{"toolCallId":"abc","result":"..."}  → 工具结果
d:{"finishReason":"stop","usage":{"promptTokens":10,"completionTokens":20}}  → 完成信号

每行的第一个字符是类型标识：

0：文本内容
2：数据（附加信息）
9：工具调用开始
a：工具调用结果
d：流完成
e：错误

你通常不需要直接处理这个协议——useChat 会自动解析。但理解它有助于调试：当流式响应出问题时，打开浏览器 DevTools 的 Network 面板，查看 Response 的原始文本，就能看到这些带类型前缀的行，快速定位是文本没传出来、还是工具调用卡住了。

5. 多模型适配

5.1 切换模型

AI SDK 最大的设计优势——切换模型只需改一行。这得益于 Provider 抽象层的设计：每个 Provider 包（@ai-sdk/openai、@ai-sdk/anthropic）都实现了统一的 LanguageModel 接口。streamText 和 generateText 只依赖这个接口，不关心底层是 OpenAI 的 REST API 还是 Anthropic 的 Messages API——协议差异全部被 Provider 封装掉了。

这种设计在实际项目中极其重要：AI 行业变化快，今天最好的模型明天可能被超越。如果你的代码和某个模型 API 耦合，切换成本很高。有了 Provider 抽象，切换模型就是改一个字符串的事。

import { openai } from '@ai-sdk/openai'
import { anthropic } from '@ai-sdk/anthropic'
import { google } from '@ai-sdk/google'
 
// 用 GPT-4o
streamText({ model: openai('gpt-4o'), messages })
 
// 切换到 Claude
streamText({ model: anthropic('claude-sonnet-4-20250514'), messages })
 
// 切换到 Gemini
streamText({ model: google('gemini-2.0-flash'), messages })

5.2 动态模型选择

让用户自己选模型——服务端根据参数实例化对应 Provider：

// lib/ai-models.ts
import { openai } from '@ai-sdk/openai'
import { anthropic } from '@ai-sdk/anthropic'
import { google } from '@ai-sdk/google'
 
const models = {
  'gpt-4o': openai('gpt-4o'),
  'gpt-4o-mini': openai('gpt-4o-mini'),
  'claude-sonnet': anthropic('claude-sonnet-4-20250514'),
  'gemini-flash': google('gemini-2.0-flash'),
} as const
 
export type ModelId = keyof typeof models
 
export function getModel(modelId: ModelId) {
  const model = models[modelId]
  if (!model) throw new Error(`不支持的模型：${modelId}`)
  return model
}

// app/api/chat/route.ts
import { streamText } from 'ai'
import { getModel } from '@/lib/ai-models'
 
export async function POST(request: Request) {
  const { messages, model: modelId } = await request.json()
 
  const result = streamText({
    model: getModel(modelId),
    messages,
  })
 
  return result.toDataStreamResponse()
}

// 客户端传递模型选择
const { messages, handleSubmit } = useChat({
  body: { model: selectedModel }, // 'gpt-4o' | 'claude-sonnet' | ...
})

5.3 模型 Fallback

主模型失败时自动降级到备选模型：

import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
import { anthropic } from '@ai-sdk/anthropic'
 
export async function POST(request: Request) {
  const { messages } = await request.json()
 
  // 定义模型优先级
  const modelChain = [
    openai('gpt-4o'),
    anthropic('claude-sonnet-4-20250514'),
    openai('gpt-4o-mini'),   // 最后的兜底模型
  ]
 
  for (const model of modelChain) {
    try {
      const result = streamText({ model, messages })
      return result.toDataStreamResponse()
    } catch (error) {
      console.warn(`模型调用失败，尝试下一个：`, error)
      continue
    }
  }
 
  return Response.json({ error: '所有模型均不可用' }, { status: 503 })
}

5.4 兼容 OpenAI 协议的服务

很多本地部署或第三方服务兼容 OpenAI API 协议（DeepSeek、Together AI、Groq、Ollama）——用 @ai-sdk/openai 配置 baseURL 即可：

import { createOpenAI } from '@ai-sdk/openai'
 
// DeepSeek
const deepseek = createOpenAI({
  baseURL: 'https://api.deepseek.com/v1',
  apiKey: process.env.DEEPSEEK_API_KEY,
})
streamText({ model: deepseek('deepseek-chat'), messages })
 
// Ollama（本地模型）
const ollama = createOpenAI({
  baseURL: 'http://localhost:11434/v1',
  apiKey: 'ollama', // Ollama 不需要真实 Key
})
streamText({ model: ollama('llama3.1'), messages })
 
// Groq（超快推理）
const groq = createOpenAI({
  baseURL: 'https://api.groq.com/openai/v1',
  apiKey: process.env.GROQ_API_KEY,
})
streamText({ model: groq('llama-3.1-70b-versatile'), messages })

6. 高级配置

6.1 自定义 Headers

给 AI 请求添加自定义 Header（用于代理、审计等）：

streamText({
  model: openai('gpt-4o'),
  messages,
  headers: {
    'X-Request-Id': crypto.randomUUID(),
    'X-Workspace-Id': workspaceId,
  },
})

6.2 请求超时与中止

const controller = new AbortController()
 
// 30 秒超时
setTimeout(() => controller.abort(), 30000)
 
const result = streamText({
  model: openai('gpt-4o'),
  messages,
  abortSignal: controller.signal,
})

6.3 onFinish 回调

流完成后执行操作——保存数据库、记录 Token 消耗：

const result = streamText({
  model: openai('gpt-4o'),
  messages,
  onFinish: async ({ text, usage, finishReason }) => {
    // 保存 AI 回复到数据库
    await db.insert(dbMessages).values({
      chatId,
      role: 'assistant',
      content: text,
      tokenCount: usage.totalTokens,
    })
 
    // 更新 Token 配额
    await db.update(tokenQuotas)
      .set({ used: sql`used + ${usage.totalTokens}` })
      .where(eq(tokenQuotas.workspaceId, workspaceId))
  },
})

onFinish 在流完成后执行

onFinish 回调在流完成后（所有 Token 发送完毕后）执行，不会阻塞流式响应。但如果回调抛出错误，客户端不会收到通知——需要自己做错误处理。

本章小结

三层架构：Core（服务端调用）→ UI（前端 Hook）→ Providers（模型适配）
核心函数：streamText（流式）、generateText（非流式）、generateObject（结构化输出）
核心 Hook：useChat（对话）、useCompletion（补全）、useObject（流式对象）
Data Stream 协议：AI SDK 自定义的流式传输格式，支持文本、工具调用、元数据
多模型适配：切换模型只改一行代码，支持 Fallback 和兼容 OpenAI 协议的服务
关键优势：几行代码替代几十行 ReadableStream 样板，且自动处理错误、重试、状态管理

下一章讲 AI Chat 对话系统的完整实战。