Agent.stream()

.stream() 方法讓 Agent 能以功能更強、格式更靈活的方式即時串流回應。此方法接受訊息及可選的串流選項，提供目前的串流體驗，並同時支援 Mastra 原生格式及 AI SDK v5+ 相容性。

使用範例
使用範例的直接連結

const stream = await agent.stream('message for agent')

資訊

模型相容性：此方法專為 V2 模型而設。V1 模型應使用 .streamLegacy() 方法。框架會自動偵測模型版本，版本不相符時會擲回錯誤。

參數
參數的直接連結

messages:

string | string[] | CoreMessage[] | AiMessageType[] | UIMessageWithMetadata[]

要傳送給 Agent 的訊息。可以是單一字串、字串陣列或結構化訊息物件。

options?:

AgentExecutionOptions<Output, Format>

串流過程的可選配置。

AgentExecutionOptions<Output, Format>

maxSteps?:

number

執行期間最多可執行的步驟數目。

scorers?:

MastraScorers | Record<string, { scorer: MastraScorer['name']; sampling?: ScoringSamplingConfig }>

針對執行結果運行的評估 scorer。

scorer:

string

要使用的 scorer 名稱。

sampling?:

ScoringSamplingConfig

scorer 的取樣配置。

type:

'none' | 'ratio'

取樣策略類型。使用 'none' 停用取樣，或使用 'ratio' 按百分比取樣。

rate?:

number

取樣率（0 至 1）。type 為 'ratio' 時必須提供。

onIterationComplete?:

(context: IterationCompleteContext) => { continue?: boolean; feedback?: string } | void | Promise<{ continue?: boolean; feedback?: string } | void>

每次反覆運算完成後呼叫的 callback 函式。可用來監察進度、提供 feedback 以引導 Agent，或提早停止執行。callback 會收到該次反覆運算的 context，包括目前文字、Tool 呼叫及完成原因。

context.iteration:

number

目前的反覆運算編號（從 1 開始）。

context.maxIterations:

number | undefined

允許的反覆運算次數上限（如有設定）。

context.text:

string

此次反覆運算的文字回應。

context.isFinal:

boolean

這是否為最後一次反覆運算。

context.finishReason:

string

此次反覆運算結束的原因（例如 'stop'、'length'、'tool-calls'）。

context.toolCalls:

ToolCall[]

此次反覆運算所作的 Tool 呼叫。

context.messages:

MastraDBMessage[]

截至目前累積的所有訊息。

return.continue?:

boolean

設為 false 可提早停止執行。

return.feedback?:

string

用來引導 Agent 下一次反覆運算的 feedback 訊息。

isTaskComplete?:

IsTaskCompleteConfig

驗證任務是否完成的任務完成評分配置。使用 Mastra 的評估 scorer，自動檢查 Agent 回應是否符合完成準則。

scorers:

MastraScorer[]

評估任務完成情況的 scorer 陣列。每個 scorer 會傳回 0（不通過）或 1（通過）。

strategy?:

'all' | 'any'

合併 scorer 結果的策略。'all' 要求所有 scorer 通過，'any' 則要求至少一個通過。

onComplete?:

(result: IsTaskCompleteRunResult) => void | Promise<void>

任務完成檢查結束時呼叫的 callback，會收到包含每個 scorer 分數的結果。

parallel?:

boolean

是否並行運行 scorer。

timeout?:

number

等待所有 scorer 完成的最長時間（毫秒）。

suppressFeedback?:

boolean

設為 true 時，會標記完成檢查的 feedback，讓使用端可將其從顯示輸出中隱藏。只有檢查失敗時才會將 feedback 加入對話，以引導下一次反覆運算。

delegation?:

DelegationConfig

子 Agent 委派的配置。用來控制及監察 Agent 何時把任務委派給其他 Agent，亦可修改或拒絕委派，以及提供 feedback 引導監督 Agent。

onDelegationStart?:

(context: DelegationStartContext) => DelegationStartResult | void | Promise<DelegationStartResult | void>

委派給子 Agent 前呼叫。可用來修改委派參數、完全拒絕委派，或改動 context.requestContext，在子 Agent 執行的 request context 中加入項目。

onDelegationComplete?:

(context: DelegationCompleteContext) => { feedback?: string } | void | Promise<{ feedback?: string } | void>

子 Agent 委派完成後呼叫。context 包含可停止後續執行的 bail() 方法；你亦可傳回 { feedback }，引導監督 Agent 的下一步操作。feedback 會以 assistant 訊息儲存至監督 Agent 的記憶體。

messageFilter?:

(context: MessageFilterContext) => MastraDBMessage[] | Promise<MastraDBMessage[]>

委派給子 Agent 前呼叫的 callback 函式。可用來篩選要傳給子 Agent 的訊息。

tracingContext?:

TracingContext

用於 span 階層及 metadata 的 Tracing context。

returnScorerData?:

boolean

是否在回應中傳回詳細評分資料。

onChunk?:

(chunk: ChunkType) => Promise<void> | void

串流期間針對每個資料區塊呼叫的 callback 函式。

onError?:

({ error }: { error: Error | string }) => Promise<void> | void

串流期間發生錯誤時呼叫的 callback 函式。

onAbort?:

(event: any) => Promise<void> | void

串流中止時呼叫的 callback 函式。

abortSignal?:

AbortSignal

讓你中止 Agent 執行的 signal 物件。signal 中止時，所有進行中的操作都會終止，包括 Agent 委派且仍在進行的子 Agent 執行。

activeTools?:

Array<keyof ToolSet> | undefined

執行期間可使用的已啟用 Tool 名稱陣列。

prepareStep?:

PrepareStepFunction<any>

多步驟執行中每個步驟開始前呼叫的 callback 函式。

context?:

ModelMessage[]

提供給 Agent 的其他 context 訊息。

structuredOutput?:

StructuredOutputOptions<S extends ZodTypeAny = ZodTypeAny>

微調結構化輸出生成的選項。

schema:

StandardJSONSchemaV1

定義預期輸出結構的標準 JSON Schema。

model?:

MastraLanguageModel

用於生成結構化輸出的語言模型。如有提供，Agent 便可透過 Tool 呼叫、文字及結構化輸出作多步驟回應

errorStrategy?:

'strict' | 'warn' | 'fallback'

處理 schema 驗證錯誤的策略。'strict' 會擲回錯誤、'warn' 會記錄警告，而 'fallback' 會使用後備值。

fallbackValue?:

schema 驗證失敗且 errorStrategy 為 'fallback' 時使用的後備值。

instructions?:

string

提供給結構化輸出模型的其他指示。

jsonPromptInjection?:

boolean | 'system' | 'inline' | 'auto'

控制 JSON schema 如何傳送至模型。設為 'auto'，支援時使用原生結構化輸出，否則使用 inline prompt injection。

providerOptions?:

ProviderOptions

傳給內部結構化 Agent 的 Provider 專用選項。可用來控制模型行為，例如思考模型的推理強度（例如 { openai: { reasoningEffort: 'low' } }）。

outputProcessors?:

Processor[]

覆寫 Agent 上設定的輸出 processor。輸出 processor 可在 Agent 訊息傳回使用者前修改或驗證訊息。必須實作 processOutputResult 或 processOutputStream 函式其中之一（或兩者）。

includeRawChunks?:

boolean

是否在串流輸出中包含原始資料區塊（並非所有模型 Provider 都支援）。

inputProcessors?:

Processor[]

覆寫 Agent 上設定的輸入 processor。輸入 processor 可在 Agent 處理訊息前修改或驗證訊息。必須實作 processInput 函式。

instructions?:

string

針對此次生成覆寫 Agent 預設指示的自訂指示。無需建立新的 Agent instance，便可動態修改 Agent 行為。

system?:

要加入 prompt 的自訂 system 訊息。可以是單一字串、訊息物件或兩者任一的陣列。system 訊息會提供額外 context 或行為指示，以補充 Agent 的主要指示。

output?:

Zod schema | JsonSchema7

**已棄用。** 使用不含 model 的 structuredOutput 可達到相同效果。定義預期的輸出結構。可以是 JSON Schema 物件或 Zod schema。

memory?:

object

記憶體配置。這是管理記憶體的建議方式。

thread:

string | { id: string; metadata?: Record<string, any>, title?: string }

對話 thread，可以是字串 ID，或包含 id 及可選 metadata 的物件。

resource:

string

與 thread 相關聯的使用者或資源標識符。

options?:

MemoryConfig

記憶體行為配置，包括 lastMessages、readOnly、semanticRecall、workingMemory 及 filterIncompleteToolCalls。

onTitleGenerated?:

(title: string) => void | Promise<void>

生成 thread 標題並持久儲存後非同步觸發的 callback。標題生成會在背景運行，可能在串流結束後才完成。只有在記憶體選項啟用 generateTitle，而且 thread 尚無標題時才會觸發。

onFinish?:

StreamTextOnFinishCallback<any> | StreamObjectOnFinishCallback<OUTPUT>

串流完成時呼叫的 callback 函式，會收到最終結果。

onStepFinish?:

StreamTextOnStepFinishCallback<any> | never

每個執行步驟後呼叫的 callback 函式，會以 JSON 字串接收步驟詳情。不適用於結構化輸出

telemetry?:

TelemetrySettings

串流期間收集 OTLP telemetry 的設定（並非 Tracing）。

isEnabled?:

boolean

啟用或停用 telemetry。實驗階段預設停用。

recordInputs?:

boolean

啟用或停用輸入記錄。預設啟用。你可能需要停用輸入記錄，以免記錄敏感資料。

recordOutputs?:

boolean

啟用或停用輸出記錄。預設啟用。你可能需要停用輸出記錄，以免記錄敏感資料。

functionId?:

string

此函式的標識符，用於按函式為 telemetry 資料分組。

modelSettings?:

CallSettings

Model-specific settings like temperature, maxOutputTokens, topP, etc. These settings control how the language model generates responses.

temperature?:

number

Controls randomness in generation (0-2). Higher values make output more random.

maxOutputTokens?:

number

Maximum number of tokens to generate in the response. Note: Use maxOutputTokens (not maxTokens) as per AI SDK v5 convention.

maxRetries?:

number

Maximum number of retry attempts for failed requests.

topP?:

number

Nucleus sampling parameter (0-1). Controls diversity of generated text.

topK?:

number

Top-k sampling parameter. Limits vocabulary to k most likely tokens.

presencePenalty?:

number

Penalty for token presence (-2 to 2). Reduces repetition.

frequencyPenalty?:

number

Penalty for token frequency (-2 to 2). Reduces repetition of frequent tokens.

stopSequences?:

string[]

Stop sequences. If set, the model will stop generating text when one of the stop sequences is generated.

toolChoice?:

'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }

控制 Agent 在串流期間如何使用 Tool。

'auto':

string

讓模型決定是否使用 Tool（預設）。

'none':

string

不使用任何 Tool。

'required':

string

要求模型使用至少一個 Tool。

{ type: 'tool'; toolName: string }:

object

要求模型使用指定名稱的 Tool。

toolsets?:

ToolsetsInput

串流期間供 Agent 使用的其他 Toolset。

clientTools?:

ToolsInput

在請求的 'client' 端執行的 Tool。這些 Tool 的定義中沒有 execute 函式。

hooks?:

ToolHooks

每次執行時，在 Tool 呼叫前後運行的 hook。針對此次執行覆寫相符的 Agent 層級 hook。beforeToolCall 可傳回 { proceed: false, output } 以略過 Tool 呼叫。

savePerStep?:

boolean

每個串流步驟完成後逐步儲存訊息（預設：false）。

requireToolApproval?:

boolean

設為 true 時，所有 Tool 呼叫都必須在執行前明確核准。串流會發出 tool-call-approval 資料區塊並暫停，直至呼叫 approveToolCall() 或 declineToolCall()。

autoResumeSuspendedTools?:

boolean

設為 true 時，使用者在同一 thread 傳送新訊息後，會自動恢復已暫停的 Tool。Agent 會根據 Tool 的 resumeSchema，從使用者訊息擷取 resumeData。必須配置記憶體。

toolCallConcurrency?:

number

可同時執行的 Tool 呼叫數目上限。可能需要核准時預設為 1，否則為 10。

providerOptions?:

Record<string, Record<string, JSONValue>>

傳遞給底層 LLM Provider 的其他 Provider 專用選項。結構為 { providerName: { optionKey: value } }。例如：{ openai: { reasoningEffort: 'high' }, anthropic: { maxTokens: 1000 } }。

openai?:

Record<string, JSONValue>

OpenAI 專用選項。例如：{ reasoningEffort: 'high' }

anthropic?:

Record<string, JSONValue>

Anthropic 專用選項。例如：{ maxTokens: 1000 }

google?:

Record<string, JSONValue>

Google 專用選項。例如：{ safetySettings: [...] }

[providerName]?:

Record<string, JSONValue>

其他 Provider 專用選項。key 是 Provider 名稱，值則是 Provider 專用選項的 record。

runId?:

string

此次生成執行的唯一 ID，適合用於追蹤及除錯。

requestContext?:

RequestContext

用於依賴注入及 context 資訊的 Request Context。

tracingContext?:

TracingContext

用於建立子 span 及加入 metadata 的 Tracing context。使用 Mastra 的 tracing 系統時會自動注入。

currentSpan?:

Span

用於建立子 span 及加入 metadata 的目前 span。可用來建立自訂子 span，或在執行期間更新 span 屬性。

tracingOptions?:

TracingOptions

Tracing 配置選項。

metadata?:

Record<string, any>

要加入根 Trace span 的 metadata。適合用來加入自訂屬性，例如使用者 ID、工作階段 ID 或功能旗標。

requestContextKeys?:

string[]

要擷取為此 Trace metadata 的其他 RequestContext key。支援以點號表示法存取巢狀值（例如 'user.id'）。

traceId?:

string

用於此次執行的 Trace ID（1 至 32 個十六進制字元）。如有提供，此 Trace 將屬於指定的 Trace。

parentSpanId?:

string

用於此次執行的父 span ID（1 至 16 個十六進制字元）。如有提供，根 span 將建立為此 span 的子項。

tags?:

string[]

要套用至此 Trace 的 tag。這些字串標籤用於分類和篩選 Trace。

versions?:

VersionOverrides

每次呼叫時覆寫子 Agent 委派的版本。此設定會合併至 Mastra instance 層級版本之上，並透過 requestContext，在子 Agent 呼叫之間自動傳遞。需要 editor 依賴套件。請參閱 Editor 版本管理。

VersionOverrides

agents?:

Record<string, VersionSelector>

Agent ID 至其版本 selector 的映射。

VersionSelector

versionId?:

string

按 ID 指定特定版本。

status?:

'draft' | 'published'

指定具有此發佈狀態的最新版本。

untilIdle?:

boolean | { maxIdleMs?: number }

設定後，串流會在背景任務延續期間保持開啟。背景任務完成時，Agent 會自動重新呼叫 LLM，並透過同一個 fullStream 串流後續輪次。傳入 true 可使用預設設定（閒置逾時 5 分鐘），或傳入包含 maxIdleMs 的物件進行配置。需要記憶體。取代獨立的 streamUntilIdle() 方法。

maxIdleMs?:

number

輪次之間閒置達到此毫秒數後關閉外層串流。計時器只會在 wrapper 處於輪次之間時運行。預設：5 分鐘。

傳回值
傳回值的直接連結

stream:

MastraModelOutput<Output>

傳回可存取串流輸出的 MastraModelOutput instance。

traceId?:

string

啟用 Tracing 時，與此次執行相關聯的 Trace ID。可用來關聯記錄及為執行流程除錯。

spanId?:

string

啟用 Tracing 時，與此次執行相關聯的根 span ID。可用於 span 層級的查閱及關聯。

延伸使用範例
延伸使用範例的直接連結

Mastra 格式（預設）
Mastra 格式（預設）的直接連結

index.ts
import { stepCountIs } from 'ai-v5'

const stream = await agent.stream('Tell me a story', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
})

// Access text stream
for await (const chunk of stream.textStream) {
  console.log(chunk)
}

// or access full stream
for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get full text after streaming
const fullText = await stream.text

AI SDK v5+ 格式
AI SDK v5+ 格式的直接連結

如要配合 AI SDK v5（及更新版本）使用串流，可使用 utility 函式 toAISdkStream 進行轉換。

index.ts
import { stepCountIs, createUIMessageStreamResponse } from 'ai'
import { toAISdkStream } from '@mastra/ai-sdk'

const stream = await agent.stream('Tell me a story', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
})

// In an API route for frontend integration
return createUIMessageStreamResponse({
  stream: toAISdkStream(stream, { from: 'agent' }),
})

使用 callback
使用 callback 的直接連結

所有 callback 函式現在都可用作頂層屬性，提供更簡潔的 API 體驗。

index.ts
const stream = await agent.stream('Tell me a story', {
  onFinish: result => {
    console.log('Streaming finished:', result)
  },
  onStepFinish: step => {
    console.log('Step completed:', step)
  },
  onChunk: chunk => {
    console.log('Received chunk:', chunk)
  },
  onError: ({ error }) => {
    console.error('Streaming error:', error)
  },
  onAbort: event => {
    console.log('Stream aborted:', event)
  },
})

// Process the stream
for await (const chunk of stream.textStream) {
  console.log(chunk)
}

使用選項的進階範例
使用選項的進階範例的直接連結

index.ts
import { z } from 'zod'
import { stepCountIs } from 'ai'

await agent.stream('message for agent', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
  memory: {
    thread: 'user-123',
    resource: 'test-app',
  },
  toolChoice: 'auto',
  // Structured output with better DX
  structuredOutput: {
    schema: z.object({
      sentiment: z.enum(['positive', 'negative', 'neutral']),
      confidence: z.number(),
    }),
    model: 'openai/gpt-5.6-sol',
    errorStrategy: 'warn',
  },
  // Output processors for streaming response validation
  outputProcessors: [
    new ModerationProcessor({ model: 'openrouter/openai/gpt-oss-safeguard-20b' }),
    new BatchPartsProcessor({ maxBatchSize: 3, maxWaitTime: 100 }),
  ],
})

Responses WebSocket 傳輸
Responses WebSocket 傳輸的直接連結

透過 Provider 選項啟用 Responses WebSocket 串流。這只適用於串流呼叫，並支援直接使用 OpenAI 模型及 Azure OpenAI Responses 部署。如果 WebSocket 串流無法使用，Mastra 會改用 HTTP 串流。串流完成時，Mastra 預設會關閉 WebSocket。

index.ts
const stream = await agent.stream('Hello', {
  providerOptions: {
    openai: {
      transport: 'websocket', // 'websocket' | 'fetch' | 'auto'
      websocket: {
        url: 'wss://api.openai.com/v1/responses',
        closeOnFinish: true, // default
      },
    },
  },
})

使用 Azure OpenAI 時，請以 useResponsesAPI: true 配置 gateway，然後使用 providerOptions.azure.transport。

index.ts
const stream = await agent.stream('Hello', {
  providerOptions: {
    azure: {
      transport: 'websocket',
      store: false,
      websocket: { closeOnFinish: true },
    },
  },
})

如要在串流完成後保持連線開啟，請設定 closeOnFinish: false，並手動關閉連線。

index.ts
const stream = await agent.stream('Hello', {
  providerOptions: {
    openai: {
      transport: 'websocket',
      websocket: { closeOnFinish: false },
    },
  },
})

// Later, when you're done with the connection:
stream.transport?.close()

Responses WebSocket 連線一次只會運行一個回應。Mastra 會拒絕在同一 WebSocket 傳輸上，包含 previous_response_id 且互相重疊的延續請求。請等待使用中的串流完成，再傳送回應鏈中的下一輪。

使用範例使用範例 的直接連結

參數參數 的直接連結

messages:

options?:

maxSteps?:

scorers?:

scorer:

sampling?:

type:

rate?:

onIterationComplete?:

context.iteration:

context.maxIterations:

context.text:

context.isFinal:

context.finishReason:

context.toolCalls:

context.messages:

return.continue?:

return.feedback?:

isTaskComplete?:

scorers:

strategy?:

onComplete?:

parallel?:

timeout?:

suppressFeedback?:

delegation?:

onDelegationStart?:

onDelegationComplete?:

messageFilter?:

tracingContext?:

returnScorerData?:

onChunk?:

onError?:

onAbort?:

abortSignal?:

activeTools?:

prepareStep?:

context?:

structuredOutput?:

schema:

model?:

errorStrategy?:

fallbackValue?:

instructions?:

jsonPromptInjection?:

providerOptions?:

outputProcessors?:

includeRawChunks?:

inputProcessors?:

instructions?:

system?:

output?:

memory?:

thread:

resource:

options?:

onTitleGenerated?:

onFinish?:

onStepFinish?:

telemetry?:

isEnabled?:

recordInputs?:

recordOutputs?:

functionId?:

modelSettings?:

temperature?:

maxOutputTokens?:

maxRetries?:

topP?:

topK?:

presencePenalty?:

frequencyPenalty?:

stopSequences?:

toolChoice?:

'auto':

'none':

'required':

{ type: 'tool'; toolName: string }:

使用範例
使用範例的直接連結

參數
參數的直接連結

傳回值
傳回值的直接連結

延伸使用範例
延伸使用範例的直接連結

Mastra 格式（預設）
Mastra 格式（預設）的直接連結

AI SDK v5+ 格式
AI SDK v5+ 格式的直接連結

使用 callback
使用 callback 的直接連結

使用選項的進階範例
使用選項的進階範例的直接連結

Responses WebSocket 傳輸
Responses WebSocket 傳輸的直接連結

相關內容
相關內容的直接連結