前几天在 GitHub 上看到 MemOS 团队开源了 Memmy定位是“统一多 Agent 记忆”。这个方向其实在 Agent 开发社区里讨论了很久单 Agent 的记忆管理还没完全统一标准多 Agent 之间的记忆共享、权限隔离、长期记忆检索就更是各写各的。Memmy 的出现算是把“记忆层”单独抽出来做成基础设施思路很值得学习。这篇文章我会从 Agent 记忆的基础概念讲起分析多 Agent 场景下记忆管理为什么难再结合一个可运行的简化示例讲讲“统一记忆服务”应该怎么设计、怎么落地。如果你是做 AI Agent 开发、想了解 Agent 记忆框架或者正在折腾多 Agent 项目这篇文章应该能给你一套比较完整的参考。1. 背景与核心概念Agent 为什么需要记忆1.1 从“无状态对话”到“有状态 Agent”早期的大模型应用大多是无状态的用户问一句模型答一句前后对话之间没有什么关联。这种方式实现简单但一旦遇到需要多轮交互、需要根据用户历史偏好做决策的场景就很难满足需求。Agent 和普通对话机器人最大的区别在于它需要在一个任务周期里持续决策、持续行动。比如一个客服 Agent它可能需要先了解用户之前的工单记录再结合当前问题判断处理方案一个编程 Agent它需要记住项目里已经改过哪些文件、哪些测试已经通过。这些信息如果不能跨轮次保留Agent 的表现就会非常“健忘”。记忆能力Memory就是解决这个问题的核心机制。它让 Agent 可以把对话历史、工具调用结果、用户偏好、任务状态等信息保存下来在后续推理和行动时重新读取从而形成连续的行为逻辑。1.2 单 Agent 记忆的常见类型在讨论多 Agent 记忆之前我们先看单 Agent 场景下的记忆分类。通常可以分成这几种记忆类型存储内容典型实现对话记忆最近几轮用户消息、模型回复滑动窗口、消息列表工作记忆当前任务的中间状态、变量、上下文内存对象、临时缓存长期记忆跨会话保存的用户偏好、事实知识数据库、向量库语义记忆概念、常识、领域知识的结构化表达知识图谱、向量索引情景记忆特定事件、历史交互记录的抽象事件日志、摘要存储这里容易混淆的是一点对话历史并不等于记忆。对话历史是原始数据记忆是从数据里提炼出来的、可供检索和复用的信息。真正设计记忆系统时通常要做一层“信息提炼”比如把冗长对话压缩成摘要再入库。1.3 多 Agent 记忆从“各自记忆”到“共享记忆”多 Agent 系统里不同 Agent 往往承担不同职责有的负责信息收集有的负责任务拆解有的负责最终回复合成。如果每个 Agent 只维护自己的私有记忆就会出现几个典型问题信息孤岛。A Agent 获取到的用户偏好B Agent 完全不知道导致任务需要重复询问。上下文重复传递。为了让下游 Agent 了解上游结果只能把大量信息塞进提示词里既浪费 token又容易超长。状态不一致。多个 Agent 并行处理同一个任务时各自保存的“任务状态”可能相互冲突。所以“统一多 Agent 记忆”并不是一个伪需求而是 Agent 走向复杂协作时的必然趋势。MemOS 团队开源 Memmy正是在这个背景下出现的他们想把不同 Agent 的记忆收口到统一的能力层让记忆的写入、读取、检索、共享都有一致规范。2. Memmy 到底解决什么问题2.1 Memmy 是什么根据公开信息Memmy 是 MemOS 团队开源的一个面向多 Agent 场景的记忆管理项目核心方向是“统一多 Agent 记忆”。它希望为不同 Agent 提供统一的记忆接口让 Agent 之间的记忆可以按需共享同时保留对记忆数据的控制能力。由于项目仍在快速迭代中我建议你以官方 GitHub 仓库的最新文档为准。本文不会编造具体 API 名称而是重点讲解它背后的设计思路并给出一套可落地的简化实现。2.2 多 Agent 记忆统一要解决的四件事我们从工程角度拆解一个“统一多 Agent 记忆”系统至少要解决四件事记忆的标准化写入。不同 Agent 产生的记忆需要统一成相同的结构而不是各自记各自的。记忆的高效检索。Agent 需要根据当前任务快速找到相关记忆不能每次全量扫描。记忆的隔离与共享。有些记忆属于单 Agent有些记忆需要全局共享两者要能区分。记忆的生命周期管理。记忆会过期、会冲突、会失效需要更新和淘汰机制。Memmy 这类项目的价值在于它把这些能力从各个 Agent 里抽出来做成一个独立的记忆基础设施。这样开发者在新增 Agent 时不需要再重复实现“记住用户偏好”“查找历史任务”这类通用能力。2.3 与 Agent 框架、MCP、Harness 的关系最近社区里 Agent 周边概念很多容易混淆我在这里做一个简单区分Agent 框架负责“怎么搭建一个 Agent”包括模型调用、工具注册、推理循环。MCP 解决的是“怎么让 Agent 调用外部工具和数据源”是一种标准化连接协议。Harness 通常指 Agent 的运行容器或执行环境负责资源管理、权限控制、执行编排。Memmy 这类记忆项目解决的是“Agent 怎么记住东西”属于 Agent 的能力组件可以和框架、MCP 配合使用。所以它们不是替代关系而是分层配合关系。你完全可以在某个 Agent 框架里开发业务逻辑同时接入记忆服务来统一管理记忆。3. 记忆层的核心设计思路3.1 记忆条目的数据结构设计无论用什么存储记忆条目本身应该有相对稳定的核心结构。我建议至少包含以下字段字段含义示例memory_id记忆唯一标识mem_20250612_001agent_id记忆所属 Agentrecruiter_agentsession_id所属会话session_1001memory_type记忆类型preference / fact / task_statecontent记忆内容文本用户偏好 Python 技术栈tags标签便于检索[user, tech_stack]visibility可见范围private / public / sharedcreated_at创建时间2025-06-12T10:00:00Zupdated_at更新时间2025-06-12T10:30:00Z这个结构是我个人在 Agent 项目中归纳的通用结构不一定和 Memmy 完全一致但它能覆盖多数需求。visibility 字段很关键它是“跨 Agent 共享”的开关。3.2 统一读写接口统一记忆服务的核心是提供一组对上层 Agent 友好的接口。大致可以抽象为write_memory写入一条记忆。read_memory根据条件读取记忆。search_memory根据语义或关键词检索记忆。update_memory更新已有记忆。delete_memory删除记忆。list_memory列出某个范围下的全部记忆。每个 Agent 在运行时只需要调用这些接口不需要关心底层到底用了 Redis、MySQL 还是向量数据库。这就是“统一”的价值。3.3 底层存储选型底层存储可以分层设计热记忆近几轮的对话和短期任务状态放 Redis读写快过期自动清理。冷记忆长期偏好、事实知识放数据库或向量库便于检索和保存。语义检索把记忆内容做 embedding存入向量库支持相似度搜索。在实际落地时冷热分离会让系统更稳定Hot path 不要碰慢存储批量检索走索引或向量。Memmy 这类项目内部大概率也参考了类似思路。4. 从零实现一个简化版多 Agent 记忆服务为了把上面的设计思路讲透我这里给出一个教学用示例。它模拟了“Memmy 风格”的统一记忆服务多个 Agent 通过 HTTP 接口写入记忆、读取记忆、检索记忆。这个示例适合本地运行用来理解记忆层的核心逻辑。需要说明这不是 Memmy 的真实 API而是我根据自己的工程经验设计的参考实现。真实项目请以官方仓库为准。4.1 项目结构agent-memory-demo/ ├── requirements.txt ├── app.py # FastAPI 入口 ├── memory_store.py # 内存存储与检索核心 └── examples/ └── agents_demo.py # 模拟多 Agent 调用4.2 环境准备建议使用 Python 3.10 或更高版本。mkdir agent-memory-demo cd agent-memory-demo python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activaterequirements.txtfastapi0.111.0 uvicorn0.30.1 pydantic2.7.4安装依赖pip install -r requirements.txt4.3 实现核心记忆存储文件memory_store.py这里实现一个内存版记忆存储并用简单的关键词重叠算法模拟语义检索。生产环境建议把这一层换成 Redis 向量数据库。import time import uuid from typing import List, Optional class MemoryItem: 记忆条目 def __init__( self, agent_id: str, content: str, memory_type: str fact, session_id: str , tags: Optional[List[str]] None, visibility: str public, ): self.memory_id uuid.uuid4().hex[:12] self.agent_id agent_id self.content content self.memory_type memory_type self.session_id session_id self.tags tags or [] self.visibility visibility self.created_at time.time() self.updated_at time.time() def to_dict(self): return { memory_id: self.memory_id, agent_id: self.agent_id, content: self.content, memory_type: self.memory_type, session_id: self.session_id, tags: self.tags, visibility: self.visibility, created_at: self.created_at, updated_at: self.updated_at, } class MemoryStore: 简化版记忆存储生产环境可替换为 Redis 向量库 def __init__(self): self._items {} # memory_id - MemoryItem def write(self, agent_id: str, content: str, **kwargs) - MemoryItem: item MemoryItem(agent_idagent_id, contentcontent, **kwargs) self._items[item.memory_id] item return item def read(self, memory_id: str) - Optional[MemoryItem]: return self._items.get(memory_id) def list_memory(self, agent_id: Optional[str] None) - List[MemoryItem]: if agent_id is None: return list(self._items.values()) return [item for item in self._items.values() if item.agent_id agent_id] def search(self, query: str, limit: int 5) - List[MemoryItem]: 基于标签和关键词重合度的简单检索。 生产环境建议替换为 embedding 向量检索。 query_tags set(query.lower().replace(, ).split()) scored [] for item in self._items.values(): score 0.0 if query.lower() in item.content.lower(): score 1.0 content_words set(item.content.lower().split()) overlap query_tags content_words score len(overlap) * 0.5 tag_overlap query_tags set(item.tags) score len(tag_overlap) * 1.0 if score 0: scored.append((score, item)) scored.sort(keylambda x: x[0], reverseTrue) return [item for _, item in scored[:limit]]这个类结构很直白。write负责写入记忆read根据 ID 读取list_memory可以按 Agent 过滤search做了一个简单的相关性打分。理解这个逻辑后你把它换成真实向量检索也只是替换search的问题。4.4 用 FastAPI 封装统一接口文件app.pyfrom typing import List, Optional import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from memory_store import MemoryStore app FastAPI(titleAgent Memory Service, version0.1.0) store MemoryStore() class MemoryWriteRequest(BaseModel): agent_id: str content: str memory_type: str fact session_id: str tags: List[str] [] visibility: str public class MemoryUpdateRequest(BaseModel): content: Optional[str] None tags: Optional[List[str]] None app.post(/memories) def write_memory(req: MemoryWriteRequest): item store.write( agent_idreq.agent_id, contentreq.content, memory_typereq.memory_type, session_idreq.session_id, tagsreq.tags, visibilityreq.visibility, ) return item.to_dict() app.get(/memories/{memory_id}) def read_memory(memory_id: str): item store.read(memory_id) if not item: raise HTTPException(status_code404, detailmemory not found) return item.to_dict() app.get(/memories) def list_memories(agent_id: Optional[str] None): items store.list_memory(agent_idagent_id) return [item.to_dict() for item in items] app.post(/memories/search) def search_memories(query: str, agent_id: Optional[str] None, limit: int 5): items store.search(query, limitlimit) if agent_id: items [item for item in items if item.agent_id agent_id] return [item.to_dict() for item in items] app.put(/memories/{memory_id}) def update_memory(memory_id: str, req: MemoryUpdateRequest): item store.read(memory_id) if not item: raise HTTPException(status_code404, detailmemory not found) if req.content is not None: item.content req.content if req.tags is not None: item.tags req.tags item.updated_at time.time() return item.to_dict() app.delete(/memories/{memory_id}) def delete_memory(memory_id: str): item store.read(memory_id) if not item: raise HTTPException(status_code404, detailmemory not found) del store._items[memory_id] return {status: deleted} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)这里有一个细节需要解释store._items是私有变量示例里直接访问了这是为了缩短代码长度。真实工程里应该在MemoryStore里增加一个delete方法封装删除逻辑避免外部直接操作内部数据结构。4.5 模拟多 Agent 写入与共享文件examples/agents_demo.py这个脚本模拟三个角色用户画像 Agent、技术筛选 Agent、沟通 Agent。它们通过同一个记忆服务写入和读取记忆。import requests BASE_URL http://127.0.0.1:8000 def post_memory(agent_id: str, content: str, tags: list, visibility: str public): resp requests.post(f{BASE_URL}/memories, json{ agent_id: agent_id, content: content, tags: tags, visibility: visibility, }) print(f[{agent_id}] 写入记忆:, resp.json()) return resp.json() def search_memory(query: str, agent_id: str None): resp requests.post(f{BASE_URL}/memories/search, json{ query: query, agent_id: agent_id, limit: 5, }) print(f检索 {query} 结果:) for item in resp.json(): print( -, item[content], | 来源 Agent:, item[agent_id]) return resp.json() if __name__ __main__: # 用户画像 Agent 写入用户偏好 post_memory( agent_iduser_profile_agent, content用户是 Python 后端开发者熟悉 FastAPI 和 PostgreSQL, tags[user, tech_stack], ) # 技术筛选 Agent 记录岗位要求 post_memory( agent_idtech_screen_agent, content当前岗位要求熟悉 Python、FastAPI、Redis, tags[job, requirement], ) # 沟通 Agent 检索共享记忆 search_memory(Python FastAPI 岗位要求)运行这个脚本前需要先启动记忆服务python app.py再开一个终端运行示例python examples/agents_demo.py预期输出大致如下[user_profile_agent] 写入记忆: {memory_id: ..., content: 用户是 Python 后端开发者熟悉 FastAPI 和 PostgreSQL, ...} [tech_screen_agent] 写入记忆: {memory_id: ..., content: 当前岗位要求熟悉 Python、FastAPI、Redis, ...} 检索 Python FastAPI 岗位要求 结果: - 当前岗位要求熟悉 Python、FastAPI、Redis | 来源 Agent: tech_screen_agent - 用户是 Python 后端开发者熟悉 FastAPI 和 PostgreSQL | 来源 Agent: user_profile_agent这个例子虽然简单但已经体现了“统一多 Agent 记忆”的核心流程不同 Agent 写入各自的记忆沟通 Agent 在需要时统一检索不需要提前约定字段格式也不需要把全部信息塞进提示词。5. 用记忆存储还是直接拼提示词很多同学可能会问多 Agent 系统里与其搞一套记忆服务不如直接把所有历史信息都拼到 prompt 里不是更简单吗在早期原型阶段确实可以但一旦 Agent 数量增多、任务变长这种方案会快速失控。对比维度全量拼入 Prompt统一记忆服务Token 成本随历史线性增长成本高只取相关记忆成本可控检索速度无检索但上下文过长影响模型效果向量检索毫秒级返回信息隔离所有 Agent 看到全部内容可控制可见范围和权限长期保存会话结束即丢失跨会话持久化多 Agent 一致性各持一份副本容易不一致统一写入和读取状态一致所以更合理的架构是短期内的小任务状态放内存或 Redis关键事实放记忆服务只有当前决策真正需要的相关内容才注入提示词。这也是记忆层存在的工程价值。6. 常见问题与排查思路在实现和使用记忆服务时最常遇到的问题如下。问题现象常见原因解决思路Agent 总是“忘事”记忆写入后没有在决策前检索检查 Agent 推理流程是否在调用模型前读取了记忆检索结果不相关只用关键词匹配没有语义能力升级为向量检索或用 embedding 计算相似度多 Agent 写到同一份记忆导致冲突缺少 agent_id 隔离维度写入时强制带 agent_id检索时按需过滤记忆越来越多成本升高缺少生命周期管理增加过期时间、定期摘要压缩、淘汰低价值记忆隐私或越权读取visibility 没有校验在读取接口层校验可见范围和权限并发写入覆盖数据读取-修改-写入没有做原子操作使用版本号或乐观锁必要时用 Lua 脚本保证原子性如果你遇到“记忆好像没生效”的问题我建议先按这个顺序排查确认写入接口是否被调用。打印日志检查 write_memory 是否真的执行。确认读取时是否传对了条件。比如当前 Agent 只读取自己写入的记忆但需要的其实是全局记忆。确认检索是否返回了结果。如果用的是向量库往往还需要先确认 embedding 是否生成成功。确认记忆注入是否真的进入提示词。很多框架层会过滤外部数据需要检查中间管道。7. 最佳实践与工程建议7.1 记忆设计要明确“写入什么、谁可见”不要在代码里随手 write 所有内容。先定义清楚哪些信息值得成为记忆用户长期偏好值得写入。一轮无关闲聊不值得写入。任务中间状态可以放短期缓存。工具调用结果按需摘要后入库。visibility 字段要尽早设计。至少区分公开记忆和私有记忆避免所有 Agent 都能读到敏感信息。7.2 把记忆存储做成可替换的抽象层我的建议是不要在业务代码里直接调用 Redis 或向量库 SDK而是先定义一层 MemoryStore 接口。这样后续换存储、加缓存、做权限控制都更容易。上面的示例就是这种思路只是实现简单了一些。7.3 记忆更新要谨慎删除要克制记忆一旦写入可能会被多个 Agent 依赖。更新记忆时最好保留历史版本或者至少记录 updated_at方便回溯。批量删除记忆在生产环境要格外谨慎建议先做标记删除再异步清理。7.4 关注记忆安全与红线上限Agent 记忆里可能包含用户隐私、密钥、授权信息。在工程上要注意对记忆内容做脱敏处理密钥和 token 不进记忆库。记忆读取接口要做鉴权不能任何 Agent 都能读取全部记忆。涉及删除和变更的操作要遵循最小权限原则。生产环境变更前先备份并在测试环境验证。7.5 监控指标如果你把记忆服务真正跑到生产建议至少监控五个指标写入 QPS 和写入耗时。检索召回率和响应耗时。单 Agent 记忆量增长速度。检索失败率和空结果率。记忆容量和存储成本。这些指标能帮你判断当前记忆策略是否健康。比如空结果率太高说明很多查询没有命中记忆需要检查写入策略和检索策略。8. 从 Memmy 到自建记忆层下一步怎么走MemOS 团队开源 Memmy 这件事给整个 Agent 生态带来的信号很明确记忆正在从“Agent 内部实现细节”变成一种独立的平台能力。以后做多 Agent 应用的团队可能不再需要从零开发记忆模块而是直接接入一个开源的记忆层。如果你准备在自己项目里落地多 Agent 记忆我建议按照下面的路径推进先盘业务场景。明确需要记忆的信息有哪些类型、哪些需要跨 Agent 共享。跑通一个最小闭环。用本文的简版示例先解决“写入-检索-注入”通路。引入真实存储。把内存存储替换成 Redis、MySQL 或向量数据库。加上权限和可视化。控制 Agent 可见范围方便调试。持续评估效果。通过实际任务的成功率来判断记忆策略是否有效。在动手实现时优先级应该放在检索质量和更新策略上。检索不准记忆存得再多也没有意义更新不严记忆就会变成垃圾场。如果你刚开始接触 Agent 记忆我建议先从最简单的“显式写入 关键词检索”开始跑通后再引入向量检索。直接上完整生产方案很容易在调试阶段被复杂链路卡住。这个领域迭代很快Memmy 只是其中一个方向后续大概率还会出现更多专注于记忆管理、记忆可视化、记忆共享协议的开源项目。保持关注官方仓库自己动手多写几个 Agent 调用记忆服务的例子会比只看概念收获大得多。