跳到主要内容

使用 AI SDK UI

AI SDK UI 是一个用于构建 AI 驱动界面的 React 工具和组件库。本指南将介绍如何使用 @mastra/ai-sdk 将 Mastra 输出转换为 AI SDK 兼容格式，以便在前端使用 AI SDK UI 的 hook 和组件。

备注

要从 AI SDK v4 迁移到 v5？请参阅迁移指南。

提示

想查看更多示例？请访问 Mastra 的 UI Dojo 或 Next.js 快速入门指南。

开始使用
开始使用的直接链接

安装 @mastra/ai-sdk 包，将 Mastra 与 AI SDK UI 结合使用。@mastra/ai-sdk 提供自定义 API 路由和工具，用于以 AI SDK 兼容格式流式传输 Mastra Agent。其中包括聊天、Workflow 和网络路由 handler，以及用于 UI 集成的工具和导出类型。

@mastra/ai-sdk 可与 AI SDK UI 的三个主要 hook 集成：useChat()、useCompletion() 和 useObject()。

安装所需包以开始使用：

npm install @mastra/ai-sdk@latest @ai-sdk/react ai

现在可以按照下方的集成指南和方案继续操作了。

集成指南
集成指南的直接链接

通常，你需要设置以 AI SDK 兼容格式流式传输 Mastra 内容的 API 路由，然后在 useChat() 等 AI SDK UI hook 中使用这些路由。请选择以下方式之一：

设置好 API 路由后，就可以在 useChat() hook 中使用它们。

Mastra Server
Mastra Server的直接链接

将 Mastra 作为独立 Server 运行，并将前端（例如使用 Vite + React）连接到其 API 端点。此方式会使用 Mastra 的自定义 API 路由功能。

信息

Mastra 的 UI Dojo 就采用了这种设置。

你可以使用 chatRoute()、workflowRoute() 和 networkRoute() 创建以 AI SDK 兼容格式流式传输 Mastra 内容的 API 路由。实现后，即可在 useChat() 中使用这些 API 路由。

此示例展示如何在 /chat 端点设置聊天路由，并使用 ID 为 weatherAgent 的 Agent。

src/mastra/index.ts
import { Mastra } from '@mastra/core'
import { chatRoute } from '@mastra/ai-sdk'

export const mastra = new Mastra({
  server: {
    apiRoutes: [
      chatRoute({
        path: '/chat',
        agent: 'weatherAgent',
      }),
    ],
  },
})

你也可以使用动态 Agent 路由，详情请参阅 chatRoute() Reference 文档。

与框架无关
与框架无关的直接链接

如果你不想运行 Mastra Server，而是使用 Next.js 或 Express 等框架，可以在自己的 API 路由 handler 中使用 handleChatStream()、handleWorkflowStream() 和 handleNetworkStream() 函数。

这些函数返回一个 ReadableStream，你可以使用 createUIMessageStreamResponse() 对其进行封装。

AI SDK v6 兼容性

与框架无关的 handler 会保留现有 AI SDK v5 默认行为。如果应用针对 AI SDK v6 进行类型定义，请传入 version: 'v6'。为使 handleChatStream() 和 handleNetworkStream() 获得最佳 TypeScript 类型推断，请将 messages 作为已安装 ai 版本中的 UIMessage[] 传入。

以下示例展示如何将这些函数与 Next.js App Router 配合使用。

此示例展示如何在 /chat 端点设置聊天路由，并使用 ID 为 weatherAgent 的 Agent。

app/chat/route.ts
import { handleChatStream } from '@mastra/ai-sdk'
import { createUIMessageStreamResponse } from 'ai'
import { mastra } from '@/src/mastra'

export async function POST(req: Request) {
  const params = await req.json()
  const stream = await handleChatStream({
    mastra,
    agentId: 'weatherAgent',
    params,
  })
  return createUIMessageStreamResponse({ stream })
}

useChat()
usechat的直接链接

无论是通过 Mastra Server 创建 API 路由，还是使用自行选择的框架，现在都可以在 useChat() hook 中使用这些 API 端点。

假设你已在 /chat 设置使用天气 Agent 的路由，就可以如下向其提问。请务必设置正确的 api URL。

import { useChat } from '@ai-sdk/react'
import { useState } from 'react'
import { DefaultChatTransport } from 'ai'

export default function Chat() {
  const [inputValue, setInputValue] = useState('')
  const { messages, sendMessage } = useChat({
    transport: new DefaultChatTransport({
      api: 'http://localhost:4111/chat',
    }),
  })

  const handleFormSubmit = (e: React.FormEvent) => {
    e.preventDefault()
    sendMessage({ text: inputValue })
  }

  return (
    <div>
      <pre>{JSON.stringify(messages, null, 2)}</pre>
      <form onSubmit={handleFormSubmit}>
        <input
          value={inputValue}
          onChange={e => setInputValue(e.target.value)}
          placeholder="Name of the city"
        />
      </form>
    </div>
  )
}

使用 prepareSendMessagesRequest 自定义发送到聊天路由的请求，例如向 Agent 传递额外配置。

使用 Mastra Memory
使用 Mastra Memory的直接链接

为 Agent 配置 Memory 后，Mastra 会在 Server 上从存储中加载对话历史。客户端只需发送新消息，而不要发送完整对话历史。

发送完整历史不仅多余，还可能导致消息顺序错误，因为客户端时间戳可能与数据库中存储的时间戳冲突。

import { useChat } from '@ai-sdk/react'
import { DefaultChatTransport } from 'ai'

const { messages, sendMessage } = useChat({
  transport: new DefaultChatTransport({
    api: 'http://localhost:4111/chat/weatherAgent',
    prepareSendMessagesRequest({ messages }) {
      return {
        body: {
          messages: [messages[messages.length - 1]],
          memory: {
            thread: 'user-thread-123',
            resource: 'user-123',
          },
        },
      }
    },
  }),
})

根据应用自身的状态设置 memory.thread 和 memory.resource，例如 URL 参数、身份验证上下文或数据库中的值。

有关 Mastra Memory 如何加载和存储消息的更多信息，请参阅消息历史。

chatRoute() 和 handleChatStream() 已支持 Memory。请配置客户端，使其只发送新消息，并包含线程和资源标识符。

useCompletion()
usecompletion的直接链接

useCompletion() hook 处理前端与 Mastra Agent 之间的单轮补全，让你可以发送提示词并通过 HTTP 接收流式响应。

前端可以如下实现：

app/page.tsx
import { useCompletion } from '@ai-sdk/react'

export default function Page() {
  const { completion, input, handleInputChange, handleSubmit } = useCompletion({
    api: '/api/completion',
  })

  return (
    <form onSubmit={handleSubmit}>
      <input name="prompt" value={input} onChange={handleInputChange} id="input" />
      <button type="submit">Submit</button>
      <div>{completion}</div>
    </form>
  )
}

请选择一种后端实现：

src/mastra/index.ts
import { Mastra } from '@mastra/core/mastra'
import { registerApiRoute } from '@mastra/core/server'
import { handleChatStream } from '@mastra/ai-sdk'
import { createUIMessageStreamResponse } from 'ai'

export const mastra = new Mastra({
  server: {
    apiRoutes: [
      registerApiRoute('/completion', {
        method: 'POST',
        handler: async c => {
          const { prompt } = await c.req.json()
          const mastra = c.get('mastra')
          const stream = await handleChatStream({
            mastra,
            agentId: 'weatherAgent',
            params: {
              messages: [
                {
                  id: '1',
                  role: 'user',
                  parts: [
                    {
                      type: 'text',
                      text: prompt,
                    },
                  ],
                },
              ],
            },
          })

          return createUIMessageStreamResponse({ stream })
        },
      }),
    ],
  },
})

自定义 UI
自定义 UI的直接链接

自定义 UI（也称为生成式 UI）允许你根据 Mastra 流式传输的数据渲染自定义 React 组件。你可以为 Tool 输出和 Workflow 进度创建可视化组件，而不是显示原始文本或 JSON；这也包括 Agent 网络执行和自定义事件。

以下场景适合使用自定义 UI：

  • 将 Tool 输出渲染为可视化组件（例如用天气卡片替代 JSON）
  • 使用状态指示器显示 Workflow 步骤进度
  • 通过逐步更新将 Agent 网络执行可视化
  • 在长时运行的操作期间显示进度指示器或状态更新

数据 part 类型
数据 part 类型的直接链接

Mastra 将数据作为消息中的“part”流式传输到前端。每个 part 都有一个 type，用于决定如何进行渲染。@mastra/ai-sdk 包将 Mastra 流转换为 AI SDK 兼容的 UI Message DataPart。

数据 part 类型来源说明
tool-{toolKey}AI SDK 内置Tool 调用及其状态：input-available、output-available、output-error
data-workflowworkflowRoute()包含步骤状态和最终输出的 Workflow 执行状态快照
data-workflow-stepworkflowRoute()Workflow 步骤增量，包含发生变化步骤的完整 payload
data-networknetworkRoute()包含有序步骤和输出的 Agent 网络执行数据
data-tool-agentTool 中的嵌套 Agent当前步骤仍在运行时的精简嵌套 Agent 快照
data-tool-agent-stepTool 中的嵌套 Agent嵌套步骤结束时发出的完整嵌套 Agent 步骤 payload
data-tool-workflowTool 中的嵌套 Workflow从 Tool 的 execute() 内部流式传输的 Workflow 输出
data-tool-networkTool 中的嵌套网络从 Tool 的 execute() 内部流式传输的网络输出
data-{custom}writer.custom()用于进度指示器、状态更新等的自定义事件

渲染 Tool 输出
渲染 Tool 输出的直接链接

当 Agent 调用 Tool 时，AI SDK 会自动创建 tool-{toolKey} part。这些 part 包含 Tool 的状态和输出，可用于渲染自定义组件。

Tool part 会依次经历以下状态：

  • input-streaming：正在流式传输 Tool 输入（启用 Tool 调用流式传输时）
  • input-available：已使用完整输入调用 Tool，正在等待执行
  • output-available：Tool 执行完成并产生输出
  • output-error：Tool 执行失败

以下示例将天气 Tool 的输出渲染为自定义 WeatherCard 组件。

使用 outputSchema 定义 Tool，使前端了解要渲染的数据结构。

src/mastra/tools/weather-tool.ts
import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const weatherTool = createTool({
  id: 'get-weather',
  description: 'Get current weather for a location',
  inputSchema: z.object({
    location: z.string().describe('The location to get the weather for'),
  }),
  outputSchema: z.object({
    temperature: z.number(),
    feelsLike: z.number(),
    humidity: z.number(),
    windSpeed: z.number(),
    conditions: z.string(),
    location: z.string(),
  }),
  execute: async inputData => {
    const response = await fetch(
      `https://api.weatherapi.com/v1/current.json?key=${process.env.WEATHER_API_KEY}&q=${inputData.location}`,
    )
    const data = await response.json()
    return {
      temperature: data.current.temp_c,
      feelsLike: data.current.feelslike_c,
      humidity: data.current.humidity,
      windSpeed: data.current.wind_kph,
      conditions: data.current.condition.text,
      location: data.location.name,
    }
  },
})
提示

Tool part 类型遵循 tool-{toolKey} 模式，其中 toolKey 是向 Agent 注册 Tool 时使用的 Key。例如，如果将 Tool 注册为 tools: { weatherTool }，part 类型将是 tool-weatherTool。

渲染 Workflow 数据
渲染 Workflow 数据的直接链接

使用 workflowRoute() 或 handleWorkflowStream() 时，Mastra 会发出表示 Workflow 状态快照的 data-workflow part，以及包含发生变化步骤完整 payload 的 data-workflow-step part。这样可以避免长时运行的 Workflow 在每个中间快照中重复所有已完成步骤的输出。

定义一个多步骤 Workflow，它会在执行过程中发出 data-workflow 和 data-workflow-step part。

src/mastra/workflows/activities-workflow.ts
import { createStep, createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const fetchWeather = createStep({
  id: 'fetch-weather',
  inputSchema: z.object({
    location: z.string(),
  }),
  outputSchema: z.object({
    temperature: z.number(),
    conditions: z.string(),
  }),
  execute: async ({ inputData }) => {
    // Fetch weather data...
    return { temperature: 22, conditions: 'Sunny' }
  },
})

const planActivities = createStep({
  id: 'plan-activities',
  inputSchema: z.object({
    temperature: z.number(),
    conditions: z.string(),
  }),
  outputSchema: z.object({
    activities: z.string(),
  }),
  execute: async ({ inputData, mastra }) => {
    const agent = mastra?.getAgent('activityAgent')
    const response = await agent?.generate(
      `Suggest activities for ${inputData.conditions} weather at ${inputData.temperature}°C`,
    )
    return { activities: response?.text || '' }
  },
})

export const activitiesWorkflow = createWorkflow({
  id: 'activities-workflow',
  inputSchema: z.object({
    location: z.string(),
  }),
  outputSchema: z.object({
    activities: z.string(),
  }),
})
  .then(fetchWeather)
  .then(planActivities)

activitiesWorkflow.commit()

向 Mastra 注册该 Workflow，并通过 workflowRoute() 将其暴露，以便将 Workflow 事件流式传输到前端。

src/mastra/index.ts
import { Mastra } from '@mastra/core'
import { workflowRoute } from '@mastra/ai-sdk'

export const mastra = new Mastra({
  workflows: { activitiesWorkflow },
  server: {
    apiRoutes: [
      workflowRoute({
        path: '/workflow/activitiesWorkflow',
        workflow: 'activitiesWorkflow',
      }),
    ],
  },
})

有关 Workflow 流式传输的更多信息，请参阅 Workflow 流式传输。

渲染网络数据
渲染网络数据的直接链接

使用 networkRoute() 或 handleNetworkStream() 时，Mastra 会发出包含 Agent 网络执行状态的 data-network part，其中包括调用了哪些 Agent 及其输出。

向 Mastra 注册 Agent，并通过 networkRoute() 暴露路由 Agent，以便将网络执行事件流式传输到前端。

src/mastra/index.ts
import { Mastra } from '@mastra/core'
import { networkRoute } from '@mastra/ai-sdk'

export const mastra = new Mastra({
  agents: { routingAgent, researchAgent, weatherAgent },
  server: {
    apiRoutes: [
      networkRoute({
        path: '/network',
        agent: 'routingAgent',
      }),
    ],
  },
})

有关 Agent 网络的更多信息，请参阅 Agent 网络。

自定义事件
自定义事件的直接链接

在 Tool 的 execute() 函数中使用 writer.custom() 发出自定义数据 part。这适用于进度指示器、状态更新或 Tool 执行期间的任何自定义 UI 更新。

自定义事件类型必须以 data- 开头，才能被识别为数据 part。

注意

必须对 writer.custom() 调用使用 await，否则可能会遇到 WritableStream is locked 错误。

在 Tool 的 execute() 函数中使用 writer.custom()，在不同执行阶段发出以 data- 为前缀的自定义事件。

src/mastra/tools/task-tool.ts
import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const taskTool = createTool({
  id: 'process-task',
  description: 'Process a task with progress updates',
  inputSchema: z.object({
    task: z.string().describe('The task to process'),
  }),
  outputSchema: z.object({
    result: z.string(),
    status: z.string(),
  }),
  execute: async (inputData, context) => {
    const { task } = inputData

    // Emit "in progress" custom event
    await context?.writer?.custom({
      type: 'data-tool-progress',
      data: {
        status: 'in-progress',
        message: 'Gathering information...',
      },
    })

    // Simulate work
    await new Promise(resolve => setTimeout(resolve, 3000))

    // Emit "done" custom event
    await context?.writer?.custom({
      type: 'data-tool-progress',
      data: {
        status: 'done',
        message: `Successfully processed "${task}"`,
      },
    })

    return {
      result: `Task "${task}" has been completed successfully!`,
      status: 'completed',
    }
  },
})

Tool 流式传输
Tool 流式传输的直接链接

Tool 还可以使用 context.writer.write() 流式传输数据以实现更底层的控制，也可以将 Agent 流直接传送到 Tool 的 writer。详情请参阅 Tool 流式传输。

示例
示例的直接链接

有关自定义 UI 模式的实时示例，请访问 Mastra UI Dojo。该仓库包含以下实现：

实用方案
实用方案的直接链接

流转换
流转换的直接链接

如需手动将 Mastra 流转换为 AI SDK 兼容格式，请使用 toAISdkStream() 工具。具体用法请参阅示例。

toAISdkStream() 会保留现有 AI SDK v5 默认行为。如果应用针对 AI SDK v6 进行类型定义，请传入 version: 'v6'。

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

const v5Stream = toAISdkStream(mastraStream, { from: 'agent' })
const v6Stream = toAISdkStream(mastraStream, { from: 'agent', version: 'v6' })

加载历史消息
加载历史消息的直接链接

从 Mastra Memory 加载消息并在聊天 UI 中显示时，请使用 toAISdkV5Messages() 或 toAISdkV4Messages()，将其转换为适合 useChat() 的 initialMessages 使用的相应 AI SDK 格式。

传递额外数据
传递额外数据的直接链接

sendMessage() 允许你从前端向 Mastra 传递额外数据。随后，可以在 Server 上将这些数据用作 RequestContext。

以下是前端代码示例：

import { useChat } from '@ai-sdk/react'
import { useState } from 'react'
import { DefaultChatTransport } from 'ai'

export function ChatAdditional() {
  const [inputValue, setInputValue] = useState('')
  const { messages, sendMessage } = useChat({
    transport: new DefaultChatTransport({
      api: 'http://localhost:4111/chat-extra',
    }),
  })

  const handleFormSubmit = (e: React.FormEvent) => {
    e.preventDefault()
    sendMessage(
      { text: inputValue },
      {
        body: {
          data: {
            userId: 'user123',
            preferences: {
              language: 'en',
              temperature: 'celsius',
            },
          },
        },
      },
    )
  }

  return (
    <div>
      <pre>{JSON.stringify(messages, null, 2)}</pre>
      <form onSubmit={handleFormSubmit}>
        <input
          value={inputValue}
          onChange={e => setInputValue(e.target.value)}
          placeholder="Name of the city"
        />
      </form>
    </div>
  )
}

请使用以下任一示例实现后端。

如上所示，在 Mastra 配置中添加 chatRoute()。然后，添加 Server 级中间件：

src/mastra/index.ts
import { Mastra } from '@mastra/core'

export const mastra = new Mastra({
  server: {
    middleware: [
      async (c, next) => {
        const requestContext = c.get('requestContext')

        if (c.req.method === 'POST') {
          const clonedReq = c.req.raw.clone()
          const body = await clonedReq.json()

          if (body?.data) {
            for (const [key, value] of Object.entries(body.data)) {
              requestContext.set(key, value)
            }
          }
        }
        await next()
      },
    ],
  },
})
信息

你可以通过 requestContext 参数在 Tool 中访问这些数据。详情请参阅 Request Context 文档。

经用户批准后暂停或恢复 Workflow
经用户批准后暂停或恢复 Workflow的直接链接

Workflow 可以暂停执行并等待用户输入，然后再继续运行。这适用于审批流、确认流程或任何人在回路场景。

该 Workflow 使用以下内容：

  • suspendSchema / resumeSchema：定义暂停 payload 和恢复输入的数据结构
  • suspend()：暂停 Workflow 并将暂停 payload 发送到 UI
  • resumeData：包含 Workflow 恢复时的用户响应
  • bail()：提前退出 Workflow（例如用户拒绝时）

创建一个暂停并等待批准的 Workflow 步骤。该步骤检查 resumeData 以判断是否正在恢复，并在首次执行时调用 suspend()。

src/mastra/workflows/approval-workflow.ts
import { createStep, createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const requestApproval = createStep({
  id: 'request-approval',
  inputSchema: z.object({ requestId: z.string(), summary: z.string() }),
  outputSchema: z.object({
    approved: z.boolean(),
    requestId: z.string(),
    approvedBy: z.string().optional(),
  }),
  resumeSchema: z.object({
    approved: z.boolean(),
    approverName: z.string().optional(),
  }),
  suspendSchema: z.object({
    message: z.string(),
    requestId: z.string(),
  }),
  execute: async ({ inputData, resumeData, suspend, bail }) => {
    // User rejected - bail out
    if (resumeData?.approved === false) {
      return bail({ message: 'Request rejected' })
    }
    // User approved - continue
    if (resumeData?.approved) {
      return {
        approved: true,
        requestId: inputData.requestId,
        approvedBy: resumeData.approverName || 'User',
      }
    }
    // First execution - suspend and wait
    return await suspend({
      message: `Please approve: ${inputData.summary}`,
      requestId: inputData.requestId,
    })
  },
})

export const approvalWorkflow = createWorkflow({
  id: 'approval-workflow',
  inputSchema: z.object({ requestId: z.string(), summary: z.string() }),
  outputSchema: z.object({
    approved: z.boolean(),
    requestId: z.string(),
    approvedBy: z.string().optional(),
  }),
}).then(requestApproval)

approvalWorkflow.commit()

注册该 Workflow。暂停或恢复需要使用 Storage 来持久化状态。

src/mastra/index.ts
import { Mastra } from '@mastra/core'
import { workflowRoute } from '@mastra/ai-sdk'
import { LibSQLStore } from '@mastra/libsql'

export const mastra = new Mastra({
  workflows: { approvalWorkflow },
  storage: new LibSQLStore({
    id: 'mastra-storage',
    url: 'file:../mastra.db',
  }),
  server: {
    apiRoutes: [
      workflowRoute({ path: '/workflow/approvalWorkflow', workflow: 'approvalWorkflow' }),
    ],
  },
})

要点：

  • 可通过 step.suspendPayload 访问暂停 payload
  • 如需恢复，请在请求正文中发送 runId、step（步骤 ID）和 resumeData
  • 必须配置 Storage，才能在暂停或恢复时持久化 Workflow 状态

完整实现请参阅 UI Dojo 中的 workflow-suspend-resume 示例。

Tool 中的嵌套 Agent 流
Tool 中的嵌套 Agent 流的直接链接

Tool 可以在内部调用 Agent，并将 Agent 输出流式传回 UI。在嵌套步骤仍在运行时，这会创建精简的 data-tool-agent 快照；嵌套步骤结束时，会创建 data-tool-agent-step part；嵌套运行结束时，则会创建一份完整的 data-tool-agent 快照。

该模式使用以下内容：

  • context.mastra.getAgent()：从 Tool 内部获取 Agent 实例
  • agent.stream()：流式传输 Agent 响应
  • stream.fullStream.pipeTo(context.writer)：将 Agent 流传送到 Tool 的 writer

创建一个调用 Agent 并将其流传送到 Tool writer 的 Tool。

src/mastra/tools/nested-agent-tool.ts
import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const nestedAgentTool = createTool({
  id: 'nested-agent-stream',
  description: 'Analyze weather using a nested agent',
  inputSchema: z.object({
    city: z.string().describe('The city to analyze'),
  }),
  outputSchema: z.object({
    summary: z.string(),
  }),
  execute: async (inputData, context) => {
    const agent = context?.mastra?.getAgent('weatherAgent')
    if (!agent) {
      return { summary: 'Weather agent not available' }
    }

    const stream = await agent.stream(
      `Analyze the weather in ${inputData.city} and provide a summary.`,
    )

    // Pipe the agent's stream to emit data-tool-agent parts
    await stream.fullStream.pipeTo(context!.writer!)

    return { summary: (await stream.text) ?? 'No summary available' }
  },
})

创建一个使用该 Tool 的 Agent。

src/mastra/agents/forecast-agent.ts
import { Agent } from '@mastra/core/agent'
import { nestedAgentTool } from '../tools/nested-agent-tool'

export const forecastAgent = new Agent({
  id: 'forecast-agent',
  instructions: 'Use the nested-agent-stream tool when asked about weather.',
  model: 'openai/gpt-5.6-sol',
  tools: { nestedAgentTool },
})

要点：

  • 将 fullStream 传送到 context.writer 会创建 data-tool-agent part
  • 需要刚完成的嵌套步骤的完整 payload 时，请读取 data-tool-agent-step
  • AgentDataPart 包含 id（位于 part 上）和 data.text（当前嵌套 Agent 文本快照）
  • 流结束后，Tool 仍会返回自己的输出

完整实现请参阅 UI Dojo 中的 tool-nested-streams 示例。

从 Workflow 步骤流式传输 Agent 文本
从 Workflow 步骤流式传输 Agent 文本的直接链接

Workflow 步骤可以将 Agent 流传送到步骤的 writer，从而实时流式传输 Agent 的文本输出。这样，用户可以在 Workflow 执行期间看到 Agent 正在“思考”，而无需等待步骤完成。

该模式使用以下内容：

  • Workflow 步骤中的 writer：将 Agent 的 fullStream 传送到步骤的 writer
  • text 和 data-workflow part：前端在接收步骤进度的同时，也会接收流式文本

创建一个 Workflow 步骤，通过将 Agent 响应传送到步骤的 writer 来进行流式传输。

src/mastra/workflows/weather-workflow.ts
import { createStep, createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'
import { weatherAgent } from '../agents/weather-agent'

const analyzeWeather = createStep({
  id: 'analyze-weather',
  inputSchema: z.object({ location: z.string() }),
  outputSchema: z.object({ analysis: z.string(), location: z.string() }),
  execute: async ({ inputData, writer }) => {
    const response = await weatherAgent.stream(
      `Analyze the weather in ${inputData.location} and provide insights.`,
    )

    // Pipe agent stream to step writer for real-time text streaming
    await response.fullStream.pipeTo(writer)

    return {
      analysis: await response.text,
      location: inputData.location,
    }
  },
})

const calculateScore = createStep({
  id: 'calculate-score',
  inputSchema: z.object({ analysis: z.string(), location: z.string() }),
  outputSchema: z.object({ score: z.number(), summary: z.string() }),
  execute: async ({ inputData }) => {
    const score = inputData.analysis.includes('sunny') ? 85 : 50
    return { score, summary: `Comfort score for ${inputData.location}: ${score}/100` }
  },
})

export const weatherWorkflow = createWorkflow({
  id: 'weather-workflow',
  inputSchema: z.object({ location: z.string() }),
  outputSchema: z.object({ score: z.number(), summary: z.string() }),
})
  .then(analyzeWeather)
  .then(calculateScore)

weatherWorkflow.commit()

使用 workflowRoute() 注册该 Workflow。默认启用文本流式传输。

src/mastra/index.ts
import { Mastra } from '@mastra/core'
import { workflowRoute } from '@mastra/ai-sdk'

export const mastra = new Mastra({
  agents: { weatherAgent },
  workflows: { weatherWorkflow },
  server: {
    apiRoutes: [workflowRoute({ path: '/workflow/weather', workflow: 'weatherWorkflow' })],
  },
})

要点：

  • 可以直接在 execute 函数中使用步骤的 writer（不通过 context）
  • workflowRoute() 上的 includeTextStreamParts 默认为 true，因此默认会流式传输文本
  • 文本 part 会实时流式传输，同时 data-workflow part 会随着步骤状态更新

完整实现请参阅 UI Dojo 中的 workflow-agent-text-stream 示例。

分支 Workflow 的多阶段进度
分支 Workflow 的多阶段进度的直接链接

对于包含条件分支的 Workflow（例如加急配送与标准配送），可以在自定义事件中加入标识符，以跟踪不同分支的进度。

UI Dojo 示例在事件数据中使用 stage 字段来标识正在执行的分支（例如 "validation"、"standard-processing"、"express-processing"）。前端按该字段对事件分组，以显示管道式进度 UI。

请参阅 UI Dojo 中的 branching-workflow.ts（后端）和 workflow-custom-events.tsx（前端）。

Agent 网络中的进度指示器
Agent 网络中的进度指示器的直接链接

使用 Agent 网络时，可以从子 Agent 所使用的 Tool 中发出自定义进度事件，以显示当前处于活动状态的 Agent。

UI Dojo 示例在事件数据中包含 stage 字段，用于标识正在运行的子 Agent（例如 "report-generation"、"report-review"）。前端按该字段对事件分组，并显示每个阶段的最新状态。

请参阅 UI Dojo 中的 report-generation-tool.ts（后端）和 agent-network-custom-events.tsx（前端）。

On this page