跳至主要內容

使用 AI SDK UI

AI SDK UI 是一套用於建立 AI 驅動介面的 React 工具及元件庫。本指南會說明如何使用 @mastra/ai-sdk，將 Mastra 輸出轉換成 AI SDK 相容格式，讓你可在前端使用其 hooks 及元件。

備註

正從 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 及網絡路由處理器，以及供 UI 整合使用的工具和匯出類型。

@mastra/ai-sdk 可與 AI SDK UI 的三個主要 hooks 整合：useChat()、useCompletion() 及 useObject()。

安裝所需依賴套件以開始使用：

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

現在你可依照以下整合指南及實作方法操作。

整合指南
整合指南 的直接連結

一般情況下，你會設定以 AI SDK 相容格式串流 Mastra 內容的 API 路由，然後在 useChat() 等 AI SDK UI hooks 中使用這些路由。請選擇以下其中一種方式：

設定 API 路由後，便可在 useChat() hook 中使用。

Mastra 伺服器
Mastra 伺服器 的直接連結

以獨立伺服器方式執行 Mastra，並將前端（例如使用 Vite + React）連接至其 API 端點。此方式會使用 Mastra 的自訂 API 路由功能。

資訊

Mastra 的 UI Dojo 是此設定的範例。

你可以使用 chatRoute()、workflowRoute() 及 networkRoute() 建立 API 路由，以 AI SDK 相容格式串流 Mastra 內容。完成實作後，你便可在 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() 參考文件。

不限框架
不限框架 的直接連結

如果你不想運行 Mastra 伺服器，而是使用 Next.js 或 Express 等框架，可以在自己的 API 路由處理器中使用 handleChatStream()、handleWorkflowStream() 及 handleNetworkStream() 函數。

這些函數會傳回 ReadableStream，你可以使用 createUIMessageStreamResponse() 封裝它。

AI SDK v6 相容性

這些不受框架限制的處理器會保留現有的 AI SDK v5／預設行為。如果你的應用程式以 AI SDK v6 進行型別定義，請傳入 version: 'v6'。如要讓 handleChatStream() 及 handleNetworkStream() 獲得最佳 TypeScript 型別推斷，請將 messages 以 UIMessage[] 傳入，並使用你已安裝之 ai 版本提供的類型。

以下範例說明如何配合 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 伺服器建立 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 記憶體
使用 Mastra 記憶體 的直接連結

當 Agent 已設定記憶體時，Mastra 會從伺服器上的儲存空間載入對話記錄。用戶端只需傳送新訊息，而非完整對話記錄。

傳送完整記錄不但多餘，亦可能造成訊息排序錯誤，因為用戶端時間戳記或會與資料庫中儲存的時間戳記衝突。

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 記憶體如何載入及儲存訊息，請參閱訊息記錄。

chatRoute() 及 handleChatStream() 已支援記憶體。請設定用戶端只傳送新訊息，並加入 thread 及 resource 識別碼。

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（亦稱 Generative UI）可根據 Mastra 串流的資料算繪自訂 React 元件。你可為 Tool 輸出及 Workflow 進度建立視覺元件，而非顯示原始文字或 JSON，當中包括 Agent 網絡執行及自訂事件。

如有以下需要，請使用自訂 UI：

  • 將 Tool 輸出算繪成視覺元件（例如以天氣卡片取代 JSON）
  • 使用狀態指示器顯示 Workflow step 進度
  • 以逐步更新方式呈現 Agent 網絡執行
  • 在長時間操作期間顯示進度指示器或狀態更新

Data part 類型
Data part 類型 的直接連結

Mastra 會以訊息內的「parts」將資料串流至前端。每個 part 都有一個 type，用來決定其算繪方式。@mastra/ai-sdk 套件會將 Mastra 串流轉換成 AI SDK 相容的 UI Message DataParts。

Data part 類型來源說明
tool-{toolKey}AI SDK 內置Tool 呼叫及其狀態：input-available、output-available、output-error
data-workflowworkflowRoute()包含 step 狀態及最終輸出的 Workflow 執行狀態快照
data-workflow-stepworkflowRoute()已變更 step 的完整 payload 差異資料
data-networknetworkRoute()包含已排序 step 及輸出的 Agent 網絡執行資料
data-tool-agentTool 內的巢狀 Agent當目前 step 仍在運行時提供精簡的巢狀 Agent 快照
data-tool-agent-stepTool 內的巢狀 Agent巢狀 step 完成時發出的完整 step 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} parts。這些 parts 包含 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。例如，如以 tools: { weatherTool } 註冊 Tools，part 類型便會是 tool-weatherTool。

算繪 Workflow 資料
算繪 Workflow 資料 的直接連結

使用 workflowRoute() 或 handleWorkflowStream() 時，Mastra 會發出 data-workflow parts 作為 Workflow 狀態快照，並發出 data-workflow-step parts，提供已變更 step 的完整 payload。這可避免長時間運行的 Workflow 在每個中途快照中重複所有已完成 step 的輸出。

定義包含多個 step 的 Workflow，並在執行時發出 data-workflow 及 data-workflow-step parts。

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 parts，當中包括曾呼叫哪些 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 網絡。

自訂事件
自訂事件 的直接連結

使用 writer.custom() 在 Tool 的 execute() 函數內發出自訂 data parts。這適合用於進度指示器、狀態更新，或 Tool 執行期間的任何自訂 UI 更新。

自訂事件類型必須以 data- 開頭，才會識別為 data parts。

注意

你必須使用 await 等候 writer.custom() 呼叫，否則可能出現 WritableStream is locked 錯誤。

使用 writer.custom() 在 Tool 的 execute() 函數內，於不同執行階段發出以 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 記憶體載入訊息並在聊天 UI 顯示時，請使用 toAISdkV5Messages() 或 toAISdkV4Messages()，將訊息轉換成適合 useChat() 的 initialMessages 所需的 AI SDK 格式。

傳遞額外資料
傳遞額外資料 的直接連結

sendMessage() 讓你從前端向 Mastra 傳遞其他資料。這些資料之後可在伺服器上用作 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()，然後加入伺服器層級的 middleware：

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()
      },
    ],
  },
})
資訊

你可在 Tools 中透過 requestContext 參數存取這些資料。詳情請參閱 Request Context 文件。

Workflow 配合使用者核准暫停／繼續
Workflow 配合使用者核准暫停／繼續 的直接連結

Workflow 可暫停執行並等待使用者輸入，然後才繼續。這適合用於核准流程、確認步驟或任何有人參與的情境。

此 Workflow 使用：

  • suspendSchema / resumeSchema — 定義暫停 payload 及繼續輸入的資料結構
  • suspend() — 暫停 Workflow 並將暫停 payload 傳送至 UI
  • resumeData — 包含 Workflow 繼續執行時的使用者回應
  • bail() — 提早結束 Workflow（例如使用者拒絕時）

建立會暫停以等待核准的 Workflow step。此 step 會檢查 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。暫停／繼續功能需要儲存空間才能保留狀態。

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
  • 如要繼續執行，請在請求 body 中傳送 runId、step（step ID）及 resumeData
  • 必須設定儲存空間，暫停／繼續功能才能保留 Workflow 狀態

如要查看完整實作，請參閱 UI Dojo 的 workflow-suspend-resume 範例。

Tool 內的巢狀 Agent 串流
Tool 內的巢狀 Agent 串流 的直接連結

Tool 可在內部呼叫 Agent，並將 Agent 的輸出串流回 UI。當巢狀 step 仍在運行時，這會建立精簡的 data-tool-agent 快照；巢狀 step 完成時會建立 data-tool-agent-step parts；巢狀運行完成時則會建立一份完整的 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 parts
  • 如需要剛完成之巢狀 step 的完整 payload，請讀取 data-tool-agent-step
  • AgentDataPart 包含 id（位於 part 上）及 data.text（目前的巢狀 Agent 文字快照）
  • 串流完成後，Tool 仍會傳回本身的輸出

如要查看完整實作，請參閱 UI Dojo 的 tool-nested-streams 範例。

從 Workflow steps 串流 Agent 文字
從 Workflow steps 串流 Agent 文字 的直接連結

Workflow steps 可將 Agent 串流傳送至 step 的 writer，即時串流 Agent 的文字輸出。這讓使用者在 Workflow 執行期間看到 Agent「思考」，毋須等待 step 完成。

此模式使用：

  • Workflow step 中的 writer — 將 Agent 的 fullStream 傳送至 step 的 writer
  • text 及 data-workflow parts — 前端會同時收到串流文字及 step 進度

建立 Workflow step，將 Agent 回應傳送至 step 的 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' })],
  },
})

重點：

  • step 的 writer 可在 execute 函數中使用（並非透過 context）
  • includeTextStreamParts 預設為 true；此設定套用於 workflowRoute()，因此預設會串流文字
  • 文字 parts 會即時串流，而 data-workflow parts 則會隨 step 狀態更新

如要查看完整實作，請參閱 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 所用的 Tools 發出自訂進度事件，以顯示目前啟用的 Agent。

UI Dojo 範例在事件資料中加入 stage 欄位，以識別正在運行的 subagent（例如 "report-generation"、"report-review"）。前端會按此欄位將事件分組，並顯示各自的最新狀態。

請參閱 UI Dojo 的 report-generation-tool.ts（後端）及 agent-network-custom-events.tsx（前端）。

On this page