AI 电子伴侣

创建时间: 2026-04-18最后更新: 2026-04-21作者: yangbo

1. 这一篇做什么

streamText 是 Core 层的主干函数，AI SDK 90% 的生产代码都绕不开它。这一篇把输入、输出、消费方式、UIMessageStream 协议这四件事一次讲透。

读完这一篇，你应该能答得上来：

result.textStream / result.fullStream / result.toUIMessageStreamResponse() 三种消费方式各自适合什么场景
UIMessageStream 里有哪些事件、每个事件什么时候出现
怎么在流上加拦截、改写、注入业务 data part
流的生命周期：开始、步骤、文本段、工具调用、完成、错误，以及每个节点怎么挂钩子

2. streamText 的完整签名

先看一次完整签名（精简了部分 provider 专属选项）：

stream-text-signature.ts

01const result = streamText({
02  // ── 必填
03  model: LanguageModelV2,
04
05  // ── 消息输入（三选一）
06  prompt?: string,
07  system?: string,
08  messages?: ModelMessage[],
09
10  // ── 工具与 Agent 循环
11  tools?: Record<string, Tool>,
12  toolChoice?: 'auto' | 'required' | 'none' | { type: 'tool', toolName: string },
13  stopWhen?: StopCondition | StopCondition[],
14
15  // ── 生成参数
16  temperature?: number,
17  maxOutputTokens?: number,
18  topP?: number,
19  topK?: number,
20  presencePenalty?: number,
21  frequencyPenalty?: number,
22  seed?: number,
23  stopSequences?: string[],
24
25  // ── 运行控制
26  abortSignal?: AbortSignal,
27  headers?: Record<string, string>,
28
29  // ── 可观测性
30  experimental_telemetry?: TelemetrySettings,
31
32  // ── 生命周期钩子
33  onChunk?: (event: { chunk: StreamPart }) => void | Promise<void>,
34  onFinish?: (event: FinishEvent) => void | Promise<void>,
35  onError?: (event: { error: unknown }) => void | Promise<void>,
36  onStepFinish?: (event: StepResult) => void | Promise<void>,
37  onAbort?: (event: { steps: StepResult[] }) => void | Promise<void>,
38
39  // ── Provider 专属
40  providerOptions?: Record<string, unknown>,
41})

参数看着多，实际用起来绝大多数时候只会碰到 5 到 6 个。按使用频度排一下：

频度	参数
必用	`model`, `messages` / `prompt`
常用	`system`, `tools`, `stopWhen`
偶尔	`temperature`, `maxOutputTokens`, `abortSignal`
工程化用	`experimental_telemetry`, `onFinish`, `onError`
极少	`seed`, `stopSequences`, `topK`, `providerOptions`

3. 三种消费方式

streamText 同步返回 result，真正的流挂在 result 的字段上。有三种主要的消费方式。

3.1 textStream —— 只要字符串

最简单的场景：后端脚本、CLI 工具，只想逐 token 打印。

text-stream.ts

1const result = streamText({ model: models.chat, prompt: '...' })
2
3for await (const chunk of result.textStream) {
4  process.stdout.write(chunk)
5}

textStream 是 AsyncIterable<string>，每个元素就是一段文本增量（delta），拼起来就是完整回复。

它的限制也很明显：只给你文本段，工具调用、思考段、完成事件都拿不到，所以只适合纯文本场景。另外这条流消费完就耗尽了，同一个 result 只能消费一次（第一次调用也提到过）。如果需要同时拿文本和 token 用量，用 fullStream，或者先消费完再 await result.text。

3.2 fullStream —— 细粒度事件

想感知流里发生的每一件事，用 fullStream：

full-stream.ts

01const result = streamText({
02  model: models.chat,
03  messages,
04  tools: { searchMemory },
05  stopWhen: stepCountIs(5),
06})
07
08for await (const part of result.fullStream) {
09  switch (part.type) {
10    case 'text-delta':
11      process.stdout.write(part.delta)
12      break
13    case 'reasoning-delta':
14      console.log('[思考]', part.delta)
15      break
16    case 'tool-call':
17      console.log('[工具调用]', part.toolName, part.input)
18      break
19    case 'tool-result':
20      console.log('[工具结果]', part.output)
21      break
22    case 'finish':
23      console.log('[完成]', part.finishReason, part.usage)
24      break
25    case 'error':
26      console.error('[错误]', part.error)
27      break
28  }
29}

fullStream 的完整事件类型：

事件 type	含义
`start`	整个流开始
`start-step`	一次 LLM 调用开始（一次 step）
`text-start` / `text-delta` / `text-end`	文本段开始 / 增量 / 结束
`reasoning-start` / `reasoning-delta` / `reasoning-end`	思考段开始 / 增量 / 结束
`tool-call`	工具调用完成（参数已就绪）
`tool-call-input-start` / `tool-call-input-delta`	工具参数流式构造（参数也是流式的）
`tool-result`	工具执行结果
`source`	引用来源（RAG / Web Search）
`file`	文件输出（图像 / 音频）
`finish-step`	一次 step 结束
`finish`	整个流结束（含总 token、finishReason）
`error`	错误
`abort`	用户中止

一般用在后端内部监控、埋点，或者把流转发到其他协议。

3.3 toUIMessageStreamResponse —— 给前端用

生产环境最常见的消费方式：把流直接返回给前端。

to-response.ts

01import { Hono } from 'hono'
02
03const app = new Hono()
04
05app.post('/api/chat', async (c) => {
06  const { messages } = await c.req.json()
07
08  const result = streamText({
09    model: models.chat,
10    messages: convertToModelMessages(messages),
11    tools: { searchMemory, updateEmotion },
12  })
13
14  return result.toUIMessageStreamResponse()
15})

toUIMessageStreamResponse() 干了三件事：

把 fullStream 按 UIMessageStream 协议序列化成 SSE 格式
设置正确的 Content-Type: text/event-stream 和 CORS 头
返回一个标准 Response 对象，Hono / Next.js Route Handler 直接 return 就行

前端用 useChat 消费时，内部协议就是这个。换句话说，后端只要调 toUIMessageStreamResponse()，前端 useChat 就能直接用。

4. UIMessageStream 协议完整对照

把 UIMessageStream 协议拆开来看。下面是一条典型的 SSE 流：

uimessage-stream-raw.txt

01data: {"type":"start"}
02
03data: {"type":"start-step"}
04
05data: {"type":"reasoning-start","id":"r_0"}
06
07data: {"type":"reasoning-delta","id":"r_0","delta":"用户问"}
08
09data: {"type":"reasoning-delta","id":"r_0","delta":"了过去"}
10
11data: {"type":"reasoning-end","id":"r_0"}
12
13data: {"type":"tool-input-start","toolCallId":"c_1","toolName":"searchMemory"}
14
15data: {"type":"tool-input-delta","toolCallId":"c_1","inputTextDelta":"{\"query\":"}
16
17data: {"type":"tool-input-delta","toolCallId":"c_1","inputTextDelta":"\"周五\"}"}
18
19data: {"type":"tool-input-available","toolCallId":"c_1","toolName":"searchMemory","input":{"query":"周五"}}
20
21data: {"type":"tool-output-available","toolCallId":"c_1","output":[{"content":"..."}]}
22
23data: {"type":"finish-step"}
24
25data: {"type":"start-step"}
26
27data: {"type":"text-start","id":"t_1"}
28
29data: {"type":"text-delta","id":"t_1","delta":"我记得"}
30
31data: {"type":"text-delta","id":"t_1","delta":"你周五"}
32
33data: {"type":"text-end","id":"t_1"}
34
35data: {"type":"finish-step"}
36
37data: {"type":"finish"}
38
39data: [DONE]

这条流讲了一个故事：

流开始
第一次 step：思考 → 调用 searchMemory → 拿到结果
第二次 step：生成文本回复
流结束

5. 注入业务数据：Data Parts

业务层经常要给前端发一些非文本信息：情绪分值、亲密度变化、检索命中条数等等。UIMessageStream 用 data part 承载这些：

data-part-server.ts

01import { createUIMessageStream, createUIMessageStreamResponse } from 'ai'
02
03const stream = createUIMessageStream({
04  execute: async ({ writer }) => {
05    // 先发一个业务数据：当前情绪
06    writer.write({
07      type: 'data-emotion',
08      data: { primary: 'calm', intensity: 0.6 },
09    })
10
11    // 再调 LLM，把流合并进来
12    const result = streamText({ model: models.chat, messages })
13    writer.merge(result.toUIMessageStream())
14
15    // 最后再发一个业务数据：本次用到的记忆条数
16    writer.write({
17      type: 'data-memories-used',
18      data: { count: 3 },
19    })
20  },
21})
22
23return createUIMessageStreamResponse({ stream })

前端消费：

data-part-client.tsx

01const { messages } = useChat()
02
03messages.map((m) => m.parts.map((p, i) => {
04  if (p.type === 'data-emotion') {
05    return <EmotionBadge key={i} data={p.data} />
06  }
07  if (p.type === 'data-memories-used') {
08    return <MemoriesHint key={i} count={p.data.count} />
09  }
10  // ... 其他 part
11}))

Data Parts 是 AI SDK 里做业务定制 UI 的主要手段：既不破坏协议，又保留类型安全（给 UIMessage 泛型传业务 data 类型就行），还能在 React 里自然渲染。

6. 生命周期钩子

除了流式消费，streamText 还支持一组回调钩子，用来做服务端副作用：

lifecycle-hooks.ts

01streamText({
02  model: models.chat,
03  messages,
04  tools: { searchMemory },
05
06  // 每次收到一个流片段（text-delta / tool-call / reasoning-delta 等）
07  onChunk: async ({ chunk }) => {
08    if (chunk.type === 'text-delta') {
09      metrics.incrementTokens(chunk.delta.length)
10    }
11  },
12
13  // 每个 step 结束（一次 LLM 调用完成）
14  onStepFinish: async (step) => {
15    console.log('[step]', step.finishReason, step.usage)
16  },
17
18  // 全流结束
19  onFinish: async ({ text, finishReason, usage, response }) => {
20    await saveMessage(sessionId, { role: 'assistant', content: text })
21    await trackUsage(sessionId, usage)
22  },
23
24  // 出错
25  onError: async ({ error }) => {
26    logger.error('chat error', { error, sessionId })
27  },
28
29  // 用户中止
30  onAbort: async ({ steps }) => {
31    console.log('用户中止，已完成', steps.length, '步')
32  },
33})

onFinish 才是你做持久化的地方。不要在 toUIMessageStreamResponse() 之后 await 什么事——那时候流已经返回给客户端了，你的 await 会挂起请求但对用户没有任何意义。onFinish 在流真正结束之后异步触发，是写数据库、发埋点、累计计费的最佳时机。

在 Cloudflare Workers 上，onFinish 的 await 一定要放在 ctx.waitUntil() 里，否则 Worker 可能在回调跑完之前就终止了：

workers-wait-until.ts

01app.post('/chat', async (c) => {
02  const result = streamText({
03    model: models.chat,
04    messages,
05    onFinish: ({ text }) => {
06      // ctx.waitUntil 允许 Worker 等这个异步任务完成
07      c.executionCtx.waitUntil(saveMessage(sessionId, text))
08    },
09  })
10
11  return result.toUIMessageStreamResponse()
12})

7. Abort、中止与运行时差异

用户点「停止生成」按钮时，前端 useChat 会调 stop() 方法——底层就是 AbortController.abort()。后端要响应这个信号，把 abortSignal 传进 streamText：

abort.ts

01app.post('/chat', async (c) => {
02  const result = streamText({
03    model: models.chat,
04    messages,
05    // 把 HTTP 请求的 abortSignal 往下传
06    abortSignal: c.req.raw.signal,
07  })
08
09  return result.toUIMessageStreamResponse()
10})

这样浏览器或中间代理一断开连接，AbortSignal 就会触发，Provider 会立刻停止向 LLM API 读流，省下剩下的 token。

onAbort 回调也会被触发，方便清理或记录：

on-abort.ts

1streamText({
2  model: models.chat,
3  messages,
4  abortSignal: c.req.raw.signal,
5  onAbort: async ({ steps }) => {
6    const partialText = steps.map((s) => s.text).join('')
7    await saveMessage(sessionId, { role: 'assistant', content: partialText, status: 'aborted' })
8  },
9})

Node 与 Edge 的差异

streamText 本身是运行时无关的，但部署到不同环境有些细节要注意。核心差异在于：Edge 类环境（Workers / Vercel Edge）的请求处理函数一旦返回 Response，运行时可能立刻回收资源，导致 onFinish 里的异步操作被中断。

按运行时梳理一下注意事项：

环境	注意点
Node.js	标准 `fetch` 可用（Node 18+），`onFinish` 里 `await` 自由使用
Cloudflare Workers	`onFinish` 的 async 副作用必须 `ctx.waitUntil()`，否则可能被提前杀
Vercel Edge Functions	类似 Workers，用 `event.waitUntil()`
Next.js Route Handler（Node runtime）	直接用，没特殊限制
Next.js Route Handler（Edge runtime）	同 Edge Functions
Bun / Deno	同 Node，但某些 Provider 可能有兼容 issue（社区 issue 跟进）

本专栏 AI 伴侣的主路径是 Cloudflare Workers，AI SDK × Hono 会具体讲部署注意事项。

8. 小结

streamText 是 Core 层主干，三种消费方式：textStream（纯字符串）、fullStream（细粒度事件）、toUIMessageStreamResponse()（给前端）
UIMessageStream 协议把原始 SSE 升级成了结构化事件流，text / reasoning / tool-* / source / file / data-* 每种都有完整生命周期
Data Parts 通过 createUIMessageStream + writer.write() 注入业务数据，前端按 type 分别渲染
生命周期钩子 onFinish / onError / onStepFinish / onAbort 是做持久化、埋点、清理的标准位置
Abort 信号把 HTTP 请求的 signal 传进 streamText，就能正确处理用户中止
运行时差异：Workers / Edge 要用 ctx.waitUntil() 保护 onFinish 的异步副作用

下一篇进入 Core 层的另一个主力函数——generateObject / streamObject。Zod schema 驱动的结构化输出，对照 Zod + LLM 做一次升级。