Skip to content
Vercel AI SDK Streaming 实现原理：从 Server 到 Client 数据流
概述
在传统 HTTP API 中，客户端发送 POST 请求后，服务端计算完毕返回完整 JSON，客户端调用 res.json() 一次性解出结果。对聊天场景而言，这个模型有两个问题：用户必须等待全部文本生成完毕；服务端已经生成的文本无法提前到达客户端。
流式响应的做法是：服务端在完整回答生成完毕之前，就开始把已生成的文本发送给客户端。浏览器和 Node.js 均支持通过 ReadableStream 读取响应体，因此客户端可以边接收边渲染。
ts
// 非流式读取
const res = await fetch('/api/chat', {
method: 'POST',
body: JSON.stringify({ messages }),
});
const data = await res.json();
// 流式读取
const res = await fetch('/api/chat', {
method: 'POST',
body: JSON.stringify({ messages }),
});
const reader = res.body.getReader();从模型 Provider 到浏览器界面，一次流式请求的完整链路如下：
- 模型 Provider 逐个产生 token；
- 服务端
streamText接收 Provider 的原始流； - AI SDK 将原始流编码为 Data Stream 协议；
- HTTP Response 以流式 body 发送到客户端；
- 客户端
useChat读取协议事件，并增量更新messages状态。
这条链路的关键在于每一层都在处理同一种东西：持续到达的数据。下面从底层概念开始逐层展开。
基本概念
Token 流
大语言模型不是一次性生成完整回答，而是逐个生成 token。对文本类模型来说，token 可能对应一个单词的一部分、一个单词或一个汉字。服务端 API 会把模型已生成的 token 以增量形式发出。
在代码层面，这表现为一个异步迭代流：每拿到一个 chunk，就得到一段新增文本。
HTTP Chunked Transfer
HTTP/1.1 允许响应不指定 Content-Length，而使用 Transfer-Encoding: chunked 分块发送。每个 chunk 自带长度，服务端可以持续写入，客户端可以持续读取。fetch 会把底层 chunk 重新组织为 ReadableStream。
这意味着“响应体仍在传输中”与“客户端已经开始读取”可以同时发生。
SSE 与 AI SDK Data Stream 的关系
SSE（Server-Sent Events）是一种常见的文本流协议，使用 Content-Type: text/event-stream，每条消息以 data: 开头。Vercel AI SDK 的 Data Stream 协议并不等同于 SSE。它使用换行分隔的 JSON（NDJSON），每条数据是一个 JSON 对象。虽然两者都依赖 HTTP 分块传输，客户端解析方式完全不同。
Web Streams API
浏览器和 Node.js 都实现了 Web Streams API。fetch 返回的 response.body 是一个 ReadableStream，可以用 reader.read() 逐块读取。
ts
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
// stream: true 表示当前 chunk 可能只是多字节字符的一部分，
// TextDecoder 需要保留内部状态，等待后续字节拼成完整字符。
const text = decoder.decode(value, { stream: true });
console.log(text);
}
decoder.decode();reader.read() 返回 { done, value }。done 为 true 时表示流结束，value 是 Uint8Array。TextDecoder 的 { stream: true } 选项用于跨 chunk 的多字节字符解码。
工作原理
服务端：streamText 对 Provider 流的封装
streamText 是 AI SDK 的服务端核心 API。它接收模型实例和消息列表，内部完成三件事：
- 调用模型 Provider 的流式接口；
- 将不同 Provider 的 chunk 结构统一为一致的文本流；
- 暴露
textStream、toDataStreamResponse等流式结果。
streamText 返回的结果对象 result 表示生成过程已经开始，但结果尚未拼接为完整字符串。result.toDataStreamResponse() 返回一个 Response 对象，其 body 是一个流。
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-mini'),
messages,
});
return result.toDataStreamResponse();
}如果直接使用 Provider 原生 API，OpenAI 的流式 chunk 结构、Anthropic 的流式事件结构、其他厂商的协议各不相同。streamText 把这些差异统一在模型实例后面。
数据流协议：AI SDK Data Stream
streamText 返回的流式响应遵循 AI SDK Data Stream 协议。该协议用于让客户端从流中区分文本增量、工具调用、错误和结束信息。
协议的基本形式是 NDJSON：每个事件占一行，每行都是一个 JSON 对象，第一个字段是 type。
text
{"type":"start","id":"msg_123","timestamp":"..."}
{"type":"text-delta","text":"你"}
{"type":"text-delta","text":"好"}
{"type":"finish","finishReason":"stop","usage":{"promptTokens":8,"completionTokens":2},"timestamp":"..."}文本增量事件
文本增量的事件类型是 text-delta，字段名是 text，不是 content。
json
{"type":"text-delta","text":"你好"}客户端每收到一个 text-delta，就把 text 字段追加到当前助手消息的末尾，从而逐字渲染输出。
工具调用事件
模型调用工具时，流中会出现 tool-call 事件。args 是工具参数的完整 JSON 对象。
json
{"type":"tool-call","toolCallId":"call_1","toolName":"getWeather","args":{"city":"Beijing"}}如果工具参数较长，AI SDK 也可以使用 tool-call-delta 将参数以文本片段的形式流式发送。工具执行完成后，会出现 tool-result 事件，携带工具返回值。
结束事件
正常结束的流会包含 finish 事件。客户端据此知道生成已完成，并读取 finishReason 和 token 用量。
json
{"type":"finish","finishReason":"stop","usage":{"promptTokens":8,"completionTokens":12}}如果连接在 finish 之前断开，说明流被中断。
错误事件
模型调用或工具执行出错时，流中可以包含 error 事件。
json
{"type":"error","message":"模型调用失败"}客户端不能只依赖 HTTP 状态码判断错误，因为在流开始之后发生的错误，HTTP 状态码可能仍然是 200。
Content-Type 与协议标识
toDataStreamResponse() 返回的响应体是 NDJSON，因此 Content-Type 是 text/plain; charset=utf-8，而不是 application/json。响应头中还会带有标识 AI SDK Data Stream 协议版本的字段。
toDataStreamResponse 与 toUIMessageStreamResponse
AI SDK 还提供 toUIMessageStreamResponse()。它返回的不是低层 Data Stream 协议，而是更高层的 UI Message Stream，用于直接把消息片段发送给客户端 UI。
useChat 默认按 Data Stream 协议解析响应。如果服务端使用 toUIMessageStreamResponse()，客户端需要把 useChat 的 streamProtocol 设置为 'ui'。两种响应协议不能混用。
传输层：Next.js Route Handler 中的响应对象
Next.js Route Handler 支持直接返回 Response 对象。streamText(...).toDataStreamResponse() 返回的正是 Response，所以 Route Handler 可以将其直接作为返回值。
在 HTTP/1.1 下，这个响应体会以 chunked transfer 方式发送。Next.js 不需要先把流收集成完整字符串，而是边生成边把字节发给客户端。
Node.js Runtime 和 Edge Runtime 都可以支持这种流式响应，关键取决于所选模型 Provider SDK 是否能在对应运行时工作。
客户端：useChat 的流式读取与状态更新
useChat 内部不会等待完整 JSON 返回。它拿到 response.body 后，用 reader.read() 持续读取字节，再按换行符切分成 JSON 行，最后根据 type 处理：
text-delta：把text追加到当前助手消息的content；tool-call/tool-result：更新工具调用状态；finish：记录结束原因和 token 用量；error：把错误信息写入error状态。
每处理一个事件，useChat 都会更新 React 状态，因此浏览器可以逐字显示模型输出。
如果不使用 useChat，也可以手动读取流。下面的循环展示了协议解析的核心过程：
ts
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() ?? '';
for (const line of lines) {
if (!line.trim()) continue;
const part = JSON.parse(line);
if (part.type === 'text-delta') {
appendToAssistantMessage(part.text); // 示意：追加到当前助手消息
}
}
}手动循环中的 appendToAssistantMessage 是示意函数，实际项目中需要维护消息 id、角色和内容。useChat 做的事情与之类似，但额外管理了这些状态。
基本用法
服务端：Route Handler 接入 streamText
在 Next.js Route Handler 中，先读取请求体，再调用 streamText，最后返回 toDataStreamResponse()。
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-mini'),
messages,
});
return result.toDataStreamResponse();
}返回的响应体是 NDJSON，因此 Content-Type 为 text/plain; charset=utf-8，在 DevTools 中看到这个类型属正常现象。
客户端：React 组件接入 useChat
useChat 是 AI SDK 在 React 客户端提供的 hook。它封装了发送消息、读取流、更新消息列表、处理错误和取消等逻辑。
tsx
'use client';
import { useChat } from '@ai-sdk/react';
export default function Chat() {
const {
messages,
input,
handleInputChange,
handleSubmit,
isLoading,
error,
stop,
} = useChat();
return (
<div>
{messages.map((m) => (
<div key={m.id}>
<strong>{m.role}: </strong>
<span>{m.content}</span>
</div>
))}
<form onSubmit={handleSubmit}>
<input value={input} onChange={handleInputChange} />
<button type="submit" disabled={isLoading}>发送</button>
<button type="button" onClick={() => stop()} disabled={!isLoading}>
停止
</button>
</form>
{error && <div>Error: {error.message}</div>}
</div>
);
}handleSubmit 会发送当前输入框中的消息，isLoading 在流式响应期间保持为 true，stop() 用于中断本次生成。
协议选择
useChat 默认按 Data Stream 协议解析。如果服务端改用 toUIMessageStreamResponse()，客户端需要同步设置协议：
tsx
const { messages } = useChat({
streamProtocol: 'ui',
});两种协议不能混用，服务端与客户端必须保持一致。
取消与超时
stop() 会中断当前读取，并中止底层 fetch 请求。服务端可以通过 streamText 的 abortSignal 选项感知中止，并停止上游模型调用。
ts
const result = streamText({
model,
messages,
abortSignal: req.signal,
});
return result.toDataStreamResponse();req.signal 是 Fetch API 的 Request 对象自带的 signal。客户端断开连接或主动取消时，这个 signal 会进入 aborted 状态。
超时控制可以通过 AbortSignal.timeout 或其他 AbortController 实现。不要在 Route Handler 中 await result.text 后再做超时处理，那样会等待整个流结束，失去流式意义。
命令：调试流式接口
curl
curl 默认会等待响应全部结束后才输出。使用 -N 或 --no-buffer 可以让 curl 边接收边打印。
bash
curl -N http://localhost:3000/api/chat \
-H "Content-Type: application/json" \
-d '{"messages":[{"role":"user","content":"你好"}]}'如果路由返回 Data Stream 协议，输出如下格式：
text
{"type":"start","id":"...","timestamp":"..."}
{"type":"text-delta","text":"你"}
{"type":"text-delta","text":"好"}
...
{"type":"finish","finishReason":"stop","usage":{...}}DevTools
在 Chrome DevTools 的 Network 面板中，流式请求会一直保持 pending，Response 面板会逐渐出现数据。如果 DevTools 对响应做了缓冲，可能看不到中间状态，此时可以用 curl 确认服务端是否真的在持续输出。
服务端日志
streamText 支持 onChunk 回调，可以在服务端观察每个流事件。
ts
const result = streamText({
model,
messages,
onChunk({ chunk }) {
if (chunk.type === 'text-delta') {
console.log(chunk.text);
}
},
});onChunk 只做观察，不会消费底层流，因此可以安全地与 toDataStreamResponse() 一起使用。
示例：自动化测试流式 endpoint
自动化测试流式 endpoint 时，不应把响应体当作 JSON 一次性解析，而应读取完整流，再按换行符切分成协议事件。
ts
import { test, expect } from 'vitest';
test('chat endpoint 返回流式 text-delta', async () => {
const res = await fetch('http://localhost:3000/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
messages: [{ role: 'user', content: 'Hello' }],
}),
});
expect(res.headers.get('content-type')).toContain('text/plain');
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
const parts = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() ?? '';
for (const line of lines) {
if (line.trim()) {
parts.push(JSON.parse(line));
}
}
}
if (buffer.trim()) {
parts.push(JSON.parse(buffer));
}
const deltas = parts.filter((part) => part.type === 'text-delta');
expect(deltas.length).toBeGreaterThan(0);
const text = deltas.map((part) => part.text).join('');
expect(text.length).toBeGreaterThan(0);
});测试的关键是验证协议事件，而不是验证具体模型输出。模型输出是不确定的，但 text-delta 事件一定存在。
注意点
流内错误
错误发生在流已启动之后，服务端无法再返回普通的 4xx/5xx JSON。AI SDK 会把错误编码为流中的 error 事件，useChat 收到后设置 error 状态。
因此，客户端判断请求是否失败时，不能只看 res.status，还要读取流内的协议事件。
不要在返回响应前 await result.text
streamText 返回的 result 带有异步文本结果。如果写成下面这样，流式效果会消失：
ts
const result = streamText({ model, messages });
const fullText = await result.text;
return Response.json({ text: fullText });await result.text 会等待全部文本生成完。后续返回的响应不再是流式，客户端也无法边接收边渲染。
协议解析必须匹配
useChat 默认按 Data Stream 协议解析。如果服务端使用 toUIMessageStreamResponse()，需要同步修改客户端 streamProtocol。
大事件可能影响解析
Data Stream 协议每个事件是一行 JSON。工具参数或工具结果很大时，单行 JSON 会被拆到多个网络 chunk 中，客户端需要等换行符出现才能完整解析。因此，不要把超大对象直接塞进工具结果。
Edge Runtime 兼容性
Next.js 可以在 route 中设置 export const runtime = 'edge'，但并非所有 Provider SDK 都支持 Edge Runtime。使用前需要确认模型 SDK 是否依赖 Node.js 专属 API。
应用：与直接调用 Provider API / 手写 SSE 的对比
直接调用 OpenAI 流式 API 时，代码大致如下：
ts
const stream = await openai.chat.completions.create({
model: 'gpt-4o-mini',
messages,
stream: true,
});
for await (const chunk of stream) {
const delta = chunk.choices[0]?.delta?.content;
if (delta) process.stdout.write(delta);
}换成 Anthropic 或其他 Provider，chunk 结构会改变，服务端处理逻辑也要跟着改。
手写 SSE 时，服务端可以自己把文本包成 data: 行，客户端再自己解析。这种做法对纯文本够用，但遇到工具调用、结束原因、token 用量这些结构化信息时，需要额外设计一套事件格式。
AI SDK 的抽象价值在于：
- 服务端用同一个
streamText统一不同 Provider 的流； - 客户端用同一个
useChat解析 Data Stream 协议； - 工具调用、错误、结束原因和 usage 都有固定的协议位置。
如果应用只接一个 Provider，也不需要工具调用，直接使用 Provider SDK 是可行的。一旦需要多 Provider 或复杂客户端状态，streamText + useChat 的协议设计会明显减少重复工作。