스트리밍

Mastra는 Agent 및 Workflow의 실시간 증분 응답을 지원하므로 사용자는 완료를 기다리지 않고 출력이 생성되는 대로 확인할 수 있습니다. 이는 채팅, 긴 형식의 콘텐츠, 다단계 Workflow 또는 즉각적인 피드백이 중요한 모든 시나리오에 유용합니다.

시작하기

Mastra의 스트리밍 API는 Model 버전에 따라 조정됩니다.

• .stream(): AI SDK v5 이상(LanguageModelV2)을 지원하는 V2 Model용입니다.
• .streamLegacy(): AI SDK v4(LanguageModelV1)를 지원하는 V1 Model용입니다.

Agent와 스트리밍

기본 Prompt에는 단일 문자열을 전달할 수 있고, 여러 컨텍스트를 제공하려면 문자열 배열을 전달할 수 있습니다. 역할과 대화 흐름을 정밀하게 제어하려면 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: 전문 응답으로 해결되는 약속입니다.
• stream.finishReason: Agent가 스트리밍을 중단한 이유입니다.
• stream.usage: 토큰 사용 정보입니다.

AI SDK v5+ 호환성

AI SDK v5 이상에서는 Model Provider에 LanguageModelV2를 사용합니다. AI SDK v4 Model을 사용 중이라는 오류가 발생하면 Model 패키지를 다음 메이저 버전으로 업그레이드해야 합니다. 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에서 스트리밍하면 증분 텍스트 청크 대신 실행 수명 주기를 설명하는 일련의 구조화된 이벤트가 반환됩니다. 이 이벤트 기반 형식을 사용하면 실행이 생성된 후 Workflow 진행 상황을 실시간으로 추적하고 응답할 수 있습니다..createRun().

사용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 실행의 총 토큰 사용량입니다.

Agent 또는 Workflow에서 스트리밍하면 LLM의 출력이나 Workflow 실행 상태에 대한 실시간 가시성이 제공됩니다. 이 피드백을 사용자에게 직접 전달하거나 애플리케이션에서 이를 사용하여 변경되는 Workflow 상태를 표시합니다.

Agent 또는 Workflow에서 발생하는 이벤트는 실행 시작, 텍스트 생성 또는 Tool 호출 등 다양한 생성 및 실행 단계를 나타냅니다.

이벤트 유형

다음은 .stream()에서 발생하는 이벤트의 전체 목록입니다. Agent 또는 Workflow 중 무엇을 스트리밍하는지에 따라 이러한 이벤트 중 일부만 발생합니다.

• 시작: Agent 또는 Workflow 실행의 시작을 표시합니다.
• 단계적으로 시작하다: Workflow 단계가 실행을 시작했음을 나타냅니다.
• 텍스트-델타: LLM에 의해 생성되는 증분 텍스트 청크입니다.
• Tool 호출: Agent가 Tool 이름과 인수를 포함하여 Tool을 사용하기로 결정한 경우입니다.
• Tool 결과: Tool 실행에서 반환된 결과입니다.
• 단계 마무리: 특정 단계가 완전히 완료되었는지 확인하고 해당 단계의 완료 이유와 같은 메타데이터를 포함할 수 있습니다.
• 마치다: 사용 통계를 포함하여 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'
    // ..
  }
}

작가 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

context.writer 객체는 Tool의 execute() 함수에서 사용할 수 있으며 활성 스트림에 맞춤 이벤트, 데이터 또는 값을 내보낼 수 있습니다. 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 argument

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