Memory processors

Memory processors 会在消息流经启用了 memory 的 agent 时对其进行转换和过滤。它们管理上下文窗口限制、移除不必要的内容，并优化发送给语言模型的信息。

当 agent 启用 memory 时，Mastra 会将 memory processors 添加到 agent 的 processor 管线中。这些 processors 检索消息历史、工作记忆以及语义相关的消息，然后在模型响应后持久化新消息。

Memory processors 是专门处理 memory 相关消息和状态的 processors。

内置 memory processors

启用 memory 时，Mastra 会自动添加以下 processors：

MemoryInputFilter

在 memory loaders 运行之前裁剪客户端回传的历史。对于现有 thread，它只保留当前用户轮次或新的工具结果。对于空 thread，它保留完整的初始输入，并从 assistant 部分移除可能引用存储中不存在项的提供器 item 元数据。

该 processor 最先运行，因此 MessageHistory、SemanticRecall 和 Observational Memory 只接收它们所需的新输入。

MessageHistory

检索消息历史并持久化新消息。

当你配置：

memory: new Memory({
  lastMessages: 10,
})

Mastra 在内部：

1. 创建一个 MessageHistory processor，limit: 10
2. 将其添加到 agent 的输入 processors（在 LLM 之前运行）
3. 将其添加到 agent 的输出 processors（在 LLM 之后运行）

它做什么：

• 输入：从存储中获取最近 10 条消息，并将它们前置到对话中
• 输出：在模型响应后将新消息持久化到存储

示例：

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { LibSQLStore } from '@mastra/libsql'
import { openai } from '@ai-sdk/openai'

const agent = new Agent({
  id: 'test-agent',
  name: 'Test Agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    storage: new LibSQLStore({
      id: 'memory-store',
      url: 'file:memory.db',
    }),
    lastMessages: 10, // MessageHistory processor automatically added
  }),
})

SemanticRecall

根据当前输入检索语义相关的消息，并为新消息创建 embeddings。

当你配置：

memory: new Memory({
  semanticRecall: { enabled: true },
  vector: myVectorStore,
  embedder: myEmbedder,
})

Mastra 在内部：

1. 创建一个 SemanticRecall processor
2. 将其添加到 agent 的输入 processors（在 LLM 之前运行）
3. 将其添加到 agent 的输出 processors（在 LLM 之后运行）
4. 需要同时配置 vector store 和 embedder

它做什么：

• 输入：执行向量相似度搜索以查找相关的历史消息，并将它们前置到对话中
• 输出：为新消息创建 embeddings 并将其存储在 vector store 中，以供将来检索

示例：

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { LibSQLStore } from '@mastra/libsql'
import { PineconeVector } from '@mastra/pinecone'
import { ModelRouterEmbeddingModel } from '@mastra/core/llm'
import { openai } from '@ai-sdk/openai'

const agent = new Agent({
  name: 'semantic-agent',
  instructions: 'You are a helpful assistant with semantic memory',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    storage: new LibSQLStore({
      id: 'memory-store',
      url: 'file:memory.db',
    }),
    vector: new PineconeVector({
      id: 'memory-vector',
      apiKey: process.env.PINECONE_API_KEY!,
    }),
    embedder: new ModelRouterEmbeddingModel('openai/text-embedding-3-small'),
    semanticRecall: { enabled: true }, // SemanticRecall processor automatically added
  }),
})

WorkingMemory

管理跨对话的工作记忆状态。

当你配置：

memory: new Memory({
  workingMemory: { enabled: true },
})

Mastra 在内部：

1. 创建一个 WorkingMemory processor
2. 将其添加到 agent 的输入 processors（在 LLM 之前运行）
3. 需要配置存储适配器

它做什么：

• 输入：检索当前 thread 的工作记忆状态，并将其前置到对话中
• 输出：无输出处理

示例：

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { LibSQLStore } from '@mastra/libsql'
import { openai } from '@ai-sdk/openai'

const agent = new Agent({
  name: 'working-memory-agent',
  instructions: 'You are an assistant with working memory',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    storage: new LibSQLStore({
      id: 'memory-store',
      url: 'file:memory.db',
    }),
    workingMemory: { enabled: true }, // WorkingMemory processor automatically added
  }),
})

手动控制和去重

如果你手动将 memory processor 添加到 inputProcessors 或 outputProcessors，Mastra 不会自动添加它。手动配置让你完全控制 processor 顺序：

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { MessageHistory } from '@mastra/core/processors'
import { TokenLimiter } from '@mastra/core/processors'
import { LibSQLStore } from '@mastra/libsql'
import { openai } from '@ai-sdk/openai'

// Custom MessageHistory with different configuration
const customMessageHistory = new MessageHistory({
  storage: new LibSQLStore({ id: 'memory-store', url: 'file:memory.db' }),
  lastMessages: 20,
})

const agent = new Agent({
  name: 'custom-memory-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    storage: new LibSQLStore({ id: 'memory-store', url: 'file:memory.db' }),
    lastMessages: 10, // This would normally add MessageHistory(10)
  }),
  inputProcessors: [
    customMessageHistory, // Your custom one is used instead
    new TokenLimiter({ limit: 4000 }), // Runs after your custom MessageHistory
  ],
})

Processor 执行顺序

在将 guardrails 与 memory 组合使用时，理解执行顺序很重要：

输入 Processors

[Memory Processors] → [Your inputProcessors]

1. Memory processors 首先运行：MemoryInputFilter，然后是 WorkingMemory、MessageHistory 和 SemanticRecall
2. 你的输入 processors 随后运行：guardrails、filters、validators

因此，memory 会在你的 processors 校验或过滤输入之前加载消息历史。

输出 Processors

[Your outputProcessors] → [Memory Processors]

1. 你的输出 processors 首先运行：guardrails、filters、validators
2. Memory processors 随后运行：SemanticRecall（embeddings）、MessageHistory（持久化）

此顺序旨在默认安全：如果你的输出 guardrail 调用 abort()，memory processors 永远不会运行，不会保存任何消息。

Guardrails 和 memory

默认执行顺序提供了安全的 guardrail 行为：

输出 guardrails（推荐）

输出 guardrails 在 memory processors 保存消息之前运行。如果 guardrail 中止：

• 触发 tripwire
• 跳过 memory processors
• 不会将任何消息持久化到存储

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { openai } from '@ai-sdk/openai'

// Output guardrail that blocks inappropriate content
const contentBlocker = {
  id: 'content-blocker',
  processOutputResult: async ({ messages, abort }) => {
    const hasInappropriateContent = messages.some(msg => containsBadContent(msg))
    if (hasInappropriateContent) {
      abort('Content blocked by guardrail')
    }
    return messages
  },
}

const agent = new Agent({
  id: 'safe-agent',
  name: 'safe-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({ lastMessages: 10 }),
  // Your guardrail runs BEFORE memory saves
  outputProcessors: [contentBlocker],
})

// If the guardrail aborts, nothing is saved to memory
const result = await agent.generate('Hello')
if (result.tripwire) {
  console.log('Blocked:', result.tripwire.reason)
  // Memory is empty - no messages were persisted
}

输入 guardrails

输入 guardrails 在 memory processors 加载历史之后运行。如果 guardrail 中止：

• 触发 tripwire
• LLM 永远不会被调用
• 跳过输出 processors（包括 memory 持久化）
• 不会将任何消息持久化到存储

// Input guardrail that validates user input
const inputValidator = {
  id: 'input-validator',
  processInput: async ({ messages, abort }) => {
    const lastUserMessage = messages.findLast(m => m.role === 'user')
    if (isInvalidInput(lastUserMessage)) {
      abort('Invalid input detected')
    }
    return messages
  },
}

const agent = new Agent({
  id: 'validated-agent',
  name: 'validated-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({ lastMessages: 10 }),
  // Your guardrail runs AFTER memory loads history
  inputProcessors: [inputValidator],
})

摘要

Guardrail 类型	何时运行	如果中止
输入	memory 加载历史之后	不调用 LLM，不保存任何内容
输出	memory 保存之前	不向存储保存任何内容

两种情况都是安全的——guardrails 会阻止不当内容被持久化到 memory。

处理大型附件

某些存储提供器会强制执行记录大小限制，base64 编码的文件附件可能超出该限制：

提供器	记录大小限制
DynamoDB	400 KB
Convex	1 MiB
Cloudflare D1	1 MiB

PostgreSQL、MongoDB 和 libSQL 的限制更高，通常不受影响。

使用输入 processor 将附件上传到外部存储，然后在消息持久化之前将其替换为 URL 引用。

src/mastra/processors/attachment-uploader.ts

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

export class AttachmentUploader implements Processor {
  id = 'attachment-uploader'

  async processInput({ messages }: { messages: MastraDBMessage[] }) {
    return Promise.all(messages.map(message => this.processMessage(message)))
  }

  async processMessage(message: MastraDBMessage) {
    const attachments = message.content.experimental_attachments
    if (!attachments?.length) return message

    const uploaded = await Promise.all(
      attachments.map(async attachment => {
        if (!attachment.url?.startsWith('data:')) return attachment

        const url = await this.upload(attachment.url, attachment.contentType)
        return { ...attachment, url }
      }),
    )

    return { ...message, content: { ...message.content, experimental_attachments: uploaded } }
  }

  async upload(dataUri: string, contentType?: string): Promise<string> {
    const base64 = dataUri.split(',')[1]
    const buffer = Buffer.from(base64, 'base64')

    throw new Error('Implement upload() with your storage provider')
  }
}

在你的 agent 中使用该 processor：

src/mastra/agents/support-agent.ts

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { AttachmentUploader } from '../processors/attachment-uploader'

export const supportAgent = new Agent({
  id: 'support-agent',
  name: 'Support agent',
  instructions: 'Answer customer support questions.',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({ lastMessages: 10 }),
  inputProcessors: [new AttachmentUploader()],
})

相关文档

• Processors：通用 processor 概念和自定义 processor 创建
• Guardrails：安全和校验 processors
• Memory 概览：memory 类型和配置

创建自定义 processors 时，请避免直接修改输入的 messages 数组或其对象。