流式传输

Mastra 支持来自 agents 和 workflows 的实时增量响应，让用户能够在输出生成过程中就看到内容，而不是等待完成。这对于聊天、长篇内容、多步骤 workflow，或任何需要即时反馈的场景都很有用。

快速开始

Agent.stream() 是 agent 的标准流式 API。它返回一个 MastraModelOutput，其中暴露用于渐进文本的 textStream，以及 text、steps 和 usage 等在流结束时解析的 promise。

Agent 流式传输

传入单个字符串即可作为基本提示词。提供多段上下文时，使用字符串数组。带有 role 和 content 的消息对象数组可以让你精确控制角色和对话流程。

使用 Agent.stream()

textStream 会在响应生成时将其拆分为 chunk，使输出能够逐步流式传输，而不是一次性全部到达。使用 for await 循环遍历 textStream 以检查每个流 chunk。

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，请将 stream() 与 untilIdle 选项（例如 untilIdle: true）一起使用，使流保持打开，直到这些任务完成且 agent 有机会对其结果作出响应。该选项需要 agent memory。

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：发出文本 chunk 的可读流。
• stream.text：解析为完整文本响应的 promise。
• stream.steps：解析为已完成的模型步骤的 promise。
• stream.finishReason：解析为 agent 停止流式传输原因的 promise。
• stream.usage：解析为 token 使用信息的 promise。
• stream.objectStream 和 stream.object：传入 structuredOutput 时的部分结构化输出和最终结构化输出。

完整属性集请参阅 MastraModelOutput 参考。

AI SDK 集成

使用 toAISdkStream() 和 toAISdkMessages() 将 Mastra 流和已存储消息转换为 AI SDK 兼容格式。为向后兼容，这些转换器默认使用 AI SDK v5。请传入与你所安装的 AI SDK 匹配的版本，例如 AI SDK v7 使用 version: 'v7'。

import { toAISdkStream } from '@mastra/ai-sdk'

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

const aiSDKStream = toAISdkStream(stream, {
  from: 'agent',
  version: 'v7',
})

使用 toAISdkMessages() 转换已存储消息，以供 useChat() 的 initialMessages 使用：

import { toAISdkMessages } from '@mastra/ai-sdk/ui'

const initialMessages = toAISdkMessages([{ role: 'user', content: 'Hello' }], { version: 'v7' })

关于路由处理程序，请参阅 AI SDK UI，其中涵盖 handleChatStream()、handleWorkflowStream() 和 handleNetworkStream()。请同样向这些处理程序传入与你所安装的 AI SDK 匹配的版本。所有选项请参阅 toAISdkStream() 和 toAISdkMessages() 参考。

Workflow 流式传输

来自 workflow 的流式传输返回描述运行生命周期的结构化事件序列，而不是增量文本 chunk。使用 .createRun() 创建运行后，这种基于事件的格式使你能够实时跟踪并响应 workflow 进度。

使用 Run.stream()

stream() 方法返回一个 WorkflowRunOutput。其 fullStream 属性是 workflow 事件的 ReadableStream。

const run = await testWorkflow.createRun()

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

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

更多信息请参阅 Run.stream()。

Run.stream() 的输出

事件在顶层包含 runId 和 from，因此你无需检查 payload 即可识别 workflow 运行。

{
  type: 'workflow-start',
  runId: '1eeaf01a-d2bf-4e3f-8d1b-027795ccd3df',
  from: 'WORKFLOW',
  payload: {
    workflowId: 'testWorkflow'
  }
}

Workflow 流的属性

workflow 流提供以下响应属性：

• stream.status：workflow 运行的状态。
• stream.result：workflow 运行的结果。
• stream.usage：workflow 运行的总 token 使用量。

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

由 agents 或 workflows 发出的事件代表生成和执行的不同阶段，例如运行开始、生成文本或调用工具时。

事件类型

Agent 流和 workflow 流在执行期间发出不同的事件类型。

Agent 事件

常见的 agent 事件包括：

• start：agent 运行开始。
• text-start、text-delta 和 text-end：开始和结束事件标记文本生成的边界。delta 事件携带增量文本。
• reasoning-start、reasoning-delta 和 reasoning-end：开始和结束事件标记推理生成的边界。delta 事件携带增量推理内容。
• tool-call 和 tool-result：工具被调用并返回结果。
• step-start 和 step-finish：模型步骤开始和结束。
• finish：agent 运行完成。

该列表并不详尽。所有 agent chunk 类型和 payload 请参阅 ChunkType 参考。

Workflow 事件

Workflow 事件包括：

• workflow-start：workflow 运行开始。
• workflow-step-start：workflow 步骤开始。
• workflow-step-output：步骤发出自定义输出。
• workflow-step-progress：步骤报告进度。
• workflow-step-result：步骤带结果完成。
• workflow-finish：workflow 运行完成。
• workflow-paused、workflow-step-suspended 和 workflow-canceled：workflow 运行被中断。

Workflow 事件详情请参阅 Run.stream() 参考。

检查 agent 流

使用 for await 循环遍历 stream.fullStream，以检查所有发出的事件 chunk。

const testAgent = mastra.getAgent('testAgent')

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

for await (const chunk of stream.fullStream) {
  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 由工具和 workflow 步骤共享。功能特定的示例请参阅 Tools 和 Workflows 文档。

使用工具的 agent

agent 流式传输可以与工具调用结合，使工具输出直接写入 agent 的流式响应中。这会将工具活动作为交互的一部分呈现出来。

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

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

注意

必须 await 对 writer.write() 的调用，否则流会被锁定并出现 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() 发出顶层流 chunk。这在集成 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: '',
    }
  },
})

瞬态数据 chunk

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

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

当数据量大、频率高且仅在实时会话期间相关时，请使用瞬态 chunk。页面刷新后，瞬态 chunk 将不再可用。只有工具的返回值和任何非瞬态 chunk 会从存储中加载。

使用 writer 参数

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

注意

必须 await 对 writer.write(...) 的调用，否则流会被锁定并出现 WritableStream is locked 错误。

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'

export const testStep = createStep({
  id: 'test-step',
  inputSchema: z.object({ url: z.url() }),
  outputSchema: z.object({ status: z.number() }),
  execute: async ({ inputData, writer }) => {
    const { url } = inputData

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

    const response = await fetch(url)

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

    return {
      status: response.status,
    }
  },
})