当多个 AI Agent 开始协作完成复杂任务时最先暴露的问题往往不是模型能力不够而是“记忆”出了问题每个 Agent 各记各的上下文窗口很快被塞满任务一重启重要信息全部丢失。最近在调研和落地多 Agent 系统时接触到了 Memento 这个开源项目它专门解决多 Agent 场景下的共享持久化记忆问题并且通过 MCP 协议对外提供标准化访问能力。这篇教程会从概念、核心设计、环境准备、代码实现到排错思路完整展开帮助你快速理解和落地这套方案。1. 为什么 AI Agent 需要共享持久化记忆1.1 单 Agent 的“失忆”困境先来看一个常见的业务场景一个团队同时运行着三个 AI Agent分别负责需求分析、代码生成、测试用例编写。这三个 Agent 彼此独立但处理的是同一个项目。项目启动时每个 Agent 都要重新读一遍需求文档代码 Agent 发现了某个模块的命名规范测试 Agent 完全不知道中途重启或更换模型后之前所有上下文全部清零。这个问题的根源在于大模型本身是无状态的。每一次对话请求都是独立计算模型只能依赖当前请求携带的上下文来理解任务。我们通常说的“记忆”本质上是把历史信息塞进 Prompt 里。但 Prompt 长度有限成本高而且不同 Agent 之间无法共享。1.2 持久化记忆要做的事持久化记忆系统要解决的核心问题有三个记忆要能跨会话保存不能因为 Agent 重启、模型切换而丢失。记忆要能被多个 Agent 共享而不是某个 Agent 的私有状态。记忆要能被高效检索只能写入不能查等于没有记忆。Memento 就是这样一个服务它把 Agent 的记忆从“上下文里的临时文字”变成“系统外部的持久化数据”并且提供结构化的写入和查询接口。这样多个 Agent 就像是共享同一个“团队知识库”每个人都能往里面存东西也能查到别人存的内容。1.3 与普通数据库的区别有人可能会问这不就是一个数据库吗直接用 Redis 或 MySQL 不就行了确实底层存储离不开数据库但 Memento 这类系统在数据库之上做了很多 Agent 场景特有的设计数据模型面向“实体-关系-观察”设计而不是面向业务表结构设计。写入时支持自动抽取实体和关系不需要业务方手动建模。查询时结合语义检索和结构化过滤而不是单纯的关键词匹配。对外通过 MCP 标准协议暴露能力任何支持 MCP 的客户端都能接入。换句话说Memento 是“为 Agent 设计的记忆中间件”数据库只是它的存储引擎。2. MCP 协议Memento 的标准化访问入口2.1 MCP 是什么MCPModel Context Protocol模型上下文协议是近年来 AI 工具链中非常重要的一个开放协议。它解决的是“AI 应用如何标准化地连接外部工具和数据源”的问题。在 MCP 的架构中有三个核心角色角色作用常见例子HostAI 应用或 IDE是用户交互的入口Claude Desktop、Cursor、Codex、自研 Web 应用Server暴露工具、资源、提示词的最小服务单元文件系统服务、数据库服务、Memento 服务ClientHost 内部与 Server 建立连接的协议客户端Host 内置的 MCP Client 组件一次完整的 MCP 调用流程是Host 中的 Client 通过 stdio 或 HTTP 与 Server 建立连接获取 Server 暴露的工具列表然后由模型决定调用哪个工具Client 把调用请求转发给 ServerServer 执行后返回结构化结果。2.2 MCP 提供的三类能力MCP 协议定义了三种核心原语理解它们对后续配置很重要Tools可执行的函数模型通过调用它来执行动作比如“写入记忆”“搜索记忆”。工具调用是模型主动发起的。Resources可读取的数据资源比如某个文档、某张表。资源通常由模型根据用户请求决定是否读取。Prompts可复用的提示词模板用于标准化常见任务的输入结构。Memento 主要通过 Tools 对外提供记忆读写能力。这意味着任何支持 MCP 的客户端只要能加载一个 Memento Server就能让模型自动获得“记忆工具”。2.3 为什么 Memento 选择 MCP在 Memento 之前很多 Agent 框架都有自己的记忆模块但接口各自为政。A 框架的记忆函数到了 B 框架里完全不能用。MCP 的出现改变了这个局面只要实现一套 MCP Server就能同时被 Claude Desktop、Codex、Cursor、自研应用等众多客户端复用。这种“一次接入处处可用”的特性让 Memento 的定位变得很清晰它不绑定任何特定框架而是作为基础设施存在。最近社区里关于 MCP 的讨论非常多从蓝湖 MCP、Figma MCP 到 Playwright MCP覆盖了设计稿、浏览器自动化、数据库等场景Memento 走的正是同一条路线只是它聚焦在“记忆”这个 AI Agent 最关键的环节上。3. Memento 核心设计拆解3.1 记忆分层从工作记忆到长期记忆Memento 在记忆设计上参考了认知科学的“工作记忆”和“长期记忆”概念。工作记忆负责保存当前任务需要的临时信息比如正在处理的对话内容、临时变量长期记忆负责保存跨任务、跨会话的稳定知识比如项目规范、用户偏好、历史决策。这种分层的好处是模型在每次请求时只需要加载与当前任务相关的工作记忆长期记忆按需检索避免把所有历史信息全部塞进上下文既节省 Token又提升响应质量。3.2 数据模型实体、关系与观察Memento 长期记忆的数据模型采用“知识图谱”形式核心是三个概念实体Entity一个具体的人、事、物、概念。比如“用户张三”“项目 Alpha”“支付模块”。关系Relation实体之间的有向关联。比如“张三负责Alpha”“Alpha依赖支付模块”。观察Observation针对实体描述的动态事实。比如“张三在 2025 年 6 月调整了支付超时时间为 30 秒”。这种模型的优势在于灵活。业务上不需要预先设计严格的表结构实体可以动态新增关系可以随时建立观察记录天然带时间维度非常适合 Agent 记忆这种“边用边写、边写边补”的场景。3.3 核心工作机制Memento 的核心机制可以概括为写入、检索、维护三部分。写入时Agent 把一段原始信息通过工具提交给 MementoMemento 会解析其中的实体和关系抽取关键观察去重后写入底层存储。检索时Agent 提交查询语句Memento 通过关键词匹配和语义相关度筛选出最相关的实体和观察记录返回给模型。维护阶段Memento 负责处理实体合并、关系更新、过时观察清理等操作保证知识图谱不会无限膨胀。从整体架构看Memento 的链路可以分为MCP 接入层负责与外部客户端通信、记忆编排层负责解析、路由、编排记忆组件的调用、数据访问层负责与底层存储交互。这种分层让每一部分都可以单独扩展比如更换存储引擎时不需要动 MCP 接入层。4. 环境准备与项目结构4.1 运行环境说明本文示例以 Python 3.10 以上版本为例重点演示 Memento 这类共享记忆服务的实现思路。版本需要根据你的项目实际情况调整核心依赖是 FastMCP 和 SQLitePython 内置无需额外安装。需要安装的 Python 包pip install mcp[cli] fastmcp如果你的机器上同时存在多个 Python 环境建议使用虚拟环境python3 -m venv .venv source .venv/bin/activate pip install mcp[cli] fastmcpWindows 下激活命令为.venv\Scripts\activate4.2 项目结构规划为了便于后续扩展我们把项目拆成以下结构memento-demo/ ├── server.py # MCP Server 入口暴露记忆工具 ├── memory_store.py # 记忆存储层封装 SQLite 操作 ├── memory_parser.py # 实体/关系/观察解析辅助模块 ├── requirements.txt # 依赖清单 └── data/ └── memory.db # SQLite 数据文件首次运行后生成这里的核心思路是MCP 接入层负责协议通信存储层负责数据落盘解析层负责把自然语言信息结构化。三者分离后续替换组件成本最低。5. 完整实战构建一个支持 MCP 访问的共享记忆服务5.1 创建数据存储层先实现最底层的memory_store.py负责 SQLite 的建表和增删改查操作。这个文件不涉及 MCP任何模块都可以直接调用它。# 文件路径memory_store.py import sqlite3 import json from datetime import datetime from pathlib import Path DB_PATH Path(__file__).parent / data / memory.db def get_connection(): 创建数据库连接开启外键约束支持。 conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) return conn def init_db(): 初始化数据表。 DB_PATH.parent.mkdir(parentsTrue, exist_okTrue) conn get_connection() conn.executescript( CREATE TABLE IF NOT EXISTS entities ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, entity_type TEXT DEFAULT generic, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS relations ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER NOT NULL, target_id INTEGER NOT NULL, relation_type TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(source_id, target_id, relation_type), FOREIGN KEY (source_id) REFERENCES entities(id) ON DELETE CASCADE, FOREIGN KEY (target_id) REFERENCES entities(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS observations ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity_id INTEGER NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, FOREIGN KEY (entity_id) REFERENCES entities(id) ON DELETE CASCADE ); ) conn.commit() conn.close() def add_entity(name: str, entity_type: str generic) - int: 新增实体已存在则返回现有实体 id。 conn get_connection() now datetime.now().isoformat() conn.execute( INSERT OR IGNORE INTO entities (name, entity_type, created_at) VALUES (?, ?, ?), (name, entity_type, now), ) row conn.execute(SELECT id FROM entities WHERE name ?, (name,)).fetchone() conn.commit() conn.close() return row[id]这里有几个细节值得注意INSERT OR IGNORE用于幂等写入实体已存在时不会报错UNIQUE约束保证同一对实体之间不会重复建立同类型关系避免知识图谱出现冗余边。5.2 实现记忆写入与检索继续完善memory_store.py增加关系和观察的写入方法以及面向 Agent 的检索方法。# 文件路径memory_store.py追加写入与检索方法 def add_relation(source_name: str, target_name: str, relation_type: str): 在两个实体之间建立关系。 source_id add_entity(source_name) target_id add_entity(target_name) conn get_connection() now datetime.now().isoformat() conn.execute( INSERT OR IGNORE INTO relations (source_id, target_id, relation_type, created_at) VALUES (?, ?, ?, ?), (source_id, target_id, relation_type, now), ) conn.commit() conn.close() def add_observation(entity_name: str, content: str): 为指定实体添加一条观察记录。 entity_id add_entity(entity_name) conn get_connection() now datetime.now().isoformat() conn.execute( INSERT INTO observations (entity_id, content, created_at) VALUES (?, ?, ?), (entity_id, content, now), ) conn.commit() conn.close() def search_entities(keyword: str): 按关键词检索实体并附带相关观察和关系。 conn get_connection() rows conn.execute( SELECT DISTINCT e.id, e.name, e.entity_type FROM entities e LEFT JOIN observations o ON o.entity_id e.id WHERE e.name LIKE ? OR o.content LIKE ? ORDER BY e.id DESC LIMIT 20 , (f%{keyword}%, f%{keyword}%), ).fetchall() result [] for row in rows: entity_id row[id] observations conn.execute( SELECT content, created_at FROM observations WHERE entity_id ? ORDER BY created_at DESC, (entity_id,), ).fetchall() relations conn.execute( SELECT e2.name, r.relation_type FROM relations r JOIN entities e2 ON e2.id r.target_id WHERE r.source_id ? , (entity_id,), ).fetchall() result.append( { entity: row[name], type: row[entity_type], observations: [dict(o) for o in observations], relations: [dict(rel) for rel in relations], } ) conn.close() return result检索方法返回的是结构化数据模型拿到后可以直接理解实体之间是谁负责谁、谁依赖谁、有哪些历史观察记录。注意这里用了LIKE进行关键词匹配实际项目中可以叠加向量检索提升召回率。5.3 封装 MCP Server接下来是核心部分通过 FastMCP 把上面的存储方法包装成 MCP 工具。server.py将作为 MCP Server 的入口。# 文件路径server.py from fastmcp import FastMCP import memory_store import memory_parser # 创建 MCP Server 实例名称会在客户端工具列表中显示 mcp FastMCP(memento-demo) mcp.tool() def add_entities(entities: list[str]) - str: 批量添加实体。entities 是实体名称列表。 added [] for name in entities: eid memory_store.add_entity(name) added.append({name: name, id: eid}) return str(added) mcp.tool() def add_relation(source: str, target: str, relation_type: str) - str: 在两个实体之间建立关系。 memory_store.add_relation(source, target, relation_type) return frelation added: {source} -[{relation_type}]- {target} mcp.tool() def add_observation(entity: str, content: str) - str: 为指定实体添加观察记录。 memory_store.add_observation(entity, content) return fobservation added to {entity} mcp.tool() def search_entities(keyword: str) - str: 搜索实体及其相关观察和关系。 result memory_store.search_entities(keyword) if not result: return no entities found return str(result) mcp.tool() def search_observations(keyword: str) - str: 跨实体搜索观察记录。 result memory_store.search_observations(keyword) if not result: return no observations found return str(result) if __name__ __main__: memory_store.init_db() mcp.run()memory_parser.py是一个辅助模块目前可以留空或放一些简单的文本处理函数。在实际项目中这里通常会接入大模型做实体抽取把一段自然语言文本自动解析成实体、关系和观察。5.4 启动 MCP Server在项目根目录下运行python server.pyFastMCP 默认会以 stdio 方式启动打印类似Started MCP server with tools: add_entities, add_relation...的日志。看到这行日志说明 Server 已经准备好了。如果你想快速调试工具效果可以使用 MCP Inspectormcp dev server.py它会启动一个本地调试面板你可以在浏览器中打开http://localhost:6274进入 Tools 页面手动调用add_observation和search_entities来验证功能。5.5 用 Python 客户端测试除了 MCP Inspector我们也可以写一个简单的 Python 客户端来测试 MCP 调用。# 文件路径test_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[server.py], cwd., ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 写入记忆 result await session.call_tool( add_observation, arguments{entity: 支付模块, content: 超时时间调整为 30 秒}, ) print(write result:, result) # 检索记忆 result await session.call_tool( search_entities, arguments{keyword: 支付}, ) print(search result:, result) if __name__ __main__: asyncio.run(main())运行客户端python test_client.py预期输出中会包含写入成功的结果以及搜索返回的实体、观察记录。这证明我们的 MCP Server 已经可以被外部程序通过标准协议访问了。6. 多 Agent 共享记忆的落地场景6.1 多个客户端接入同一个 ServerMemento 的共享特性体现在多个客户端可以同时连接同一个 MCP Server而 Server 底层使用同一个 SQLite 数据库。也就是说Agent A 写入的记忆Agent B 可以立刻检索到。比如在 Claude Desktop 配置文件中加入{ mcpServers: { memento: { command: python, args: [/path/to/memento-demo/server.py] } } }在 Codex 或其他支持 MCP 的客户端中配置方式类似只需要指向同一个server.py。这样不同客户端中的 Agent 就共享了同一个记忆库。6.2 多进程并发写安全SQLite 支持多进程读写但在高并发写入场景下可能出现database is locked错误。Memento 这类记忆服务通常不需要极高的写入吞吐SQLite 完全可以胜任。如果后续并发量上来有两种扩展路线继续使用 SQLite但开启 WAL 模式减少读写锁冲突。替换底层存储为 PostgreSQL保留上层 MCP 接口不变。这正好体现分层设计的好处替换存储层不影响 Agent 侧的任何代码。6.3 记忆隔离与权限控制多 Agent 共享记忆不意味着所有 Agent 都能看到所有内容。实际项目中可以按 Agent 角色或项目维度做隔离。一种简单的做法是在数据表中增加scope字段写入时标记记忆归属检索时按当前 Agent 的 scope 过滤。如果使用 Memento 官方版本还可以结合 MCP 的资源权限机制做访问控制。生产环境中使用时务必遵循最小权限原则每个 Agent 只授予它完成工作所需的最小记忆访问范围避免敏感信息跨 Agent 泄露。7. 常见问题与排查思路在实际使用 MCP 和 Memento 的过程中最常遇到的问题集中在连接、注册和检索三个方面。问题现象常见原因解决思路工具在客户端中注册不上MCP Server 启动失败或输出格式不符合协议先用mcp dev server.py单独调试客户端提示找不到 MCP Server路径配置错误或 Python 环境不一致确认command指向正确的 Python 解释器调用工具时报Connection closedstdio 通信中断可能是代码中误用了print检查 Server 代码删除调试用print写入成功后搜索不到数据未提交或检索关键词不匹配检查 commit尝试使用部分关键词多个客户端同时写入时报锁SQLite 并发写冲突开启 WAL 模式或切换 PostgreSQL中文内容乱码终端编码或 JSON 序列化问题检查终端编码设置Python 3 默认 UTF-8 通常没问题模型不主动调用记忆工具工具描述不够明确优化 tool 的 description让模型理解何时该调用7.1 工具注册不上的问题这里重点说一下工具注册问题。很多同学在 Codex 或 Cursor 中配置 MCP Server 时遇到“工具总是注册不上”根本原因通常是 Server 启动时抛出了异常但客户端无法看到详细的错误日志。排查顺序建议如下先在命令行手动运行python server.py确认能正常输出 MCP 启动日志。使用mcp dev server.py打开调试面板确认工具列表里能看到add_observation。检查客户端的配置文件路径确保args中写的是绝对路径。检查 Python 环境客户端进程使用的可能是另一个 Python 解释器需要确认依赖已安装到该环境中。7.2 stdio 传输的调试技巧MCP 的 stdio 传输模式中Server 的标准输出被协议占用不能在代码里随便print调试信息否则会破坏协议通信。调试时建议使用logging模块把日志输出到标准错误流或独立日志文件。import logging logging.basicConfig( levellogging.DEBUG, filenamememento.log, filemodea, format%(asctime)s %(levelname)s %(message)s, )这样既不影响协议通信又能保留运行日志排查问题时非常有用。8. 最佳实践与工程建议8.1 记忆数据模型设计使用 Memento 时不要一上来就追求复杂的关系模型。推荐先按以下顺序演进第一阶段只存观察记录Observation检索靠关键词。第二阶段引入实体Entity把观察挂到实体下检索按实体聚合。第三阶段引入关系Relation建立实体之间的语义连接。过早建模会增加解析器的复杂度而且 Agent 写入时对关系的定义往往不稳定。先用简单模型跑通再根据实际检索效果逐步完善。8.2 检索质量优化Memento 的核心价值在于“能查到”。提高检索质量有几个方向写入时尽量结构化让 Agent 在调用工具前先拆分实体和观察而不是整段文本塞入。多路召回关键词检索 向量语义检索结合关键词负责精确匹配向量负责语义相关。时间衰减观察记录加上时间权重较新的观察优先返回避免旧信息干扰模型判断。实体去重同一实体可能存在多个名称变体写入时做归一化处理。8.3 生产环境部署注意把 Memento 部署到生产环境前需要关注以下问题持久化存储必须配置定期备份SQLite 文件直接复制或使用 VACUUM 方式导出。数据库文件路径使用绝对路径或环境变量注入避免部署时路径错乱。如果通过 HTTP 暴露 MCP 服务必须配置认证和访问控制不能裸奔在公网。监控工具调用频率和失败率为 MCP Server 增加健康检查接口。写入操作尽量走异步队列避免 Agent 等待数据库写入影响响应速度。8.4 避免记忆污染共享记忆最大的风险是“错误记忆”被多个 Agent 复用。一个 Agent 写入了错误信息其他 Agent 检索到后可能进一步放大错误。建议在记忆系统中增加“来源标记”记录每条观察是由哪个 Agent 写入的对重要决策类记忆可以增加审核机制必要时人工确认后再写入。9. 总结与学习路线这篇文章围绕 Memento 的核心目标——“为多个 AI Agent 提供共享持久化记忆并通过 MCP 访问”——从概念到实战做了一个完整的梳理。你可以重点回顾以下几个关键点记忆分层的思路工作记忆和长期记忆分离避免上下文无限膨胀。数据模型的灵活性实体、关系、观察三者结合能覆盖大部分 Agent 记忆场景。MCP 的价值一次实现、多处复用的标准化接入能力。项目分层的工程意义MCP 接入层、记忆编排层、数据访问层解耦便于替换和升级。接下来你可以继续深入研究的方向包括语义检索与向量数据库的结合让记忆检索更准确。实体解析与大模型抽取能力的集成让写入更自动化。MCP 协议中 Resources 和 Prompts 的更多用法。多 Agent 协作框架中如何设计记忆的权限与共享边界。建议你先把本文的示例代码跑通然后尝试改造成自己的业务场景比如把“支付模块超时时间”换成“用户偏好”“项目规范”等实际数据。只有在真实数据上反复调整写入和检索逻辑才能真正理解 Memento 这类记忆中间件的设计取舍。如果这篇文章对你有帮助可以收藏备用后续再继续分享 MCP 生态中的更多实战经验。