ClassifierProcessor

The ClassifierProcessor is a hybrid processor that runs a Classifier over message text and passes the typed answers to onResult. In onResult, call abort(reason) to stop the request with a tripwire, call filter() to drop the content and continue, or do nothing to let it through. The classifier provides evidence such as probabilities, choices, and scores. onResult owns application policy.

Abort reasons are always the string supplied by the caller. Provider-generated text is never included in the abort message.

Usage example

import { Classifier } from '@mastra/core/classifier'
import { typeSafeAi } from '@ai-sdk/typesafe-ai'
import { ClassifierProcessor } from '@mastra/core/processors'

const safety = new Classifier({
  id: 'safety',
  model: typeSafeAi.decisionModel('jev-latest'),
  questions: {
    unsafe: {
      type: 'boolean',
      criteria: { true: 'The message is unsafe', false: 'The message is safe' },
    },
  },
})

const processor = new ClassifierProcessor({
  classifier: safety,
  onResult: (answers, { abort }) => {
    if ((answers.unsafe.probability ?? 1) > 0.8) abort('Message rejected by safety policy')
  },
  lastMessageOnly: true,
})

A decision model can refuse a question. A refused answer is { type: 'refusal' } and has no choice, score, or probability. The example treats a refusal as unsafe.

Several conditions can share one onResult, or use separate processors when they need different classifiers. Give each processor a unique id. Processor instances with the same id share per-run state, including accumulated stream chunks.

inputProcessors: [
  new ClassifierProcessor({
    id: 'safety-check',
    classifier: safety,
    onResult: (a, { abort }) => {
      if ((a.unsafe.probability ?? 1) > 0.8) abort('Rejected by safety policy')
    },
  }),
  new ClassifierProcessor({
    id: 'topic-check',
    classifier: topic,
    onResult: (a, { abort }) => {
      if (a.topic.type === 'refusal' || a.topic.choice === 'other') abort('Support questions only')
    },
  }),
]

Constructor parameters

options:

ClassifierProcessorOptions

Configuration for the processor

ClassifierProcessorOptions

classifier:

Classifier | string

A Classifier instance with configured questions, or the key or ID of a classifier registered on the Mastra instance. Registered classifiers are resolved lazily through mastra.getClassifierById() on first use.

onResult:

(answers, context) => void | Promise<void>

Called after each classification. answers is typed from the classifier questions. context provides abort(reason, { metadata }?) to stop the request with a tripwire, filter() to drop the message or chunk, phase, and the full ClassifierResult including usage and provider metadata. The metadata is set on the tripwire the caller receives.

onDecision?:

(decision: ClassifierProcessorDecision) => void | Promise<void>

Called after each classification, including failed classifier calls. Use it to write an audit record. See Audit decisions.

id?:

string

Processor identifier. Use a unique ID for each ClassifierProcessor in the same phase because instances with the same ID share per-run processor state.

errorStrategy?:

'warn' | 'strict'

How to handle decision model failures. 'strict' is fail-closed: it aborts the request, including during streaming, with tripwire metadata { processorId, classifierId, code: 'CLASSIFIER_FAILED' }. 'warn' is fail-open: it logs the failure and lets the content through. Classifier configuration errors always throw.

lastMessageOnly?:

boolean

When true, only the last message is classified in the input and output phases.

chunkWindow?:

number

Non-negative integer number of trailing stream chunks to classify together, including the current text-delta chunk. A value of 0 classifies only the current chunk. Each text chunk triggers a classifier call unless the windowed text is whitespace-only, so larger windows increase cost.

maxInputLength?:

number

Non-negative integer maximum number of characters sent to the classifier. Longer text is truncated.

providerOptions?:

SharedV4ProviderOptions

Provider-specific options forwarded to the decision model.

Behavior by phase

Phase	no call	abort(reason)	filter()
Input (inputProcessors)	message passes	abort before the LLM	message removed from context
Output result (outputProcessors)	message passes	abort the response	message removed from the response
Stream (outputProcessors)	chunk emitted	abort the stream	chunk not emitted

Messages with no text content are passed through without calling the classifier.

Audit decisions

onDecision receives a ClassifierProcessorDecision after each classifier call, once onResult has run. A failed classifier call produces a decision too.

const processor = new ClassifierProcessor({
  classifier: safety,
  onResult: (answers, { abort }) => {
    if ((answers.unsafe.probability ?? 1) > 0.8) {
      abort('Message rejected by safety policy', { metadata: { policy: 'unsafe' } })
    }
  },
  onDecision: async decision => {
    await auditLog.insert({
      threadId: decision.ids.threadId,
      messageId: decision.ids.messageId,
      tenantId: decision.requestContext?.get('tenantId'),
      outcome: decision.outcome,
      answers: decision.result ? decision.result.answers : null,
      error: decision.result ? null : String(decision.error),
      durationMs: decision.durationMs,
    })
  },
})

ClassifierProcessorDecision has these fields:

type ClassifierProcessorDecision = {
  processorId: string
  classifierId: string
  phase: 'input' | 'output' | 'stream'
  // 'aborted' when onResult called abort(), or when the call failed under errorStrategy 'strict'
  outcome: 'allowed' | 'filtered' | 'aborted'
  // Text sent to the classifier, after maxInputLength truncation
  state: ClassifierState
  // Set when the classifier answered
  result?: ClassifierResult
  // Set when the classifier call failed
  error?: unknown
  // Set when outcome is 'aborted', with the same reason and metadata as the tripwire the caller receives
  tripwire?: { reason: string; metadata?: unknown }
  // Duration of the classifier call, including retries
  durationMs: number
  ids: {
    runId?: string
    threadId?: string
    resourceId?: string
    messageId?: string
    traceId?: string
    spanId?: string
  }
  requestContext?: RequestContext
}

The ids that are set depend on the phase:

Phase	Ids
Input and output result	messageId, and threadId and resourceId when the agent uses memory
Stream	runId, and threadId and resourceId when the agent uses memory

threadId and resourceId come from the agent's memory options. Without memory, the processor reads them from the message, then from the MASTRA_THREAD_ID_KEY and MASTRA_RESOURCE_ID_KEY request context keys. traceId and spanId are set when the run is traced, and spanId is the processor span.

The processor waits for onDecision before it passes the content on. In the stream phase it runs once per text chunk. A slow write delays every chunk. Queue writes that take long instead of awaiting them. Errors thrown by onDecision are logged and don't change the outcome.

onDecision isn't called when onResult throws an error that isn't a tripwire, or when the user stops a streaming reply during the classifier call. Stopping the reply also cancels the classifier call, except on durable agents.

Extended usage examples

Topic scoping

src/mastra/agents/support-agent.ts

import { Agent } from '@mastra/core/agent'
import { Classifier } from '@mastra/core/classifier'
import { typeSafeAi } from '@ai-sdk/typesafe-ai'
import { ClassifierProcessor } from '@mastra/core/processors'

const topic = new Classifier({
  id: 'topic',
  model: typeSafeAi.decisionModel('jev-latest'),
  questions: {
    topic: {
      type: 'choice',
      criteria: {
        billing: 'Questions about invoices or payments',
        account: 'Questions about account settings',
        other: 'Anything else',
      },
    },
  },
})

export const agent = new Agent({
  id: 'support-agent',
  name: 'support-agent',
  instructions: 'You help customers with billing and account questions.',
  model: 'openai/gpt-6.1-sol',
  inputProcessors: [
    new ClassifierProcessor({
      classifier: topic,
      lastMessageOnly: true,
      onResult: (answers, { abort }) => {
        if (answers.topic.type === 'refusal' || answers.topic.choice === 'other') {
          abort('This assistant only handles billing and account questions')
        }
      },
    }),
  ],
})

Output quality gate

src/mastra/agents/quality-gated-agent.ts

import { Agent } from '@mastra/core/agent'
import { Classifier } from '@mastra/core/classifier'
import { typeSafeAi } from '@ai-sdk/typesafe-ai'
import { ClassifierProcessor } from '@mastra/core/processors'

const quality = new Classifier({
  id: 'quality',
  model: typeSafeAi.decisionModel('jev-latest'),
  questions: {
    quality: {
      type: 'score',
      instructions: 'Rate how well the response answers the user',
      criteria: ['Off-topic', 'Partial', 'Complete'],
    },
  },
})

export const agent = new Agent({
  id: 'quality-gated-agent',
  name: 'quality-gated-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-6.1-sol',
  outputProcessors: [
    new ClassifierProcessor({
      classifier: quality,
      onResult: (answers, { filter }) => {
        if ((answers.quality.score ?? 0) < 1) filter()
      },
    }),
  ],
})

Registered classifier

Register the classifier on the Mastra instance and reference it by key or ID. The processor resolves it on first use.

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { Classifier } from '@mastra/core/classifier'
import { typeSafeAi } from '@ai-sdk/typesafe-ai'

export const mastra = new Mastra({
  classifiers: {
    safety: new Classifier({
      id: 'safety',
      model: typeSafeAi.decisionModel('jev-latest'),
      questions: {/* ... */},
    }),
  },
})

src/mastra/processors/safety-processor.ts

import { ClassifierProcessor } from '@mastra/core/processors'

type SafetyQuestions = {
  unsafe: { type: 'boolean' }
}

const processor = new ClassifierProcessor<SafetyQuestions>({
  classifier: 'safety',
  onResult: (answers, { abort }) => {
    if ((answers.unsafe.probability ?? 1) > 0.8) abort('Message rejected by safety policy')
  },
})

When using a registered classifier by string, answer types aren't inferred from the registered instance. Pass its question map as the ClassifierProcessor type argument to type the answers parameter.

Related

• Classifier
• Guardrails