串流
Mastra 支援 agents 及 workflows 的即時增量回應，讓使用者在輸出產生時即可查看，而毋須等待完成。這適合聊天、長篇內容、多步驟 workflows，或任何重視即時回饋的情況。
開始使用開始使用 的直接連結
Mastra 的串流 API 會按模型版本調整：
.stream()：適用於 V2 模型，支援 AI SDK v5 及更新版本（LanguageModelV2）。.streamLegacy()：適用於 V1 模型，支援 AI SDK v4（LanguageModelV1）。
Agent 串流Agent 串流 的直接連結
基本 prompts 可傳入單一字串；提供多段上下文時可傳入字串陣列；如要精確控制 roles 及對話流程，則可傳入具有 role 及 content 的訊息物件陣列。
使用 Agent.stream()using-agentstream 的直接連結
textStream 會在回應產生時將其分成多個 chunks，讓輸出逐步串流，而非一次過抵達。使用 for await loop 疊代 textStream，即可檢查每個 stream 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 會分派背景任務，請使用 Agent.streamUntilIdle() 保持 stream 開啟，直至這些任務完成，而且 agent 已有機會回應其結果。
Agent.stream() 的輸出output-from-agentstream 的直接連結
輸出會串流 agent 產生的回應。
Of course!
To help you organize your day effectively, I need a bit more information.
Here are some questions to consider:
...
Agent stream 屬性Agent stream 屬性 的直接連結
Agent stream 可存取以下回應屬性：
stream.textStream：發出文字 chunks 的可讀 stream。stream.text：resolve 為完整文字回應的 Promise。stream.finishReason：agent 停止串流的原因。stream.usage：Token 使用量資料。
AI SDK v5+ 相容性AI SDK v5+ 相容性 的直接連結
AI SDK v5（及更新版本）的模型 Providers 使用 LanguageModelV2。如錯誤訊息指出你正在使用 AI SDK v4 模型，便需要將模型套件升級至下一個 major version。
如要與 AI SDK v5+ 整合，請使用 @mastra/ai-sdk 的 toAISdkV5Stream() utility，將 Mastra streams 轉換成 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 串流 的直接連結
Workflow 串流會傳回描述執行生命週期的一連串結構化 events，而不是增量文字 chunks。使用 .createRun() 建立執行後，這種 event-based 格式可讓你即時追蹤及回應 workflow 進度。
使用 Run.stream()using-runstream 的直接連結
stream() 方法會直接傳回 events 的 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() 的輸出output-from-runstream 的直接連結
Event 結構在頂層包含 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 stream 屬性Workflow stream 屬性 的直接連結
Workflow stream 可存取以下回應屬性：
stream.status：workflow 執行狀態。stream.result：workflow 執行結果。stream.usage：workflow 執行的 token 總使用量。
Agents 或 workflows 的串流可即時顯示 LLM 輸出或 workflow 執行狀態。你可將此回饋直接傳給使用者，或在應用程式中於 workflow 狀態變更時顯示最新狀態。
Agents 或 workflows 發出的 events 代表產生及執行的不同階段，例如執行開始、產生文字或呼叫 Tool。
Event 類型Event 類型 的直接連結
以下是 .stream() 發出 events 的完整清單。
實際出現的 events 只會是其中一部分，取決於你串流的是 agent 還是 workflow：
- start：標示 agent 或 workflow 開始執行。
- step-start：表示 workflow step 已開始執行。
- text-delta：LLM 產生的增量文字 chunks。
- tool-call：Agent 決定使用 Tool 時發出，包含 Tool 名稱及引數。
- tool-result：Tool 執行傳回的結果。
- step-finish：確認特定 step 已完全結束，並可能包含該 step 結束原因等 metadata。
- finish：Agent 或 workflow 完成時發出，包含使用量統計資料。
檢查 agent streams檢查 agent streams 的直接連結
使用 for await loop 疊代 stream，檢查所有發出的 event chunks。
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 輸出範例Agent 輸出範例 的直接連結
以下是可能發出的 events 範例。每個 event 必定包含 type，亦可包含 from 及 payload 等其他 fields。
{
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 APIWriter API 的直接連結
Tools 及 workflow steps 共用 writer API。特定功能的範例請參閱 Tools 及 Workflows 文件。
Agent 使用 ToolAgent 使用 Tool 的直接連結
Agent 串流可與 tool calls 結合，讓 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.writerusing-contextwriter 的直接連結
context.writer 物件可在 Tool 的 execute() 函數中使用，並可向目前 stream 發出自訂 events、資料或值。Tools 使用這些 events，在執行期間提供中途結果或狀態更新。
你必須對 writer.write() 的呼叫使用 await，否則會鎖定 stream，並出現 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() 發出頂層 stream chunks。與 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: '',
}
},
})
暫時性資料 chunks暫時性資料 chunks 的直接連結
使用 writer.custom() 發出的 data-* chunks 預設會作為訊息記錄的一部分保存至儲存空間。如 chunks 只在即時串流期間需要，例如進度更新或詳細 log 輸出，請設定 transient: true，略過儲存持久化。暫時性 chunks 仍會即時串流至用戶端，但不會儲存至資料庫。
await context?.writer?.custom({
type: 'data-build-log',
data: { line: 'Compiling module 3 of 12...' },
transient: true,
})
如資料量大、頻率高，而且只與即時 session 有關，請使用暫時性 chunks。重新整理頁面後，暫時性 chunks 將不再可用。從儲存空間載入的只有 Tool 傳回值及所有非暫時性 chunks。
使用 writer 引數using-the-writer-argument 的直接連結
writer 引數會傳入 workflow step 的 execute 函數，並可向目前 stream 發出自訂 events、資料或值。Workflow steps 使用這些 events，在執行期間提供中途結果或狀態更新。
你必須對 writer.write(...) 的呼叫使用 await，否則會鎖定 stream，並出現 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: ""
};
},
});