在实际 AI Agent 工程里一个经常被低估的问题是记忆。模型本身没有持久状态每次对话、每个任务完成后之前的上下文就随着会话结束被丢弃了。单个 Agent 还可以通过把历史写入外部存储来缓解但当多个 Agent 同时参与一个项目各自维护自己的会话历史时问题会指数级放大Agent A 记录的架构决策Agent B 看不到Agent C 上一轮实验的结论下次冷启动时完全丢失。Memento 就是为这类场景设计的共享、持久化记忆层它把 Agent 的记忆统一收口到独立存储中并通过 MCPModel Context Protocol暴露给多个 AI Agent 使用。下面的内容会从多 Agent 记忆问题的根源讲起说明 MCP 为什么适合做存储访问层然后给出一个最小可运行的 Memento MCP Server 实现覆盖写入、检索、更新、删除和命名空间隔离最后补充 MCP 工具注册不上、记忆检索不到等高频问题的排查链路以及生产环境落地的检查清单。1. 先理解 Memento 解决什么问题1.1 单 Agent 的上下文限制与记忆缺失大语言模型每次调用本质上是无状态的。输入一段文本模型根据权重预测输出调用结束之后这段对话对模型来说就不存在了。所谓“模型会记住上下文”靠的是把历史消息拼进新的请求里而请求长度受上下文窗口限制。一旦窗口被占满或者会话结束旧信息就会被挤出或清空。产品层面的“记忆”不是模型自带的能力而是工程上额外设计的外部存储。常见做法包括把关键事实写入数据库或键值存储。把长历史的摘要重新注入后续请求。用向量库保存语义片段按相关性召回。这些做法都是把模型的短期工作记忆转存到应用侧的长期存储里。Memento 这个名字来自经典设计模式中的备忘录模式在不破坏对象封装的前提下把对象内部状态保存成快照之后可以恢复。AI Agent 的记忆系统做的是同一件事保存 Agent 在运行过程中产生的事实、决策和结论之后可以被同一个 Agent 或其他 Agent 恢复使用。1.2 多 Agent 场景为什么更需要共享记忆多 Agent 系统的难点不在于单个 Agent 能写多少代码而在于多个 Agent 之间如何传递状态。Agent 和 Agent 之间没有天然共享的上下文如果不引入外部记忆就会出现三类典型问题任务拆解后信息断层。规划 Agent 把项目拆成清单执行 Agent 却看不到清单的完整背景只能重新猜测。重复劳动。Agent A 已经确认了数据库选型和连接方式Agent B 又从零调研一遍给出冲突结论。无法恢复中断。一个会话被关掉所有中间产物、踩坑记录、结论都消失下次只能重新开始。共享记忆在工程上就是把 Agent 之间的通信从“临时口述”改成“持久化书面记录”。规划 Agent 写入任务状态执行 Agent 读取任务状态测试 Agent 写回验证结果评审 Agent 查看全部历史。每个 Agent 不需要知道对方内部发生了什么只需要读写同一个记忆空间。最常见的落地场景包括编码协作多个 Agent 分工实现不同模块共享架构决策、接口约定和环境信息。数据调研多个 Agent 并行收集资料把阶段性结论写入公共记忆最后由汇总 Agent 整合。运维处置故障处理 Agent 把排查步骤和结论写入记忆后续 Agent 在处理同类问题时直接复用。1.3 Memento 在系统架构中的定位Memento 在架构上处于 Agent 和存储之间扮演一个标准化访问层。它不是一个模型不是推理框架而是一个数据服务接收 Agent 发来的记忆写入请求经过处理落到持久化存储再响应 Agent 的检索请求。一个完整的 Memento 服务由四部分组成写入路径Agent 调用记忆写入接口传入命名空间、内容、标签和过期时间。存储层负责持久化保证重启后数据不丢。读取路径Agent 提交查询检索结果经过排序和过滤后返回。访问协议层通过 MCP 把记忆能力封装成标准工具让不同 Agent 客户端都能接入。这种设计的关键收益是解耦。Agent 不直接连接数据库不需要知道表结构也不关心存储用的是 SQLite 还是向量库。记忆系统独立演进存储可以替换检索逻辑可以升级Agent 侧几乎不用改动。2. MCP 为什么适合做 Agent 存储访问层2.1 MCP 协议解决什么问题MCP全称 Model Context Protocol是一套用于连接 AI 模型与外部工具、数据源的开放协议。它解决的问题非常直接如果没有统一协议每个 AI 应用都要为每个工具单独写一套接入代码工具方也要为每个客户端适配不同接口形成 N 乘 M 的集成成本。MCP 定义了 Host、Client、Server 三层结构把工具、数据资源和提示词模板统一成标准能力客户端按同一套规则发现和调用。目前 MCP 生态已经覆盖了大量场景从浏览器自动化、数据库查询、设计工具到数据分析、测试执行等。可以看到一个明显的趋势工具接入正在从“每个 Agent 写一套集成”变成“每个工具提供一个 MCP Server”。对于 Memento 这类记忆系统MCP 的价值尤其明显。记忆服务不是给某一个 Agent 用的而是要同时服务多个不同客户端的 Agent。提供 MCP Server 之后Claude Desktop、Codex、各种支持 MCP 的 IDE 和自研 Agent 框架都可以通过同一种协议接入不需要为每种客户端单独开发集成层。2.2 Tools、Resources、Prompts 三种能力如何分工MCP 协议定义了三种核心能力原语理解它们的分工才能正确设计记忆服务的对外接口。MCP 能力定位Memento 中的用法Tools让模型执行一个动作有输入参数和返回结果remember、recall、update、forget 都作为 Tool 暴露Resources以 URI 方式暴露可读数据用于向模型注入上下文暴露记忆统计、命名空间列表、记忆导出文件Prompts预置的用户操作模板引导模型按固定流程处理提供“记录项目决策”“汇总近期记忆”等模板在 Memento 实现里主要使用的是 Tools。因为记忆操作是典型的读写动作模型需要传入参数并拿到结构化结果。Resources 适合做只读数据暴露比如把某个项目的全部记忆导出成一个文本资源让模型在启动时直接读取。Prompts 则适合把常见操作封装成套路降低模型调用工具时的不确定性。2.3 MCP、Agent Skill、RAG、Prompt 到底有什么区别这几个概念经常被放在一起讨论但它们不在同一个层次。MCP 是协议解决“能力怎么被访问”的问题。它规定工具如何声明、如何调用、如何传输结果。Agent Skill 是 Agent 侧的技能封装通常包含一段执行流程、提示词和可能调用到的工具组合解决“执行什么步骤”的问题。RAG 是检索增强生成的方法论解决“如何用外部知识提升生成质量”的问题。Prompt 则是纯文本指令解决“怎么向模型描述任务”的问题。四者的关系可以用一个例子说明。一个 Agent 接收到“总结项目本周进展”的任务它的 Skill 定义了这个任务的执行流程先调用 Memento 的 recall 工具拉取本周记忆再调用某个 MCP 数据库工具查询提交记录然后把结果拼接成结构化 Prompt 交给模型生成总结。这里 MCP 提供工具访问能力RAG 是 recall 背后的检索逻辑Prompt 是最后组织输入的方式。概念层次核心问题典型载体MCP协议层工具和数据如何被统一访问MCP Server、Tools、ResourcesAgent Skill能力封装层完成一件任务需要哪些步骤流程、提示词、工具组合RAG方法论如何用检索到的知识增强生成向量库、召回排序、注入模板Prompt表达层如何把任务描述清楚指令文本、示例、约束条件在开发 Memento 时可以在 MCP Server 内实现 RAG 风格的检索逻辑也可以在 Agent Skill 中编排记忆操作但它们解决的问题不同不能互相替代。3. 设计一个共享持久化记忆系统的核心模块3.1 记忆数据模型设计共享记忆系统第一步是定义“一条记忆长什么样”。记忆不是聊天记录它应该具备可检索、可过滤、可过期的结构化属性。一个最小可用的记忆记录包含以下字段字段类型说明idstring记忆唯一标识建议使用 UUIDnamespacestring命名空间用于逻辑隔离不同项目或团队agent_idstring写入该记忆的 Agent 标识用于来源追踪contentstring记忆正文通常是事实、决策、结论tagsarray标签用于分类和过滤created_atnumber创建时间Unix 时间戳updated_atnumber最后更新时间ttl_secondsnumber可选过期时间为空表示永不过期在 SQLite 里存储时一条记忆对应的记录大致如下{ id: f7c3a41e-8b2f-4de1-9c67-2e6f0b0c3a12, namespace: project-x, agent_id: planner-01, content: 数据库选型项目采用 PostgreSQL 16连接串保存在配置中心禁止写入代码仓库。, tags: [database, decision], created_at: 1710000000.123, updated_at: 1710000000.123, ttl_seconds: 2592000 }这里 namespace 和 agent_id 是两个容易被忽略但非常重要的字段。namespace 决定记忆属于哪个逻辑空间多项目共用一个 Memento 服务时没有 namespace 会导致数据互相污染。agent_id 用于回答“这条记忆是谁写的”它既可以是写入方的身份标识也可以作为读取时的过滤条件。3.2 写入路径记忆写入不只是往数据库插一行它需要经过一条固定链路参数校验content 不能为空namespace 不能为空tags 必须是字符串数组。规范化统一内容格式比如去除首尾空白、统一时间格式。去重判断同一命名空间、相似内容在短时间内重复写入时优先更新旧记录而不是插入新记录。持久化把记录写入存储确保事务提交成功后接口才返回。返回结果返回记忆 id方便后续更新和删除。写入接口设计成幂等会比较稳妥。所谓幂等是指同一个写入请求执行多次最终状态和执行一次一致。在记忆场景里Agent 可能因为网络重试而重复调用写入接口如果每次都插入新记录会产生大量冗余记忆。注意记忆写入接口的幂等性要在设计阶段考虑。最常见的做法是让调用方传入可选的外部请求 id服务端根据请求 id 判断是否已经处理过避免重试导致重复写入。3.3 读取路径读取路径决定 Agent 拿到的是“有用的记忆”还是“一堆噪声”。一次完整的检索流程包括解析查询参数命名空间、查询文本、过滤条件、返回条数。过滤先按 namespace 过滤再按 agent_id、过期时间过滤。排序有查询文本时按相关性排序没有查询文本时按更新时间倒序。截断返回 top-k 结果避免一次性把大量记忆塞进上下文。相关性排序有两种典型实现。最小实现是关键词重叠度计算查询词和记忆内容的共现比例生产级实现是把内容映射成向量用余弦相似度排序。向量检索对“意思相近但用词不同”的查询更友好但需要额外引入 embedding 模型和向量索引。读取路径还需要注意一个原则返回的每条记忆都应该尽量独立完整。Agent 的上下文窗口有限如果一条记忆本身写得模糊模型拿到后还要反复追溯记忆的价值就大打折扣。3.4 记忆生命周期记忆不是越久越好。过期的架构决策、废弃的环境信息、已经修完的故障记录长期留在记忆里反而会误导 Agent。完整生命周期包括创建Agent 写入新记忆。更新记忆内容变化时修改并刷新 updated_at。过期超过 ttl 后被排除出检索结果。归档对有价值但不再活跃的记忆可以移入冷存储。删除显式清理错误或敏感记忆。ttl 的设计要结合业务场景。稳定性高的信息比如“项目使用的技术栈”可以设很长甚至不过期时效性强的信息比如“当前线上告警状态”应该设短过期时间。默认值常见做法是对所有记忆设置一个基础 TTL再允许调用方按场景覆盖。4. 从零实现一个最小可运行的 Memento MCP Server4.1 环境准备示例使用 Python 实现依赖尽量少方便在本地快速跑通。需要准备的环境如下组件版本要求用途Python3.10 或更高运行示例代码mcp 包最新稳定版提供 MCP Server 开发能力Node.js可选运行 MCP Inspector 调试工具安装依赖pip install mcp如果之后要接入向量检索可以追加安装 embedding 相关依赖。示例阶段先用标准库和 SQLite 完成闭环避免一开始引入过多组件。4.2 项目结构建议目录结构如下memento/ ├── requirements.txt ├── memory_store.py ├── memento_server.py └── data/ └── memento.dbmemory_store.py 负责存储层memento_server.py 负责把存储能力包装成 MCP 工具。分开写的好处是存储逻辑可以独立测试不依赖 MCP 环境。4.3 存储层SQLite 文件数据库保证持久化SQLite 对原型和中小规模场景足够数据落在单个文件里重启不丢。核心实现如下import json import sqlite3 import time import uuid class MemoryStore: 基于 SQLite 的共享记忆存储实现。 def __init__(self, db_path: str): self.db_path db_path self._init_db() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self) - None: with self._connect() as conn: conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, namespace TEXT NOT NULL, agent_id TEXT NOT NULL DEFAULT any, content TEXT NOT NULL, tags TEXT NOT NULL DEFAULT [], created_at REAL NOT NULL, updated_at REAL NOT NULL, ttl_seconds REAL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_memories_ns_time ON memories(namespace, updated_at DESC) ) def remember( self, namespace: str, content: str, agent_id: str any, tags: list[str] | None None, ttl_seconds: float | None None, ) - str: memory_id str(uuid.uuid4()) now time.time() tags_json json.dumps(tags or [], ensure_asciiFalse) with self._connect() as conn: conn.execute( INSERT INTO memories (id, namespace, agent_id, content, tags, created_at, updated_at, ttl_seconds) VALUES (?, ?, ?, ?, ?, ?, ?, ?) , (memory_id, namespace, agent_id, content, tags_json, now, now, ttl_seconds), ) return memory_id def recall( self, namespace: str, query: str , limit: int 5, agent_id: str | None None, ) - list[dict]: now time.time() with self._connect() as conn: rows conn.execute( SELECT * FROM memories WHERE namespace ? AND (ttl_seconds IS NULL OR updated_at ttl_seconds ?) ORDER BY updated_at DESC LIMIT ? , (namespace, now, min(limit * 5, 100)), ).fetchall() results [] for row in rows: if agent_id and row[agent_id] not in (any, agent_id): continue record dict(row) record[tags] json.loads(record[tags]) record[score] self._score(query, record[content]) if query else 0 results.append(record) results.sort(keylambda r: r[score], reverseTrue) return results[:limit] def update( self, namespace: str, memory_id: str, content: str | None None, tags: list[str] | None None, ttl_seconds: float | None None, ) - bool: fields [updated_at ?] values: list [time.time()] if content is not None: fields.append(content ?) values.append(content) if tags is not None: fields.append(tags ?) values.append(json.dumps(tags, ensure_asciiFalse)) if ttl_seconds is not None: fields.append(ttl_seconds ?) values.append(ttl_seconds) values.extend([memory_id, namespace]) with self._connect() as conn: cursor conn.execute( fUPDATE memories SET {, .join(fields)} WHERE id ? AND namespace ?, values, ) return cursor.rowcount 0 def forget(self, namespace: str, memory_id: str) - bool: with self._connect() as conn: cursor conn.execute( DELETE FROM memories WHERE id ? AND namespace ?, (memory_id, namespace), ) return cursor.rowcount 0 def _score(self, query: str, content: str) - float: q_tokens set(_tokenize(query)) c_tokens set(_tokenize(content)) if not q_tokens: return 0.0 overlap len(q_tokens c_tokens) return overlap / len(q_tokens) def _tokenize(text: str) - list[str]: 简单分词中文按单字切分英文按单词切分。仅用于演示生产环境应使用正式分词器。 text text.lower() tokens: list[str] [] current for ch in text: if \u4e00 ch \u9fff: if current: tokens.append(current) current tokens.append(ch) elif ch.isalnum(): current ch else: if current: tokens.append(current) current if current: tokens.append(current) return tokens这个实现里update 和 forget 都带上了 namespace 条件这是必须的。否则一个 Agent 只要拿到 memory_id就可以跨命名空间修改或删除别人的记忆。命名空间校验放在 SQL 条件里比在应用层判断更可靠。_score 方法用于演示关键词重叠度排序中文按单字切分英文按单词切分。这个方法在真实项目中不够用但足以解释“相关性排序”在读取路径里的作用。生产环境建议替换成 embedding 向量检索。4.4 MCP Server用 FastMCP 挂载记忆工具存储层写好后MCP Server 本身非常薄。借助 mcp 包提供的 FastMCP可以用装饰器直接注册工具import json import os from mcp.server.fastmcp import FastMCP from memory_store import MemoryStore DB_PATH os.environ.get(MEMENTO_DB, data/memento.db) mcp FastMCP( memento, instructions( Memento 是一个共享记忆服务。remember 用于写入记忆 recall 用于检索记忆update_memory 用于修改记忆forget 用于删除记忆。 所有操作都必须指定 namespace避免不同项目的数据互相干扰。 ), ) store MemoryStore(DB_PATH) mcp.tool() def remember( namespace: str, content: str, tags: list[str] | None None, agent_id: str any, ttl_seconds: float | None None, ) - str: 写入一条记忆到指定命名空间返回记忆 id。 memory_id store.remember( namespacenamespace, contentcontent, tagstags or [], agent_idagent_id, ttl_secondsttl_seconds, ) return json.dumps({ok: True, memory_id: memory_id}, ensure_asciiFalse) mcp.tool() def recall( namespace: str, query: str , limit: int 5, agent_id: str | None None, ) - str: 在命名空间中检索记忆返回 JSON 数组。 records store.recall( namespacenamespace, queryquery, limitlimit, agent_idagent_id, ) return json.dumps(records, ensure_asciiFalse, defaultstr) mcp.tool() def update_memory( namespace: str, memory_id: str, content: str, ) - str: 更新一条指定记忆的内容返回是否成功。 ok store.update(namespacenamespace, memory_idmemory_id, contentcontent) return json.dumps({ok: ok}, ensure_asciiFalse) mcp.tool() def forget( namespace: str, memory_id: str, ) - str: 删除一条指定记忆返回是否成功。 ok store.forget(namespacenamespace, memory_idmemory_id) return json.dumps({ok: ok}, ensure_asciiFalse) if __name__ __main__: os.makedirs(os.path.dirname(DB_PATH), exist_okTrue) mcp.run(transportstdio)启动前要确认 data 目录存在代码里用 os.makedirs 自动创建。工具函数返回的都是 JSON 字符串而不是 Python 对象这是为了让模型侧的结果更稳定可解析。启动命令cd memento python memento_server.pystdio 模式下服务不会打印任何内容这是正常现象。它监听标准输入输出通过 JSON-RPC 与客户端通信。4.5 用 MCP Inspector 验证工具是否可用强烈建议先使用 MCP Inspector 做一次协议层验证再接入业务客户端。启动命令npx modelcontextprotocol/inspector python memento_server.pyInspector 会打开一个本地调试页面列出服务声明的所有工具。依次验证四个操作调用 remember传入 namespaceproject-x、content“数据库选型项目采用 PostgreSQL 16”、tags[database,decision]。调用 recall查询“数据库”确认能返回刚写入的记忆。调用 update_memory修改记忆内容再次 recall 确认更新生效。调用 forget删除记忆确认 recall 结果变空。最后直接检查 SQLite 文件确认数据真的落盘了sqlite3 data/memento.db select id, namespace, substr(content, 1, 40), updated_at from memories;这一步能同时验证协议层和数据层两边都通了才说明 Memento 服务本身没有问题。5. 把 Memento 接入多个 Agent5.1 在支持 MCP 的客户端中配置本地服务以 Claude Desktop 这类支持 MCP 的客户端为例配置文件里增加一个 mcpServers 节点。stdio 模式指向本地启动命令{ mcpServers: { memento: { command: python, args: [/absolute/path/to/memento/memento_server.py], env: { MEMENTO_DB: /absolute/path/to/memento/data/memento.db } } } }配置里使用绝对路径避免客户端工作目录不同导致找不到脚本或数据库文件。改完配置后必须完全重启客户端MCP Server 列表通常只在启动时加载一次。5.2 多进程共享改用 SSE 传输stdio 传输适合单个客户端进程因为 stdin/stdout 只能被一个进程占用。多个 Agent 同时连接同一个记忆服务时需要把传输方式改成 SSEServer-Sent Events让服务监听一个 HTTP 端口多个客户端通过 HTTP 连接。启动方式简单改一行if __name__ __main__: os.makedirs(os.path.dirname(DB_PATH), exist_okTrue) mcp.run(transportsse, host0.0.0.0, port8765)客户端配置改为 URL 方式{ mcpServers: { memento: { url: http://127.0.0.1:8765/sse } } }启动后可以先用 curl 确认 SSE 端点可访问curl -N http://127.0.0.1:8765/sse注意SSE 服务不要直接绑定 0.0.0.0 暴露到公网。示例里没有任何鉴权逻辑生产环境至少需要前置网关做身份认证和访问控制。5.3 命名空间与权限隔离设计多个 Agent 共享一个服务后必须通过命名空间做逻辑隔离。推荐两种命名方式模式示例适用场景按项目隔离project-x、project-y不同项目完全隔离按团队加项目隔离team-a/project-x团队内多项目协作agent_id 字段可以记录写入者读取时可以通过 agent_id 参数过滤。比如规划 Agent 想只看自己写的记忆就传入自己的 agent_id想了解项目全貌就不传。这个字段在调试和审计时也非常有用能回答“这条记忆到底是哪个 Agent 写的”这个问题。生产环境如果要做得更严格建议在 MCP Server 前面增加一层访问令牌校验在 MCP 工具内部根据令牌解析出可访问的 namespace 集合服务端强制过滤而不是依赖 Agent 自觉传参。6. 常见问题与排查链路6.1 MCP 工具始终注册不上现象是客户端配置完成后工具列表为空或者只出现部分工具。这个问题在 MCP 生态里非常高频社区里反馈的 Figma MCP 在 Codex 中工具注册不上、某些 MCP Server 在客户端中偶发失败多数属于同类原因。按以下顺序排查先确认服务本身能启动。直接执行python memento_server.py看是否有语法错误或依赖缺失。确认依赖安装完整。示例依赖只有 mcp缺失时 import 阶段就会失败。检查 stdio 模式下是否往 stdout 打印了调试日志。stdio 传输依赖标准输入输出传递 JSON-RPC如果服务端用 print 输出日志会破坏数据帧导致客户端解析失败。日志必须写到 stderr 或文件。用 MCP Inspector 启动服务确认工具列表能正常展示。Inspector 能展示不代表业务客户端能展示但能排除服务端问题。检查工具命名是否冲突。多个 MCP Server 同时注册同名工具时客户端可能丢弃冲突项。修改配置后完全重启客户端。很多客户端只在启动时执行一次工具发现热加载不一定生效。问题现象常见原因检查方式处理建议工具列表为空服务启动崩溃或依赖缺失手动执行启动命令安装依赖修复语法错误stdio 连接后无响应stdout 被日志污染检查是否有 print 输出 多 Agent 共享记忆：Memento MCP Server 实现指南
在实际 AI Agent 工程里一个经常被低估的问题是记忆。模型本身没有持久状态每次对话、每个任务完成后之前的上下文就随着会话结束被丢弃了。单个 Agent 还可以通过把历史写入外部存储来缓解但当多个 Agent 同时参与一个项目各自维护自己的会话历史时问题会指数级放大Agent A 记录的架构决策Agent B 看不到Agent C 上一轮实验的结论下次冷启动时完全丢失。Memento 就是为这类场景设计的共享、持久化记忆层它把 Agent 的记忆统一收口到独立存储中并通过 MCPModel Context Protocol暴露给多个 AI Agent 使用。下面的内容会从多 Agent 记忆问题的根源讲起说明 MCP 为什么适合做存储访问层然后给出一个最小可运行的 Memento MCP Server 实现覆盖写入、检索、更新、删除和命名空间隔离最后补充 MCP 工具注册不上、记忆检索不到等高频问题的排查链路以及生产环境落地的检查清单。1. 先理解 Memento 解决什么问题1.1 单 Agent 的上下文限制与记忆缺失大语言模型每次调用本质上是无状态的。输入一段文本模型根据权重预测输出调用结束之后这段对话对模型来说就不存在了。所谓“模型会记住上下文”靠的是把历史消息拼进新的请求里而请求长度受上下文窗口限制。一旦窗口被占满或者会话结束旧信息就会被挤出或清空。产品层面的“记忆”不是模型自带的能力而是工程上额外设计的外部存储。常见做法包括把关键事实写入数据库或键值存储。把长历史的摘要重新注入后续请求。用向量库保存语义片段按相关性召回。这些做法都是把模型的短期工作记忆转存到应用侧的长期存储里。Memento 这个名字来自经典设计模式中的备忘录模式在不破坏对象封装的前提下把对象内部状态保存成快照之后可以恢复。AI Agent 的记忆系统做的是同一件事保存 Agent 在运行过程中产生的事实、决策和结论之后可以被同一个 Agent 或其他 Agent 恢复使用。1.2 多 Agent 场景为什么更需要共享记忆多 Agent 系统的难点不在于单个 Agent 能写多少代码而在于多个 Agent 之间如何传递状态。Agent 和 Agent 之间没有天然共享的上下文如果不引入外部记忆就会出现三类典型问题任务拆解后信息断层。规划 Agent 把项目拆成清单执行 Agent 却看不到清单的完整背景只能重新猜测。重复劳动。Agent A 已经确认了数据库选型和连接方式Agent B 又从零调研一遍给出冲突结论。无法恢复中断。一个会话被关掉所有中间产物、踩坑记录、结论都消失下次只能重新开始。共享记忆在工程上就是把 Agent 之间的通信从“临时口述”改成“持久化书面记录”。规划 Agent 写入任务状态执行 Agent 读取任务状态测试 Agent 写回验证结果评审 Agent 查看全部历史。每个 Agent 不需要知道对方内部发生了什么只需要读写同一个记忆空间。最常见的落地场景包括编码协作多个 Agent 分工实现不同模块共享架构决策、接口约定和环境信息。数据调研多个 Agent 并行收集资料把阶段性结论写入公共记忆最后由汇总 Agent 整合。运维处置故障处理 Agent 把排查步骤和结论写入记忆后续 Agent 在处理同类问题时直接复用。1.3 Memento 在系统架构中的定位Memento 在架构上处于 Agent 和存储之间扮演一个标准化访问层。它不是一个模型不是推理框架而是一个数据服务接收 Agent 发来的记忆写入请求经过处理落到持久化存储再响应 Agent 的检索请求。一个完整的 Memento 服务由四部分组成写入路径Agent 调用记忆写入接口传入命名空间、内容、标签和过期时间。存储层负责持久化保证重启后数据不丢。读取路径Agent 提交查询检索结果经过排序和过滤后返回。访问协议层通过 MCP 把记忆能力封装成标准工具让不同 Agent 客户端都能接入。这种设计的关键收益是解耦。Agent 不直接连接数据库不需要知道表结构也不关心存储用的是 SQLite 还是向量库。记忆系统独立演进存储可以替换检索逻辑可以升级Agent 侧几乎不用改动。2. MCP 为什么适合做 Agent 存储访问层2.1 MCP 协议解决什么问题MCP全称 Model Context Protocol是一套用于连接 AI 模型与外部工具、数据源的开放协议。它解决的问题非常直接如果没有统一协议每个 AI 应用都要为每个工具单独写一套接入代码工具方也要为每个客户端适配不同接口形成 N 乘 M 的集成成本。MCP 定义了 Host、Client、Server 三层结构把工具、数据资源和提示词模板统一成标准能力客户端按同一套规则发现和调用。目前 MCP 生态已经覆盖了大量场景从浏览器自动化、数据库查询、设计工具到数据分析、测试执行等。可以看到一个明显的趋势工具接入正在从“每个 Agent 写一套集成”变成“每个工具提供一个 MCP Server”。对于 Memento 这类记忆系统MCP 的价值尤其明显。记忆服务不是给某一个 Agent 用的而是要同时服务多个不同客户端的 Agent。提供 MCP Server 之后Claude Desktop、Codex、各种支持 MCP 的 IDE 和自研 Agent 框架都可以通过同一种协议接入不需要为每种客户端单独开发集成层。2.2 Tools、Resources、Prompts 三种能力如何分工MCP 协议定义了三种核心能力原语理解它们的分工才能正确设计记忆服务的对外接口。MCP 能力定位Memento 中的用法Tools让模型执行一个动作有输入参数和返回结果remember、recall、update、forget 都作为 Tool 暴露Resources以 URI 方式暴露可读数据用于向模型注入上下文暴露记忆统计、命名空间列表、记忆导出文件Prompts预置的用户操作模板引导模型按固定流程处理提供“记录项目决策”“汇总近期记忆”等模板在 Memento 实现里主要使用的是 Tools。因为记忆操作是典型的读写动作模型需要传入参数并拿到结构化结果。Resources 适合做只读数据暴露比如把某个项目的全部记忆导出成一个文本资源让模型在启动时直接读取。Prompts 则适合把常见操作封装成套路降低模型调用工具时的不确定性。2.3 MCP、Agent Skill、RAG、Prompt 到底有什么区别这几个概念经常被放在一起讨论但它们不在同一个层次。MCP 是协议解决“能力怎么被访问”的问题。它规定工具如何声明、如何调用、如何传输结果。Agent Skill 是 Agent 侧的技能封装通常包含一段执行流程、提示词和可能调用到的工具组合解决“执行什么步骤”的问题。RAG 是检索增强生成的方法论解决“如何用外部知识提升生成质量”的问题。Prompt 则是纯文本指令解决“怎么向模型描述任务”的问题。四者的关系可以用一个例子说明。一个 Agent 接收到“总结项目本周进展”的任务它的 Skill 定义了这个任务的执行流程先调用 Memento 的 recall 工具拉取本周记忆再调用某个 MCP 数据库工具查询提交记录然后把结果拼接成结构化 Prompt 交给模型生成总结。这里 MCP 提供工具访问能力RAG 是 recall 背后的检索逻辑Prompt 是最后组织输入的方式。概念层次核心问题典型载体MCP协议层工具和数据如何被统一访问MCP Server、Tools、ResourcesAgent Skill能力封装层完成一件任务需要哪些步骤流程、提示词、工具组合RAG方法论如何用检索到的知识增强生成向量库、召回排序、注入模板Prompt表达层如何把任务描述清楚指令文本、示例、约束条件在开发 Memento 时可以在 MCP Server 内实现 RAG 风格的检索逻辑也可以在 Agent Skill 中编排记忆操作但它们解决的问题不同不能互相替代。3. 设计一个共享持久化记忆系统的核心模块3.1 记忆数据模型设计共享记忆系统第一步是定义“一条记忆长什么样”。记忆不是聊天记录它应该具备可检索、可过滤、可过期的结构化属性。一个最小可用的记忆记录包含以下字段字段类型说明idstring记忆唯一标识建议使用 UUIDnamespacestring命名空间用于逻辑隔离不同项目或团队agent_idstring写入该记忆的 Agent 标识用于来源追踪contentstring记忆正文通常是事实、决策、结论tagsarray标签用于分类和过滤created_atnumber创建时间Unix 时间戳updated_atnumber最后更新时间ttl_secondsnumber可选过期时间为空表示永不过期在 SQLite 里存储时一条记忆对应的记录大致如下{ id: f7c3a41e-8b2f-4de1-9c67-2e6f0b0c3a12, namespace: project-x, agent_id: planner-01, content: 数据库选型项目采用 PostgreSQL 16连接串保存在配置中心禁止写入代码仓库。, tags: [database, decision], created_at: 1710000000.123, updated_at: 1710000000.123, ttl_seconds: 2592000 }这里 namespace 和 agent_id 是两个容易被忽略但非常重要的字段。namespace 决定记忆属于哪个逻辑空间多项目共用一个 Memento 服务时没有 namespace 会导致数据互相污染。agent_id 用于回答“这条记忆是谁写的”它既可以是写入方的身份标识也可以作为读取时的过滤条件。3.2 写入路径记忆写入不只是往数据库插一行它需要经过一条固定链路参数校验content 不能为空namespace 不能为空tags 必须是字符串数组。规范化统一内容格式比如去除首尾空白、统一时间格式。去重判断同一命名空间、相似内容在短时间内重复写入时优先更新旧记录而不是插入新记录。持久化把记录写入存储确保事务提交成功后接口才返回。返回结果返回记忆 id方便后续更新和删除。写入接口设计成幂等会比较稳妥。所谓幂等是指同一个写入请求执行多次最终状态和执行一次一致。在记忆场景里Agent 可能因为网络重试而重复调用写入接口如果每次都插入新记录会产生大量冗余记忆。注意记忆写入接口的幂等性要在设计阶段考虑。最常见的做法是让调用方传入可选的外部请求 id服务端根据请求 id 判断是否已经处理过避免重试导致重复写入。3.3 读取路径读取路径决定 Agent 拿到的是“有用的记忆”还是“一堆噪声”。一次完整的检索流程包括解析查询参数命名空间、查询文本、过滤条件、返回条数。过滤先按 namespace 过滤再按 agent_id、过期时间过滤。排序有查询文本时按相关性排序没有查询文本时按更新时间倒序。截断返回 top-k 结果避免一次性把大量记忆塞进上下文。相关性排序有两种典型实现。最小实现是关键词重叠度计算查询词和记忆内容的共现比例生产级实现是把内容映射成向量用余弦相似度排序。向量检索对“意思相近但用词不同”的查询更友好但需要额外引入 embedding 模型和向量索引。读取路径还需要注意一个原则返回的每条记忆都应该尽量独立完整。Agent 的上下文窗口有限如果一条记忆本身写得模糊模型拿到后还要反复追溯记忆的价值就大打折扣。3.4 记忆生命周期记忆不是越久越好。过期的架构决策、废弃的环境信息、已经修完的故障记录长期留在记忆里反而会误导 Agent。完整生命周期包括创建Agent 写入新记忆。更新记忆内容变化时修改并刷新 updated_at。过期超过 ttl 后被排除出检索结果。归档对有价值但不再活跃的记忆可以移入冷存储。删除显式清理错误或敏感记忆。ttl 的设计要结合业务场景。稳定性高的信息比如“项目使用的技术栈”可以设很长甚至不过期时效性强的信息比如“当前线上告警状态”应该设短过期时间。默认值常见做法是对所有记忆设置一个基础 TTL再允许调用方按场景覆盖。4. 从零实现一个最小可运行的 Memento MCP Server4.1 环境准备示例使用 Python 实现依赖尽量少方便在本地快速跑通。需要准备的环境如下组件版本要求用途Python3.10 或更高运行示例代码mcp 包最新稳定版提供 MCP Server 开发能力Node.js可选运行 MCP Inspector 调试工具安装依赖pip install mcp如果之后要接入向量检索可以追加安装 embedding 相关依赖。示例阶段先用标准库和 SQLite 完成闭环避免一开始引入过多组件。4.2 项目结构建议目录结构如下memento/ ├── requirements.txt ├── memory_store.py ├── memento_server.py └── data/ └── memento.dbmemory_store.py 负责存储层memento_server.py 负责把存储能力包装成 MCP 工具。分开写的好处是存储逻辑可以独立测试不依赖 MCP 环境。4.3 存储层SQLite 文件数据库保证持久化SQLite 对原型和中小规模场景足够数据落在单个文件里重启不丢。核心实现如下import json import sqlite3 import time import uuid class MemoryStore: 基于 SQLite 的共享记忆存储实现。 def __init__(self, db_path: str): self.db_path db_path self._init_db() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self) - None: with self._connect() as conn: conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, namespace TEXT NOT NULL, agent_id TEXT NOT NULL DEFAULT any, content TEXT NOT NULL, tags TEXT NOT NULL DEFAULT [], created_at REAL NOT NULL, updated_at REAL NOT NULL, ttl_seconds REAL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_memories_ns_time ON memories(namespace, updated_at DESC) ) def remember( self, namespace: str, content: str, agent_id: str any, tags: list[str] | None None, ttl_seconds: float | None None, ) - str: memory_id str(uuid.uuid4()) now time.time() tags_json json.dumps(tags or [], ensure_asciiFalse) with self._connect() as conn: conn.execute( INSERT INTO memories (id, namespace, agent_id, content, tags, created_at, updated_at, ttl_seconds) VALUES (?, ?, ?, ?, ?, ?, ?, ?) , (memory_id, namespace, agent_id, content, tags_json, now, now, ttl_seconds), ) return memory_id def recall( self, namespace: str, query: str , limit: int 5, agent_id: str | None None, ) - list[dict]: now time.time() with self._connect() as conn: rows conn.execute( SELECT * FROM memories WHERE namespace ? AND (ttl_seconds IS NULL OR updated_at ttl_seconds ?) ORDER BY updated_at DESC LIMIT ? , (namespace, now, min(limit * 5, 100)), ).fetchall() results [] for row in rows: if agent_id and row[agent_id] not in (any, agent_id): continue record dict(row) record[tags] json.loads(record[tags]) record[score] self._score(query, record[content]) if query else 0 results.append(record) results.sort(keylambda r: r[score], reverseTrue) return results[:limit] def update( self, namespace: str, memory_id: str, content: str | None None, tags: list[str] | None None, ttl_seconds: float | None None, ) - bool: fields [updated_at ?] values: list [time.time()] if content is not None: fields.append(content ?) values.append(content) if tags is not None: fields.append(tags ?) values.append(json.dumps(tags, ensure_asciiFalse)) if ttl_seconds is not None: fields.append(ttl_seconds ?) values.append(ttl_seconds) values.extend([memory_id, namespace]) with self._connect() as conn: cursor conn.execute( fUPDATE memories SET {, .join(fields)} WHERE id ? AND namespace ?, values, ) return cursor.rowcount 0 def forget(self, namespace: str, memory_id: str) - bool: with self._connect() as conn: cursor conn.execute( DELETE FROM memories WHERE id ? AND namespace ?, (memory_id, namespace), ) return cursor.rowcount 0 def _score(self, query: str, content: str) - float: q_tokens set(_tokenize(query)) c_tokens set(_tokenize(content)) if not q_tokens: return 0.0 overlap len(q_tokens c_tokens) return overlap / len(q_tokens) def _tokenize(text: str) - list[str]: 简单分词中文按单字切分英文按单词切分。仅用于演示生产环境应使用正式分词器。 text text.lower() tokens: list[str] [] current for ch in text: if \u4e00 ch \u9fff: if current: tokens.append(current) current tokens.append(ch) elif ch.isalnum(): current ch else: if current: tokens.append(current) current if current: tokens.append(current) return tokens这个实现里update 和 forget 都带上了 namespace 条件这是必须的。否则一个 Agent 只要拿到 memory_id就可以跨命名空间修改或删除别人的记忆。命名空间校验放在 SQL 条件里比在应用层判断更可靠。_score 方法用于演示关键词重叠度排序中文按单字切分英文按单词切分。这个方法在真实项目中不够用但足以解释“相关性排序”在读取路径里的作用。生产环境建议替换成 embedding 向量检索。4.4 MCP Server用 FastMCP 挂载记忆工具存储层写好后MCP Server 本身非常薄。借助 mcp 包提供的 FastMCP可以用装饰器直接注册工具import json import os from mcp.server.fastmcp import FastMCP from memory_store import MemoryStore DB_PATH os.environ.get(MEMENTO_DB, data/memento.db) mcp FastMCP( memento, instructions( Memento 是一个共享记忆服务。remember 用于写入记忆 recall 用于检索记忆update_memory 用于修改记忆forget 用于删除记忆。 所有操作都必须指定 namespace避免不同项目的数据互相干扰。 ), ) store MemoryStore(DB_PATH) mcp.tool() def remember( namespace: str, content: str, tags: list[str] | None None, agent_id: str any, ttl_seconds: float | None None, ) - str: 写入一条记忆到指定命名空间返回记忆 id。 memory_id store.remember( namespacenamespace, contentcontent, tagstags or [], agent_idagent_id, ttl_secondsttl_seconds, ) return json.dumps({ok: True, memory_id: memory_id}, ensure_asciiFalse) mcp.tool() def recall( namespace: str, query: str , limit: int 5, agent_id: str | None None, ) - str: 在命名空间中检索记忆返回 JSON 数组。 records store.recall( namespacenamespace, queryquery, limitlimit, agent_idagent_id, ) return json.dumps(records, ensure_asciiFalse, defaultstr) mcp.tool() def update_memory( namespace: str, memory_id: str, content: str, ) - str: 更新一条指定记忆的内容返回是否成功。 ok store.update(namespacenamespace, memory_idmemory_id, contentcontent) return json.dumps({ok: ok}, ensure_asciiFalse) mcp.tool() def forget( namespace: str, memory_id: str, ) - str: 删除一条指定记忆返回是否成功。 ok store.forget(namespacenamespace, memory_idmemory_id) return json.dumps({ok: ok}, ensure_asciiFalse) if __name__ __main__: os.makedirs(os.path.dirname(DB_PATH), exist_okTrue) mcp.run(transportstdio)启动前要确认 data 目录存在代码里用 os.makedirs 自动创建。工具函数返回的都是 JSON 字符串而不是 Python 对象这是为了让模型侧的结果更稳定可解析。启动命令cd memento python memento_server.pystdio 模式下服务不会打印任何内容这是正常现象。它监听标准输入输出通过 JSON-RPC 与客户端通信。4.5 用 MCP Inspector 验证工具是否可用强烈建议先使用 MCP Inspector 做一次协议层验证再接入业务客户端。启动命令npx modelcontextprotocol/inspector python memento_server.pyInspector 会打开一个本地调试页面列出服务声明的所有工具。依次验证四个操作调用 remember传入 namespaceproject-x、content“数据库选型项目采用 PostgreSQL 16”、tags[database,decision]。调用 recall查询“数据库”确认能返回刚写入的记忆。调用 update_memory修改记忆内容再次 recall 确认更新生效。调用 forget删除记忆确认 recall 结果变空。最后直接检查 SQLite 文件确认数据真的落盘了sqlite3 data/memento.db select id, namespace, substr(content, 1, 40), updated_at from memories;这一步能同时验证协议层和数据层两边都通了才说明 Memento 服务本身没有问题。5. 把 Memento 接入多个 Agent5.1 在支持 MCP 的客户端中配置本地服务以 Claude Desktop 这类支持 MCP 的客户端为例配置文件里增加一个 mcpServers 节点。stdio 模式指向本地启动命令{ mcpServers: { memento: { command: python, args: [/absolute/path/to/memento/memento_server.py], env: { MEMENTO_DB: /absolute/path/to/memento/data/memento.db } } } }配置里使用绝对路径避免客户端工作目录不同导致找不到脚本或数据库文件。改完配置后必须完全重启客户端MCP Server 列表通常只在启动时加载一次。5.2 多进程共享改用 SSE 传输stdio 传输适合单个客户端进程因为 stdin/stdout 只能被一个进程占用。多个 Agent 同时连接同一个记忆服务时需要把传输方式改成 SSEServer-Sent Events让服务监听一个 HTTP 端口多个客户端通过 HTTP 连接。启动方式简单改一行if __name__ __main__: os.makedirs(os.path.dirname(DB_PATH), exist_okTrue) mcp.run(transportsse, host0.0.0.0, port8765)客户端配置改为 URL 方式{ mcpServers: { memento: { url: http://127.0.0.1:8765/sse } } }启动后可以先用 curl 确认 SSE 端点可访问curl -N http://127.0.0.1:8765/sse注意SSE 服务不要直接绑定 0.0.0.0 暴露到公网。示例里没有任何鉴权逻辑生产环境至少需要前置网关做身份认证和访问控制。5.3 命名空间与权限隔离设计多个 Agent 共享一个服务后必须通过命名空间做逻辑隔离。推荐两种命名方式模式示例适用场景按项目隔离project-x、project-y不同项目完全隔离按团队加项目隔离team-a/project-x团队内多项目协作agent_id 字段可以记录写入者读取时可以通过 agent_id 参数过滤。比如规划 Agent 想只看自己写的记忆就传入自己的 agent_id想了解项目全貌就不传。这个字段在调试和审计时也非常有用能回答“这条记忆到底是哪个 Agent 写的”这个问题。生产环境如果要做得更严格建议在 MCP Server 前面增加一层访问令牌校验在 MCP 工具内部根据令牌解析出可访问的 namespace 集合服务端强制过滤而不是依赖 Agent 自觉传参。6. 常见问题与排查链路6.1 MCP 工具始终注册不上现象是客户端配置完成后工具列表为空或者只出现部分工具。这个问题在 MCP 生态里非常高频社区里反馈的 Figma MCP 在 Codex 中工具注册不上、某些 MCP Server 在客户端中偶发失败多数属于同类原因。按以下顺序排查先确认服务本身能启动。直接执行python memento_server.py看是否有语法错误或依赖缺失。确认依赖安装完整。示例依赖只有 mcp缺失时 import 阶段就会失败。检查 stdio 模式下是否往 stdout 打印了调试日志。stdio 传输依赖标准输入输出传递 JSON-RPC如果服务端用 print 输出日志会破坏数据帧导致客户端解析失败。日志必须写到 stderr 或文件。用 MCP Inspector 启动服务确认工具列表能正常展示。Inspector 能展示不代表业务客户端能展示但能排除服务端问题。检查工具命名是否冲突。多个 MCP Server 同时注册同名工具时客户端可能丢弃冲突项。修改配置后完全重启客户端。很多客户端只在启动时执行一次工具发现热加载不一定生效。问题现象常见原因检查方式处理建议工具列表为空服务启动崩溃或依赖缺失手动执行启动命令安装依赖修复语法错误stdio 连接后无响应stdout 被日志污染检查是否有 print 输出