> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# 流式传输

Mastra 支持来自 Agent 和 Workflow 的实时增量响应，让用户在内容生成时即可看到输出，无需等待全部完成。这适用于聊天、长篇内容、多步骤 Workflow，以及任何重视即时反馈的场景。

## 快速开始

Mastra 的流式 API 会根据模型版本自动适配：

- **`.stream()`**：用于 V2 模型，支持 **AI SDK v5** 及更高版本（`LanguageModelV2`）。
- **`.streamLegacy()`**：用于 V1 模型，支持 **AI SDK v4**（`LanguageModelV1`）。

## 使用 Agent 进行流式传输

你可以为基本提示词传入单个字符串；提供多段上下文时传入字符串数组；或者传入包含 `role` 和 `content` 的消息对象数组，以精确控制角色和对话流程。

### 使用 `Agent.stream()`

`textStream` 会在生成响应时将其拆分成多个数据块，使输出逐步流式返回，而不是一次性到达。使用 `for await` 循环遍历 `textStream`，即可检查每个流数据块。

```typescript
const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}
```

有关更多信息，请参阅 [Agent.stream()](https://mastra.zisheng.pro/reference/streaming/agents/stream)。

> **提示:** 对于会分派[后台任务](https://mastra.zisheng.pro/docs/long-running-agents/background-tasks)的 Agent，请使用 [`Agent.streamUntilIdle()`](https://mastra.zisheng.pro/reference/streaming/agents/streamUntilIdle) 保持流打开，直到这些任务完成且 Agent 有机会响应其结果。

### `Agent.stream()` 的输出

输出会流式返回 Agent 生成的响应。

```text
Of course!
To help you organize your day effectively, I need a bit more information.
Here are some questions to consider:
...
```

### Agent 流属性

Agent 流可以访问以下响应属性：

- **`stream.textStream`**：发出文本数据块的可读流。
- **`stream.text`**：解析为完整文本响应的 Promise。
- **`stream.finishReason`**：Agent 停止流式传输的原因。
- **`stream.usage`**：token 用量信息。

### AI SDK v5+ 兼容性

AI SDK v5（及更高版本）的模型 Provider 使用 `LanguageModelV2`。如果收到正在使用 AI SDK v4 模型的错误，需要将模型软件包升级到下一个主版本。

要与 AI SDK v5+ 集成，请使用 `@mastra/ai-sdk` 中的 `toAISdkV5Stream()` utility，将 Mastra 流转换为兼容 AI SDK 的格式：

```typescript
import { toAISdkV5Stream } from '@mastra/ai-sdk'

const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

// Convert to AI SDK v5+ compatible stream
const aiSDKStream = toAISdkV5Stream(stream, { from: 'agent' })
```

要将消息转换为 AI SDK v5+ 格式，请使用 `@mastra/ai-sdk/ui` 中的 `toAISdkV5Messages()` utility：

```typescript
import { toAISdkV5Messages } from '@mastra/ai-sdk/ui'

const messages = [{ role: 'user', content: 'Hello' }]
const aiSDKMessages = toAISdkV5Messages(messages)
```

## 使用 Workflow 进行流式传输

Workflow 的流式传输返回一系列描述运行生命周期的结构化事件，而不是增量文本数据块。使用 `.createRun()` 创建运行后，这种基于事件的格式可以实时跟踪并响应 Workflow 进度。

### 使用 `Run.stream()`

`stream()` 方法直接返回事件的 `ReadableStream`。

```typescript
const run = await testWorkflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream) {
  console.log(chunk)
}
```

有关更多信息，请参阅 [Run.stream()](https://mastra.zisheng.pro/reference/streaming/workflows/stream)。

### `Run.stream()` 的输出

事件结构在顶层包含 `runId` 和 `from`，无需深入 payload 即可更轻松地识别和跟踪 Workflow 运行。

```typescript
{
  type: 'workflow-start',
  runId: '1eeaf01a-d2bf-4e3f-8d1b-027795ccd3df',
  from: 'WORKFLOW',
  payload: {
    stepName: 'step-1',
    args: { value: 'initial data' },
    stepCallId: '8e15e618-be0e-4215-a5d6-08e58c152068',
    startedAt: 1755121710066,
    status: 'running'
  }
}
```

### Workflow 流属性

Workflow 流可以访问以下响应属性：

- **`stream.status`**：Workflow 运行的状态。
- **`stream.result`**：Workflow 运行的结果。
- **`stream.usage`**：Workflow 运行的 token 总用量。

来自 Agent 或 Workflow 的流式传输可以实时查看 LLM 输出或 Workflow 运行状态。可以将这些反馈直接传给用户，或在应用中显示不断变化的 Workflow 状态。

Agent 或 Workflow 发出的事件代表生成和执行的不同阶段，例如运行开始、生成文本或调用 Tool 时。

## 事件类型

以下是 `.stream()` 发出的完整事件列表。 根据流式传输来自 **Agent** 还是 **Workflow**，只会出现其中一部分事件：

- **start**：表示 Agent 或 Workflow 运行开始。
- **step-start**：表示 Workflow 步骤已开始执行。
- **text-delta**：LLM 生成的增量文本数据块。
- **tool-call**：Agent 决定使用 Tool 时发出，包括 Tool 名称和参数。
- **tool-result**：Tool 执行返回的结果。
- **step-finish**：确认某个步骤已完全结束，可能包括该步骤结束原因等 metadata。
- **finish**：Agent 或 Workflow 完成时发出，包括用量统计信息。

## 检查 Agent 流

使用 `for await` 循环遍历 `stream`，检查发出的所有事件数据块。

```typescript
const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

for await (const chunk of stream) {
  console.log(chunk)
}
```

有关更多信息，请参阅 [Agent.stream()](https://mastra.zisheng.pro/reference/streaming/agents/stream)。

### Agent 输出示例

以下是可能发出的事件示例。每个事件始终包含 `type`，还可以包含 `from` 和 `payload` 等其他字段。

```typescript
{
  type: 'start',
  from: 'AGENT',
  // ..
}
{
  type: 'step-start',
  from: 'AGENT',
  payload: {
    messageId: 'msg-cdUrkirvXw8A6oE4t5lzDuxi',
    // ...
  }
}
{
  type: 'tool-call',
  from: 'AGENT',
  payload: {
    toolCallId: 'call_jbhi3s1qvR6Aqt9axCfTBMsA',
    toolName: 'testTool'
    // ..
  }
}
```

## Writer API

`writer` API 由 Tool 和 Workflow 步骤共享。各功能的具体示例请参阅 Tool 和 Workflow 文档。

## Agent 使用 Tool

Agent 流式传输可与 Tool 调用结合使用，让 Tool 输出直接写入 Agent 的流式响应，从而在交互中呈现 Tool 活动。

```typescript
import { Agent } from '@mastra/core/agent'
import { testTool } from '../tools/test-tool'

export const testAgent = new Agent({
  id: 'test-agent',
  name: 'Test Agent',
  instructions: 'You are a weather agent.',
  model: 'openai/gpt-5.6-sol',
  tools: { testTool },
})
```

### 使用 `context.writer`

Tool 的 `execute()` 函数可以使用 `context.writer` 对象，向活动流发出自定义事件、数据或值。Tool 使用这些事件在执行期间提供中间结果或状态更新。

> **注意:** 必须对 `writer.write()` 的调用使用 `await`，否则会锁定流并收到 `WritableStream is locked` 错误。

```typescript
import { createTool } from '@mastra/core/tools'

export const testTool = createTool({
  execute: async (inputData, context) => {
    const { value } = inputData

    await context?.writer?.write({
      type: 'custom-event',
      status: 'pending',
    })

    const response = await fetch()

    await context?.writer?.write({
      type: 'custom-event',
      status: 'success',
    })

    return {
      value: '',
    }
  },
})
```

还可以使用 `writer.custom()` 发出顶层流数据块。这在与 UI 框架集成时很有用。

```typescript
import { createTool } from '@mastra/core/tools'

export const testTool = createTool({
  execute: async (inputData, context) => {
    const { value } = inputData

    await context?.writer?.custom({
      type: 'data-tool-progress',
      status: 'pending',
    })

    const response = await fetch()

    await context?.writer?.custom({
      type: 'data-tool-progress',
      status: 'success',
    })

    return {
      value: '',
    }
  },
})
```

### 临时数据块

默认情况下，通过 `writer.custom()` 发出的 `data-*` 数据块会作为消息历史的一部分持久化到存储。对于只在实时流式传输期间需要的数据块，例如进度更新或详细日志输出，请设置 `transient: true` 以跳过存储持久化。临时数据块仍会实时流式发送到客户端，但不会保存到数据库。

```typescript
await context?.writer?.custom({
  type: 'data-build-log',
  data: { line: 'Compiling module 3 of 12...' },
  transient: true,
})
```

如果数据量大或频率高，并且只与实时会话相关，请使用临时数据块。页面刷新后，临时数据块不再可用。只有 Tool 的返回值和非临时数据块会从存储加载。

## 使用 `writer` 参数

`writer` 参数会传给 Workflow 步骤的 `execute` 函数，可向活动流发出自定义事件、数据或值。Workflow 步骤使用这些事件在执行期间提供中间结果或状态更新。

> **注意:** 必须对 `writer.write(...)` 的调用使用 `await`，否则会锁定流并收到 `WritableStream is locked` 错误。

```typescript
import { createStep } from "@mastra/core/workflows";

export const testStep = createStep({
  execute: async ({ inputData, writer }) => {
    const { value } = inputData;

    await writer?.write({
      type: "custom-event",
      status: "pending"
    });

    const response = await fetch(...);

    await writer?.write({
      type: "custom-event",
      status: "success"
    });

    return {
      value: ""
    };
  },
});
```