📚 AI 工程与智能体教程站

第 5 章 · 文本生成与流式输出 ​

本章目标：

  • 掌握 generateText 与 streamText 两大核心函数的适用场景
  • 学会读取生成结果中的文本、用量（usage）、finish reason 与 sources
  • 理解 Blocking UI 与 Streaming UI 的体验差异与权衡
  • 掌握 onError / onChunk / onEnd 回调的用法

5.1 两大核心函数 ​

大语言模型（LLM）可以根据 prompt 生成文本，prompt 中可以包含指令和需要处理的信息。例如让模型想出一个菜谱、起草一封邮件或总结一份文档。

AI SDK Core 提供了两个函数来生成和流式输出 LLM 的文本：

  • generateText：为给定的 prompt 和模型生成完整文本
  • streamText：为给定的 prompt 和模型流式输出文本

高级特性如 tool calling 和结构化数据生成都构建在文本生成之上。

本章示例统一使用 Vercel AI Gateway 构造模型（同样适用于自定义 provider，见第 3 章）：

ts
import { createGateway } from 'ai';

export const gateway = createGateway({
  apiKey: process.env.AI_GATEWAY_API_KEY ?? '',
});

// 自定义 OpenAI 兼容 Provider 等价写法：
// import { createOpenAICompatible } from '@ai-sdk/openai-compatible';
// export const myProvider = createOpenAICompatible({
//   name: 'my-provider',
//   baseURL: process.env.OPENAI_COMPATIBLE_BASE_URL ?? '',
//   apiKey: process.env.OPENAI_COMPATIBLE_API_KEY ?? '',
// });

5.2 generateText：一次性生成 ​

使用 generateText 函数可以生成文本。该函数非常适合非交互式场景——你需要写一段文字（如起草邮件、总结网页）以及使用工具的 agent 场景：

tsx
import { generateText } from 'ai';
import { gateway } from './provider';

const { text } = await generateText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a vegetarian lasagna recipe for 4 people.',
});

你也可以使用更高级的 prompt 来执行更复杂的指令：

tsx
import { generateText } from 'ai';
import { gateway } from './provider';

const { text } = await generateText({
  model: gateway('openai/gpt-5'),
  instructions:
    'You are a professional writer. ' +
    'You write simple, clear, and concise content.',
  prompt: `Summarize the following article in 3-5 sentences: ${article}`,
});

generateText 的结果对象包含生成的输出和元数据：

  • result.content：所有步骤中生成的内容
  • result.text：最后一步生成的文本
  • result.files：所有步骤中生成的文件
  • result.sources：被用作参考的来源（仅部分模型支持）
  • result.toolCalls / result.toolResults：所有步骤中的工具调用及其结果
  • result.finishReason：模型结束生成的原因
  • result.rawFinishReason：来自 provider 的原始结束原因
  • result.usage：所有步骤的总用量（多步生成时）
  • result.warnings：来自模型 provider 的警告（如不支持的设置）
  • result.steps：所有步骤的详情，含每步 performance
  • result.finalStep：最后一步的详情，含 performance
  • result.output：通过 output 规范生成的结构化输出

每个步骤都包含 performance 性能信息，例如：

  • effectiveOutputTokensPerSecond：有效输出 token/秒（outputTokens / requestSeconds）
  • stepTimeMs：步骤总耗时（含模型响应时间与工具执行时间），单位毫秒
  • responseTimeMs：等待语言模型响应的耗时，单位毫秒
  • timeToFirstOutputMs：（仅流式步骤）收到第一个输出块的耗时；对 generateText 为 undefined

访问响应头与响应体 ​

有时你需要访问 provider 返回的完整响应，比如读取特定于 provider 的响应头或响应体。可以通过 finalStep.response 属性获取：

ts
import { generateText } from 'ai';

const result = await generateText({
  // ...
});

console.log(JSON.stringify(result.finalStep.response.headers, null, 2));
console.log(JSON.stringify(result.finalStep.response.body, null, 2));

5.3 为什么需要流式输出 ​

流式对话文本 UI（类似 ChatGPT）在过去几年广受欢迎。LLM 虽然强大，但生成长输出时的速度可能远慢于你习惯的延迟——如果构建传统的阻塞式 UI，用户可能要盯着 loading 图标等上 5 秒、10 秒甚至 40 秒才能看到完整回复。

  • Blocking UI：阻塞式响应会等到全部内容可用后才展示
  • Streaming UI：流式响应可以在内容可用的同时逐段传输展示

当然，流式并非总是必需的：如果用更小更快的模型就能满足需求且不需要流式，开发过程往往更简单可控。但无论模型速度如何，AI SDK 都把实现流式 UI 变得极其简单——下面用不到 10 行代码实现文本流式生成：

ts
import { streamText } from 'ai';
import { gateway } from './provider';

const { textStream } = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a poem about embedding models.',
});

for await (const textPart of textStream) {
  console.log(textPart);
}

💡 result.textStream 同时是 ReadableStream 和 AsyncIterable。

5.4 streamText 深入 ​

AI SDK Core 提供的 streamText 函数简化了从 LLM 流式获取文本的过程：

ts
import { streamText } from 'ai';
import { gateway } from './provider';

const result = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Invent a new holiday and describe its traditions.',
});

// 示例：将 textStream 作为异步可迭代对象使用
for await (const textPart of result.textStream) {
  console.log(textPart);
}

⚠️ streamText 会立即开始流式传输并抑制错误以防止服务器崩溃，请使用 onError 回调记录错误。

result.stream 可以传给多个独立 helper 以便集成到 AI SDK UI：

  • createUIMessageStreamResponse({ stream: toUIMessageStream({ stream: result.stream }) })：创建 UI Message 流 HTTP 响应（含工具调用等），可用于 Next.js App Router API route
  • pipeUIMessageStreamToResponse({ ... , response })：将 UI Message 流增量写入 Node.js response 对象
  • createTextStreamResponse({ stream: toTextStream({ stream: result.stream }) })：创建纯文本流的 HTTP 响应
  • pipeTextStreamToResponse({ ... , response })：将文本增量写入 Node.js response 对象

💡 streamText 使用背压（backpressure）机制，只在被请求时才生成 token——你必须消费流它才会完成。

流结束后以下 promise 会 resolve：result.content、result.text、result.finalStep、result.files、result.sources、result.toolCalls、result.toolResults、result.finishReason、result.rawFinishReason、result.usage、result.warnings、result.steps 等。

对 streamText 来说，timeToFirstOutputMs 在收到某步的第一个输出块时确定；当至少收到两个块时，timeBetweenOutputChunksMs 还会给出 min、p10、median、avg、p90、max 统计。

5.5 streamText 回调 ​

onError ​

streamText 为了能不等模型就发送数据而立即开始流式传输，错误会成为流的一部分而不是抛出异常（防止服务器崩溃）。要记录错误请提供 onError 回调：

tsx
import { streamText } from 'ai';
import { gateway } from './provider';

const result = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Invent a new holiday and describe its traditions.',
  onError({ error }) {
    console.error(error); // 你的错误记录逻辑
  },
});

onChunk ​

onChunk 回调会在流的每个 chunk 上触发，接收 stream 的所有事件类型，包括：start、start-step、text-start、text-delta、text-end、reasoning-start、reasoning-delta、reasoning-end、source、file、tool-call、tool-input-start、tool-input-delta、tool-input-end、tool-result、tool-error、finish-step、finish、error、raw 等：

tsx
import { streamText } from 'ai';
import { gateway } from './provider';

const result = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Invent a new holiday and describe its traditions.',
  onChunk({ chunk }) {
    // 自定义逻辑，例如：
    if (chunk.type === 'text-delta') {
      console.log(chunk.text);
    }
  },
});

onEnd ​

onEnd 回调在流结束时触发，包含文本、用量、finish reason、消息、步骤等信息：

tsx
import { streamText } from 'ai';
import { gateway } from './provider';

const result = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Invent a new holiday and describe its traditions.',
  onEnd({ text, finishReason, usage, responseMessages, steps, totalUsage }) {
    // 自定义逻辑，例如保存聊天历史或记录用量
    const messages = responseMessages; // 已生成的消息
  },
});

5.6 实验性生命周期回调 ​

streamText 与 generateText 都提供了一组实验性生命周期回调，用于在生成过程的不同阶段挂钩子，适用于日志、可观测性、调试与自定义遥测。回调内部抛出的错误会被静默捕获，不会中断流程（第 6 章将系统展开）：

tsx
import { streamText } from 'ai';
import { gateway } from './provider';

const result = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'What is the weather in San Francisco?',

  onStart({ modelId }) {
    console.log('Streaming started', { modelId });
  },

  onStepEnd({ finishReason, usage }) {
    console.log('Step finished', { finishReason, usage });
  },
});

5.7 stream 属性与流变换 ​

如果想自己实现 UI 或以其他方式处理流，可以用 stream 属性读取全部事件：

tsx
import { streamText } from 'ai';
import { gateway } from './provider';
import { z } from 'zod';

const result = streamText({
  model: gateway('openai/gpt-5'),
  tools: {
    cityAttractions: {
      inputSchema: z.object({ city: z.string() }),
      execute: async ({ city }) => ({
        attractions: ['attraction1', 'attraction2', 'attraction3'],
      }),
    },
  },
  prompt: 'What are some San Francisco tourist attractions?',
});

for await (const part of result.stream) {
  switch (part.type) {
    case 'text-delta': {
      // 处理文本增量
      break;
    }
    case 'tool-call': {
      switch (part.toolName) {
        case 'cityAttractions': {
          // 处理工具调用
          break;
        }
      }
      break;
    }
    case 'tool-result': {
      switch (part.toolName) {
        case 'cityAttractions': {
          // 处理工具结果
          break;
        }
      }
      break;
    }
    case 'finish': {
      // 流结束
      break;
    }
    // 其他事件类型同理
  }
}

smoothStream 平滑输出 ​

可以使用 experimental_transform 选项对流进行变换，常用于过滤、修改或平滑文本流。变换会在回调触发和 promise resolve 之前应用——如果你有一个全转大写的变换，onEnd 收到的将是变换后的文本。

AI SDK 内置 smoothStream 函数用于平滑文本和 reasoning 的流式输出：

tsx
import { smoothStream, streamText } from 'ai';

const result = streamText({
  model,
  prompt,
  experimental_transform: smoothStream(),
});

自定义变换 ​

也可以实现自己的变换。下面的例子把所有文本转为大写：

ts
import { streamText, type TextStreamPart, type ToolSet } from 'ai';

const upperCaseTransform =
  <TOOLS extends ToolSet>() =>
  (options: { tools: TOOLS; stopStream: () => void }) =>
    new TransformStream<TextStreamPart<TOOLS>, TextStreamPart<TOOLS>>({
      transform(chunk, controller) {
        controller.enqueue(
          // 对 text-delta 块，把文本转为大写：
          chunk.type === 'text-delta'
            ? { ...chunk, text: chunk.text.toUpperCase() }
            : chunk,
        );
      },
    });

还可以通过 stopStream() 提前终止流（例如违反 guardrails 时）。调用 stopStream 时应模拟 finish-step 和 finish 事件以保证流格式完整、回调都被触发。多个变换按提供顺序依次应用：

tsx
const result = streamText({
  model,
  prompt,
  experimental_transform: [firstTransform, secondTransform],
});

5.8 Sources 来源 ​

部分 provider（如 Perplexity、Google）会在响应中附带 sources。目前 sources 仅限为回答提供依据的网页，可通过结果的 sources 属性访问。每个 url 类型的 source 包含 id、url、title（可选）与 providerMetadata：

ts
const result = await generateText({
  model: gateway('google/gemini-2.5-flash'),
  prompt: 'List the top 5 San Francisco news from the past week.',
});

for (const source of result.sources) {
  if (source.sourceType === 'url') {
    console.log('ID:', source.id);
    console.log('Title:', source.title);
    console.log('URL:', source.url);
    console.log();
  }
}

使用 streamText 时，sources 既可以从 stream 属性中以 source 事件读取，也可以通过 result.sources promise 获得。

本章小结 ​

  • generateText 适合非交互式的一次性生成；结果对象提供 text、usage、finishReason、steps、performance 等完整元数据
  • streamText 通过 textStream 异步迭代消费文本增量；必须消费流，否则因背压机制不会完成
  • 流式场景的错误不会抛出而是进入流内，务必配置 onError 记录
  • onChunk / onEnd 回调覆盖每个 chunk 与流结束时机；实验性生命周期回调（onStart/onStepStart/onStepEnd 等）适合日志与遥测
  • experimental_transform 可平滑或自定义变换流，stopStream() 可提前终止（需模拟 finish 事件）
  • 部分模型支持 sources 返回引用来源

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 关于 generateText 与 streamText 的选择，下列说法正确的是？

2. 为什么必须消费 streamText 返回的流？

3. streamText 过程中发生错误时，默认行为是？

4. 在自定义流变换中调用 stopStream() 提前终止时，为什么还要手动 enqueue finish-step 和 finish 事件？

🛠️ 动手实践 ​

  1. 用 generateText 让模型为你的一个真实项目写一段 README 简介，打印 result.text 与 result.usage.totalTokens，并观察 finalStep.performance.responseTimeMs。
  2. 用 streamText 实现终端打字机效果：消费 textStream 时逐字符 process.stdout.write(delta)，并用 AbortSignal.timeout(8000) 加 8 秒超时。
  3. 编写一个自定义变换：检测文本流中出现「敏感词」即调用 stopStream() 终止并模拟 finish 事件；再用 smoothStream() 组合测试两个 transform 的顺序效果。