串流

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() 工具函式，將 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() 工具函式：

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，無須深入檢查酬載，就能更輕鬆地識別及追蹤 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：確認特定步驟已完全結束，並可能包含該步驟結束原因等中繼資料。
• 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 文件。

使用 Tool 的 Agent

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