Processors
Processors 在消息流经 agent 时对其进行转换、校验或控制。它们在 agent 执行管线的特定位置运行，使你能够在输入到达语言模型之前修改输入，或在输出返回用户之前修改输出。
Processors 通过以下方式配置：
inputProcessors：在消息到达语言模型之前运行。outputProcessors：在语言模型生成响应之后、返回用户之前运行。
可以使用单个 Processor 对象，也可以使用 Mastra 的 workflow 原语将它们组合为 workflows。Workflows 让你对执行顺序拥有高级控制，并支持条件或并行处理。
有些处理器同时实现输入和输出逻辑，可以根据转换发生的位置放入任一数组。
某些内置处理器还会发送隐藏的系统提醒信号。这些信号会持久化到原始 memory 历史中，并在下一次模型调用前转换为 <system-reminder>...</system-reminder> 上下文，但标准的面向前端的消息转换和默认 memory 召回会隐藏它们，除非你明确选择启用。要在不保留信号的情况下为当前调用传递信号，请使用 transient: true 发送。
何时使用 processors何时使用 processors的直接链接
使用 processors 来：
- 规范化或校验用户输入
- 为 agent 添加 guardrails
- 检测并防止提示词注入或越狱尝试
- 为安全或合规目的审核内容
- 转换消息（例如翻译语言、过滤工具调用）
- 限制 token 用量或消息历史长度
- 脱敏敏感信息（PII）
- 对消息应用自定义业务逻辑
Mastra 内置了多个用于常见场景的处理器。你也可以为应用特定需求创建自定义处理器。
快速开始快速开始的直接链接
导入并实例化处理器，然后将其传入 agent 的 inputProcessors 或 outputProcessors 数组：
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',
}),
],
})
执行顺序执行顺序的直接链接
Processors 按照它们在数组中出现的顺序运行：
inputProcessors: [new UnicodeNormalizer(), new PromptInjectionDetector(), new ModerationProcessor()]
对于输出处理器，顺序决定了应用于模型响应的转换序列。
启用 memory 时启用 memory 时的直接链接
当 agent 启用 memory 时，memory processors 会自动添加到管线中：
输入处理器：
[Memory Processors] → [Your inputProcessors]
Memory 首先加载消息历史，然后运行你的处理器。
输出处理器：
[Your outputProcessors] → [Memory Processors]
你的处理器先运行，然后 memory 持久化消息。
在这种顺序下，调用 abort() 的输出 guardrail 会跳过 memory processors，阻止消息被保存。详情请参阅 Memory Processors。
将 processors 附加到 agent将 processors 附加到 agent的直接链接
Processors 通过三个数组在 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()],
})
inputProcessors在 LLM 之前运行。outputProcessors在 LLM 响应期间和之后运行。errorProcessors在 LLM API 调用抛出异常时运行，因此它们可以从提供器错误中恢复。
错误处理器默认启用错误处理器默认启用的直接链接
每个 agent 都会自动补齐以下四个中你的 errorProcessors 列表尚未包含的处理器：
ProviderHistoryCompat：修复出站提示词中提供器特定的历史不兼容问题PrefillErrorHandler：从 assistant 响应预填充拒绝中恢复UnsupportedFileHandler：将模型拒绝的文件从提示词中移除，在存在 sandbox 时上传到 agent 的 sandbox，然后再次调用模型StreamErrorRetryProcessor：重试瞬时的 API 和流故障
只有当你的列表中没有任何处理器携带相同 id 时，才会添加默认处理器，因此你自己的实例始终优先且不会被重复。添加的默认处理器会按其 id 在默认序列中的位置放置，因此只命名后面的处理器不会颠倒两者的顺序。即使只提供你自己的 StreamErrorRetryProcessor，最终解析结果仍会在它之前包含 ProviderHistoryCompat、PrefillErrorHandler 和 UnsupportedFileHandler。
错误处理器会在第一个返回 { retry: true } 的处理器处停止，因此相对顺序决定了哪种修复会运行。默认顺序将修复处理器放在重试处理器之前。如果放在更前面，具有宽泛匹配器的实例（例如 bad-request 匹配器会匹配任何 400）会抢先处理本可由修复处理器解决的请求，并将其原样重发。默认实例仅重试瞬时故障：连接重置、可重试的 OpenAI 流错误，以及提供器标记为可重试的错误。你自己的实例可以配置更宽泛的匹配器，顺序同样会保持修复处理器在它之前。
你的处理器之间的相对顺序永远不会被重排。id 与默认处理器匹配的处理器会参与该排序。使用自定义 id 的处理器则保持在你列表中的位置。
要在不丢失其余默认项的情况下自定义其中一个，请传入具有相同 id 的处理器：
import { Agent } from '@mastra/core/agent'
import { StreamErrorRetryProcessor } from '@mastra/core/processors'
const agent = new Agent({
id: 'tuned-agent',
name: 'tuned-agent',
model: 'openai/gpt-5',
instructions: '...',
errorProcessors: [new StreamErrorRetryProcessor({ maxRetries: 5 })],
maxProcessorRetries: 5,
})
上面的 StreamErrorRetryProcessor 是该 agent 上唯一的流重试处理器，它会解析到通常的位置：最后，位于你的列表从未命名的 ProviderHistoryCompat、PrefillErrorHandler 和 UnsupportedFileHandler 默认项之后。若要运行完全自定义的集合，请传入全部四个 id。
处理器的 maxRetries 不能提高 agent 的限制。如果没有 maxProcessorRetries，错误处理器重试会在 3 次后停止（请参阅错误处理器重试限制），因此当处理器需要更多重试时请设置它。
空的 errorProcessors 列表与其他列表一样会合并，因此仍会获得四个默认项。要关闭默认项，请设置 errorProcessorDefaults: false。此时 agent 只运行你配置的处理器；如果没有配置任何处理器，则一个也不运行：
new Agent({
id: 'custom-error-recovery-agent',
name: 'custom-error-recovery-agent',
model: 'openai/gpt-5',
instructions: '...',
errorProcessorDefaults: false,
errorProcessors: [new StreamErrorRetryProcessor({ maxRetries: 5 })],
maxProcessorRetries: 5,
})
传给单次 generate() 或 stream() 调用的 errorProcessors 列表工作方式不同：它会替换该调用的 agent 解析后列表（包括默认项）。
Agent 的 memory 也可以贡献自己的错误处理器，例如在请求超出模型上下文窗口时进行恢复。它们在上述处理器之后运行，并在调用传入自己的 errorProcessors 列表时保持不变。
每个数组还可以接受一个返回数组的函数，因此可以基于 RequestContext 为每次请求构建处理器：
new Agent({
id: 'processors-agent',
inputProcessors: ({ requestContext }) => {
const limit = requestContext.get('tokenLimit') ?? 4000
return [new TokenLimiter(limit)]
},
})
按调用覆盖 processors按调用覆盖 processors的直接链接
agent.generate() 和 agent.stream() 接受相同的三个数组。当你传入一个数组时，它只在该调用中替换 agent 上对应的数组。Memory、workspace 和其他由框架管理的 processors 仍会在你的数组周围运行。
await agent.stream('Summarize this', {
inputProcessors: [new TokenLimiter(2000)],
maxProcessorRetries: 5,
})
创建自定义 processors创建自定义 processors的直接链接
自定义处理器需实现 Processor 接口。
处理器方法接收两个用于访问对话的参数：
messages：当前阶段的MastraDBMessage对象快照数组。messageList：实时的MessageList实例。可用于读取其他阶段，或就地添加、删除、替换消息。
文本位于 message.content.parts 中，而不是 message.content 本身。遍历 parts 并按 part.type === 'text' 过滤以读取用户或 assistant 文本。为兼容旧版，还存在一个扁平化的 message.content.content 字符串，可用作回退。完整细节请参阅 Processor 参考中的 消息参数。
转换输入消息转换输入消息的直接链接
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() 在 agentic 循环的每一步运行（包括工具调用后的继续）。它支持逐步的配置变更，例如运行时切换模型或修改工具选择。
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 {}
}
}
返回 toolChoice: 'none' 会阻止模型调用工具，但仍会发送工具定义，因此先前的工具调用和结果会保留在对话中。只有当该步骤还向模型发送结构化输出 schema 时，Mastra 才会移除工具定义。
该方法接收当前的 stepNumber、model、tools、toolChoice、messages 等。返回一个对象，包含你想在该步骤中覆盖的任何属性，例如 { model, toolChoice, tools, systemMessages }。
所有可用参数和返回类型请参阅 Processor 参考。
在提供器调用前重写 LLM 请求在提供器调用前重写 LLM 请求的直接链接
当你需要重写 Mastra 发送给模型的最终提示词时，请使用 processLLMRequest()。该 hook 在 Mastra 将 MessageList 转换为面向提供器的提示词格式（LanguageModelV2Prompt）之后、提供器调用之前立即运行。
对话变更请使用基于消息的 hooks：
processInput()：在 agentic 循环开始前更改一次对话。processInputStep()：在每次 LLM 调用前更改消息或步骤配置。processLLMRequest()：仅更改当前提供器调用的出站提示词。
从 processLLMRequest() 返回的更改是瞬时的。它们不会持久化回 MessageList、memory、UI 历史或未来的提供器调用。因此该 hook 适合提供器兼容性重写、角色/内容规范化，或其他不应改变已存储对话历史的模型特定提示词更改。
该方法接收 prompt、model、stepNumber、steps、state 和共享的 processor 上下文。从 processLLMRequest() 调用 abort() 会发出正常的 tripwire 响应并停止调用。
错误处理器也会在此通道中运行。放入 errorProcessors 且实现了 processLLMRequest() 的处理器会获得该 hook，这正是默认的 ProviderHistoryCompat 能够修复出站提示词、而不是等待提供器拒绝的原因。仅实现 processAPIError() hook 的处理器（StreamErrorRetryProcessor 和 PrefillErrorHandler）不实现此方法，因此在此处不起作用。同时出现在 inputProcessors 和 errorProcessors 中的处理器只会运行一次，位置为其 inputProcessors 中的位置。
所有可用参数和返回类型请参阅 Processor 参考。
在提供器调用后处理 LLM 响应在提供器调用后处理 LLM 响应的直接链接
使用 processLLMResponse() 在步骤完成且流片段收集完毕后处理已完成的 LLM 响应。该 hook 与 processLLMRequest() 配对使用：在请求 hook 中暂存状态（例如缓存键），然后在响应 hook 中读回，以执行诸如写入缓存之类的副作用。
state 对象与同一步骤传给 processLLMRequest() 的是同一个实例。当 fromCache 为 true 时，响应是从缓存重放的，而不是由实时模型调用产生的，此时写入缓存的处理器应跳过写入。
该方法接收 chunks、model、stepNumber、steps、state、fromCache 和共享的 processor 上下文。
所有可用参数和返回类型请参阅 Processor 参考。
使用 prepareStep() 回调使用-preparestep-回调的直接链接
generate() 或 stream() 上的 prepareStep() 回调是 processInputStep() 的简写形式。在内部，Mastra 将其包装在一个 processor 中，在每一步调用你的函数。它接受与 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' }
}
},
})
转换输出消息转换输出消息的直接链接
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')
}
}
该方法还会接收一个 result 对象，包含完整生成数据、text、usage（token 计数）、finishReason 和 steps（每个包含 toolCalls、toolResults 等）。可用于跟踪用量或检查工具调用：
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() 方法在流式片段到达客户端之前对其进行转换或过滤：
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()。
若要同时接收工具通过 writer.custom() 发出的自定义 data-* 片段，请在处理器上设置 processDataParts = true。这样你就能在工具发出的数据片段到达客户端之前检查、修改或拦截它们。
校验每个响应校验每个响应的直接链接
processOutputStep() 方法在每个 LLM 步骤之后运行，允许你校验响应并可选地请求重试：
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 对象，它在单次请求的生命周期内持续存在。State 按处理器的 id 隔离。每个处理器只能看到自己的数据，并且在 processOutputStream、processOutputStep 和 processOutputResult 之间共享。每次新的 agent.generate() 或 agent.stream() 调用都会创建新的 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
}
async processOutputResult({ messages, state }) {
console.log(`Total words: ${state.wordCount}`)
return messages
}
}
内置实用处理器内置实用处理器的直接链接
Mastra 提供用于常见任务的实用处理器：
安全与校验处理器请参阅 Guardrails 页面，了解输入/输出 guardrails 和审核处理器。 Memory 专用处理器请参阅 Memory Processors 页面，了解处理消息历史、语义召回和工作记忆的处理器。
TokenLimitertokenlimiter的直接链接
当总 token 数超过指定限制时，通过移除较旧的消息来防止上下文窗口溢出。它优先保留最近的消息，并保留系统消息。
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 参考。
ToolCallFiltertoolcallfilter的直接链接
从发送给 LLM 的消息中移除工具调用和结果，节省冗长工具交互的 token。可以选择仅排除特定工具。此过滤器只影响 LLM 输入，被过滤的消息仍会保存到 memory。
默认情况下，ToolCallFilter 会过滤先前历史中的工具调用，而保留当前 agent 循环中进行的工具调用。使用 filterAfterToolSteps 也可以在循环期间进行过滤，同时保留最近的产生工具的步骤。
new ToolCallFilter({
filterAfterToolSteps: 2,
})
设置 preserveModelOutput: true 可为被过滤的已完成工具结果保留紧凑的 toModelOutput 历史。过滤器只保留面向模型的输出，并移除原始工具参数和原始结果。
new ToolCallFilter({
preserveModelOutput: true,
})
配置选项请参阅 ToolCallFilter 参考，memory 前端过滤请参阅 Memory Processors 页面。
ToolSearchProcessortoolsearchprocessor的直接链接
为拥有大量工具库的 agents 提供运行时工具发现功能。该处理器不会预先提供所有工具，而是为 agent 提供 search_tools 和 load_tool 元工具，以便按需通过关键字查找和加载工具，从而减少上下文 token 用量。
配置选项和使用示例请参阅 ToolSearchProcessor 参考。
ProviderHistoryCompatproviderhistorycompat的直接链接
当 agents 跨模型提供器复用消息时，处理提供器特定的历史不兼容问题。它可以在提供器调用前重写出站 LLM 请求，或从已知的提供器 API 错误中恢复并重试。
Mastra 默认将其添加到每个 agent 的 errorProcessors 中。当你需要自定义兼容规则或可预测的列表位置时，可提供自己的实例。对于使用观察式记忆的 Azure OpenAI agents，该处理器会重写出站提示词中的 <system-reminder> 包装，以避免内容审核拒绝。已存储的历史保持不变。
内置的 Anthropic 工具调用 ID 规则只会在出站请求中重写不符合 Anthropic ^[a-zA-Z0-9_-]+$ 模式的 ID，调用及其配对结果会同时处理，因此配对关系得以保留。持久化历史保留原始 ID。调用失败后的响应式恢复仍作为回退方案，用于通过预检查无法识别的提供器（例如 Vertex）提供的 Claude。
设置、内置规则和自定义规则选项请参阅 ProviderHistoryCompat 参考。
响应缓存响应缓存的直接链接
在 API 稳定之前，可能会在没有主版本号提升的情况下发生破坏性变更。
当 agent 收到完全相同的请求时，响应缓存会跳过 LLM 调用并重放先前缓存的响应。可用于降低延迟并避免为重复调用付费。
缓存通过 ResponseCache 输入处理器实现。Mastra 不提供 agent 级选项。要启用缓存，请显式注册该处理器。这样在 Mastra 收集反馈期间可以保持 API 表面精简。按调用覆盖通过 RequestContext 传递。
何时使用响应缓存何时使用响应缓存的直接链接
当相同请求在不同用户或会话之间重复出现时使用缓存。示例包括建议提示词按钮和重复搜索，或对相同输入进行分类的 guardrail LLM。当调用通过工具触发外部副作用时应跳过缓存，因为缓存命中会重放工具调用而不会重新执行它们。
快速开始快速开始的直接链接
将 ResponseCache 添加到 agent 的 inputProcessors，并传入任意 MastraServerCache 作为后端。开发时，InMemoryServerCache 开箱即用：
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 并将响应写入缓存。后续具有相同解析后提示词的调用会返回缓存响应，而不会调用 LLM。
通过 RequestContext 按调用覆盖通过 RequestContext 按调用覆盖的直接链接
按调用配置通过 RequestContext 传递。使用 ResponseCache.context() 构建新的 context，或使用 ResponseCache.applyContext() 合并到已有的 context：
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 会在 request context 上查找 MASTRA_RESOURCE_ID_KEY 并将其用作缓存作用域。已经填充 resource id 的 agent（例如通过 memory）会自动获得按用户隔离。用户永远不会看到彼此的缓存响应。
当你需要不同的作用域时，请显式覆盖：
new Agent({
id: 'processors-agent',
inputProcessors: [
new ResponseCache({
cache,
scope: 'org-123', // explicit tenant scope
}),
],
})
传入 scope: null 可有意在所有调用者之间共享条目。仅将此用于已知公开、非个性化的内容。
自定义缓存后端自定义缓存后端的直接链接
ResponseCache 接受任何 MastraServerCache。生产环境请使用 @mastra/redis 中的 RedisServerCache：
import { Agent } from '@mastra/core/agent'
import { ResponseCache } from '@mastra/core/processors'
import { RedisServerCache } from '@mastra/redis'
import { createClient } from 'redis'
const client = createClient({ url: process.env.REDIS_URL })
await client.connect()
const cache = new RedisServerCache({ client })
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（完成时写入缓存）。两者都在 agentic 循环内运行，位于 memory 加载和更早的输入处理器转换提示词之后。
缓存键由 Mastra 即将发送给模型的解析后 LanguageModelV2Prompt 派生。键是在 memory 加载和更早的输入处理器运行之后创建的，agentic 工具循环中的每个步骤都会独立缓存。
缓存键包含的内容缓存键包含的内容的直接链接
当你未提供 key 时，处理器会根据在该步骤改变 LLM 响应的输入确定性地派生一个键：agentId、stepNumber（因此工具循环中的每个步骤都有自己的缓存条目）、scope、模型标识（provider、modelId、spec 版本）以及解析后的 prompt（memory 后 + processors 后）。这些输入的任何变化都会自动使缓存失效。
多模态提示词也会包含在内。对于图像和文件部分，键包含完整 URL 或内联二进制数据（Uint8Array、ArrayBuffer）的字节摘要。仅引用的图像不同的请求因此会获得不同的缓存条目。
自定义缓存键自定义缓存键的直接链接
在构造函数上或按调用将 key 作为函数传入，以从这些输入的任意子集派生你自己的缓存键。该函数接收确定性哈希本会使用的相同输入，并返回字符串（或 Promise<string>）：
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 调用。Agentic 循环会从这些片段合成流，而不是调用模型。agent.generate() 会将它们收集为 FullOutput；agent.stream() 返回一个 MastraModelOutput，其片段来自缓存缓冲区，因此遍历 fullStream 或等待 text、usage 和 finishReason 的消费者会看到缓存值。
缓存写入发生在响应完成之后。失败的运行（错误、tripwire 触发）不会缓存。下次调用会干净地重试。
高级模式高级模式的直接链接
使用 maxSteps 确保最终响应使用-maxsteps-确保最终响应的直接链接
使用 maxSteps 限制 agent 执行时，如果 agent 在最后一步尝试工具调用，可能会返回空响应。使用带 sendSignal 的 processInputStep() 在最后一步注入响应式提醒。此方法会追加信号而不是修改系统消息，因此保留了提示词缓存。
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 },
})
}
}
信号会作为模型可见的内联 <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，加入解释信号标签的系统提示词，并将相同的 maxSteps 值传给 generate() 或 stream()：
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'。关于处理器发出的信号，请访问 Signals。
传递提醒但不保留传递提醒但不保留的直接链接
默认情况下，从处理器发送的信号会成为对话的一部分：它被写入存储，并在后续轮次重新进入提示词。对于每轮都重新注入的指令来说，这是不希望的，因为副本会堆积，模型会开始将自己过去的提醒当作模仿的上下文。设置 transient: true 可仅针对当前调用向模型传递信号而不保留它。
**何时使用：**当对话增长时，你希望将简短的控制指令保持在模型的近期窗口中，例如 "保持当前任务"、"回答保持在三句话以内"，或依赖实时应用状态的每轮约束。每轮重新注入，使其始终靠近最新消息。
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,
})
}
}
瞬态信号仍会出现在当前调用的提示词中。模型会在最新轮次附近看到它。由于不会被保留，每轮重新发送只会在上下文中保持一份新的副本，而不是累积历史，存储的 thread 历史也永远不会包含它。由于不写入任何内容，它还能在轮次之间保持稳定的提示词缓存前缀。
发出自定义流事件发出自定义流事件的直接链接
输出处理器接收一个 writer 对象，允许你在流式传输期间向客户端发出自定义数据片段。这对于流式传输审核结果或发送 UI 更新信号而不阻塞原始流很有用。
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- 前缀（例如 data-moderation-update、data-status）。
默认情况下，processOutputStream() 会跳过 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 中为消息添加自定义元数据。该元数据可通过响应对象访问：
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 片段 payload 或 stream.response promise 访问元数据。
将 workflows 用作 processors将 workflows 用作 processors的直接链接
你可以将 Mastra workflows 用作 processors，以创建具有并行执行、条件分支和错误处理的复杂处理管线：
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 时，processor workflows 会自动注册为 workflows，允许你在 Studio 中查看和调试它们。
重试机制重试机制的直接链接
Processors 可以请求 LLM 根据反馈重试其响应。这对于实现质量检查或输出校验（包括迭代改进）很有用：
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, // Step retry limit for processInputStep/processOutputStep. Unset by default, which disables those retries.
})
重试机制：
- 适用于
processOutputStep()和processInputStep()方法 - 重放该步骤，并将中止原因原样追加到对话末尾作为系统提醒。重试会将先前的请求保留为前缀，从而保留提供器提示词缓存
- 通过
retryCount参数跟踪重试次数 - 需要在 agent 或调用上显式设置
maxProcessorRetries限制
错误处理器重试限制错误处理器重试限制的直接链接
processAPIError() 拥有单独的重试预算。当 errorProcessors 解析为非空列表且省略 maxProcessorRetries 时，运行时允许最多 3 次错误处理器重试。错误处理器在每个 agent 上默认启用（请参阅错误处理器默认启用），但 maxProcessorRetries 本身保持未设置，因此 processInputStep() 和 processOutputStep() 的重试仍需要显式限制。
对于 StreamErrorRetryProcessor，其自身的 maxRetries（默认 1）独立于 agent 上限来限制重试。当该处理器是请求的唯一重试机制时，请将模型重试保持为 0。
违规回调违规回调的直接链接
所有 processors 都暴露 onViolation 属性，在检测到策略违规时触发，包括调用 abort()（block 策略）和处理器发出警告（warn 策略）时。可用于告警、日志记录或副作用，而不影响处理器的主要逻辑：
import { ModerationProcessor, TokenCostControl } 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 tokenCostControl = new TokenCostControl({
maxCost: 10.0,
scope: 'resource',
window: '30d',
})
tokenCostControl.onViolation = ({ processorId, message, detail }) => {
alertSystem.notify(`[${processorId}] ${message}`)
}
回调接收一个 ProcessorViolation 对象，包含：
processorId：检测到违规的处理器的 IDmessage：对违规内容的人类可读描述detail：处理器特定的元数据（例如成本用量、检测到的 PII 类型、审核类别）
onViolation 是基础 Processor 接口的一部分，因此任何自定义处理器也可以使用它。当任何处理器调用 abort() 时，runner 会自动调用它。回调内部抛出的错误会被捕获，以防止干扰处理器管线。
Abort 和 tripwire 片段Abort 和 tripwire 片段的直接链接
调用 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.tripwire 暴露相同信息，同时 result.finishReason === 'other'。
abort 接受第二个 options 参数：
retry: true请求 agent 重试而不是结束。输入和输出处理器重试需要在 agent 或调用上设置maxProcessorRetries。metadata将结构化数据附加到tripwire片段，以便下游消费者根据pii、quality或moderation等类别进行分支处理。
API 错误处理API 错误处理的直接链接
processAPIError 方法处理 LLM API 拒绝，即 API 拒绝请求的错误（例如 400 或 422 状态码），而不是网络或服务器故障。这让你可以在 API 拒绝消息格式时修改请求并重试。
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 包含一个内置的 PrefillErrorHandler，可自动处理 Anthropic 的 "assistant message prefill" 错误。该处理器会自动注入，无需配置。
从附件下载失败中恢复从附件下载失败中恢复的直接链接
当模型无法自行获取附件 URL 时，Mastra 会在调用模型之前下载该附件。如果下载失败，processAPIError() 会收到一个 MastraError，其 id 为 DOWNLOAD_ASSETS_FAILED，附件 URL 位于 error.details.url。错误处理器可以利用这一点来替换附件（例如使用新的签名 URL），并返回 { retry: true }。接下来会轮到回退模型。
如果没有办法恢复下载，agent 也不会使该轮次失败。它会重试一次，将附件替换为 [Attachment unavailable: <name>] 文本占位符，并记录警告。Mastra 会在存储的用户消息的 metadata.mastra.unavailableAttachments 下记录该 URL。后续轮次会显示相同的占位符，而不会再次下载附件。附件本身保留在 thread 的历史中。
无法解码的 data URL 会立即被替换为占位符，因为重试无法修复。
接收后台工作通知接收后台工作通知的直接链接
使用 createBackgroundWorkSignalProcessor() 为符合条件的后台工具调用保留活动调用者的信号能力：
import { Agent } from '@mastra/core/agent'
import { createBackgroundWorkSignalProcessor } from '@mastra/core/processors'
const agent = new Agent({
id: 'background-agent',
name: 'Background agent',
instructions: 'Complete tasks using the available tools.',
model: 'openai/gpt-5.6-sol',
inputProcessors: [createBackgroundWorkSignalProcessor()],
})
Mastra 的后台运行时仍负责执行、持久化、重试、结果协调和继续处理。该处理器仅保留调用者作用域的通知访问权限。当后台工作开始时，它会发出 work-deferred 或 work-awaited，状态为 status: 'running'。在权威工具结果被协调后，它会发出 work-completed 或 work-failed。
通知是进程本地的，并采用尽力而为的方式。如果原始运行已经结束或信号无法传递（例如任务在另一个进程上恢复），Mastra 会保留权威结果，而不会重建运行或重试通知。前台调用永远不会发出后台工作通知，因为没有可报告的分离工作。
相关文档相关文档的直接链接
- Agent 生命周期：完整运行的顺序和
RequestContext可见性 - Guardrails：安全和校验处理器
- Memory Processors：memory 专用处理器和自动集成
- Processor 接口：processors 的完整 API 参考
- ToolSearchProcessor 参考：运行时工具搜索的 API 参考