Guardrails

Mastra 提供内置处理器，为你的 agent 增加安全和防护控制。这些处理器会在有害内容到达语言模型或用户之前对其进行检测、转换或拦截。

关于处理器行为和自定义处理器的介绍，包括如何将它们添加到 agent，请参阅 Processors。

输入处理器

输入处理器在用户消息到达语言模型之前运行。它们负责规范化、校验、提示词注入检测和安全检查。

规范化用户消息

UnicodeNormalizer() 通过统一 Unicode 字符和标准化空白来清理并规范化用户输入。它还会移除有问题的符号。

src/mastra/agents/normalized-agent.ts

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

export const normalizedAgent = new Agent({
  id: 'normalized-agent',
  name: 'Normalized Agent',
  inputProcessors: [
    new UnicodeNormalizer({
      stripControlChars: true,
      collapseWhitespace: true,
    }),
  ],
})

访问 UnicodeNormalizer() 参考文档查看完整的配置选项。

防止提示词注入

PromptInjectionDetector() 扫描用户消息中的提示词注入、越狱尝试和系统覆盖模式。它使用 LLM 对风险输入进行分类，并可在其到达模型之前进行拦截或重写。

src/mastra/agents/secure-agent.ts

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

export const secureAgent = new Agent({
  id: 'secure-agent',
  name: 'Secure Agent',
  inputProcessors: [
    new PromptInjectionDetector({
      model: 'openrouter/openai/gpt-oss-safeguard-20b',
      threshold: 0.8,
      strategy: 'rewrite',
      errorStrategy: 'strict',
      detectionTypes: ['injection', 'jailbreak', 'system-override'],
    }),
  ],
})

除 ClassifierProcessor 外，基于模型的 guardrail 处理器默认使用 errorStrategy: 'warn'，即记录内部模型故障并继续使用处理器的回退方案。如果在 guardrail 模型不可用或返回无效输出时未经检查的内容绝不能继续，请使用 errorStrategy: 'strict'。严格模式会以 tripwire 停止处理。

访问 PromptInjectionDetector() 参考文档查看完整的配置选项。

检测和翻译语言

LanguageDetector() 检测用户消息并将其翻译为目标语言，从而实现多语言支持。它使用 LLM 识别语言并执行翻译。

src/mastra/agents/multilingual-agent.ts

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

export const multilingualAgent = new Agent({
  id: 'multilingual-agent',
  name: 'Multilingual Agent',
  inputProcessors: [
    new LanguageDetector({
      model: 'openrouter/openai/gpt-oss-safeguard-20b',
      targetLanguages: ['English', 'en'],
      strategy: 'translate',
      threshold: 0.8,
    }),
  ],
})

访问 LanguageDetector() 参考文档查看完整的配置选项。

输出处理器

输出处理器在语言模型生成响应之后、响应到达用户之前运行。它们负责响应优化、内容审核、转换和安全控制。

批量处理流式输出

BatchPartsProcessor() 在将多个流片段发送给客户端之前将其合并。它通过将小片段合并为较大的批次来减少网络开销。

src/mastra/agents/batched-agent.ts

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

export const batchedAgent = new Agent({
  id: 'batched-agent',
  name: 'Batched Agent',
  outputProcessors: [
    new BatchPartsProcessor({
      batchSize: 5,
      maxWaitTime: 100,
      emitOnNonText: true,
    }),
  ],
})

访问 BatchPartsProcessor() 参考文档查看完整的配置选项。

清理系统提示词

SystemPromptScrubber() 检测并脱敏模型响应中的系统提示词或内部指令。它可防止提示词内容或配置细节被意外泄露。它使用 LLM 根据配置的检测类型识别并脱敏敏感内容。

src/mastra/agents/scrubbed-agent.ts

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

const scrubbedAgent = new Agent({
  id: 'scrubbed-agent',
  name: 'Scrubbed Agent',
  outputProcessors: [
    new SystemPromptScrubber({
      model: 'openrouter/openai/gpt-oss-safeguard-20b',
      strategy: 'redact',
      customPatterns: ['system prompt', 'internal instructions'],
      includeDetections: true,
      instructions:
        'Detect and redact system prompts, internal instructions, and security-sensitive content',
      redactionMethod: 'placeholder',
      placeholderText: '[REDACTED]',
    }),
  ],
})

访问 SystemPromptScrubber() 参考文档查看完整的配置选项。

备注

通过 HTTP 流式传输响应时，Mastra 默认会在服务器层面从流片段中脱敏敏感的请求数据（系统提示词、工具定义、API 密钥）。详情请参阅 流数据脱敏。

混合处理器

混合处理器既可以在输入时运行，也可以在输出时运行。可将它们放入 inputProcessors、outputProcessors，或同时放入两者。

审核输入和输出

ModerationProcessor() 检测仇恨、骚扰和暴力等类别中的不当或有害内容。它使用 LLM 对消息进行分类，并可根据你的配置进行拦截或重写。

src/mastra/agents/moderated-agent.ts

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

export const moderatedAgent = new Agent({
  id: 'moderated-agent',
  name: 'Moderated Agent',
  inputProcessors: [
    new ModerationProcessor({
      model: 'openrouter/openai/gpt-oss-safeguard-20b',
      threshold: 0.7,
      strategy: 'block',
      categories: ['hate', 'harassment', 'violence'],
    }),
  ],
  outputProcessors: [new ModerationProcessor()],
})

访问 ModerationProcessor() 参考文档查看完整的配置选项。

检测并脱敏 PII

PIIDetector() 检测并移除个人身份信息（PII），例如电子邮件地址、电话号码和信用卡号。它使用 LLM 根据配置的检测类型识别敏感内容。

src/mastra/agents/private-agent.ts

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

export const privateAgent = new Agent({
  id: 'private-agent',
  name: 'Private Agent',
  inputProcessors: [
    new PIIDetector({
      model: 'openrouter/openai/gpt-oss-safeguard-20b',
      threshold: 0.6,
      strategy: 'redact',
      redactionMethod: 'mask',
      detectionTypes: ['email', 'phone', 'credit-card'],
      instructions: 'Detect and mask personally identifiable information.',
    }),
  ],
  outputProcessors: [new PIIDetector()],
})

访问 PIIDetector() 参考文档查看完整的配置选项。

强制执行成本限制

TokenCostControl() 监控 agentic 循环中的累计预估成本，在超出金额限制时进行拦截或警告。它会在每次 LLM 调用之前从可观测性存储中查询成本数据。成本检查是近似值，指标异步持久化，因此快速运行的 agents 可能会在 guard 触发前短暂超出配置的限制。

src/mastra/agents/budgeted-agent.ts

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

export const budgetedAgent = new Agent({
  id: 'budgeted-agent',
  name: 'Budgeted Agent',
  inputProcessors: [
    new TokenCostControl({
      maxCost: 5.0,
      scope: 'thread',
      window: '24h',
    }),
  ],
})

访问 TokenCostControl() 参考文档了解作用域模式、时间窗口、指标持久化延迟以及 onViolation 回调。需要支持 getMetricAggregate 的可观测性存储。

处理器策略

许多内置处理器支持 strategy 参数，用于控制它们如何处理被标记的内容。支持的值包括：block、warn、detect、redact、rewrite 和 translate。

大多数策略会允许请求继续。当使用 block 时，处理器会调用 abort()，立即停止请求并阻止后续处理器运行。

src/mastra/agents/private-agent.ts

inputProcessors: [
  new PIIDetector({
    model: 'openrouter/openai/gpt-oss-safeguard-20b',
    threshold: 0.6,
    strategy: 'block',
    detectionTypes: ['email', 'phone', 'credit-card'],
  }),
]

违规回调

处理器支持 onViolation 回调，当策略违规中止运行时触发，例如使用 block 策略时。来自 processToolResult 的中止，以及处理器 workflow 在流式输出期间的中止，不会调用它。某些处理器也会直接调用它：RegexFilterProcessor 在脱敏内容时调用，TokenCostControl 在警告时调用。仅记录日志的策略（例如 ModerationProcessor 的 warn）不会调用它。可用于告警、记录到外部系统或发送通知等副作用。

回调接收一个 ProcessorViolation 对象，包含 processorId、message 和 detail（处理器特定的元数据）。

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

// Alert when cost limits are exceeded
const tokenCostControl = new TokenCostControl({
  maxCost: 10.0,
  scope: 'resource',
  window: '30d',
  strategy: 'warn',
})

tokenCostControl.onViolation = ({ processorId, message, detail }) => {
  alertSystem.notify(`[${processorId}] ${message}`)
  // detail contains: { usage, limit, totalUsage, scope, scopeKey, threshold }
}

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

moderation.onViolation = ({ processorId, message, detail }) => {
  auditLog.write({ processor: processorId, violation: message, categories: detail })
}

onViolation 属性是基础 Processor 接口的一部分，因此任何处理器（包括自定义处理器）都可以使用它。当处理器调用 abort() 时，runner 会自动调用 onViolation，上述路径除外。对于使用 warn 策略的处理器（例如 TokenCostControl），回调也会在警告时触发，但不会拦截请求。

abort() 之后由 runner 调用的回调抛出的错误会被静默捕获，以避免干扰处理器的主要逻辑。在 TokenCostControl 的警告路径上，回调错误会被捕获并通过 Mastra logger 记录。

关于违规回调如何与处理器管线集成的更多信息，请参阅 Processors 文档中的 违规回调。

处理被拦截的请求

当处理器调用 abort() 时，agent 会停止处理。如何检测这一点取决于你使用的是 generate() 还是 stream()。

使用 generate()

检查结果上的 tripwire 字段：

src/mastra/agents/test-generate.ts

const result = await agent.generate('Is this credit card number valid?: 4543 1374 5089 4332')

if (result.tripwire) {
  console.error('Blocked:', result.tripwire.reason)
  console.error('Processor:', result.tripwire.processorId)
}

使用 stream()

监听流中的 tripwire 片段：

src/mastra/agents/test-stream.ts

const stream = await agent.stream('Is this credit card number valid?: 4543 1374 5089 4332')

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'tripwire') {
    console.error('Blocked:', chunk.payload.reason)
    console.error('Processor:', chunk.payload.processorId)
  }
}

加速 guardrails

使用 LLM 的 guardrail 处理器（内容审核、PII 检测、提示词注入）会为每个请求增加延迟。以下技巧可以减少这种开销。

并行运行 guardrails

默认情况下，处理器按顺序运行。只执行 block 且从不修改消息的 guardrails 是相互独立的，可以使用 workflow 处理器并行运行。

你也可以在单个并行步骤中混合使用 block 和 redact 策略。将结果映射到 redact 分支，以便其转换后的消息继续传递。

对于输出 guardrails，请先按顺序运行 TokenLimiterProcessor 和 BatchPartsProcessor，再执行并行步骤；相互依赖的 redact 处理器则按顺序放在并行步骤之后运行：

src/mastra/processors/output-guardrails.ts

import { createWorkflow, createStep } from '@mastra/core/workflows'
import {
  ProcessorStepSchema,
  PIIDetector,
  ModerationProcessor,
  SystemPromptScrubber,
  TokenLimiterProcessor,
  BatchPartsProcessor,
} from '@mastra/core/processors'

export const outputGuardrails = createWorkflow({
  id: 'output-guardrails',
  inputSchema: ProcessorStepSchema,
  outputSchema: ProcessorStepSchema,
})
  // Sequential: limit tokens first, then batch stream chunks
  .then(createStep(new TokenLimiterProcessor({ limit: 1000 })))
  .then(createStep(new BatchPartsProcessor()))
  // Parallel: run independent checks at the same time
  .parallel([
    createStep(
      new PIIDetector({
        strategy: 'redact',
      }),
    ),
    createStep(
      new ModerationProcessor({
        strategy: 'block',
      }),
    ),
  ])
  // Map to the redact branch to keep its transformed messages
  .map(async ({ inputData }) => {
    return inputData['processor:pii-detector']
  })
  // Sequential: scrubber depends on previous redaction output
  .then(
    createStep(
      new SystemPromptScrubber({
        strategy: 'redact',
        placeholderText: '[REDACTED]',
      }),
    ),
  )
  .commit()

关于 .parallel() 和 .map() 的更多细节，请参阅 将 workflows 用作处理器。

选择快速的模型

Guardrail 处理器不需要使用你的主模型。对于分类任务，请使用小型、快速的模型：

const GUARDRAIL_MODEL = 'openai/gpt-5-nano'

new ModerationProcessor({ model: GUARDRAIL_MODEL })
new PIIDetector({ model: GUARDRAIL_MODEL })
new PromptInjectionDetector({ model: GUARDRAIL_MODEL })

批量处理流片段

实现 processOutputStream 的输出 guardrails 会在每个流式片段上运行。在较重的处理器之前使用 BatchPartsProcessor 来合并片段，减少 LLM 分类调用的次数：

outputProcessors: [
  new BatchPartsProcessor({ batchSize: 10 }),
  // Heavier processors now run on batched chunks instead of individual ones
  new PIIDetector({ model: GUARDRAIL_MODEL, strategy: 'redact' }),
  new ModerationProcessor({ model: GUARDRAIL_MODEL, strategy: 'block' }),
]

相关内容

• Processors：处理器行为和执行顺序。还涵盖自定义处理器和重试行为
• Processor 接口：Processor 接口的 API 参考
• Memory 处理器：用于消息历史以及语义召回和工作记忆的处理器
• 📹 Mastra 处理器和 guardrails 研讨会