基于MCP协议的多Agent共享持久记忆服务实现

很多同学在搭建多 Agent 项目时都会遇到一个很现实的问题每个 Agent 在单个会话里表现得还挺好可一旦开始多轮交互、多个角色协作记忆就断了。上一轮已经明确告诉 Agent 的信息这一轮它又像第一次见面一样重新问对话窗口有长度限制历史记录既不共享也不持久整个系统很快就会被重复劳动拖垮。要解决这个问题不能只靠无限扩大上下文窗口更合理的思路是在所有 Agent 之下加一层“共享且持久”的记忆服务。本文要聊的 Memento就是这样一个典型方向为多个 AI Agent 提供共享、持久、可通过 MCP 协议访问的记忆能力。下面的内容会从核心概念讲起然后给出一个最小可运行的 MCP Server 实现包含存储层、工具接口、多 Agent 接入配置以及常见问题的排查思路。无论你是刚开始接触 Agent 开发还是已经在做多 Agent 工程化都可以参考这套方案落地自己的记忆层。1. 背景与核心概念1.1 为什么多 Agent 需要共享持久记忆多 Agent 系统目前最容易被吐槽的点就是“记不住”。单个 Agent 在会话窗口内只要上下文没有超过长度限制它还能记住用户刚说的要求但一旦会话轮数变多或者需要把任务拆给另一个 Agent 处理前一个 Agent 积累的上下文就很难继续传递。这里有两个核心矛盾。第一个是窗口有上限token 成本也有上限。如果每一轮都把全部历史记录塞进上下文既慢又贵而且越长的上下文越容易让模型忽略关键信息。第二个矛盾是多个 Agent 之间的记忆天然隔离。Agent A 已经确认了用户的偏好Agent B 并不知道Agent C 查到了关键结论Agent D 又得重新查一遍。这种“集体失忆”会让整个系统表现得像一群没有默契的独立程序而不是一个协作团队。所以共享、持久、可编程访问的记忆层就成了多 Agent 架构里非常关键的基础设施。它让 Agent 不必把所有历史都塞进窗口而是只把“值得长期记住的事实”写入外部存储在需要时再通过检索取回。这样既控制了上下文长度又让多个 Agent 之间有了共同的工作记忆。1.2 Memento 是什么Memento 的核心定位就是给多个 AI Agent 提供一套“共享且持久”的记忆能力。从命名上也能看出它的设计意图memento 在英文里有“纪念品、回忆”的意思对应到系统中就是让 Agent 把值得记住的信息存下来并在后续对话或协作任务中随时取用。它并不是要替代 Agent 的上下文窗口而是作为一种补充窗口负责“眼前”记忆库负责“长期”。从实现角度看一个成熟的 Memento 服务通常包含三层能力。第一层是存储层用数据库或文件系统保存 Agent 记忆确保服务重启后数据不丢第二层是抽象层提供记忆的写入、查询、更新、过期、权限等语义而不是让 Agent 直接操作底层数据库第三层是接入层通过 MCP 把记忆能力暴露成标准化工具让任何支持 MCP 的 Agent 都能直接调用。这样做最大的价值是“协议统一”。过去每个 Agent 框架都有自己的记忆实现接入新 Agent 时要重新写一套对接逻辑。现在 Memento 把记忆能力做成 MCP Server客户端只需要按 MCP 协议发现工具、调用工具就能复用同一套记忆基础设施。1.3 MCP 在 Memento 中的角色MCP 是 Model Context Protocol 的缩写可以理解为 AI 应用与外部工具、数据源之间的一套开放通信协议。它定义了客户端比如 Agent 宿主程序和服务端比如提供记忆能力的工具服务如何发现服务、如何调用工具、以及如何传输结果。Memento 选择用 MCP 暴露记忆能力最大的好处是“一次接入处处可用”。这里需要区分两个容易混淆的概念MCP 和 Agent Skill。Skill 通常描述的是 Agent 本身具备的一种技能或一套行为模板例如“会写 SQL”“会做代码审查”而 MCP 描述的是外部工具如何被 Agent 调用。可以简单理解Skill 是 Agent 内部的“能力”MCP 是 Agent 与外部的“连接器”。Memento 本身属于外部服务所以更适合以 MCP Server 的形式存在。这个区分在团队协作时很重要因为它决定了能力应该写进 Agent 配置还是独立部署成服务并用协议访问。2. 整体架构与消息模型2.1 架构分层一个完整的 Memento 服务在架构上可以拆成四层看层级职责典型组件接入层暴露 MCP Server处理 Agent 请求MCP Server、工具注册业务层实现记忆写入、搜索、更新、逻辑删除等语义MemoryStore 服务类存储层持久化记忆数据SQLite、PostgreSQL、Redis辅助能力向量检索、权限校验、审计日志Embedding 模型、API 网关最底层其实是 Agent 客户端。Agent 宿主程序例如 Claude Desktop、Cursor、Codex、自研 Agent 框架以 MCP Client 身份连接 Memento Server调用memory_store、memory_search这类工具。Server 接收到请求后经过业务层处理再读写底层存储。在早期落地时不需要把架构做得太重。一个 SQLite 文件加一个 MCP Server 进程已经足够支撑开发环境和小规模使用。等到 Agent 数量变多、记忆数据量变大再逐步把存储层替换为 PostgreSQL或者引入向量数据库做语义检索。2.2 Memory 数据模型记忆数据如何建模直接决定了 Memento 服务好不好扩展。下面是一份比较通用的记忆模型设计后续实现会参考它。记忆字段定义如下字段类型含义idstring唯一标识由服务生成agentstring写入该记忆的 Agent 标识namespacestring命名空间用于项目/租户隔离memory_typestring记忆类型目录如 episodic、semantic、proceduralcontentstring记忆正文metadatastring扩展元信息JSON 格式created_atnumber创建时间戳updated_atnumber更新时间戳revokedinteger是否逻辑删除0 表示有效1 表示已失效为什么要加namespace因为多 Agent 场景下可能存在多个项目、多个团队、多条业务线。如果不做隔离Agent A 写入的记忆很可能在 Agent B 的检索中造成干扰甚至泄露敏感信息。namespace可以认为是记忆中最小粒度的“租户边界”。memory_type则用来区分不同性质的记忆。semantic可以存用户偏好、项目背景、业务规则这类长期事实episodic可以存某次任务的过程和结论procedural可以存操作步骤和工作流。类型分离之后Agent 在搜索记忆时也可以按类型过滤提升准确率。revoked字段建议使用逻辑删除而不是物理删除。记忆可能存在引用关系或者事后需要审计物理删除会让追责和回溯变得困难。逻辑删除保留了历史痕迹只是在查询时默认过滤掉revoked 1的记录。2.3 工具接口设计Memento 通过 MCP Tools 向外暴露能力。工具设计要尽量贴近记忆操作的自然语义同时保证参数清晰。建议暴露以下工具memory_store写入一条记忆。memory_search按关键词搜索记忆。memory_get按 ID 获取单条记忆。memory_update更新记忆内容或元信息。memory_revoke逻辑删除一条记忆。memory_stats查看某个命名空间的记忆统计信息。这套接口覆盖了日常使用的主要场景。写到这里的读者可能会发现这套设计与 REST 接口的 CRUD 有些相似但区别在于通过 MCP 暴露后Agent 不需要理解 HTTP 路由、鉴权头、序列化细节只需要在工具列表里选一个工具并传入参数即可。这也是 MCP 能快速普及的原因之一。3. 环境准备与版本说明3.1 运行环境在开始写代码之前需要准备一个可运行的 Python 环境。本示例实现以下列环境作为参考实际项目请根据你的现状调整版本操作系统Windows / macOS / Linux 均可。Python3.10 或更高版本。MCP SDK使用官方mcpPython SDK 的最新稳定版。数据库SQLitePython 自带不需要额外安装数据库服务。客户端工具支持 MCP 的桌面客户端或 MCP Inspector用于验证工具调用。版本需要根据你的项目实际情况调整本文重点演示配置思路和完整实现。如果你的 MCP SDK 版本较新或较旧API 可能有细微差异以下代码以当前主流用法为例。3.2 项目结构建议按下面的目录结构组织项目。将存储层与 MCP Server 入口分离后续维护起来更清晰。memento-project/ ├── memento_server.py # MCP Server 入口注册工具 ├── memento_store.py # SQLite 存储层实现 ├── self_test.py # 不依赖 MCP 的自测脚本 ├── requirements.txt # 依赖清单 └── config/ └── mcp-client.json # 客户端 MCP 配置示例memento_store.py专注于数据持久化不依赖 MCP SDK因此可以单独写单元测试也可以方便地替换成 PostgreSQL 实现。memento_server.py只做协议转换把 MCP 工具调用翻译成MemoryStore的方法调用。3.3 安装依赖创建一个requirements.txt文件内容如下mcp然后执行安装命令pip install -r requirements.txtMCP SDK 版本迭代比较快建议安装后确认导入是否正常python -c import mcp; print(mcp.__version__)如果执行没有报错说明环境就绪。如果mcp的 API 与你当前版本不一致可以优先参考官方 SDK 文档中对FastMCP的说明。本文使用的FastMCP是官方 Python SDK 提供的简化封装适合快速注册工具。4. 基于 MCP 实现 Memento 核心服务4.1 存储层实现首先实现存储层。这里选择 SQLite原因是它不需要额外部署、单文件即可持久化对开发环境和中小规模项目非常友好。代码如下# memento_store.py import json import sqlite3 import threading import time import uuid from pathlib import Path class MemoryStore: 基于 SQLite 的轻量记忆存储实现。 def __init__(self, db_path: str memento.db): self.db_path Path(db_path) self.db_path.parent.mkdir(parentsTrue, exist_okTrue) self._lock threading.Lock() self._conn sqlite3.connect(str(self.db_path), check_same_threadFalse) self._conn.row_factory sqlite3.Row self._init_schema() def _init_schema(self): with self._lock: self._conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, agent TEXT NOT NULL, namespace TEXT NOT NULL DEFAULT default, memory_type TEXT NOT NULL DEFAULT semantic, content TEXT NOT NULL, metadata TEXT NOT NULL DEFAULT {}, created_at REAL NOT NULL, updated_at REAL NOT NULL, revoked INTEGER NOT NULL DEFAULT 0 ) ) self._conn.execute( CREATE INDEX IF NOT EXISTS idx_mem_namespace ON memories(namespace, revoked) ) self._conn.commit() def store(self, agent: str, content: str, namespace: str default, memory_type: str semantic, metadata: dict None) - str: memory_id str(uuid.uuid4()) now time.time() meta_text json.dumps(metadata or {}, ensure_asciiFalse) with self._lock: self._conn.execute( INSERT INTO memories (id, agent, namespace, memory_type, content, metadata, created_at, updated_at, revoked) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0) , (memory_id, agent, namespace, memory_type, content, meta_text, now, now), ) self._conn.commit() return memory_id def search(self, query: str, namespace: str default, agent: str None, limit: int 10) - list: sql SELECT * FROM memories WHERE revoked 0 AND namespace ? params [namespace] if agent: sql AND agent ? params.append(agent) sql ORDER BY updated_at DESC LIMIT ? params.append(limit) with self._lock: rows self._conn.execute(sql, params).fetchall() words [w.lower() for w in query.split() if w] results [] for row in rows: content_lower (row[content] or ).lower() metadata_lower (row[metadata] or ).lower() score 0 for w in words: if w in content_lower: score 2 elif w in metadata_lower: score 1 if not words or score 0: item dict(row) item[score] score results.append(item) results.sort(keylambda x: (x[score], x[updated_at]), reverseTrue) return results def get(self, memory_id: str): with self._lock: row self._conn.execute( SELECT * FROM memories WHERE id ? AND revoked 0, (memory_id,), ).fetchone() return dict(row) if row else None def update(self, memory_id: str, content: str None, metadata: dict None) - bool: updates [] params [] if content is not None: updates.append(content ?) params.append(content) if metadata is not None: updates.append(metadata ?) params.append(json.dumps(metadata, ensure_asciiFalse)) if not updates: return False updates.append(updated_at ?) params.append(time.time()) params.append(memory_id) with self._lock: cur self._conn.execute( fUPDATE memories SET {, .join(updates)} WHERE id ? AND revoked 0, params, ) self._conn.commit() return cur.rowcount 0 def revoke(self, memory_id: str) - bool: with self._lock: cur self._conn.execute( UPDATE memories SET revoked 1, updated_at ? WHERE id ?, (time.time(), memory_id), ) self._conn.commit() return cur.rowcount 0 def stats(self, namespace: str default) - dict: with self._lock: row self._conn.execute( SELECT COUNT(*) AS total, COUNT(DISTINCT agent) AS agents FROM memories WHERE revoked 0 AND namespace ? , (namespace,), ).fetchone() return dict(row)这段代码有几个关键点需要解释。第一check_same_threadFalse允许连接被多线程复用。MCP Server 在处理请求时可能会使用线程池如果不开这个参数跨线程访问 SQLite 连接会报错。第二所有写操作都通过self._lock加锁。SQLite 虽然对并发读比较友好但并发写容易出现database is locked。加线程锁可以避免同一进程内的并发写冲突。第三search方法自己实现了一个简单的关键词打分逻辑命中content得 2 分命中metadata得 1 分按分数和更新时间排序。这个策略在数据量小的时候完全够用生产环境建议替换为向量检索。4.2 MCP Server 入口接下来创建 MCP Server 入口把MemoryStore封装成 MCP 工具。完整代码如下# memento_server.py import json import os from mcp.server.fastmcp import FastMCP from memento_store import MemoryStore store MemoryStore(os.getenv(MEMENTO_DB_PATH, memento.db)) mcp FastMCP(memento) mcp.tool def memory_store( agent: str, content: str, namespace: str default, memory_type: str semantic, metadata: str {}, ) - str: 将一条记忆写入共享记忆库。 try: meta json.loads(metadata or {}) except json.JSONDecodeError: meta {raw: metadata} memory_id store.store( agentagent, contentcontent, namespacenamespace, memory_typememory_type, metadatameta, ) return json.dumps({ok: True, memory_id: memory_id}, ensure_asciiFalse) mcp.tool def memory_search( query: str, namespace: str default, agent: str , limit: int 10, ) - str: 搜索与 query 相关的记忆。 rows store.search( queryquery, namespacenamespace, agentagent or None, limitlimit, ) return json.dumps(rows, ensure_asciiFalse, indent2) mcp.tool def memory_get(memory_id: str) - str: 根据 memory_id 获取一条记忆详情。 row store.get(memory_id) if row is None: return json.dumps({ok: False, error: not found}, ensure_asciiFalse) return json.dumps(row, ensure_asciiFalse) mcp.tool def memory_update(memory_id: str, content: str , metadata: str ) - str: 更新一条记忆的内容或元信息。 meta None if metadata: try: meta json.loads(metadata) except json.JSONDecodeError: meta {raw: metadata} ok store.update( memory_id, contentcontent or None, metadatameta, ) return json.dumps({ok: ok}, ensure_asciiFalse) mcp.tool def memory_revoke(memory_id: str) - str: 逻辑删除一条记忆。 ok store.revoke(memory_id) return json.dumps({ok: ok}, ensure_asciiFalse) mcp.tool def memory_stats(namespace: str default) - str: 查看某个命名空间的记忆统计信息。 return json.dumps(store.stats(namespace), ensure_asciiFalse) if __name__ __main__: mcp.run(transportstdio)FastMCP会自动根据函数签名生成 JSON Schema并注册为 MCP 工具。每个mcp.tool装饰的函数都对应一个可由 Agent 调用的工具。这里有一个值得注意的设计metadata参数使用字符串而不是 dict因为 MCP 工具的参数在跨进程传输时通常是 JSON 基础类型字符串格式最稳妥。函数内部再json.loads解析成 dict。当你启动这个脚本时它会通过 stdio 与客户端通信。stdio 模式非常适合 MCP 的本地客户端场景例如 Claude Desktop、Cursor 等工具都会以子进程方式拉起 MCP Server。4.3 运行与自测为了不依赖 MCP 客户端可以先直接对存储层做一次自测。创建self_test.py# self_test.py from memento_store import MemoryStore def main(): store MemoryStore(:memory:) memory_id store.store( agentagent-a, content用户偏好使用 Python 编写数据处理脚本, namespaceproject-demo, memory_typesemantic, ) print(stored:, memory_id) results store.search(Python, namespaceproject-demo) print(search results:, results) print(stats:, store.stats(project-demo)) if __name__ __main__: main()运行命令python self_test.py预期输出中stored会打印一条 UUIDsearch results会包含刚才写入的记忆stats会显示总条数为 1。这说明存储层工作正常。接着验证 MCP Server 能否正常拉起。在项目目录执行python memento_server.py正常情况下脚本会保持运行等待 stdio 输入。如果代码和依赖安装正确不会有任何输出也不会报错。如果报错说明环境或依赖有问题。4.4 通过 MCP Inspector 验证要完整验证 MCP 工具是否能被客户端发现和调用可以使用 MCP Inspector 或支持 MCP 的桌面客户端。如果你安装了mcpCLI可以尝试mcp dev memento_server.py不同版本的 MCP CLI 子命令可能不同运行前先查看帮助mcp --help启动后在 Inspector 界面里应该能看到一个名为memento的服务工具列表包含memory_store、memory_search、memory_get、memory_update、memory_revoke、memory_stats。调用memory_store写入一条记忆再用memory_search搜索就可以验证完整链路。如果本地没有 Inspector也可以直接用手边的 MCP 客户端配置项目地址启动服务在客户端的工具列表中确认。下面一节会给出多 Agent 接入的配置示例。5. 多 Agent 接入实践5.1 配置 MCP 客户端让多个 Agent 接入同一个 Memento Server核心就是让它们指向同一个 MCP 配置。以常见的桌面客户端为例MCP 客户端配置通常是一个 JSON 文件格式大致如下{ mcpServers: { memento: { command: python, args: [/absolute/path/to/memento_server.py], cwd: /absolute/path/to/project, env: { MEMENTO_DB_PATH: /absolute/path/to/memento_data/memento.db } } } }这里有几个容易踩坑的细节。第一command建议写上 Python 环境的绝对路径。如果你同时装了好几个 Python 版本直接用python可能会拉起没有安装mcp包的环境。第二args同样建议使用绝对路径否则客户端工作目录不同可能找不到脚本。第三MEMENTO_DB_PATH用绝对路径可以避免“写入后重启数据丢失”的问题。两个不同的 Agent只要都使用这份配置指向同一个数据库路径就能共享同一份记忆。也就是说Agent A 写入project-demo命名空间的记忆Agent B 在相同的namespace下搜索时可以直接检索到。5.2 两个 Agent 协作示例假设现在有 Agent A 负责需求分析Agent B 负责写代码。它们的协作流程可以是Agent A 在 Memento 中写入一条记忆用户要求所有接口统一返回 { code, data, message } 结构。Agent B 在开始编码前调用memory_search查询关键词接口返回。Agent B 得到记忆后按统一结构生成接口代码。这个流程看起来很简单但它解决了一个很实际的问题过去 Agent A 和 Agent B 之间如果需要传递项目规范只能通过共享文件、人工粘贴或数据库直连现在只需要让它们调用同一个记忆工具即可。为了让协作更可控建议每个 Agent 在写入时都用namespace区分业务线例如project-demo、team-shared、agent-private-AGENT_ID。读的权限可以放宽到业务线但写的权限最好做限制。比如 Agent 只能删除自己写入的记忆公共记忆只能由特定管理员 Agent 删除。5.3 命名空间与权限设计权限设计是 Memento 从 Demo 走向生产环境必须跨过的一道门槛。建议按下面三层来设计。第一层是命名空间隔离。不同项目、不同租户使用不同namespace查询和写入都强制带namespace。第二层是 Agent 身份识别。每个 Agent 在调用工具时应当通过环境变量或请求头传递自己的agent_idServer 在写入时自动记录避免客户端伪造。第三层是操作权限。可以用一个简单的 ACL 表规定哪些 Agent 可以写、哪些 Agent 只能读、哪些 Agent 可以执行逻辑删除。初期不需要做得太复杂先把“命名空间”和“agent 字段”这两个基础打好。后续如果出现跨团队共用同一个 Memento 服务再考虑引入独立的认证中心或 API 网关。6. 常见问题与排查思路6.1 MCP 工具注册不上工具注册不上是多 Agent 接入时最头疼的问题。尤其是同时接入多个 MCP Server 时经常出现某个 Server 的工具列表为空或者列表里看不到memory_store。常见原因和解决思路如下表问题现象常见原因解决思路工具列表为空服务名都看不到MCP Server 进程启动即报错用命令行直接运行脚本确认无异常输出能看到服务名但工具列表为空FastMCP 版本不兼容或装饰器未生效升级mcp包检查 SDK 文档中的 FastMCP 用法部分工具注册不上工具函数名重复或参数类型不被 SDK 支持检查工具函数名唯一性避免使用复杂自定义类型同一个客户端接多个 MCP 服务时冲突工具名称全局冲突给 MCP Server 起唯一名称工具名增加前缀在 Codex 中注册不上Codex 客户端对工具数量或命名有额外限制查看客户端日志缩小暴露工具范围保持 tool name 简短唯一排查时先做“最小化验证”写一个hello工具能注册说明 Server 本身没问题问题大概率出在参数模型或依赖版本上如果hello也注册不了优先检查 MCP SDK 版本和 Python 环境。6.2 服务启动失败如果运行python memento_server.py时报ModuleNotFoundError: No module named mcp说明当前 Python 环境没有安装依赖。解决方案是确认客户端实际使用的 Python 路径和命令行环境一致然后在该环境重新安装依赖。如果报语法错误很可能是因为 Python 版本过低。代码中使用了类型注解和 f-string 等特性建议使用 Python 3.10 以上版本。如果报sqlite3.OperationalError: unable to open database file检查数据库路径是否存在且拥有写权限尤其是 Windows 下路径可能包含反斜杠建议使用绝对路径或标准库pathlib处理。6.3 记忆写入后查不到写入成功后查询为空可能是搜索逻辑或命名空间不一致导致的。首先确认写入和查询使用的namespace是否一致其次确认agent参数是否传入最后检查search的关键词是否与记忆内容完全无关。还有一个小细节容易被忽略store方法会把metadata序列化成 JSON 字符串存储搜索时也会对metadata做文本匹配。如果你把内容放在 metadata 里建议用content字段保存主要文本metadata只存标签、来源等结构化信息否则检索效果会偏差。6.4 并发写入冲突与数据膨胀当多个 Agent 同时写入时SQLite 可能报database is locked。解决思路有三个一是打开 WAL 模式在连接建立后执行PRAGMA journal_modeWAL;二是缩短事务长度避免在一个事务里执行过多操作三是如果并发量仍然很大就升级到 PostgreSQL 或 MySQL。数据膨胀方面最直接的办法是给记忆加上metadata.ttl过期时间并定期运行清理任务将过期记忆批量revoke。还可以按namespace设置额度避免单个项目写入无限制的数据量。7. 最佳实践与工程建议7.1 设计原则做 Memento 这类共享记忆服务我建议先把下面三条原则记在心里。第一记忆要分层。不要把用户偏好、项目背景、任务过程全部混在一起。用memory_type区分方便按类型检索也方便后续对不同类型做不同的生命周期管理。第二检索要可解释。关键词匹配虽然简单但至少能让开发者说清楚“为什么返回这条记忆”。如果一上来就接向量库反而可能让问题和排错变得复杂。第三写入要先于检索优化。很多团队上来就死磕检索准确率却忽略了记忆写入不到位的根本问题Agent 根本不知道该在什么时候记录什么。建议先设计“什么内容值得写入”的提示词模板然后再调检索。7.2 安全与合规多 Agent 共享记忆意味着记忆内容可能被多个 Agent 看到安全边界必须提前设计。重要项目建议在 Memento Server 入口增加一层敏感信息过滤在写入前对内容做一次脱敏检查拦截手机号、身份证号、密钥等敏感字段在查询时根据请求方身份限制可见的数据范围。同时开启审计日志。每条记忆的写入、更新、删除都应该记录操作人、操作时间、操作内容。这样即使发生数据异常也能快速定位是哪条 Agent、哪个命名空间出的问题。记忆数据在生产环境中属于用户数据删除要遵循“最小必要”原则能逻辑删除就不要物理删除能保留审计就不要直接清空。7.3 扩展方向当数据规模上来之后可以考虑引入向量检索。思路很简单写入记忆时用 Embedding 模型生成向量存储到向量数据库查询时把 query 转成向量做相似度检索。关键词匹配和向量检索可以并存采用“向量召回 关键词精排”的混合策略。另一个扩展方向是记忆折叠。随着时间推移同一命名空间下会产生大量细碎记录例如“用户今天说要改 A”“用户明天说要改 A”。如果不处理检索时会出现大量重复内容。可以考虑定期用大模型把历史记忆做归纳压缩合并成更高层次的长期记忆。还有分层缓存。热记忆放 Redis冷记忆放 SQLite 或 PostgreSQL可以降低访问延迟同时保证大数据量下的可扩展性。不过这些优化都要等基础链路跑通之后再逐步加入不要一开始就追求全套架构。8. 总结与学习路线这篇文章从多 Agent 的“集体失忆”问题出发介绍了 Memento 的核心定位并给出了一个基于 MCP 的完整实现。通过阅读和实践你应该掌握了以下内容理解多 Agent 为什么需要共享持久记忆了解 MCP 与 Agent Skill 的区别以及 MCP Server 的基本工作方式掌握 Memento 服务的分层架构和记忆数据模型能够用 Python 实现存储层和 MCP Server并暴露memory_store、memory_search等工具知道如何让多个 Agent 通过 MCP 客户端配置接入同一份记忆掌握常见问题排查思路比如工具注册不上、数据丢失、并发写入冲突等。下一步学习建议按这条路线推进先深入读一遍 MCP 协议规范搞清楚工具发现、工具调用、资源暴露的底层消息格式再学习 Agent Skill 与 MCP 的配合方式思考哪些能力应该内建在 Agent 里哪些应该外置成服务接着研究 RAG 与向量数据库把记忆检索从关键词匹配升级到语义召回最后再考虑多 Agent 编排框架让不同 Agent 之间的记忆协作更加自动化。如果你最近正在折腾多 Agent 协同先别急着把记忆层做得花里胡哨。把最小闭环跑通一个 SQLite 文件、一个 MCP Server、两个 Agent足够验证这个思路是否适合你的场景。等跑通了再逐步加上向量检索、权限和监控。希望这篇文章能帮你省下一点试错时间也祝你的 Agent 都能真正“记住该记住的事”。