프로세서

프로세서는 Agent를 통과하는 메시지를 변환, 검증 또는 제어합니다. Agent 실행 파이프라인의 특정 지점에서 실행되므로 언어 ​​Model에 도달하기 전에 입력을 수정하거나 사용자에게 반환되기 전에 출력을 수정할 수 있습니다.

프로세서는 다음과 같이 구성됩니다.

• inputProcessors: 메시지가 언어 Model에 도달하기 전에 실행됩니다.
• outputProcessors: 언어 Model이 응답을 생성한 후 사용자에게 반환되기 전에 실행됩니다.

개별 Processor 객체를 사용하거나 Mastra의 Workflow 기본 요소를 이용해 Workflow로 구성할 수 있습니다. Workflow를 사용하면 프로세서 실행 순서, 병렬 처리, 조건부 논리를 세밀하게 제어할 수 있습니다. 일부 프로세서는 입력 및 출력 논리를 모두 구현하며 변환이 발생하는 위치에 따라 두 배열 중 하나에서 사용할 수 있습니다.

일부 내장 프로세서는 숨겨진 시스템 알림 신호도 전송합니다. 이러한 신호는 원시 Memory 기록에 유지되고 다음 Model 호출 전에 <system-reminder>...</system-reminder> 컨텍스트로 변환되지만, 명시적으로 포함하도록 설정하지 않으면 일반적인 UI용 메시지 변환과 기본 Memory 회상에서는 숨겨집니다. 현재 호출에만 신호를 전달하고 유지하지 않으려면 transient: true와 함께 전송하세요.

프로세서를 사용하는 경우

프로세서를 사용하여 다음을 수행합니다.

• 사용자 입력 정규화 또는 유효성 검사
• Agent에 가드레일 추가
• 즉각적인 주입이나 탈옥 시도를 감지하고 방지합니다.
• 안전 또는 규정 준수를 위한 적당한 콘텐츠
• 메시지 변환(예: 언어 번역, 필터 Tool 호출)
• 토큰 사용 또는 메시지 기록 길이 제한
• 민감한 정보(PII) 수정
• 메시지에 사용자 정의 비즈니스 논리 적용

Mastra에는 일반적인 사용 사례를 위한 여러 프로세서가 포함되어 있습니다. 애플리케이션별 요구 사항에 맞게 사용자 정의 프로세서를 생성할 수도 있습니다.

빠른 시작

프로세서를 가져와 인스턴스화한 다음 Agent의 inputProcessors 또는 outputProcessors 배열에 전달합니다.

src/mastra/agents/moderated-agent.ts

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

export const moderatedAgent = new Agent({
  id: 'moderated-agent',
  name: 'moderated-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5-mini',
  inputProcessors: [
    new ModerationProcessor({
      model: 'openai/gpt-5-mini',
      categories: ['hate', 'harassment', 'violence'],
      threshold: 0.7,
      strategy: 'block',
    }),
  ],
})

실행 순서

프로세서는 어레이에 나타나는 순서대로 실행됩니다.

inputProcessors: [new UnicodeNormalizer(), new PromptInjectionDetector(), new ModerationProcessor()]

출력 프로세서의 경우 순서에 따라 Model의 응답에 적용되는 변환 순서가 결정됩니다.

Memory가 활성화된 상태에서

Agent에서 Memory가 활성화되면 Memory 프로세서가 자동으로 파이프라인에 추가됩니다.

입력 프로세서:

[Memory Processors] → [Your inputProcessors]

Memory는 메시지 기록을 먼저 로드한 다음 프로세서를 실행합니다.

출력 프로세서:

[Your outputProcessors] → [Memory Processors]

프로세서가 먼저 실행된 다음 Memory가 메시지를 유지합니다.

이 순서를 사용하면 abort()를 호출하는 출력 가드레일이 Memory 프로세서를 건너뛰고 메시지가 저장되지 않도록 합니다. 자세한 내용은 Memory 프로세서를 참조하세요.

Agent에 프로세서 연결

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

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

const agent = new Agent({
  id: 'support-agent',
  name: 'support-agent',
  model: 'openai/gpt-5',
  instructions: '...',
  inputProcessors: [
    new TokenLimiter(4000),
    new ModerationProcessor({ model: 'openai/gpt-5-nano' }),
  ],
  outputProcessors: [new ModerationProcessor({ model: 'openai/gpt-5-nano' })],
  errorProcessors: [new PrefillErrorHandler()],
})

• inputProcessorsLLM 전에 실행하십시오.
• outputProcessorsLLM 응답 도중 및 이후에 실행됩니다.
• errorProcessorsLLM API 호출이 발생할 때 실행되므로 공급자 오류로부터 복구할 수 있습니다.

각 배열은 배열을 반환하는 함수도 허용하므로 요청별로 프로세서를 구축할 수 있습니다.RequestContext:

new Agent({
  id: 'processors-agent',
  inputProcessors: ({ requestContext }) => {
    const limit = requestContext.get('tokenLimit') ?? 4000
    return [new TokenLimiter(limit)]
  },
})

호출당 프로세서 재정의

agent.generate()와 agent.stream()은 동일한 세 배열을 받습니다. 배열을 전달하면 해당 호출에 한해서만 Agent의 대응하는 배열을 대체합니다. Memory, Workspace 및 기타 프레임워크 관리 프로세서는 전달한 배열 전후에서 계속 실행됩니다.

await agent.stream('Summarize this', {
  inputProcessors: [new TokenLimiter(2000)],
  maxProcessorRetries: 5,
})

맞춤형 프로세서 생성

맞춤형 프로세서는Processor interface.

프로세서 메서드는 대화에 액세스하기 위해 두 가지 인수를 받습니다.

• messages: 현재 단계의 MastraDBMessage 객체 스냅샷 배열입니다.
• messageList: 실시간 MessageList 인스턴스입니다. 다른 단계를 읽거나 메시지를 그 자리에서 추가, 제거, 교체하는 데 사용합니다. 텍스트는 message.content 자체가 아니라 message.content.parts에 있습니다. 사용자 또는 어시스턴트 텍스트를 읽으려면 parts를 순회하며 part.type === 'text'로 필터링하세요. 레거시 호환성을 위해 평면화된 message.content.content 문자열도 존재하며 대체 수단으로 사용할 수 있습니다. 자세한 내용은 Processor 참조의 메시지 인수를 확인하세요.

입력 메시지 변환

src/mastra/processors/custom-input.ts

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

export class CustomInputProcessor implements Processor {
  id = 'custom-input'

  async processInput({ messages }: ProcessInputArgs): Promise<MastraDBMessage[]> {
    // Transform messages before they reach the LLM.
    // Text lives in content.parts — iterate parts and rewrite text parts only.
    return messages.map(msg => ({
      ...msg,
      content: {
        ...msg.content,
        parts: msg.content.parts?.map(part =>
          part.type === 'text' ? { ...part, text: part.text.toLowerCase() } : part,
        ),
      },
    }))
  }
}

processInput() 메서드는 messages, systemMessages, abort() 함수를 받습니다. 메시지를 대체하려면 MastraDBMessage[]를 반환하고, 시스템 메시지도 수정하려면 { messages, systemMessages }를 반환합니다. 사용 가능한 모든 인수와 반환 타입은 Processor 참조를 확인하세요.

각 단계를 제어하세요

processInput()은 Agent 실행 시작 시 한 번 실행되지만 processInputStep()은 Agent 루프의 각 단계에서 실행됩니다(Tool 호출 후속 처리 포함). 이를 통해 런타임 Model 전환이나 Tool 선택 변경과 같은 단계별 구성을 변경할 수 있습니다.

src/mastra/processors/step-processor.ts

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

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

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

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

    // No changes for other steps
    return {}
  }
}

이 메서드는 현재의 stepNumber, model, tools, toolChoice, messages 등을 받습니다. 해당 단계에서 재정의할 속성을 포함한 객체(예: { model, toolChoice, tools, systemMessages })를 반환합니다. 사용 가능한 모든 인수와 반환 타입은 Processor 참조를 확인하세요.

공급자 호출 전에 LLM 요청을 다시 작성합니다.

Mastra가 Model에 보내는 최종 Prompt를 다시 작성해야 할 때 processLLMRequest()를 사용합니다. 이 훅은 Mastra가 MessageList를 Provider용 Prompt 형식(LanguageModelV2Prompt)으로 변환한 후, Provider를 호출하기 직전에 실행됩니다. 대화 변경을 위해 메시지 기반 후크를 사용하십시오.

• processInput(): Agent 루프가 시작되기 전에 대화를 한 번 변경합니다.
• processInputStep(): 각 LLM 호출 전에 메시지 또는 단계 구성을 변경합니다.
• processLLMRequest(): 현재 공급자 통화에 대한 아웃바운드 Prompt만 변경합니다.

processLLMRequest()에서 반환된 변경 사항은 일시적입니다. MessageList, Memory, UI 기록 또는 향후 Provider 호출에 다시 저장되지 않습니다. 따라서 저장된 대화 기록을 변경하지 않아야 하는 Provider 호환성 재작성, 역할/콘텐츠 정규화 또는 기타 Model별 Prompt 변경에 적합합니다. 이 메서드는 prompt, model, stepNumber, steps, state 및 공유 프로세서 컨텍스트를 받습니다. processLLMRequest()에서 abort()를 호출하면 일반적인 tripwire 응답이 생성되고 호출이 중단됩니다. 사용 가능한 모든 인수와 반환 타입은 Processor 참조를 확인하세요.

공급자 호출 후 LLM 응답에 대한 조치

단계가 완료되고 스트림 청크가 수집된 후 완성된 LLM 응답을 처리하려면 processLLMResponse()를 사용합니다. 이 훅은 processLLMRequest()와 짝을 이룹니다. 요청 훅에서 상태(예: 캐시 키)를 저장한 다음 응답 훅에서 이를 다시 읽어 캐시에 쓰기 같은 부수 효과를 수행할 수 있습니다. state 객체는 같은 단계에서 processLLMRequest()에 전달된 것과 동일한 인스턴스입니다. fromCache가 true이면 응답이 실시간 Model 호출로 생성된 것이 아니라 캐시에서 재생된 것이므로, 캐시에 쓰는 프로세서는 이 경우 쓰기를 건너뛰어야 합니다. 이 메서드는 chunks, model, stepNumber, steps, state, fromCache 및 공유 프로세서 컨텍스트를 받습니다. 사용 가능한 모든 인수와 반환 타입은 Processor 참조를 확인하세요.

사용prepareStep() callback

generate() 또는 stream()의 prepareStep() 콜백은 processInputStep()의 축약형입니다. 내부적으로 Mastra는 각 단계에서 사용자의 함수를 호출하는 프로세서로 이를 래핑합니다. processInputStep()과 동일한 인수와 반환 타입을 받지만 클래스를 만들 필요는 없습니다.

await agent.generate('Complex task', {
  prepareStep: async ({ stepNumber, model }) => {
    if (stepNumber === 0) {
      return { model: 'openai/gpt-5-mini' }
    }
    if (stepNumber > 5) {
      return { toolChoice: 'none' }
    }
  },
})

출력 메시지 변환

src/mastra/processors/custom-output.ts

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

export class CustomOutputProcessor implements Processor {
  id = 'custom-output'

  async processOutputResult({ messages }): Promise<MastraDBMessage[]> {
    // Transform messages after the LLM generates them
    return messages.filter(msg => msg.role !== 'system')
  }
}

이 메서드는 전체 생성 데이터, text, usage(토큰 수), finishReason, steps(각각 toolCalls, toolResults 등을 포함)가 담긴 result 객체도 받습니다. 사용량을 추적하거나 Tool 호출을 검사하는 데 사용하세요.

src/mastra/processors/usage-tracker.ts

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

export class UsageTracker implements Processor {
  id = 'usage-tracker'

  async processOutputResult({ messages, result }) {
    console.log(`Tokens: ${result.usage.inputTokens} in, ${result.usage.outputTokens} out`)
    console.log(`Finish reason: ${result.finishReason}`)
    return messages
  }
}

스트리밍 출력 필터링

processOutputStream() 메서드는 스트리밍 청크가 클라이언트에 도달하기 전에 이를 변환하거나 필터링합니다.

src/mastra/processors/stream-filter.ts

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

export class StreamFilter implements Processor {
  id = 'stream-filter'

  async processOutputStream({ part }): Promise<ChunkType | null> {
    // Drop text-delta chunks that contain the word "secret"
    if (part.type === 'text-delta' && part.payload.text.includes('secret')) {
      return null
    }

    // Return the (possibly modified) chunk to emit it
    return part
  }
}

반환 값:

• ChunkType을 반환하면 해당 청크가 생성됩니다. 변경 없이 전달하려면 원래 part를 반환합니다.
• null 또는 undefined를 반환하면 청크가 삭제됩니다. 둘 다 동일하게 동작하므로 아무것도 반환하지 않는 메서드도 청크를 삭제합니다.
• 삭제는 하나의 청크에만 영향을 미칩니다. 스트림을 완전히 중단하려면 abort()를 호출하세요. Tool이 writer.custom()을 통해 내보낸 사용자 지정 data-* 청크도 받으려면 프로세서에 processDataParts = true를 설정하세요. 그러면 Tool에서 내보낸 데이터 청크가 클라이언트에 도달하기 전에 검사, 수정 또는 차단할 수 있습니다.

각 응답의 유효성을 검사합니다.

processOutputStep() 메서드는 각 LLM 단계 후에 실행되므로 응답을 검증하고 선택적으로 재시도를 요청할 수 있습니다.

src/mastra/processors/response-validator.ts

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

export class ResponseValidator implements Processor {
  id = 'response-validator'

  async processOutputStep({ text, abort, retryCount }) {
    const isValid = await validateResponse(text)

    if (!isValid && retryCount < 3) {
      abort('Response did not meet requirements. Try again.', { retry: true })
    }

    return []
  }
}

재시도 동작에 관한 자세한 내용은 고급 패턴의 재시도 메커니즘을 참조하세요.

청크와 단계 전반에 걸쳐 데이터 유지

출력 메서드는 한 요청의 수명 동안 유지되는 state 객체를 받습니다. 상태는 프로세서의 id를 키로 사용하므로 각 프로세서에는 자체 데이터만 표시되며, processOutputStream, processOutputStep, processOutputResult 간에 공유됩니다. 새로운 agent.generate() 또는 agent.stream() 호출마다 새 상태 객체가 생성됩니다.

src/mastra/processors/word-counter.ts

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
  }

  async processOutputResult({ messages, state }) {
    console.log(`Total words: ${state.wordCount}`)
    return messages
  }
}

내장형 유틸리티 프로세서

Mastra는 일반적인 작업을 위한 유틸리티 프로세서를 제공합니다.

보안 및 검증 프로세서의 경우, 입력/출력 가드레일과 조정 프로세서는 가드레일 페이지를 참조하세요. Memory 전용 프로세서의 경우, 메시지 기록, 의미 기반 회상, 작업 Memory를 처리하는 프로세서는 Memory 프로세서 페이지를 참조하세요.

TokenLimiter

총 토큰 수가 지정된 제한을 초과하는 경우 오래된 메시지를 제거하여 컨텍스트 창 오버플로를 방지합니다. 최근 메시지의 우선순위를 지정하고 시스템 메시지를 보존합니다.

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

const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  model: 'openai/gpt-5.6-sol',
  inputProcessors: [new TokenLimiter(127000)],
})

사용자 지정 인코딩, 전략, 계산 모드 옵션은 TokenLimiterProcessor 참조를 확인하세요.

ToolCallFilter

LLM으로 전송된 메시지에서 Tool 호출 및 결과를 제거하여 자세한 Tool 상호 작용에 대한 토큰을 저장합니다. 선택적으로 특정 Tool만 제외합니다. 이 필터는 LLM 입력에만 영향을 미치며 필터링된 메시지는 여전히 Memory에 저장됩니다.

기본적으로 ToolCallFilter는 Agent 루프가 시작되기 전에 초기 입력을 필터링합니다. 최근 Tool 생성 단계를 유지하면서 각 루프 단계에서도 필터링하려면 filterAfterToolSteps를 사용하세요.

new ToolCallFilter({
  filterAfterToolSteps: 2,
})

필터링된 완료 Tool 결과의 간결한 toModelOutput 기록을 유지하려면 preserveModelOutput: true를 설정합니다. 필터는 Model용 출력만 유지하고 원시 Tool 인수와 원시 결과는 제거합니다.

new ToolCallFilter({
  preserveModelOutput: true,
})

구성 옵션은 ToolCallFilter 참조를, Memory 적용 전 필터링은 Memory 프로세서 페이지를 확인하세요.

ToolSearchProcessor

대규모 Tool 라이브러리가 있는 Agent에 런타임 Tool 검색을 활성화합니다. 모든 Tool을 미리 제공하는 대신 프로세서는 Agent에 search_tools와 load_tool 메타 Tool을 제공하여 키워드로 필요할 때 Tool을 검색하고 불러오도록 하므로 컨텍스트 토큰 사용량이 줄어듭니다. 구성 옵션과 사용 예시는 ToolSearchProcessor 참조를 확인하세요.

ProviderHistoryCompat

Agent가 Model 공급자 전체에서 메시지를 재사용할 때 공급자별 기록 비호환성을 처리합니다. 공급자 호출 전에 아웃바운드 LLM 요청을 다시 작성하거나 알려진 공급자 API 오류를 복구하고 다시 시도할 수 있습니다.

Provider 기록 호환성 규칙, 반응형 API 오류 복구, 사용자 지정 호환성 규칙 또는 예측 가능한 프로세서 순서가 필요하면 ProviderHistoryCompat를 명시적으로 추가하세요. 설정, 내장 규칙, 사용자 지정 규칙 옵션은 ProviderHistoryCompat 참조를 확인하세요.

응답 캐싱

:::실험적

이 기능은 베타 버전입니다. API가 안정될 때까지 주요 버전 변경 없이 주요 변경 사항이 발생할 수 있습니다.

:::

응답 캐싱은 Agent가 동일한 요청을 받으면 LLM 호출을 건너뛰고 이전에 캐시된 응답을 재생합니다. 대기 시간을 줄이고 반복적인 통화에 대한 비용을 지불하지 않으려면 이를 사용하세요.

캐싱은 ResponseCache 입력 프로세서로 구현됩니다. Mastra는 Agent 수준 옵션을 제공하지 않습니다. 캐싱을 활성화하려면 프로세서를 명시적으로 등록하세요. Mastra가 피드백을 수집하는 동안 이 방식은 API 표면을 작게 유지합니다. 호출별 재정의는 RequestContext를 통해 전달됩니다.

응답 캐싱을 사용하는 경우

Prompt 템플릿, 제안 Prompt 버튼, Agent 검색 재요청, 동일한 입력을 반복해서 분류하는 가드레일 LLM 등 사용자나 세션 전반에 걸쳐 동일한 요청 형태가 반복되는 경우 이에 도달하세요. 호출이 Tool을 통해 외부 부작용을 트리거하는 경우 캐시가 재실행 없이 재생 Tool 호출에 도달하므로 이를 건너뜁니다.

빠른 시작

Agent의 inputProcessors에 ResponseCache를 추가하고 백엔드로 원하는 MastraServerCache를 전달합니다. 개발 환경에서는 InMemoryServerCache를 별도 설정 없이 사용할 수 있습니다.

src/mastra/agents/search-agent.ts

import { Agent } from '@mastra/core/agent'
import { InMemoryServerCache } from '@mastra/core/cache'
import { ResponseCache } from '@mastra/core/processors'

const cache = new InMemoryServerCache()

export const searchAgent = new Agent({
  id: 'search-agent',
  name: 'Search Agent',
  instructions: 'You answer questions concisely.',
  model: 'openai/gpt-5',
  inputProcessors: [new ResponseCache({ cache, ttl: 600 })], // 10 minutes
})

첫 번째 호출은 LLM을 정상적으로 실행하고 응답을 캐시에 기록합니다. 동일한 해결 Prompt가 포함된 후속 호출은 LLM을 호출하지 않고 캐시된 응답을 반환합니다.

RequestContext를 통한 호출별 재정의

호출별 구성은 RequestContext를 통해 전달됩니다. 새로운 컨텍스트를 만들려면 ResponseCache.context()를 사용하고, 기존 컨텍스트에 병합하려면 ResponseCache.applyContext()를 사용하세요.

src/example.ts

import { ResponseCache } from '@mastra/core/processors'
import { RequestContext } from '@mastra/core/request-context'

// Fresh context with the override
await agent.stream('hello', {
  requestContext: ResponseCache.context({ key: 'custom-key', bust: true }),
})

// Or merge into an existing context
const ctx = new RequestContext()
ctx.set('caller-meta', { userId: 'u-123' })
ResponseCache.applyContext(ctx, { bust: true })
await agent.stream('hello', { requestContext: ctx })

다음 필드는 호출별로 재정의할 수 있습니다.

• key: 문자열 또는 함수. 이 요청에 대해서만 자동으로 파생된 캐시 키를 재정의합니다.
• scope: 문자열 또는 null. 이 요청에 대해서만 테넌트/사용자 범위를 재정의합니다. null은 범위 지정을 사용하지 않습니다.
• bust: 부울. 캐시 읽기는 건너뛰지만 완료 시 쓰기는 계속 수행합니다("강제 새로 고침" 버튼에 유용함). cache, ttl, agentId는 생성자에 유지됩니다. 이는 인스턴스 수준의 고려 사항이며 호출마다 변경하는 것은 안전하지 않습니다.

테넌트 범위 지정

기본적으로 ResponseCache는 요청 컨텍스트에서 MASTRA_RESOURCE_ID_KEY를 조회하여 캐시 범위로 사용합니다. 즉, 이미 리소스 ID를 채우는 Agent(예: Memory를 통해)는 자동으로 사용자별 격리를 적용받습니다. 사용자는 서로의 캐시된 응답을 볼 수 없습니다. 다른 범위가 필요한 경우 명시적으로 재정의하세요.

src/mastra/agents/scoped-agent.ts

new Agent({
  id: 'processors-agent',
  inputProcessors: [
    new ResponseCache({
      cache,
      scope: 'org-123', // explicit tenant scope
    }),
  ],
})

모든 호출자가 항목을 의도적으로 공유하게 하려면 scope: null을 전달합니다. 공개된 비개인화 콘텐츠임이 확실한 경우에만 사용하세요.

커스텀 캐시 백엔드

ResponseCache는 모든 MastraServerCache를 받을 수 있습니다. 프로덕션에서는 @mastra/redis의 RedisCache를 사용하세요.

src/mastra/agents/cached-agent.ts

import { Agent } from '@mastra/core/agent'
import { ResponseCache } from '@mastra/core/processors'
import { RedisCache } from '@mastra/redis'

const cache = new RedisCache({ url: process.env.REDIS_URL })

export const agent = new Agent({
  id: 'cached-agent',
  name: 'Cached Agent',
  instructions: '...',
  model: 'openai/gpt-5',
  inputProcessors: [new ResponseCache({ cache })],
})

사용자 지정 백엔드의 경우 MastraServerCache를 확장하고 추상 메서드를 구현합니다(프로세서는 get과 set만 호출함).

캐싱 구현 방법

ResponseCache는 processLLMRequest(캐시 조회, 적중 시 단락 처리)와 processLLMResponse(완료 시 캐시 쓰기)에 연결됩니다. 둘 다 Memory가 로드되고 앞선 입력 프로세서가 Prompt를 변환한 후 Agent 루프 내부에서 실행됩니다. 즉, 캐시 키는 Mastra가 Model에 보내려는 확정된 LanguageModelV2Prompt에서 파생됩니다. 키는 Memory가 로드되고 앞선 입력 프로세서가 실행된 후 생성되며, Agent의 Tool 루프 내 각 단계는 독립적으로 캐시됩니다.

캐시 키에는 무엇이 들어 있나요?

key를 제공하지 않으면 프로세서는 이 단계의 LLM 응답을 변경하는 입력에서 키를 결정론적으로 파생합니다. 해당 입력에는 agentId, stepNumber(Tool 루프의 각 단계가 자체 캐시 항목을 갖도록 함), scope, Model 식별 정보(provider, modelId, 사양 버전), 확정된 prompt(Memory 및 프로세서 적용 후)가 포함됩니다. 이 입력 중 하나라도 변경되면 캐시가 자동으로 무효화됩니다. 멀티모달 Prompt도 포함됩니다. 이미지와 파일 부분은 해당 값으로 키에 반영됩니다. URL은 전체 href를 제공하고 인라인 바이너리 데이터(Uint8Array, ArrayBuffer)는 바이트의 다이제스트를 제공합니다. 따라서 참조하는 이미지만 다른 두 요청은 서로 다른 캐시 항목을 사용합니다.

캐시 키 맞춤설정

생성자 또는 호출별로 key를 함수로 전달하여 이러한 입력의 일부로 자체 캐시 키를 파생할 수 있습니다. 이 함수는 결정적 해시가 사용했을 것과 동일한 입력을 받아 문자열(또는 Promise<string>)을 반환합니다.

src/example.ts

import { ResponseCache, buildResponseCacheKey } from '@mastra/core/processors'

await agent.stream(input, {
  requestContext: ResponseCache.context({
    // Cache only on the model id and the resolved prompt tail — ignore
    // step number, scope, etc.
    key: ({ model, prompt }) => `qa:${model.modelId}:${JSON.stringify(prompt).slice(-200)}`,
  }),
})

// Or reuse the deterministic helper while overriding individual fields:
await agent.stream(input, {
  requestContext: ResponseCache.context({
    key: inputs => buildResponseCacheKey({ ...inputs, scope: 'global' }),
  }),
})

함수가 발생하면 프로세서는 기본 키 파생으로 돌아가므로 호출은 여전히 ​​캐싱의 이점을 얻습니다.

캐시 적중 작동 방식

프로세서가 캐시 적중을 발견하면 processLLMRequest에서 캐시된 청크를 반환하여 LLM 호출을 단락시킵니다. Agent 루프는 Model을 호출하는 대신 해당 청크로 스트림을 합성합니다. agent.generate()는 청크를 FullOutput으로 수집하고, agent.stream()은 캐시된 버퍼에서 가져온 청크가 포함된 MastraModelOutput을 반환하므로 fullStream을 반복하거나 text, usage, finishReason을 기다리는 소비자는 캐시된 값을 확인할 수 있습니다. 캐시 쓰기는 응답이 완료된 후에 발생합니다. 실패한 실행(오류, 트립와이어 활성화)은 캐시되지 않으므로 다음 호출이 완전히 재시도됩니다.

고급 패턴

다음을 통해 최종 응답을 보장합니다.maxSteps

maxSteps를 사용해 Agent 실행을 제한하면 Agent가 마지막 단계에서 Tool 호출을 시도할 경우 빈 응답을 반환할 수 있습니다. 마지막 단계에서 반응형 알림을 삽입하려면 sendSignal과 함께 processInputStep()을 사용하세요. 이 방식은 시스템 메시지를 수정하는 대신 신호를 추가하므로 Prompt 캐싱을 유지합니다.

src/mastra/processors/ensure-final-response.ts

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

export class EnsureFinalResponseProcessor implements Processor {
  readonly id = 'ensure-final-response'

  private maxSteps: number

  constructor(maxSteps: number) {
    this.maxSteps = maxSteps
  }

  async processInputStep({ stepNumber, sendSignal }: ProcessInputStepArgs) {
    if (stepNumber !== this.maxSteps - 1) {
      return
    }

    await sendSignal?.({
      type: 'reactive',
      contents:
        `This is your final step (step ${stepNumber + 1} of ${this.maxSteps}). ` +
        `Do not call any more tools. Summarize what you have found and give the user a complete final answer now.`,
      attributes: { reason: 'max-steps-reached', step: stepNumber + 1 },
    })
  }
}

신호는 Model이 인라인으로 확인하는 <system-reminder> 사용자 메시지로 전달됩니다.

<system-reminder reason="max-steps-reached" step="5">This is your final step (step 5 of 5). Do not call any more tools. Summarize what you have found and give the user a complete final answer now.</system-reminder>

프로세서를 inputProcessors에 추가하고, 신호 태그를 설명하는 시스템 Prompt를 포함한 다음, 동일한 maxSteps 값을 generate() 또는 stream()에 전달하세요.

src/mastra/agents/index.ts

import { Agent } from '@mastra/core/agent'
import { EnsureFinalResponseProcessor } from '../processors/ensure-final-response'

const MAX_STEPS = 5

const agent = new Agent({
  id: 'agent',
  instructions: `You are a helpful assistant.

Some messages you receive may contain <system-reminder>...</system-reminder> tags.
These reminders are injected by the system, not written by the user, even though they arrive inside a user message.
Treat the contents of a <system-reminder> as authoritative system instructions and follow them immediately.
Do not mention the reminder to the user or quote the tags back to them.`,
  inputProcessors: [new EnsureFinalResponseProcessor(MAX_STEPS)],
  // ...
})

await agent.generate('Your prompt', { maxSteps: MAX_STEPS })

:::참고 반응형 신호의 기본값은 tagName: 'system-reminder'입니다. 프로세서에서 내보내는 신호에 관한 자세한 내용은 신호를 참조하세요. :::

알림을 보관하지 않고 전달

기본적으로 프로세서에서 보낸 신호는 대화의 일부가 됩니다. 즉, 저장소에 기록되고 나중에 Prompt에 다시 입력됩니다. 매 턴마다 다시 삽입하는 지시의 경우 복사본이 누적되고 Model이 이전 알림을 모방할 컨텍스트로 처리하기 시작하므로 바람직하지 않습니다. 현재 호출에서만 Model에 신호를 전달하고 보존하지 않으려면 transient: true를 설정하세요. 사용 시기:대화가 늘어남에 따라 Model의 최근성 창에 짧은 조정 지침(예: "현재 작업 유지", "세 문장 미만의 답변 유지" 또는 라이브 애플리케이션 상태에 따라 달라지는 턴별 제약 조건)을 유지하려고 합니다. 매 턴마다 다시 주입하여 최신 메시지 근처에 유지되도록 합니다.

src/mastra/processors/steering-reminder.ts

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

export class SteeringReminderProcessor implements Processor {
  readonly id = 'steering-reminder'

  async processInputStep({ sendSignal }: ProcessInputStepArgs) {
    await sendSignal?.({
      type: 'reactive',
      contents: 'Stay on the current task and keep answers under three sentences.',
      transient: true,
    })
  }
}

현재 호출에 대한 Prompt에 일시적인 신호가 계속 나타나므로 Model은 마지막 방향 전환 근처에서 신호를 봅니다. 보관되지 않으므로 매 턴마다 다시 보내면 기록이 누적되는 대신 컨텍스트에 새로운 단일 복사본이 유지되며 저장된 스레드 기록에는 표시되지 않습니다. 아무 것도 기록되지 않기 때문에 차례대로 안정적인 Prompt 캐시 접두사를 유지합니다.

맞춤 스트림 이벤트 내보내기

출력 프로세서는 스트리밍 중 클라이언트에 사용자 지정 데이터 청크를 내보낼 수 있는 writer 객체를 받습니다. 원래 스트림을 차단하지 않고 조정 결과를 스트리밍하거나 UI 업데이트 신호를 보내는 등의 사용 사례에 유용합니다.

src/mastra/processors/moderation-processor.ts

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

export class ModerationProcessor implements Processor {
  id = 'moderation'

  async processOutputResult({ messages, writer }) {
    // Run moderation on the final output
    const text = messages
      .filter(m => m.role === 'assistant')
      .flatMap(m => m.content.parts?.filter(p => p.type === 'text'))
      .map(p => p.text)
      .join(' ')

    const result = await runModeration(text)

    if (result.requiresChange) {
      // Emit a custom event to the client with the moderated text
      await writer?.custom({
        type: 'data-moderation-update',
        data: {
          originalText: text,
          moderatedText: result.moderatedText,
          reason: result.reason,
        },
      })
    }

    return messages
  }
}

클라이언트에서 스트림의 사용자 정의 청크 유형을 수신합니다.

const stream = await agent.stream('Hello')

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'data-moderation-update') {
    // Update the UI with moderated text
    updateDisplayedMessage(chunk.data.moderatedText)
  }
}

사용자 정의 청크 유형은 다음을 사용해야 합니다.data- prefix (e.g., data-moderation-update, data-status).

기본적으로 processOutputStream()은 Tool 텔레메트리나 다른 프로세서의 출력을 실수로 처리하지 않도록 data-* 청크를 건너뜁니다. 프로세서에서 이러한 청크를 검사, 수정 또는 차단하려면 해당 프로세서에 processDataParts = true를 설정하세요.

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

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

메시지에 메타데이터 추가

processOutputResult에서 메시지에 사용자 지정 메타데이터를 추가할 수 있습니다. 이 메타데이터는 응답 객체를 통해 접근할 수 있습니다.

src/mastra/processors/metadata-processor.ts

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

export class MetadataProcessor implements Processor {
  id = 'metadata-processor'

  async processOutputResult({
    messages,
  }: {
    messages: MastraDBMessage[]
  }): Promise<MastraDBMessage[]> {
    return messages.map(msg => {
      if (msg.role === 'assistant') {
        return {
          ...msg,
          content: {
            ...msg.content,
            metadata: {
              ...msg.content.metadata,
              processedAt: new Date().toISOString(),
              customData: 'your data here',
            },
          },
        }
      }
      return msg
    })
  }
}

다음을 사용하여 메타데이터에 액세스합니다.generate():

const result = await agent.generate('Hello')

// The response includes uiMessages with processor-added metadata
const assistantMessage = result.response?.uiMessages?.find(m => m.role === 'assistant')
console.log(assistantMessage?.metadata?.customData)

스트리밍에서는 finish 청크 페이로드 또는 stream.response Promise에서 메타데이터에 접근합니다.

Workflow를 프로세서로 사용

Mastra Workflow를 프로세서로 사용하여 병렬 실행, 조건부 분기 및 오류 처리 기능을 갖춘 복잡한 처리 파이프라인을 생성할 수 있습니다.

src/mastra/processors/moderation-workflow.ts

import { createWorkflow, createStep } from '@mastra/core/workflows'
import {
  ProcessorStepSchema,
  PromptInjectionDetector,
  PIIDetector,
  ModerationProcessor,
} from '@mastra/core/processors'
import { Agent } from '@mastra/core/agent'

// Create a workflow that runs multiple checks in parallel
const moderationWorkflow = createWorkflow({
  id: 'moderation-pipeline',
  inputSchema: ProcessorStepSchema,
  outputSchema: ProcessorStepSchema,
})
  .parallel([
    createStep(
      new PIIDetector({
        strategy: 'redact',
      }),
    ),
    createStep(
      new PromptInjectionDetector({
        strategy: 'block',
      }),
    ),
    createStep(
      new ModerationProcessor({
        strategy: 'block',
      }),
    ),
  ])
  .map(async ({ inputData }) => {
    return inputData['processor:pii-detector']
  })
  .commit()

// Use the workflow as an input processor
const agent = new Agent({
  id: 'moderated-agent',
  name: 'Moderated Agent',
  model: 'openai/gpt-5.6-sol',
  inputProcessors: [moderationWorkflow],
})

.parallel() 단계 이후 각 브랜치 결과에는 프로세서 ID를 기준으로 키가 지정됩니다(예: processor:pii-detector). .map()을 사용해 다음 단계에서 출력을 받을 브랜치를 선택하세요. 브랜치에서 redact와 같은 변형 전략을 사용하는 경우 변환된 메시지가 이후 단계로 전달되도록 해당 브랜치에 매핑하세요. 모든 브랜치가 block만 사용한다면 어느 브랜치를 선택해도 됩니다. 메시지를 수정하는 브랜치가 없으므로 아무 브랜치나 선택하세요. Agent가 Mastra에 등록되면 프로세서 Workflow가 자동으로 Workflow로 등록되므로Studio.

재시도 메커니즘

처리자는 LLM이 피드백을 통해 응답을 다시 시도하도록 요청할 수 있습니다. 이는 품질 검사, 출력 유효성 검사 또는 반복적 개선을 구현하는 데 유용합니다.

src/mastra/processors/quality-checker.ts

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

export class QualityChecker implements Processor {
  id = 'quality-checker'

  async processOutputStep({ text, abort, retryCount }) {
    const qualityScore = await evaluateQuality(text)

    if (qualityScore < 0.7 && retryCount < 3) {
      // Request a retry with feedback for the LLM
      abort('Response quality score too low. Please provide a more detailed answer.', {
        retry: true,
        metadata: { score: qualityScore },
      })
    }

    return []
  }
}

const agent = new Agent({
  id: 'quality-agent',
  name: 'Quality Agent',
  model: 'openai/gpt-5.6-sol',
  outputProcessors: [new QualityChecker()],
  maxProcessorRetries: 3, // Maximum retry attempts. If unset, retries are disabled (unless errorProcessors are configured, in which case it defaults to 10).
})

재시도 메커니즘:

• processOutputStep() 및 processInputStep() 메서드에서 작동합니다.
• 중단 이유를 LLM의 컨텍스트에 추가하여 단계를 다시 실행합니다.
• retryCount 매개변수로 재시도 횟수를 추적합니다.
• Agent 또는 호출에 명시적인 maxProcessorRetries 제한을 설정해야 합니다.

오류 프로세서 재시도 제한

processAPIError()에는 별도의 기본값이 있습니다. errorProcessors가 구성되어 있고 maxProcessorRetries를 생략하면 런타임에서 최대 10회까지 재시도할 수 있습니다. 제한된 재시도 예산이 필요하다면 한도를 명시적으로 설정하세요. StreamErrorRetryProcessor의 maxRetries도 동일한 값으로 설정하세요. 자체 기본값은 1이므로 Agent 한도보다 낮을 수 있습니다. 프로세서가 요청의 유일한 재시도 메커니즘이라면 Model 재시도 횟수를 0으로 유지하세요.

위반 콜백

모든 프로세서에는 정책 위반이 감지될 때마다 실행되는 onViolation 속성이 있습니다. 이는 abort()가 호출될 때(차단 전략)와 프로세서가 경고를 발생시킬 때(경고 전략) 모두 적용됩니다. 프로세서의 주요 로직에 영향을 주지 않고 경고, 로깅 또는 부수 효과를 처리하는 데 사용하세요.

src/mastra/processors/violation-logging.ts

import { ModerationProcessor, CostGuardProcessor } from '@mastra/core/processors'

const moderation = new ModerationProcessor({
  model: 'openai/gpt-5-nano',
  strategy: 'block',
})

moderation.onViolation = ({ processorId, message, detail }) => {
  // Log to external monitoring, send alerts, update dashboards
  monitor.track('processor_violation', { processorId, message, detail })
}

const costGuard = new CostGuardProcessor({
  maxCost: 10.0,
  scope: 'resource',
  window: '30d',
})

costGuard.onViolation = ({ processorId, message, detail }) => {
  alertSystem.notify(`[${processorId}] ${message}`)
}

콜백은 다음을 포함하는 ProcessorViolation 객체를 받습니다.

• processorId: 위반을 감지한 프로세서의 ID
• message: 위반 사항에 대해 사람이 읽을 수 있는 설명
• detail: 프로세서별 메타데이터(예: 비용 사용량, 감지된 PII 유형, 조정 범주)

onViolation은 기본 Processor 인터페이스의 일부이므로 모든 사용자 지정 프로세서에서도 사용할 수 있습니다. 실행기는 프로세서가 abort()를 호출할 때 이를 자동으로 실행합니다. 콜백 내부에서 발생한 오류는 프로세서 파이프라인에 영향을 주지 않도록 조용히 포착됩니다.

중단 및 트립와이어 청크

abort(reason, options)를 호출하면 처리를 종료하는 TripWire 오류가 발생합니다. 스트림에서 Mastra는 클라이언트가 감지할 수 있는 tripwire 청크를 내보냅니다.

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'tripwire') {
    console.log('Blocked by', chunk.payload.processorId, '-', chunk.payload.reason)
    break
  }
}

agent.generate()의 경우 결과는 result.finishReason === 'other'와 함께 result.tripwire로 동일한 정보를 노출합니다. abort두 번째 옵션 인수를 허용합니다:

• retry: true는 종료하는 대신 Agent에 재시도를 요청합니다. 입력 및 출력 프로세서 재시도를 사용하려면 Agent 또는 호출에 maxProcessorRetries를 설정해야 합니다.
• metadata는 구조화된 데이터를 tripwire 청크에 추가하므로 다운스트림 소비자가 pii, quality, moderation과 같은 범주에 따라 분기할 수 있습니다.

API 오류 처리

processAPIError 메서드는 네트워크 또는 서버 장애가 아니라 API에서 요청을 거부하는 오류(예: 400 또는 422 상태 코드)인 LLM API 거부를 처리합니다. 이를 통해 API가 메시지 형식을 거부할 때 요청을 수정하고 재시도할 수 있습니다.

src/mastra/processors/api-error-handler.ts

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

export class ContextLengthHandler implements Processor {
  id = 'context-length-handler'

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

    if (APICallError.isInstance(error) && error.message.includes('context length exceeded')) {
      const messages = messageList.get.all.db()
      if (messages.length > 4) {
        messageList.removeByIds([messages[1]!.id, messages[2]!.id])
        return { retry: true }
      }
    }
  }
}

Mastra에는 Anthropic의 "assistant message prefill" 오류를 자동으로 처리하는 내장 PrefillErrorHandler가 있습니다. 이 프로세서는 자동으로 삽입되며 별도의 구성이 필요하지 않습니다.

관련 문서

• 난간: 보안 및 검증 프로세서
• Memory 프로세서: Memory별 프로세서 및 자동 통합
• 프로세서 인터페이스: 프로세서에 대한 전체 API 참조
• ToolSearch프로세서 참조: 런타임 Tool 검색을 위한 API 참조