Agent.스트림()

그만큼.stream()이 방법을 사용하면 향상된 기능과 형식 유연성을 통해 Agent의 응답을 실시간 스트리밍할 수 있습니다. 이 방법은 메시지와 선택적 스트리밍 옵션을 허용하여 Mastra의 기본 형식과 AI SDK v5+ 호환성을 모두 지원하는 현재 스트리밍 환경을 제공합니다.

사용예

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

정보

Model 호환성: 이 메서드는 V2 Model용으로 설계되었습니다. V1 Model에는 .streamLegacy() 메서드를 사용하세요. 프레임워크가 Model 버전을 자동으로 감지하며 버전이 일치하지 않으면 오류를 발생시킵니다.

매개변수

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

하위 Agent 위임을 위한 구성입니다. Agent가 다른 Agent에게 작업을 위임하는 시점을 제어하고 모니터링하며, 위임을 수정하거나 거부하고 감독자를 안내하는 피드백을 제공할 수 있습니다.

onDelegationStart?:

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

하위 Agent에게 위임하기 전에 호출됩니다. 위임 매개변수를 수정하거나 위임을 완전히 거부하거나 context.requestContext를 변경하여 하위 Agent 실행의 요청 컨텍스트에 항목을 추가할 수 있습니다.

onDelegationComplete?:

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

하위 Agent 위임이 완료된 후 호출됩니다. 컨텍스트에는 추가 실행을 중지하는 bail() 메서드가 포함되며, { feedback }을 반환하여 감독자의 다음 작업을 안내할 수 있습니다. 피드백은 어시스턴트 메시지로 감독자의 Memory에 저장됩니다.

messageFilter?:

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

하위 Agent에게 위임하기 전에 호출되는 콜백 함수입니다. 하위 Agent에 전달되는 메시지를 필터링하는 데 사용합니다.

tracingContext?:

TracingContext

스팬 계층 구조 및 메타데이터를 위한 추적 컨텍스트입니다.

returnScorerData?:

boolean

응답에 상세한 채점 데이터를 반환할지 여부입니다.

onChunk?:

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

스트리밍 중 각 청크에 대해 호출되는 콜백 함수입니다.

onError?:

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

스트리밍 중 오류가 발생하면 호출되는 콜백 함수입니다.

onAbort?:

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

스트림이 중단되면 호출되는 콜백 함수입니다.

abortSignal?:

AbortSignal

Agent 실행을 중단할 수 있는 신호 객체입니다. 신호가 중단되면 Agent가 위임하여 현재 실행 중인 모든 하위 Agent 실행을 포함해 진행 중인 모든 작업이 종료됩니다.

activeTools?:

Array<keyof ToolSet> | undefined

실행 중 사용할 수 있는 활성 Tool 이름의 배열입니다.

prepareStep?:

PrepareStepFunction<any>

다단계 실행의 각 단계 전에 호출되는 콜백 함수입니다.

context?:

ModelMessage[]

Agent에 제공할 추가 컨텍스트 메시지입니다.

structuredOutput?:

StructuredOutputOptions<S extends ZodTypeAny = ZodTypeAny>

구조화된 출력 생성을 세밀하게 조정하는 옵션입니다.

schema:

StandardJSONSchemaV1

예상 출력 구조를 정의하는 표준 JSON Schema입니다.

model?:

MastraLanguageModel

구조화된 출력 생성에 사용할 Model입니다. 제공하면 Agent가 Tool 호출, 텍스트, 구조화된 출력을 포함하는 여러 단계로 응답할 수 있습니다.

errorStrategy?:

'strict' | 'warn' | 'fallback'

스키마 검증 오류를 처리하는 전략입니다. 'strict'는 오류를 발생시키고, 'warn'은 경고를 기록하며, 'fallback'은 대체 값을 사용합니다.

fallbackValue?:

<S extends ZodTypeAny>

스키마 검증에 실패하고 errorStrategy가 'fallback'일 때 사용할 대체 값입니다.

instructions?:

string

구조화된 출력 Model에 제공할 추가 지침입니다.

jsonPromptInjection?:

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

JSON 스키마가 Model에 전달되는 방식을 제어합니다. 'auto'로 설정하면 지원되는 경우 네이티브 구조화 출력을 사용하고, 그렇지 않으면 인라인 Prompt 삽입을 사용합니다.

providerOptions?:

ProviderOptions

내부 구조화 Agent에 전달되는 Provider별 옵션입니다. 사고 Model의 추론 수준과 같은 Model 동작을 제어할 때 사용합니다(예: { openai: { reasoningEffort: 'low' } }).

outputProcessors?:

Processor[]

Agent에 설정된 출력 프로세서를 재정의합니다. 출력 프로세서는 사용자에게 반환되기 전에 Agent의 메시지를 수정하거나 검증할 수 있습니다. processOutputResult 및 processOutputStream 함수 중 하나 이상을 구현해야 합니다.

includeRawChunks?:

boolean

스트림 출력에 원시 청크를 포함할지 여부입니다(일부 Model Provider에서는 사용할 수 없음).

inputProcessors?:

Processor[]

Agent에 설정된 입력 프로세서를 재정의합니다. 입력 프로세서는 메시지가 Agent에서 처리되기 전에 메시지를 수정하거나 검증할 수 있습니다. processInput 함수를 구현해야 합니다.

instructions?:

string

이 특정 생성 작업에서 Agent의 기본 지침을 재정의하는 사용자 지정 지침입니다. 새 Agent 인스턴스를 만들지 않고 Agent 동작을 동적으로 수정할 때 유용합니다.

system?:

string | string[] | CoreSystemMessage | SystemModelMessage | CoreSystemMessage[] | SystemModelMessage[]

Prompt에 포함할 사용자 지정 시스템 메시지입니다. 단일 문자열, 메시지 객체 또는 두 형식 중 하나의 배열을 사용할 수 있습니다. 시스템 메시지는 Agent의 기본 지침을 보완하는 추가 컨텍스트나 동작 지침을 제공합니다.

output?:

Zod schema | JsonSchema7

**사용 중단됨.** 동일한 결과를 얻으려면 Model 없이 structuredOutput을 사용하세요. 예상되는 출력 구조를 정의합니다. JSON Schema 객체 또는 Zod 스키마를 사용할 수 있습니다.

memory?:

object

Memory 구성입니다. Memory를 관리하는 데 권장되는 방식입니다.

thread:

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

문자열 ID 또는 id와 선택적 metadata를 포함하는 객체로 지정하는 대화 스레드입니다.

resource:

string

스레드와 연결된 사용자 또는 리소스의 식별자입니다.

options?:

MemoryConfig

lastMessages, readOnly, semanticRecall, workingMemory, filterIncompleteToolCalls를 포함한 Memory 동작 구성입니다.

onTitleGenerated?:

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

스레드 제목이 생성되어 스토리지에 저장되면 비동기적으로 실행되는 콜백입니다. 제목 생성은 백그라운드에서 실행되며 스트림이 종료된 후 완료될 수도 있습니다. Memory 옵션에서 generateTitle이 활성화되어 있고 스레드에 기존 제목이 없을 때만 실행됩니다.

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 사용 여부를 Model이 결정하도록 합니다(기본값).

'none':

string

어떤 Tool도 사용하지 않습니다.

'required':

string

Model이 하나 이상의 Tool을 사용하도록 요구합니다.

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

object

Model이 이름으로 지정된 특정 Tool을 사용하도록 요구합니다.

toolsets?:

ToolsetsInput

스트리밍 중 Agent가 사용할 수 있도록 제공할 추가 Tool 세트입니다.

clientTools?:

ToolsInput

요청의 'client' 측에서 실행되는 Tool입니다. 이러한 Tool의 정의에는 execute 함수가 없습니다.

hooks?:

ToolHooks

Tool 호출 전후에 실행되는 실행별 훅입니다. 이 실행에 대해 일치하는 Agent 수준 훅을 재정의합니다. beforeToolCall은 { proceed: false, output }을 반환하여 Tool 호출을 건너뛸 수 있습니다.

savePerStep?:

boolean

각 스트림 단계가 완료된 후 메시지를 점진적으로 저장합니다(기본값: false).

requireToolApproval?:

boolean

true이면 모든 Tool 호출을 실행하기 전에 명시적인 승인이 필요합니다. 스트림은 tool-call-approval 청크를 내보내고 approveToolCall() 또는 declineToolCall()이 호출될 때까지 일시 중지됩니다.

autoResumeSuspendedTools?:

boolean

true이면 사용자가 같은 스레드에 새 메시지를 보낼 때 일시 중단된 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

의존성 주입 및 컨텍스트 정보를 위한 요청 컨텍스트입니다.

tracingContext?:

TracingContext

하위 스팬을 생성하고 메타데이터를 추가하기 위한 Tracing 컨텍스트입니다. Mastra의 Tracing 시스템을 사용할 때 자동으로 주입됩니다.

currentSpan?:

Span

하위 스팬을 생성하고 메타데이터를 추가하기 위한 현재 스팬입니다. 실행 중 사용자 지정 하위 스팬을 생성하거나 스팬 속성을 업데이트할 때 사용합니다.

tracingOptions?:

TracingOptions

Tracing 구성 옵션입니다.

metadata?:

Record<string, any>

루트 Trace 스팬에 추가할 메타데이터입니다. 사용자 ID, 세션 ID 또는 기능 플래그와 같은 사용자 지정 속성을 추가할 때 유용합니다.

requestContextKeys?:

string[]

이 Trace의 메타데이터로 추출할 추가 RequestContext 키입니다. 중첩 값에는 점 표기법을 지원합니다(예: 'user.id').

traceId?:

string

이 실행에 사용할 Trace ID입니다(1~32자의 16진수 문자). 제공하면 이 Trace는 지정된 Trace의 일부가 됩니다.

parentSpanId?:

string

이 실행에 사용할 상위 스팬 ID입니다(1~16자의 16진수 문자). 제공하면 루트 스팬이 이 스팬의 하위 스팬으로 생성됩니다.

tags?:

string[]

이 Trace에 적용할 태그입니다. Trace를 분류하고 필터링하기 위한 문자열 레이블입니다.

versions?:

VersionOverrides

하위 Agent 위임을 위한 호출별 버전 재정의입니다. Mastra 인스턴스 수준 버전 위에 병합되며 requestContext를 통해 하위 Agent 호출에 자동으로 전파됩니다. editor 패키지가 필요합니다. Editor 버전 관리를 참조하세요.

VersionOverrides

agents?:

Record<string, VersionSelector>

Agent ID와 해당 버전 선택기를 매핑한 객체입니다.

VersionSelector

versionId?:

string

ID로 특정 버전을 지정합니다.

status?:

'draft' | 'published'

이 게시 상태에 해당하는 최신 버전을 지정합니다.

untilIdle?:

boolean | { maxIdleMs?: number }

설정하면 백그라운드 작업이 이어지는 동안 스트림을 열린 상태로 유지합니다. 백그라운드 작업이 완료되면 Agent가 LLM을 자동으로 다시 호출하고 동일한 fullStream을 통해 후속 턴을 스트리밍합니다. 기본 설정(유휴 제한 시간 5분)을 사용하려면 true를 전달하고, 사용자 지정하려면 maxIdleMs가 포함된 객체를 전달하세요. Memory가 필요합니다. 독립형 streamUntilIdle() 메서드를 대체합니다.

maxIdleMs?:

number

턴 사이의 유휴 상태가 지정된 밀리초 동안 지속되면 외부 스트림을 닫습니다. 타이머는 래퍼가 턴 사이에 있을 때만 작동합니다. 기본값: 5분.

보고

stream:

MastraModelOutput<Output>

스트리밍 출력에 접근할 수 있는 MastraModelOutput 인스턴스를 반환합니다.

traceId?:

string

Tracing이 활성화된 경우 이 실행과 연결된 Trace ID입니다. 로그를 연계하고 실행 흐름을 디버깅할 때 사용합니다.

spanId?:

string

Tracing이 활성화된 경우 이 실행과 연결된 루트 스팬 ID입니다. 스팬 수준 조회 및 연계에 사용합니다.

확장된 사용 예

마스트라 형식(기본값)

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(이상)에서 스트림을 사용하려면 유틸리티 함수를 사용하여 변환할 수 있습니다.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 }),
  ],
})

응답 WebSocket 전송

공급자 옵션을 사용하여 응답 WebSocket 스트리밍을 선택합니다. 이는 스트리밍 호출에만 적용되며 직접 OpenAI Model 및 Azure OpenAI 응답 배포에 지원됩니다. 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를 사용하여 게이트웨이를 구성한 다음 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를 포함하는 중복 후속 요청을 거부합니다. 응답 체인의 다음 턴을 보내기 전에 활성 스트림이 완료될 때까지 기다리세요.

관련된

• 응답 생성
• 스트리밍 응답
• 대리인 승인