流式传输

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，即可检查每个流数据块。

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()。

提示

对于会分派后台任务的 Agent，请使用 Agent.streamUntilIdle() 保持流打开，直到这些任务完成且 Agent 有机会响应其结果。

Agent.stream() 的输出

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

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 的格式：

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：

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。

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()。

Run.stream() 的输出

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

{
  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，检查发出的所有事件数据块。

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()。

Agent 输出示例

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

{
  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 活动。

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 错误。

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 框架集成时很有用。

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 以跳过存储持久化。临时数据块仍会实时流式发送到客户端，但不会保存到数据库。

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 错误。

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: ""
    };
  },
});