프로세서 인터페이스

그만큼Processor인터페이스는 Mastra의 모든 프로세서에 대한 계약을 정의합니다. 프로세서는 Agent 실행 파이프라인의 다양한 단계를 처리하기 위해 하나 이상의 메서드를 구현할 수 있습니다.

프로세서 메서드가 실행될 때

프로세서 메서드는 Agent 실행 수명 주기의 다양한 지점에서 실행됩니다.

┌────────────────────────────────────────────────────────────────────┐
│                        Agent Execution Flow                        │
├────────────────────────────────────────────────────────────────────┤
│                                                                    │
│  User Input                                                        │
│      │                                                             │
│      ▼                                                             │
│  ┌────────────────────────┐                                        │
│  │  processInput          │  ← Runs ONCE at start                  │
│  └───────────┬────────────┘                                        │
│              │                                                     │
│              ▼                                                     │
│  ┌──────────────────────────────────────────────────────────────┐  │
│  │                        Agentic Loop                          │  │
│  │                                                              │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processInputStep      │  ← Runs at EACH step             │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processLLMRequest     │  ← Before provider call          │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │        LLM Execution ──── API Error? ───┐                    │  │
│  │              │                          │                    │  │
│  │              │              ┌───────────┴──────────┐         │  │
│  │              │              │  processAPIError     │         │  │
│  │              │              └──────────────────────┘         │  │
│  │              │                (retry loops back to LLM)      │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processOutputStream   │  ← Runs on EACH stream chunk     │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processLLMResponse    │  ← After stream completes        │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processOutputStep     │  ← Runs after EACH LLM step      │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │        Tool Execution (if needed)                            │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processToolResult     │  ← Runs per tool, after each     │  │
│  │  └───────────┬────────────┘    tool.execute() returns        │  │
│  │              │                                               │  │
│  │              └──────── Loop back if tools called ────────────│  │
│  │                                                              │  │
│  └──────────────────────────────────────────────────────────────┘  │
│              │                                                     │
│              ▼                                                     │
│  ┌────────────────────────┐                                        │
│  │  processOutputResult   │  ← Runs ONCE after completion          │
│  └────────────────────────┘                                        │
│              │                                                     │
│              ▼                                                     │
│        Final Response                                              │
│                                                                    │
└────────────────────────────────────────────────────────────────────┘

방법	실행 시점	사용 사례
processInput	Agent 루프 시작 시 한 번	초기 사용자 입력 검증/변환, 컨텍스트 추가
processInputStep	Agent 루프의 각 단계에서 각 LLM 호출 전	단계 사이의 메시지 변환, Tool 결과 처리
processLLMRequest	LLM 요청 변환 후 Provider 호출 전	변경 사항을 유지하지 않고 현재 호출의 아웃바운드 LanguageModelV2Prompt 다시 작성
processAPIError	LLM API 호출 실패 시	API 거부 검사, 선택적으로 상태/메시지 변경 및 재시도 요청
processOutputStream	LLM 응답 중 각 스트리밍 청크에서	스트리밍 콘텐츠 필터링/수정, 실시간 패턴 감지
processLLMResponse	LLM 단계가 완료되고 스트림 청크가 수집된 후	전체 응답 캡처 또는 캐시, processLLMRequest 호출 후 부수 효과 실행
processOutputStep	각 LLM 응답 후, Tool 실행 전	출력 품질 검증, 재시도를 사용하는 가드레일 구현
processToolResult	각 Tool의 tool.execute() 이후, 결과가 메시지 목록에 추가되기 전	Prompt 삽입 탐지를 위한 Tool 출력 검사, 민감한 필드 수정, 정책 위반 시 중단
processOutputResult	생성 완료 후 한 번	최종 응답 후처리, 결과 로깅

인터페이스 정의

interface Processor<TId extends string = string, TTripwireMetadata = unknown> {
  readonly id: TId
  readonly name?: string
  readonly description?: string
  /** Index of this processor in the workflow (set at runtime when combining processors). */
  processorIndex?: number
  /** When true, processOutputStream also receives `data-*` chunks. Default: false. */
  processDataParts?: boolean

  /** Callback invoked when this processor detects a violation, regardless of strategy. */
  onViolation?: (violation: ProcessorViolation) => void | Promise<void>

  processInput?(
    args: ProcessInputArgs<TTripwireMetadata>,
  ): Promise<ProcessInputResult> | ProcessInputResult

  processInputStep?(
    args: ProcessInputStepArgs<TTripwireMetadata>,
  ):
    | Promise<ProcessInputStepResult | MessageList | MastraDBMessage[] | undefined | void>
    | ProcessInputStepResult
    | MessageList
    | MastraDBMessage[]
    | void
    | undefined

  processLLMRequest?(
    args: ProcessLLMRequestArgs<TTripwireMetadata>,
  ): Promise<ProcessLLMRequestResult> | ProcessLLMRequestResult

  processLLMResponse?(
    args: ProcessLLMResponseArgs<TTripwireMetadata>,
  ): Promise<ProcessLLMResponseResult> | ProcessLLMResponseResult

  processAPIError?(
    args: ProcessAPIErrorArgs<TTripwireMetadata>,
  ): Promise<ProcessAPIErrorResult | void> | ProcessAPIErrorResult | void

  processOutputStream?(
    args: ProcessOutputStreamArgs<TTripwireMetadata>,
  ): Promise<ChunkType | null | undefined>

  processOutputStep?(args: ProcessOutputStepArgs<TTripwireMetadata>): ProcessorMessageResult

  processToolResult?(args: ProcessToolResultArgs<TTripwireMetadata>): ProcessorMessageResult

  processOutputResult?(args: ProcessOutputResultArgs<TTripwireMetadata>): ProcessorMessageResult
}

속성

id:

string

프로세서의 고유 식별자입니다. 추적 및 디버깅에 사용됩니다.

name?:

string

프로세서의 선택적 표시 이름입니다. 제공하지 않으면 id를 사용합니다.

description?:

string

추적 및 Studio에 표시되는 선택적 설명입니다.

processorIndex?:

number

결합된 프로세서 목록에서 이 프로세서의 위치입니다. 프로세서가 Memory, Workspace 및 호출별 재정의와 병합될 때 Mastra가 런타임에 설정합니다. 직접 설정하지 마세요.

processDataParts?:

boolean

true이면 processOutputStream 메서드가 Tool이 writer.custom()을 통해 내보낸 data-* 청크도 수신합니다. 기본값은 false입니다.

onViolation?:

(violation: ProcessorViolation) => void | Promise<void>

전략(block 또는 warn)과 관계없이 프로세서가 정책 위반을 감지할 때 호출되는 선택적 콜백입니다. 알림 전송, 외부 시스템 로깅 또는 사용자 이메일 발송과 같은 부수 효과에 사용합니다. 프로세서 로직을 방해하지 않도록 이 콜백에서 발생한 오류는 조용히 포착됩니다. 위반 객체에는 processorId, message 및 프로세서별 detail 필드가 포함됩니다.

메시지 인수

대부분의 프로세서 메서드는 messages와 messageList를 모두 받습니다. 두 값은 동일한 기본 대화를 가리키지만 서로 다른 방식으로 노출합니다.

messages대messageList

• messages: 현재 단계로 범위가 지정된 일반 MastraDBMessage 객체 배열입니다. processInput과 processInputStep에서는 시스템 메시지를 제외합니다. processOutputResult와 processOutputStep에서는 최신 LLM 응답을 포함합니다. 이 배열은 messageList를 기반으로 하므로 메시지의 content.parts를 제자리에서 편집하면 후속 프로세서와 영속화에도 반영됩니다.
• messageList: 실행을 뒷받침하는 라이브 MessageList 인스턴스입니다. 필터링된 뷰(input, response, remembered, all), 여러 출력 형식(db, ui, core), 대화를 변경하는 메서드를 제공합니다. 현재 단계의 메시지를 읽거나 순회하거나 필드를 가볍게 편집하기만 한다면 messages를 사용하세요. 다음 작업이 필요하면 messageList를 사용하세요.
• 출력을 처리하는 동안 입력 메시지 등 다른 단계의 메시지를 읽습니다.
• 전체 메시지를 추가, 제거 또는 교체합니다.
• 타사 API용 UI 또는 핵심 메시지와 같은 다른 형식으로 변환합니다.

messages는 항상 messageList에서 파생되므로 메시지를 추가, 제거 또는 재정렬하는 표준 방법은 messageList를 변경하는 것입니다. 메시지 콘텐츠를 제자리에서 편집하는 경우(예: content.parts 다시 작성)에는 messages를 직접 변경해도 동일합니다. messages에서 새 배열을 반환하면 Mastra가 현재 단계의 messageList와 조정합니다.

고집

Memory가 활성화된 경우 모든 프로세서가 완료된 후의 최종 messageList만 스토리지에 영속화됩니다. 두 반환 방식의 영속화 결과는 동일합니다.

• messageList를 직접 변경하거나 동일한 MessageList 인스턴스를 반환하면 기록된 변경 사항이 제자리에 적용되므로 저장된 대화에 변경 사항이 반영됩니다.
• MastraDBMessage[] 또는 { messages, systemMessages }를 반환하면 Mastra가 반환된 배열을 현재 단계의 messageList와 조정하여 누락된 메시지를 제거하고 시스템 메시지를 교체합니다. 다른 MessageList 인스턴스를 반환하면 오류가 발생합니다. 항상 프로세서에 전달된 인스턴스를 변경하세요.

메시지에서 텍스트 읽기

MastraDBMessage.content구조화된 객체를 사용합니다. 문자열은 지원되지 않습니다. 사용자 또는 보조자 텍스트를 읽는 정식 방법은 다음과 같습니다.content.parts:

import type { MastraDBMessage } from '@mastra/core/memory'

function getText(message: MastraDBMessage): string {
  let text = ''

  if (message.content.parts) {
    for (const part of message.content.parts) {
      if (part.type === 'text' && typeof part.text === 'string') {
        text += part.text
      }
    }
  }

  // Fallback for legacy messages that only have the flattened `content` string
  if (!text && typeof message.content.content === 'string') {
    text = message.content.content
  }

  return text
}

핵심 사항:

• message.content.parts가 기본 데이터 원본입니다. 하나의 메시지는 Tool 호출, Tool 결과, 파일 부분 등 텍스트가 아닌 부분을 비롯해 여러 부분을 포함할 수 있습니다. part.text를 읽기 전에 part.type === 'text'를 기준으로 필터링하세요.
• message.content.content는 이전 버전과의 호환성을 위해 유지되는 병합 문자열입니다. parts가 비어 있거나 없는 경우에만 대체 수단으로 사용하세요.
• MastraDBMessage에서 message.content 자체는 일반 문자열이 아닙니다. 레거시 CoreMessage 형태에서는 문자열일 수 있지만, 프로세서는 항상 MastraDBMessage를 받습니다.

행동 양식

processInput

LLM으로 전송되기 전에 입력 메시지를 처리합니다. Agent 실행 시작 시 한 번 실행됩니다.

processInput?(args: ProcessInputArgs): Promise<ProcessInputResult> | ProcessInputResult;

ProcessInputArgs

messages:

MastraDBMessage[]

처리할 사용자 및 assistant 메시지입니다(시스템 메시지 제외).

systemMessages:

CoreMessage[]

모든 시스템 메시지입니다(Agent 지침, Memory 컨텍스트, 사용자가 제공한 메시지). 수정하여 반환할 수 있습니다.

messageList:

MessageList

고급 메시지 관리를 위한 전체 MessageList 인스턴스입니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다. 실행을 중지하는 TripWire 오류를 발생시킵니다. LLM이 피드백을 반영해 단계를 재시도하도록 요청하려면 retry: true를 전달하세요.

retryCount:

number

이 생성에서 프로세서가 재시도를 트리거한 횟수입니다. 재시도 횟수를 제한할 때 사용합니다. Mastra가 항상 전달하며 0에서 시작합니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

requestContext?:

RequestContext

threadId 및 resourceId 같은 실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

ProcessInputResult

이 메서드는 세 가지 유형 중 하나를 반환할 수 있습니다.

MastraDBMessage[]:

array

변환된 메시지 배열입니다. 시스템 메시지는 변경되지 않습니다.

MessageList:

MessageList

전달받은 것과 동일한 messageList 인스턴스입니다. 해당 인스턴스를 직접 변경했음을 나타냅니다.

{ messages, systemMessages }:

object

변환된 메시지와 수정된 시스템 메시지를 모두 포함하는 객체입니다.

---

processInputStep

LLM으로 전송하기 전에 Agent 루프의 각 단계에서 입력 메시지를 처리합니다. 시작할 때 한 번 실행되는 processInput과 달리 Tool 호출 후속 단계를 포함한 모든 단계에서 실행됩니다.

processInputStep?<TTripwireMetadata = unknown>(
  args: ProcessInputStepArgs<TTripwireMetadata>,
):
  | Promise<ProcessInputStepResult | MessageList | MastraDBMessage[] | void | undefined>
  | ProcessInputStepResult
  | MessageList
  | MastraDBMessage[]
  | void
  | undefined;

Agent 루프의 실행 순서

1. processInput(시작할 때 한 번)
2. processInputStepinputProcessors에서(각 단계에서, LLM 호출 전)
3. prepareStep콜백(inputProcessors 다음에 processInputStep 파이프라인의 일부로 실행됨)
4. processLLMRequestinputProcessors에서(Prompt 변환 후, 공급자 호출 전)
5. LLM 실행
6. processOutputStreamOutputProcessors에서(각 스트리밍 청크에서)
7. processLLMResponseinputProcessors에서(스트림이 완료된 후 다음과 쌍을 이룹니다.)processLLMRequest)
8. processOutputStepOutputProcessors에서(LLM 응답 후, Tool 실행 전)
9. Tool 실행(필요한 경우)
10. Tool이 호출된 경우 2단계부터 반복하세요.

ProcessInputStepArgs

messages:

MastraDBMessage[]

이전 단계의 Tool 호출과 결과를 포함한 모든 메시지입니다(읽기 전용 스냅샷).

messageList:

MessageList

메시지를 관리하는 MessageList 인스턴스입니다. 직접 변경하거나 결과로 반환할 수 있습니다.

stepNumber:

number

현재 단계 번호입니다(0부터 시작). 0단계는 최초 LLM 호출입니다.

steps:

StepResult[]

text, toolCalls 및 toolResults를 포함한 이전 단계의 결과입니다.

systemMessages:

CoreMessage[]

모든 시스템 메시지입니다(읽기 전용 스냅샷). 교체하려면 결과로 반환하세요.

model:

MastraLanguageModelV2

현재 사용 중인 Model입니다. 전환하려면 결과에 다른 Model을 반환하세요.

toolChoice?:

ToolChoice

현재 Tool 선택 설정입니다('auto', 'none', 'required' 또는 특정 Tool).

activeTools?:

string[]

현재 활성화된 Tool 이름입니다. Tool을 제한하려면 필터링된 배열을 반환하세요.

tools?:

ToolSet

이 단계에서 현재 사용할 수 있는 Tool입니다. Tool을 추가하거나 교체하려면 결과로 반환하세요.

providerOptions?:

SharedV2ProviderOptions

Provider별 옵션입니다(예: Anthropic cacheControl, OpenAI reasoningEffort).

modelSettings?:

CallSettings

temperature, maxTokens, topP 같은 Model 설정입니다.

structuredOutput?:

StructuredOutputOptions

구조화된 출력 구성입니다(schema, 출력 모드). 수정하려면 결과로 반환하세요.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다. 실행을 중지하는 TripWire 오류를 발생시킵니다. LLM이 피드백을 반영해 단계를 재시도하도록 요청하려면 retry: true를 전달하세요.

retryCount:

number

ProcessorContext의 현재 재시도 횟수입니다. 0에서 시작하며, 프로세서가 트리거하는 재시도 횟수를 제한하는 데 사용합니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

ProcessInputStepResult

processInputStep여러 모양을 반환할 수 있습니다.

• ProcessInputStepResult 객체: 이 단계에서 아래 속성의 조합을 재정의합니다(다음에 설명).
• MessageList: 메시지를 제자리에서 변경했음을 나타내려면 동일한 messageList 인스턴스를 반환합니다.
• MastraDBMessage[]: 변환된 메시지 배열을 반환합니다. 해당 단계의 메시지를 교체합니다.
• void 또는 undefined: 단계를 변경하지 않으려면 아무것도 반환하지 않습니다. 개체 양식은 다음 속성의 모든 조합을 반환할 수 있습니다.

model?:

LanguageModelV2 | string

이 단계의 Model을 변경합니다. Model 인스턴스 또는 'openai/gpt-5.5' 같은 라우터 ID를 사용할 수 있습니다.

toolChoice?:

ToolChoice

이 단계의 Tool 선택 동작을 변경합니다.

activeTools?:

string[]

이 단계에서 사용할 수 있는 Tool을 필터링합니다.

tools?:

ToolSet

이 단계의 Tool을 교체하거나 수정합니다. 병합하려면 스프레드를 사용하세요: { tools: { ...tools, newTool } }.

messages?:

MastraDBMessage[]

모든 메시지를 교체합니다. messageList와 함께 사용할 수 없습니다.

messageList?:

MessageList

동일한 messageList 인스턴스를 반환합니다(해당 인스턴스를 변경했음을 나타냄). messages와 함께 사용할 수 없습니다.

systemMessages?:

CoreMessage[]

이 단계에 한해 모든 시스템 메시지를 교체합니다.

providerOptions?:

SharedV2ProviderOptions

이 단계의 Provider별 옵션을 변경합니다.

modelSettings?:

CallSettings

이 단계의 Model 설정을 변경합니다.

structuredOutput?:

StructuredOutputOptions

이 단계의 구조화된 출력 구성을 변경합니다.

프로세서 체인

여러 프로세서가 processInputStep을 구현하면 순서대로 실행되며 변경 사항이 연쇄적으로 전달됩니다.

Processor 1: receives { model: 'gpt-5.4' } → returns { model: 'gpt-5.4-mini' }
Processor 2: receives { model: 'gpt-5.4-mini' } → returns { toolChoice: 'none' }
Final: model = 'gpt-5.4-mini', toolChoice = 'none'

시스템 메시지 격리

시스템 메시지는 각 단계가 시작될 때 원래 값으로 재설정됩니다. processInputStep에서 수정한 내용은 현재 단계에만 영향을 주며 이후 단계에는 영향을 주지 않습니다.

사용 사례

• 단계 번호 또는 상황에 따른 동적 Model 전환
• 특정 단계 이후 Tool 비활성화
• 대화 컨텍스트에 따라 Tool을 동적으로 추가하거나 교체
• Provider 간 메시지 부분 유형 변환(예: Anthropic의 reasoning → thinking)
• 단계 번호 또는 누적된 컨텍스트를 기반으로 메시지 수정
• 단계별 시스템 지침 추가
• 단계별 Provider 옵션 조정(예: 캐시 제어)
• 단계 컨텍스트를 기반으로 구조화된 출력 schema 수정

---

processLLMRequest

Mastra가 MessageList를 LanguageModelV2Prompt로 변환한 후, Provider를 호출하기 전에 최종 LLM 요청을 처리합니다. 현재 아웃바운드 요청에만 영향을 주는 일시적인 Model 인식 재작성에 이 메서드를 사용하세요. 반환된 Prompt 변경 사항은 현재 호출에 한해 Model에 전달됩니다. MessageList, Memory, UI 기록 또는 이후 Provider 호출에는 다시 영속화되지 않습니다.

processLLMRequest?(
  args: ProcessLLMRequestArgs,
): Promise<ProcessLLMRequestResult> | ProcessLLMRequestResult;

ProcessLLMRequestArgs

prompt:

LanguageModelV2Prompt

이 호출에서 Provider로 전송할 LLM 요청 Prompt입니다.

model:

MastraLanguageModel

Prompt를 받을 결정된 Model입니다. Provider별 재작성 범위를 지정하는 데 사용하세요.

stepNumber:

number

현재 단계 번호입니다(0부터 시작). 0단계는 최초 LLM 호출입니다.

steps:

StepResult[]

text, toolCalls 및 toolResults를 포함한 이전 단계의 결과입니다.

state:

Record<string, unknown>

이 요청 내 모든 메서드 호출에서 유지되는 프로세서별 상태입니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다. 실행을 중지하고 tripwire 청크를 내보내는 TripWire 오류를 발생시킵니다.

retryCount:

number

ProcessorContext의 현재 재시도 횟수입니다. 0에서 시작하며, 프로세서가 트리거하는 재시도 횟수를 제한하는 데 사용합니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

writer?:

ProcessorStreamWriter

스트리밍 중 사용자 정의 데이터 청크를 내보내는 스트림 writer입니다. data-* 청크를 내보내려면 writer.custom()을 호출하세요.

abortSignal?:

AbortSignal

작업 취소를 위한 신호입니다.

반환 값

processLLMRequest는 { prompt?: LanguageModelV2Prompt } | undefined | void 형식인 ProcessLLMRequestResult를 반환합니다.

• 현재 Provider 호출의 아웃바운드 Prompt를 교체하려면 { prompt }를 반환합니다.
• 원래 Prompt를 변경하지 않고 전달하려면 undefined 또는 void를 반환합니다.

사용 사례

• Model 호출 전에 제공자별 Prompt 부분 제거 또는 재구성
• 공급자의 입력 요구 사항에 맞게 역할 또는 콘텐츠 정규화
• 루프 중간에 공급자를 전환할 때 Tool 결과 형식 적용

---

processLLMResponse

단계가 완료된 후(또는 캐시된 응답이 재생된 후), 출력 프로세서가 응답 청크를 수집하고 나면 LLM 응답을 처리합니다. 이 후크는 processLLMRequest와 쌍을 이룹니다. Provider 호출 전에 processLLMRequest를 사용해 캐시 키 같은 상태를 저장하고, 완료된 응답에 대한 작업(예: 캐시에 쓰기)은 processLLMResponse에서 수행하세요. state 객체는 같은 단계에서 processLLMRequest에 전달된 것과 동일한 인스턴스이므로 프로세서가 호출 전후 작업을 연계할 수 있습니다.

processLLMResponse?(
  args: ProcessLLMResponseArgs,
): Promise<ProcessLLMResponseResult> | ProcessLLMResponseResult;

ProcessLLMResponseArgs

chunks:

CachedLLMStepChunk[]

이 단계의 LLM 호출에서 생성되거나 캐시에서 재생된 청크이며, 축약된 형식({ type, payload })입니다.

model:

MastraLanguageModel

응답을 생성했거나 생성했을 Model입니다.

stepNumber:

number

현재 단계 번호입니다(0부터 시작).

steps:

StepResult[]

현재 단계를 포함하여 지금까지 완료된 모든 단계입니다.

state:

Record<string, unknown>

같은 단계의 processLLMRequest와 공유되는 프로세서별 상태입니다. 두 후크 간에 데이터(예: 캐시 키)를 전달할 때 사용합니다.

fromCache:

boolean

true이면 processLLMRequest가 { response }를 반환하여 응답이 캐시에서 재생된 것입니다. 캐시에 쓰는 프로세서는 이 값이 true일 때 쓰기를 건너뛰어야 합니다.

warnings?:

LanguageModelV2CallWarning[]

언어 Model 호출에서 보고한 경고입니다(예: 지원되지 않는 설정).

request?:

unknown

사용 가능한 경우의 Provider 요청 본문입니다. 추적에 유용합니다.

rawResponse?:

unknown

사용 가능한 경우의 원시 Provider 응답입니다. 추적에 유용합니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다. 실행을 중지하는 TripWire 오류를 발생시킵니다.

retryCount:

number

현재 재시도 횟수입니다. 0에서 시작하며, 프로세서가 트리거하는 재시도 횟수를 제한하는 데 사용합니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

writer?:

ProcessorStreamWriter

사용자 정의 데이터 청크를 내보내는 스트림 writer입니다.

abortSignal?:

AbortSignal

작업 취소를 위한 신호입니다.

반환 값

processLLMResponse는 undefined | void 형식인 ProcessLLMResponseResult를 반환합니다. 반환 값은 향후 확장성을 위해 예약되어 있습니다.

사용 사례

• 실시간 호출 후 캐시에 LLM 응답 쓰기(캐시 키 파생과 쌍을 이룸)processLLMRequest)
• 분석을 위한 전체 응답 로깅 또는 기록
• 완료된 응답을 기반으로 부작용 유발

---

processAPIError

LLM API 거부 오류가 최종 오류로 표시되기 전에 처리합니다. 재시도할 수 없는 오류(예: 400 또는 422 상태 코드)로 API 호출이 실패할 때 실행됩니다. 응답이 성공한 후 실행되는 processOutputStep과 달리 API가 요청을 거부할 때 실행됩니다. processAPIError를 구현하는 프로세서를 Agent의 errorProcessors 배열에 추가하세요. 프로세서는 오류를 검사하고 요청을 수정할 수 있습니다. 예를 들어 messageList에서 메시지를 제거할 수 있습니다. 수정된 상태로 재시도하려면 { retry: true }를 반환하세요.

processAPIError?(args: ProcessAPIErrorArgs): Promise<ProcessAPIErrorResult | void> | ProcessAPIErrorResult | void;

ProcessAPIErrorArgs

error:

unknown

LLM API 호출 중 발생한 오류입니다.

messages:

MastraDBMessage[]

오류 발생 시점의 모든 메시지입니다.

messageList:

MessageList

메시지를 관리하는 MessageList 인스턴스입니다. 재시도 전에 요청을 변경하려면 이 인스턴스를 수정하세요.

stepNumber:

number

현재 단계 번호입니다(0부터 시작).

steps:

StepResult[]

지금까지 완료된 모든 단계입니다.

state:

Record<string, unknown>

이 요청 내 모든 메서드 호출에서 유지되는 프로세서별 상태입니다.

retryCount:

number

오류 처리기의 현재 재시도 횟수입니다. 재시도 횟수를 제한하는 데 사용하세요.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다.

writer?:

ProcessorStreamWriter

스트리밍 중 사용자 정의 데이터 청크를 내보내는 스트림 writer입니다. data-* 청크를 내보내려면 writer.custom()을 호출하세요.

requestContext?:

RequestContext

Agent 호출에서 전달된 요청 컨텍스트입니다.

abortSignal?:

AbortSignal

작업 취소를 위한 신호입니다.

ProcessAPIErrorResult

retry:

boolean

수정 사항을 적용한 후 LLM 호출을 재시도할지 여부입니다.

사용 사례

• 요청을 수정하고 재시도하여 API 관련 거부 처리
• 요청 수정을 통해 재시도할 수 없는 오류를 재시도 가능한 오류로 변환
• Model별 오류 복구 전략 구현

예: 사용자 정의 오류 복구

src/mastra/processors/error-recovery.ts

import { APICallError } from '@ai-sdk/provider'
import type { Processor, ProcessAPIErrorArgs, ProcessAPIErrorResult } from '@mastra/core/processors'

export class ErrorRecoveryProcessor implements Processor {
  id = 'error-recovery'

  processAPIError({
    error,
    messageList,
    retryCount,
  }: ProcessAPIErrorArgs): ProcessAPIErrorResult | void {
    // Only retry once
    if (retryCount > 0) return

    // Check for a specific API error
    if (APICallError.isInstance(error) && error.message.includes('context length exceeded')) {
      // Trim older messages to fit within context
      const messages = messageList.get.all.db()
      if (messages.length > 4) {
        messageList.removeByIds([messages[1]!.id, messages[2]!.id])
        return { retry: true }
      }
    }
  }
}

---

processOutputStream

내장된 상태 관리를 통해 스트리밍 출력 청크를 처리합니다. 프로세서가 청크를 축적하고 더 큰 컨텍스트를 기반으로 결정을 내릴 수 있습니다.

processOutputStream?(args: ProcessOutputStreamArgs): Promise<ChunkType | null | undefined>;

ProcessOutputStreamArgs

part:

ChunkType

현재 처리 중인 스트림 청크입니다.

streamParts:

ChunkType[]

지금까지 스트림에서 확인된 모든 청크입니다.

state:

Record<string, unknown>

단일 요청 내 모든 청크와 모든 메서드 호출에서 유지되는 변경 가능한 프로세서별 상태입니다. 새 generate 또는 stream 호출마다 새로운 상태 객체가 생성됩니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

스트림을 중단하는 함수입니다. 스트림을 종료하고 tripwire 청크를 내보내는 TripWire 오류를 발생시킵니다. 종료하는 대신 LLM 재시도를 요청하려면 retry: true를 전달하세요.

retryCount:

number

ProcessorContext의 현재 재시도 횟수입니다. 0에서 시작하며, 프로세서가 트리거하는 재시도 횟수를 제한하는 데 사용합니다.

messageList?:

MessageList

대화 기록에 액세스하기 위한 MessageList 인스턴스입니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

writer?:

ProcessorStreamWriter

사용자 정의 데이터 청크를 클라이언트로 내보내는 스트림 writer입니다. data-* 유형 청크를 내보내려면 writer.custom()을 호출하세요. 스트리밍 중에 사용할 수 있습니다.

반환 값

processOutputStream보고Promise<ChunkType | null | undefined>.

• 청크를 내보내려면 ChunkType을 반환합니다. 변경 없이 내보내려면 원래 part를 반환하고, 수정된 청크를 내보내려면 새 ChunkType을 반환합니다.
• 청크를 삭제하려면 null을 반환합니다. 다음 프로세서나 클라이언트에는 아무것도 전송되지 않습니다.
• 청크를 삭제하려면 undefined를 반환합니다(return; 문에서 암시적으로 반환되거나 메서드가 끝까지 실행되어 반환되는 undefined 포함). null과 undefined는 동일하게 동작합니다. 청크를 삭제하면 해당 단일 청크에만 영향을 미칩니다. 스트림이 계속되고 다음 청크가 계속 처리됩니다. 스트림을 완전히 중지하려면 다음을 호출하세요.abort().

---

processOutputResult

스트리밍이나 생성이 완료된 후 전체 출력 결과를 처리합니다.

processOutputResult?(args: ProcessOutputResultArgs): ProcessorMessageResult;

ProcessOutputResultArgs

messages:

MastraDBMessage[]

생성된 응답 메시지입니다.

messageList:

MessageList

메시지를 관리하는 MessageList 인스턴스입니다.

state:

Record<string, unknown>

이 요청 내 모든 메서드 호출에서 유지되는 프로세서별 상태입니다. processOutputStream 및 기타 메서드와 공유됩니다.

result:

OutputResult

text(누적된 텍스트), usage(inputTokens, outputTokens, totalTokens를 포함한 토큰 사용량), finishReason(생성이 종료된 이유), steps(각각 toolCalls, toolResults, reasoning, sources, files 등을 포함하는 모든 LLM 단계 결과)를 포함하는 결정된 생성 결과입니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다. 실행을 중지하고 tripwire 청크를 내보내는 TripWire 오류를 발생시킵니다.

retryCount:

number

ProcessorContext의 현재 재시도 횟수입니다. 0에서 시작하며, 프로세서가 트리거하는 재시도 횟수를 제한하는 데 사용합니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

writer?:

ProcessorStreamWriter

사용자 정의 데이터 청크를 클라이언트로 내보내는 스트림 writer입니다. data-* 유형 청크를 내보내려면 writer.custom()을 호출하세요. 스트리밍 중에 사용할 수 있습니다.

---

processOutputStep

Tool 실행 전에 Agent 루프의 각 LLM 응답 후 출력을 처리합니다. 마지막에 한 번 실행되는 processOutputResult와 달리 모든 단계에서 실행됩니다. 재시도를 트리거할 수 있는 가드레일을 구현하기에 가장 적합한 메서드입니다.

processOutputStep?(args: ProcessOutputStepArgs): ProcessorMessageResult;

ProcessOutputStepArgs

messages:

MastraDBMessage[]

최신 LLM 응답을 포함한 모든 메시지입니다.

messageList:

MessageList

메시지를 관리하는 MessageList 인스턴스입니다.

stepNumber:

number

현재 단계 번호입니다(0부터 시작).

finishReason?:

string

LLM의 종료 이유입니다(stop, tool-use, length 등).

providerMetadata?:

ProviderMetadata

종료 단계의 Provider별 메타데이터입니다(예: AWS Bedrock 가드레일 Trace). Model 단계에서 Provider 메타데이터가 생성된 경우 표시되며, steps가 비어 있는 콘텐츠 필터 차단의 경우도 포함됩니다.

toolCalls?:

ToolCallInfo[]

이 단계에서 실행한 Tool 호출입니다(있는 경우).

text?:

string

이 단계에서 생성된 텍스트입니다.

usage:

LanguageModelUsage

현재 단계의 토큰 사용량입니다(inputTokens, outputTokens, totalTokens).

systemMessages:

CoreMessage[]

읽기/수정할 수 있는 모든 시스템 메시지입니다.

steps:

StepResult[]

현재 단계를 포함하여 지금까지 완료된 모든 단계입니다.

state:

Record<string, unknown>

이 요청 내 모든 메서드 호출에서 유지되는 프로세서별 상태입니다. processOutputStream 및 processOutputResult와 공유됩니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

처리를 중단하는 함수입니다. LLM이 단계를 재시도하도록 요청하려면 retry: true를 전달하세요.

retryCount:

number

프로세서가 재시도를 트리거한 횟수입니다. 재시도 횟수를 제한할 때 사용합니다. Mastra가 항상 전달하며 0에서 시작합니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

사용 사례

• 재시도를 요청할 수 있는 품질 가드레일 구현
• Tool 실행 전 LLM 출력 검증
• 단계별 로깅 또는 측정항목 추가
• 재시도 기능으로 출력 조정 구현

예: 재시도가 포함된 품질 가드레일

src/mastra/processors/quality-guardrail.ts

import type { Processor } from '@mastra/core/processors'

export class QualityGuardrail implements Processor {
  id = 'quality-guardrail'

  async processOutputStep({ text, abort, retryCount }) {
    const score = await evaluateResponseQuality(text)

    if (score < 0.7) {
      if (retryCount < 3) {
        // Request retry with feedback for the LLM
        abort('Response quality too low. Please provide more detail.', {
          retry: true,
          metadata: { qualityScore: score },
        })
      } else {
        // Max retries reached, block the response
        abort('Response quality too low after multiple attempts.')
      }
    }

    return []
  }
}

processToolResult

tool.execute()가 반환된 후, 결과가 메시지 목록에 추가되거나 다음 LLM 호출에 전달되기 전에 Tool 결과를 처리합니다. Tool 실행 전에 발생하는 processOutputStep과 대칭을 이룹니다. 이 메서드를 사용하여 Tool 출력에서 Prompt 삽입을 검사하거나, 민감한 필드를 마스킹하거나, abort('reason', { retry: true })로 실행을 중단하세요. Tool 결과를 변경하려면 messageList.updateToolInvocation을 통해 messageList를 제자리에서 변경하세요. 런타임은 프로세서 처리 후의 결과를 메시지 목록에서 다시 읽고, 큐에 추가하기 전에 후속 Tool 결과 스트림 청크를 덮어쓰므로 스트리밍 클라이언트에 처리된 값이 표시됩니다. 이 메서드는 tool.execute()가 오류를 발생시키는 경우 실행되지 않으며, 결과를 사용할 수 있는 성공적인 Tool 실행에만 호출됩니다.

processToolResult?(args: ProcessToolResultArgs): ProcessorMessageResult;

ProcessToolResultArgs

messages:

MastraDBMessage[]

Tool 호출이 포함된 현재 assistant 메시지를 비롯한 모든 메시지입니다.

messageList:

MessageList

메시지를 관리하는 MessageList 인스턴스입니다. Tool 결과를 마스킹하거나 변환된 값으로 교체하려면 updateToolInvocation을 호출하세요.

stepNumber:

number

현재 단계 번호입니다(0부터 시작).

toolName:

string

실행된 Tool의 이름입니다.

toolCallId:

string

이 특정 Tool 호출의 고유 식별자입니다.

args:

unknown

LLM이 Tool에 전달한 인수입니다.

result:

unknown

Tool이 반환한 값입니다. 클라이언트에서 실행되는 Tool의 경우 ensureSerializable을 거친 tool.execute()의 출력입니다. Provider에서 실행되는 Tool(예: Anthropic web_search)의 경우 ensureSerializable을 거치지 않은 Provider 스트림의 원시 결과입니다.

providerExecuted?:

boolean

이 결과가 Anthropic web_search처럼 Provider에서 실행된 Tool에서 온 것인지 여부입니다. 클라이언트에서 실행되는 Tool의 기본값은 undefined입니다.

systemMessages:

CoreMessage[]

읽을 수 있는 모든 시스템 메시지입니다.

steps:

StepResult[]

지금까지 완료된 모든 단계입니다.

state:

Record<string, unknown>

이 요청 내 모든 메서드 호출에서 유지되는 프로세서별 상태입니다. 동일한 프로세서의 다른 프로세서 메서드와 공유됩니다.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

실행을 중단하는 함수입니다. 중단 이유를 피드백으로 제공하여 LLM이 단계를 재시도하도록 요청하려면 retry: true를 전달하세요.

retryCount:

number

프로세서가 재시도를 트리거한 횟수입니다. 0에서 시작합니다.

tracingContext?:

TracingContext

Observability를 위한 추적 컨텍스트입니다.

requestContext?:

RequestContext

실행 메타데이터가 포함된 요청 범위 컨텍스트입니다.

사용 사례

• LLM이 보기 전에 신속한 주입을 위한 스캐닝 Tool 출력입니다.
• Tool 반환에서 민감한 필드(PII, 비밀, 자격 증명)를 수정합니다.
• Tool이 정책을 위반하는 콘텐츠를 반환하면 실행을 중단합니다.
• 규정 준수 또는 감사를 위한 로깅 또는 계측 Tool 반환.

예: 민감한 필드 수정

src/mastra/processors/redact-tool-result.ts

import type { Processor } from '@mastra/core/processors'

export class RedactToolResult implements Processor {
  id = 'redact-tool-result'

  async processToolResult({ toolName, toolCallId, args, result, messageList }) {
    if (toolName !== 'lookup-customer') return

    const redacted = {
      ...(result as Record<string, unknown>),
      ssn: '[REDACTED]',
      email: '[REDACTED]',
    }

    messageList.updateToolInvocation({
      type: 'tool-invocation',
      toolInvocation: {
        state: 'result',
        toolCallId,
        toolName,
        args,
        result: redacted,
      },
    })
  }
}

예: Tool 출력에서 ​​Prompt 삽입 차단

src/mastra/processors/scan-tool-result.ts

import type { Processor } from '@mastra/core/processors'

export class ScanToolResult implements Processor {
  id = 'scan-tool-result'

  async processToolResult({ result, abort }) {
    const text = typeof result === 'string' ? result : JSON.stringify(result)
    if (containsPromptInjection(text)) {
      abort('blocked by scan-tool-result: suspected prompt injection')
    }
  }
}

function containsPromptInjection(text: string): boolean {
  return /ignore (all )?(previous|prior) instructions/i.test(text)
}

프로세서 유형

Mastra는 프로세서가 필요한 메소드를 구현하도록 유형 별칭을 제공합니다.

// Must implement processInput, processInputStep, processLLMRequest, or processLLMResponse (or any combination)
type InputProcessor = Processor &
  (
    | { processInput: required }
    | { processInputStep: required }
    | { processLLMRequest: required }
    | { processLLMResponse: required }
  )

// Must implement processOutputStream, processOutputStep, OR processOutputResult (or any combination)
type OutputProcessor = Processor &
  (
    | { processOutputStream: required }
    | { processOutputStep: required }
    | { processOutputResult: required }
  )

// Must implement processAPIError
type ErrorProcessor = Processor & { processAPIError: required }

processAPIError를 구현하는 프로세서는 errorProcessors에 구성하세요.

const agent = new Agent({
  id: 'agent',
  errorProcessors: [new PrefillErrorHandler()],
})

사용 예

기본 입력 프로세서

src/mastra/processors/lowercase.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class LowercaseProcessor implements Processor {
  id = 'lowercase'

  async processInput({ messages }): Promise<MastraDBMessage[]> {
    return messages.map(msg => ({
      ...msg,
      content: {
        ...msg.content,
        parts: msg.content.parts?.map(part =>
          part.type === 'text' ? { ...part, text: part.text.toLowerCase() } : part,
        ),
      },
    }))
  }
}

단계별 프로세서processInputStep

src/mastra/processors/dynamic-model.ts

import type {
  Processor,
  ProcessInputStepArgs,
  ProcessInputStepResult,
} from '@mastra/core/processors'

export class DynamicModelProcessor implements Processor {
  id = 'dynamic-model'

  async processInputStep({
    stepNumber,
    steps,
    toolChoice,
  }: ProcessInputStepArgs): Promise<ProcessInputStepResult> {
    // Use a fast model for initial response
    if (stepNumber === 0) {
      return { model: 'openai/gpt-5-mini' }
    }

    // Switch to powerful model after tool calls
    if (steps.length > 0 && steps[steps.length - 1].toolCalls?.length) {
      return { model: 'openai/gpt-5.6-sol' }
    }

    // Disable tools after 5 steps to force completion
    if (stepNumber > 5) {
      return { toolChoice: 'none' }
    }

    return {}
  }
}

메시지 변환기processInputStep

src/mastra/processors/reasoning-transformer.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class ReasoningTransformer implements Processor {
  id = 'reasoning-transformer'

  async processInputStep({ messages, messageList }) {
    // Transform reasoning parts to thinking parts at each step
    // This is useful when switching between model providers
    for (const msg of messages) {
      if (msg.role === 'assistant' && msg.content.parts) {
        for (const part of msg.content.parts) {
          if (part.type === 'reasoning') {
            ;(part as any).type = 'thinking'
          }
        }
      }
    }
    return messageList
  }
}

하이브리드 프로세서(입력 및 출력)

src/mastra/processors/content-filter.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'
import type { ChunkType } from '@mastra/core/stream'

export class ContentFilter implements Processor {
  id = 'content-filter'
  private blockedWords: string[]

  constructor(blockedWords: string[]) {
    this.blockedWords = blockedWords
  }

  async processInput({ messages, abort }): Promise<MastraDBMessage[]> {
    for (const msg of messages) {
      const text = msg.content.parts
        ?.filter(p => p.type === 'text')
        .map(p => p.text)
        .join(' ')

      if (this.blockedWords.some(word => text?.includes(word))) {
        abort('Blocked content detected in input')
      }
    }
    return messages
  }

  async processOutputStream({ part, abort }): Promise<ChunkType | null> {
    if (part.type === 'text-delta') {
      if (this.blockedWords.some(word => part.payload.text.includes(word))) {
        abort('Blocked content detected in output')
      }
    }
    return part
  }
}

상태가 있는 스트림 누산기

src/mastra/processors/word-counter.ts

import type { Processor } from '@mastra/core/processors'
import type { ChunkType } from '@mastra/core/stream'

export class WordCounter implements Processor {
  id = 'word-counter'

  async processOutputStream({ part, state }): Promise<ChunkType> {
    // Initialize state on first chunk
    if (!state.wordCount) {
      state.wordCount = 0
    }

    // Count words in text chunks
    if (part.type === 'text-delta') {
      const words = part.payload.text.split(/\s+/).filter(Boolean)
      state.wordCount += words.length
    }

    // Log word count on finish
    if (part.type === 'finish') {
      console.log(`Total words: ${state.wordCount}`)
    }

    return part
  }
}

상태 수명주기

모든 프로세서는 processLLMRequest, processLLMResponse, processOutputStream, processOutputStep, processOutputResult 및 processAPIError에서 state 객체를 받습니다. 상태에는 세 가지 중요한 속성이 있습니다.

• 프로세서별: 각 프로세서는 프로세서의 id를 키로 하는 자체 state 객체를 가집니다. id가 다른 프로세서는 서로의 상태를 읽거나 덮어쓸 수 없습니다.
• 요청별: 각 agent.generate() 또는 agent.stream() 호출이 시작될 때마다 새 상태 객체가 생성됩니다. 상태는 요청 간 또는 사용자 간에 유출되지 않습니다.
• 여러 메서드에서 공유됨: 하나의 요청 내에서 동일한 state 객체가 processLLMRequest(Provider 호출 전), processLLMResponse(단계 완료 후), processOutputStream(모든 청크), processOutputStep(모든 LLM 단계 후), processOutputResult(마지막에 한 번), processAPIError(LLM 호출 실패 시)에 전달됩니다. 예를 들어 processLLMRequest에서 캐시 키를 저장하고 processLLMResponse에서 다시 읽어 응답을 기록할 수 있습니다. state는 빈 객체로 시작하므로 처음 액세스할 때 필드를 방어적으로 초기화하세요.

import type { Processor } from '@mastra/core/processors'

export class WordCounter implements Processor {
  id = 'word-counter'

  async processOutputStream({ part, state }) {
    state.wordCount ??= 0
    if (part.type === 'text-delta') {
      state.wordCount += part.payload.text.split(/\s+/).filter(Boolean).length
    }
    return part
  }
}

중단 및 트립와이어 청크

각 메서드의 abort 함수는 처리를 중지하고 출력 스트림에 tripwire 청크를 내보내는 TripWire 오류를 발생시킵니다. 클라이언트는 이 청크를 감지하여 차단된 응답과 정상 종료를 구분할 수 있습니다.

abort('Blocked content detected', { retry: false, metadata: { category: 'pii' } })

• reason: 사람이 읽을 수 있는 설명입니다. tripwire.payload.reason에 표시됩니다.
• retry: true이면 Agent가 reason을 피드백으로 전달하여 같은 단계를 재시도합니다. Agent 또는 호출에 maxProcessorRetries가 설정된 경우에만 재시도가 실행되며, 그렇지 않으면 요청이 중단됩니다. errorProcessors가 구성된 경우 해당 호출의 maxProcessorRetries 기본값은 10입니다.
• metadata: 후속 소비자를 위해 tripwire 청크에 첨부되는 선택적 구조화 데이터입니다. 내보낸 tripwire 청크의 형태는 다음과 같습니다.

type TripwireChunk = {
  type: 'tripwire'
  runId: string
  from: 'AGENT'
  payload: {
    reason: string
    retry?: boolean
    metadata?: unknown
    processorId: string
  }
}

비스트리밍 호출(agent.generate())에서는 결과의 result.tripwire와 result.finishReason === 'other'를 통해 동일한 정보가 노출됩니다.

커스텀 데이터 청크 내보내기

writer에 액세스할 수 있는 프로세서는 writer.custom(chunk)을 호출하여 사용자 정의 data-* 청크를 클라이언트로 스트리밍할 수 있습니다. Tool도 자체 writer를 통해 같은 작업을 수행할 수 있습니다. 프로세서가 일반 텍스트 및 Tool 청크 외의 콘텐츠를 내보낼 수 있는 유일한 방법입니다.

await writer.custom({
  type: 'data-moderation',
  runId,
  from: 'AGENT',
  data: { level: 'warn', reason: 'Possibly unsafe' },
})

Memory가 구성된 경우 processOutputStream 또는 processOutputResult에서 내보낸 사용자 정의 data-* 청크가 assistant 메시지의 일부로 저장됩니다. Memory에 저장하지 않고 스트리밍하려면 청크 객체에 transient: true를 설정하세요.

await writer.custom({
  type: 'data-progress',
  data: { status: 'Processing' },
  transient: true,
})

청크의 속성으로 transient를 전달하세요. writer.custom()의 두 번째 인수로 전달하면 안 됩니다. 두 번째 인수에는 messageId와 같은 writer 옵션이 포함됩니다. 기본적으로 프로세서는 Tool 원격 측정 데이터나 자체 출력을 실수로 처리하지 않도록 processOutputStream에서 data-* 청크를 확인하지 않습니다. 프로세서에서 processDataParts: true를 설정하여 명시적으로 활성화하세요.

class ModerationCollector implements Processor {
  id = 'moderation-collector'
  processDataParts = true

  async processOutputStream({ part, state }) {
    if (part.type === 'data-moderation') {
      state.warnings ??= []
      state.warnings.push(part.data)
    }
    return part
  }
}

사용자 지정 데이터 청크로 처리하려면 청크의 type이 data-로 시작해야 합니다. processOutputStream에서 null 또는 undefined를 반환하면 여전히 청크가 삭제되므로, 프로세서는 텍스트 청크를 필터링하는 것과 같은 방식으로 사용자 지정 데이터를 검사하거나 수정하거나 필터링할 수 있습니다.

Agent에서 프로세서 구성

프로세서는 세 가지 어레이를 통해 Agent에 연결됩니다.

import { Agent } from '@mastra/core/agent'
import { PrefillErrorHandler } from '@mastra/core/processors'

const agent = new Agent({
  id: 'support-agent',
  name: 'support-agent',
  model: 'openai/gpt-5',
  instructions: '...',
  inputProcessors: [new ContentFilter(['secret'])],
  outputProcessors: [new WordCounter()],
  errorProcessors: [new PrefillErrorHandler()],
  maxProcessorRetries: 3,
})

• inputProcessors: LLM 이전에 실행합니다. 입력 메시지를 받습니다.
• outputProcessors: LLM 응답 중 또는 응답 후에 실행됩니다. 출력 청크 또는 메시지를 수신합니다.
• errorProcessors: LLM API 호출이 발생하면 실행됩니다. 원시 오류를 수신합니다.

각 어레이는 또한 요청에 따라 프로세서를 구축할 수 있도록 기능을 허용합니다.RequestContext:

new Agent({
  id: 'processor-interface-agent',
  inputProcessors: ({ requestContext }) => {
    const blockedWords = requestContext.get('blockedWords') ?? []
    return [new ContentFilter(blockedWords)]
  },
})

통화별 재정의

agent.generate()와 agent.stream()은 inputProcessors, outputProcessors, errorProcessors, maxProcessorRetries를 허용합니다. 호출에서 프로세서 배열을 설정하면 해당 요청에 대해 Agent에 구성된 대응 배열을 대체합니다. Mastra가 자동으로 추가하는 Memory, Workspace, Skill, 채널, 브라우저 프로세서는 항상 유지되며 사용자 배열 전후에 실행됩니다.

await agent.stream('Summarize this', {
  outputProcessors: [new StreamFilter()],
  maxProcessorRetries: 5,
})

maxProcessorRetries전달된 통화는 Agent 기본값을 재정의합니다. 둘 다 설정되지 않은 경우 프로세서가 요청한 재시도는 중단으로 처리됩니다.

관련된

• 프로세서 개요: 프로세서에 대한 개념 가이드
• 난간: 보안 및 검증 프로세서
• Memory 프로세서: Memory 전용 프로세서