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:

string

要使用的評分器名稱。

sampling?:

ScoringSamplingConfig

評分器的取樣設定。

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>

每次迭代完成後呼叫的回呼函式。可用來監控進度、提供意見以引導 Agent，或提早停止執行。此回呼會接收迭代的上下文，包含目前文字、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 下一次迭代的意見訊息。

isTaskComplete?:

IsTaskCompleteConfig

用於驗證任務是否完成的完成度評分設定。它會使用 Mastra 的評估評分器，自動檢查 Agent 回應是否符合完成條件。

scorers:

MastraScorer[]

評估任務是否完成的評分器陣列。每個評分器會回傳 0（失敗）或 1（通過）。

strategy?:

'all' | 'any'

合併評分結果的策略。'all' 要求所有評分器都通過，'any' 則至少需有一個通過。

onComplete?:

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

任務完成檢查結束時呼叫的回呼。會接收包含各評分器分數的結果。

parallel?:

boolean

是否平行執行評分器。

timeout?:

number

等待所有評分器完成的時間上限（毫秒）。

suppressFeedback?:

boolean

設為 true 時，會標記完成檢查的意見，讓取用端可將其從顯示的輸出中隱藏。只有檢查失敗時才會把意見加入對話，以引導下一次迭代。

delegation?:

DelegationConfig

subagent 委派的設定。可用來控制及監控 Agent 何時將任務委派給其他 Agent，並能修改或拒絕委派，以及提供意見來引導 supervisor。

onDelegationStart?:

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

委派給 subagent 前呼叫。可用來修改委派參數、完全拒絕委派，或變更 context.requestContext，將項目加入 subagent 執行的請求上下文。

onDelegationComplete?:

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

subagent 委派完成後呼叫。上下文包含可停止後續執行的 bail() 方法；你也可以回傳 { feedback } 來引導 supervisor 的下一個動作。意見會以 assistant 訊息儲存在 supervisor memory 中。

messageFilter?:

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

委派給 subagent 前呼叫的回呼函式。可用來篩選要傳給 subagent 的訊息。

tracingContext?:

TracingContext

用於 span 階層與中繼資料的 Tracing 上下文。

returnScorerData?:

boolean

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

onChunk?:

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

串流期間每個 chunk 都會呼叫的回呼函式。

onError?:

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

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

onAbort?:

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

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

abortSignal?:

AbortSignal

可讓你中止 Agent 執行的 signal 物件。signal 中止時，所有進行中的作業都會終止，包含 Agent 委派且仍在執行的所有 subagent run。

activeTools?:

Array<keyof ToolSet> | undefined

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

prepareStep?:

PrepareStepFunction<any>

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

context?:

ModelMessage[]

提供給 Agent 的額外上下文訊息。

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' 時，若支援便使用原生結構化輸出，否則將其內嵌注入提示。

providerOptions?:

ProviderOptions

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

outputProcessors?:

Processor[]

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

includeRawChunks?:

boolean

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

inputProcessors?:

Processor[]

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

instructions?:

string

針對這次生成覆寫 Agent 預設指示的自訂指示。適合用來動態調整 Agent 行為，而不必建立新的 Agent 執行個體。

system?:

要納入提示的自訂 system 訊息。可以是單一字串、訊息物件，或由兩者任一組成的陣列。system 訊息會提供額外的上下文或行為指示，補充 Agent 的主要指示。

output?:

Zod schema | JsonSchema7

**已棄用。**請改用未指定 model 的 structuredOutput 來達成相同效果。定義預期的輸出結構。可以是 JSON Schema 物件或 Zod schema。

memory?:

object

memory 的設定。這是管理 memory 的建議方式。

thread:

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

對話 thread，可以是字串 ID，也可以是包含 id 與選用 metadata 的物件。

resource:

string

與 thread 關聯的使用者或資源識別碼。

options?:

MemoryConfig

memory 行為的設定，包含 lastMessages、readOnly、semanticRecall、workingMemory 與 filterIncompleteToolCalls。

onTitleGenerated?:

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

thread 標題產生並持久化至儲存空間時，以非同步方式觸發的回呼。標題生成會在背景執行，可能在串流結束後才完成。只有在 memory 選項中啟用 generateTitle，且 thread 尚無標題時才會觸發。

onFinish?:

StreamTextOnFinishCallback<any> | StreamObjectOnFinishCallback<OUTPUT>

串流完成時呼叫的回呼函式。會接收最終結果。

onStepFinish?:

StreamTextOnStepFinishCallback<any> | never

每個執行步驟完成後呼叫的回呼函式。會以 JSON 字串接收步驟詳細資訊。結構化輸出無法使用此函式

telemetry?:

TelemetrySettings

串流期間的 OTLP 遙測收集設定（不是 Tracing）。

isEnabled?:

boolean

啟用或停用遙測。實驗階段預設為停用。

recordInputs?:

boolean

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

recordOutputs?:

boolean

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

functionId?:

string

此函式的識別碼。用於依函式將遙測資料分組。

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

在請求的「使用者端」執行的 Tool。這些 Tool 的定義中沒有 execute 函式。

hooks?:

ToolHooks

每次執行專用、會在 Tool 呼叫前後執行的 hook。會覆寫這次執行中相符的 Agent 層級 hook。beforeToolCall 可以回傳 { proceed: false, output } 以略過 Tool 呼叫。

savePerStep?:

boolean

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

requireToolApproval?:

boolean

設為 true 時，所有 Tool 呼叫在執行前都需要明確核准。串流會發出 tool-call-approval chunk 並暫停，直到呼叫 approveToolCall() 或 declineToolCall()。

autoResumeSuspendedTools?:

boolean

設為 true 時，當使用者在同一個 thread 傳送新訊息，就會自動恢復已暫停的 Tool。Agent 會依據 Tool 的 resumeSchema，從使用者訊息擷取 resumeData。必須設定 memory。

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 專用選項。鍵是 Provider 名稱，值則是 Provider 專用選項的記錄。

runId?:

string

這次生成執行的唯一 ID，適合用於追蹤與偵錯。

requestContext?:

RequestContext

用於相依性注入與上下文資訊的 Request Context。

tracingContext?:

TracingContext

用於建立子 span 與加入中繼資料的 Tracing 上下文。使用 Mastra 的 tracing 系統時會自動注入。

currentSpan?:

Span

用於建立子 span 與加入中繼資料的目前 span。可用來建立自訂子 span，或在執行期間更新 span 屬性。

tracingOptions?:

TracingOptions

Tracing 設定選項。

metadata?:

Record<string, any>

要加入根 trace span 的中繼資料。適合用來加入使用者 ID、session ID 或功能旗標等自訂屬性。

requestContextKeys?:

string[]

要擷取為此 trace 中繼資料的其他 RequestContext 鍵。巢狀值支援點記法（例如 'user.id'）。

traceId?:

string

這次執行要使用的 Trace ID（1–32 個十六進位字元）。若有提供，此 trace 會成為指定 trace 的一部分。

parentSpanId?:

string

這次執行要使用的父 span ID（1–16 個十六進位字元）。若有提供，根 span 會建立為此 span 的子項。

tags?:

string[]

要套用至此 trace 的標籤。這些字串標籤用於分類與篩選 trace。

versions?:

VersionOverrides

每次叫用時針對 sub-Agent 委派設定的版本覆寫。它會合併至 Mastra 執行個體層級版本之上，並透過 requestContext 在 sub-Agent 呼叫間自動傳遞。需要 editor 套件。請參閱 Editor 版本控制。

VersionOverrides

agents?:

Record<string, VersionSelector>

Agent ID 與其版本選擇器的對應表。

VersionSelector

versionId?:

string

依 ID 指定特定版本。

status?:

'draft' | 'published'

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

untilIdle?:

boolean | { maxIdleMs?: number }

設定後，串流會在背景任務的接續回合之間保持開啟。背景任務完成時，Agent 會自動重新叫用 LLM，並透過同一個 fullStream 串流接續回合。傳入 true 會使用預設設定（閒置逾時 5 分鐘），也可以傳入含 maxIdleMs 的物件進行設定。需要 memory。此選項取代獨立的 streamUntilIdle() 方法。

maxIdleMs?:

number

回合之間閒置達指定毫秒數後關閉外層串流。計時器只會在 wrapper 位於回合之間時執行。預設：5 分鐘。

回傳值
「回傳值」的直接連結

stream:

MastraModelOutput<Output>

回傳可存取串流輸出的 MastraModelOutput 執行個體。

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（及後續版本）使用此串流，可以透過工具函式 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' }),
})

使用回呼
「使用回呼」的直接連結

所有回呼函式現在都可作為頂層屬性使用，讓 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 時，請為 gateway 設定 useResponsesAPI: true，接著使用 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 連線一次只會執行一個回應。在同一個 WebSocket 傳輸上，Mastra 會拒絕重疊且包含 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+ 格式」的直接連結

使用回呼
「使用回呼」的直接連結

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

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

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