Agent 持久化记忆层实战：用 mem0 + COS Vectors 解决跨会话失忆

如果你跑过几个真正的 Agent 项目大概率会遇到同一个尴尬模型本身再聪明换一个会话就彻底“失忆”。同一个用户刚说完的偏好下一次对话里再问它就跟第一次听见一样。我给自研助手项目 OpenClaw内部代号“小龙虾”做了一套持久化记忆层核心思路是用 mem0 做记忆的抽取与管理用 COS Vectors 做向量存储与语义召回。这套东西跑完以后“小龙虾”终于能记住用户是谁、喜欢什么、讨厌什么。这篇文章主要讲记忆层怎么设计、关键代码怎么写、哪些坑是我实测踩过的适合正在被 Agent 失忆问题折腾的开发者参考。1. 先说清问题Agent 的“失忆”到底缺在哪1.1 会话内“伪记忆”和跨会话“真失忆”很多人一开始觉得“模型记不住事情”是因为上下文窗口不够大于是无脑加大窗口。但窗口再大也只是当前的 session 内有效。你可以在一次对话里反复说“我不吃香菜”模型每次都能准确避开香菜看起来好像它记住了。可只要这个会话结束下一次开新会话它又会踩进同一个坑用户说“来碗面不要香菜”它照样给你推荐香菜拌牛肉。这时候你才意识到之前的“记住”只是上下文窗口里的短期存在不是真正的记忆。我习惯用一个类比来解释这个问题想象一个餐厅服务员服务期间能记住你这桌的忌口和偏好但你下次再来换了一个服务员他又得从头问一遍。Agent 没有长期记忆就相当于餐厅永远不给你建立客户档案每次都是第一次来。OpenClaw 这个项目在早期就是这个状态。当时团队内部开玩笑给它起名“小龙虾”意思是有很多只脚功能多、壳硬能抗 bug但脑子只有七秒记忆。后来发现这不是讽刺是事实。用户连续问三遍同一个问题它三遍的回答都像第一次这是最让人崩溃的现场。所以记忆层这件事不是锦上添花是能不能继续用的底线。1.2 直接把历史聊天记录全塞进 Prompt 不现实既然上下文窗口能存住信息那有人会想把所有历史聊天记录全塞进 Prompt 不就行了我早期也这么干过把最近几十轮对话全部拼进 system prompt 里最后发现三个问题。第一个是 token 成本。一段 30 分钟的对话光历史记录可能就有几千个 token每次请求都要重新计算费用肉眼可见地涨。第二个是噪声。历史里大量信息是无用的比如“今天天气不错”“你现在几点睡”这种寒暄混在真正的偏好里模型处理起来更费劲。第三个是长上下文的注意力分散问题。模型对长文本中间部分的内容关注度明显偏低那些对用户最重要的偏好信息如果恰好落在中间照样可能被忽略。这不是模型能力问题是处理长文本时的通病。记忆层要解决的不是“把全部内容存下来”而是“把值得记住的内容提炼出来并在需要的时候精准捞回来”。这跟人脑的记忆机制很像你不会记住每天的早餐吃了什么但你会记住“朋友对花生过敏”这种重要信息。1.3 三种方案的取舍市面上真正可行的方案大致有三类我做一个简单的对比方案实现方式优点缺点A. 原文存储 关键词检索所有聊天记录入库用关键词或正则匹配实现简单适合演示语义理解差用户换个说法就匹配不到B. 长上下文硬扛把全部历史拼进 Prompt效果直接短期可用成本高噪声大无法规模化C. 抽取型记忆 向量语义检索从对话中提炼结构化记忆向量化后存储按相似度召回召回准确成本可控能跨会话需要额外开发有架构复杂度我最终选了 C。虽然前期开发成本最高但这是唯一能在真实用户场景下长期跑通的方案。具体的组合就是标题里写的mem0 负责“抽记忆、管记忆、更新记忆”COS Vectors 负责“存向量、检索向量、过滤向量”。2. 记忆层的整体设计与数据流2.1 记忆的分层管理在动手写代码之前必须先想清楚一个问题到底要把什么“记”下来如果只是存聊天记录那我们绕了一圈还是在做日志不是记忆。我参考了认知科学里工作记忆和长期记忆的区分把 OpenClaw 的记忆分成几个层级用户级记忆跨会话长期保存例如用户的姓名、职业、饮食忌口、明显偏好。这类记忆优先级最高除非用户主动纠正否则基本不删。会话级记忆只在当次会话内有效例如“这次帮用户查了三个城市的房价”“用户刚才提到下周要去出差”。会话结束后可留作参考但不作为长期事实。Agent 级记忆记录 Agent 自身的行为规则和使用经验比如“用户不喜欢太长的回复”“遇到复杂问题先列大纲再回答”。知识级记忆从文档、知识库里向量化后的内容更接近 RAG 里的东西。在一开始做的时候我的建议是不要贪多先做用户级和会话级两种就够了。Agent 级记忆看起来很炫但实际上提取难度高容易把模型带偏等前面的基础打牢了再搞。每条记忆记录的结构也不只是存一句话那么简单。实际存储时会带上这些字段{ id: mem_0001, user_id: user_123, session_id: session_456, type: user_preference, # fact / preference / task_state content: 用户不吃香菜偏好清淡口味, importance: 0.9, created_at: 2025-01-01T10:00:00Z, expires_at: None, # None 表示长期有效 source_session: session_456 }这些字段不只是为了记录后面做向量检索的时候过滤条件全靠它们。2.2 写入链路从对话到记忆记忆不是凭空生成的需要从用户对话中抽取。抽取这个环节我强烈建议直接用 LLM而不是写规则。原因很简单用户的表达千奇百怪规则永远写不完。比如用户说“我讨厌香菜”“别放香菜”“香菜闻着就恶心”这三句话表达了同一个意思但写法完全不同。用规则匹配你得写多少条 pattern用 LLM几行 prompt 就解决了。抽取的触发时机也有讲究。第一版我是每条消息进来都跑一次抽取结果对应服务的调用量直接翻倍成本一下子失控。后来改成对话完成一轮后才异步触发抽取也就是“用户消息 Agent 回复”作为一组输入让模型判断这里面有没有值得长期保存的信息。抽取 Prompt 我大致是这么设计的你是记忆抽取助手。请从下面的对话中找出值得长期记住的信息。 值得记住的信息包括 1. 用户明确的个人偏好 2. 用户提到的关键事实家庭、工作、健康等 3. 当前任务的关键状态 不值得记住的信息包括 1. 一次性的寒暄 2. 临时性提问 3. 模型自身的思考过程 请输出 JSON 数组每个元素包含 content、type、importance 三个字段。 importance 范围 0~1数值越高表示越值得记住。抽取完成后每条内容会被 embedding 模型转成向量然后写入 COS Vectors。这里有一个容易被忽略的点写入前要做相似度检查。如果库里面已经有一条“用户不喜欢香菜”你又写入一条“用户讨厌香菜”两条高度相似却互不关联过几天就会产生一堆重复记忆。正确的做法是先拿这条新记忆的向量去检索如果相似度超过设定阈值就走“更新”逻辑而不是“新增”逻辑。2.3 读取链路先检索后注入读取链路是每次 Agent 接收用户消息时执行的顺序大概是这样的用户发来一条消息。系统先把这条消息做 embedding得到查询向量。带着 user_id 过滤条件去 COS Vectors 里检索 top-k 条记忆。按相似度阈值过滤掉低相关结果。把命中的记忆拼进 system prompt最后交给 Agent 生成回复。这里有一个特别重要的经验寄回来的记忆不能当成用户“亲口说的话”直接使用。模型容易过度信任注入的内容如果某条记忆已经过期或者被纠正过模型却还拿它当依据就会出现很荒谬的回答。我的做法是在 prompt 里把记忆区明确包裹成“参考信息”并加上一句提示这些是之前对话中的历史印象可能已经变化请以用户当前的真实输入为准。以下是系统长期保存的用户记忆仅供参考不一定仍然准确 memory - 用户不吃香菜 - 用户偏好清淡口味 - 用户近期在关注租房信息 /memory 请根据你的判断结合当前对话内容回答用户的问题。这个细节看着不起眼但能避免很多“AI 固执己见”的尴尬场景。3. 实操从 0 到 1 给“小龙虾”装上记忆3.1 环境准备与依赖先说环境。我用的是 Python 3.10这是 mem0 运行时要求的最低版本。如果你的项目还在用 Python 3.8 或更早建议先升级不然后续兼容性问题会让你怀疑人生。基础依赖安装pip install mem0 pip install sentence-transformers pip install cos-vectors-client这里需要说明一下不同版本可能 SDK 名称和接口有细微变化以官方文档为准。我写这篇文章时用的是这些包的稳定版本跑起来没有问题。接着是 embedding 模型的选择。这块同样没有标准答案完全看你的数据场景。如果业务以中文为主我建议用中文优化的通用 embedding 模型例如 BGE 系列。我选了一个 768 维的模型这个维度属于“够用且不吃力”的范围检索效果不错向量体积也不算大写入和查询的延迟均可接受。如果你嫌本地模型推理太慢也可以调 API效果其实差不多就是每次调用多一笔费用。我后来为了控制成本还是换回了本地部署。3.2 把 mem0 的向量后端切换为 COS Vectorsmem0 本身的架构里向量存储是独立抽象层。默认配置用的可能是本地向量库但对需要跨机器共享的 Agent 服务来说本地存储根本没法用多实例部署时每个实例各自一套记忆数据根本不同步。于是我把注意力转向 COS Vectors。它的定位是托管式向量检索服务可以把向量数据和元数据放在同一个地方管理省去自建向量数据库的运维成本。mem0 的向量存储层暴露了一套接口我们需要实现一个自己的存储类然后把它注入 mem0 的配置。大致代码如下from mem0.vector_store import VectorStore class CosVectorsStore(VectorStore): def __init__(self, endpoint, collection_name, dim768): # 初始化客户端 self.client create_cos_vectors_client(endpoint) self.collection collection_name # 如果集合不存在就创建 if not self.client.has_collection(collection_name): self.client.create_collection( namecollection_name, dimensiondim, metriccosine ) def insert(self, vectors, metadata, ids): # 批量写入向量metadata 会作为过滤字段保存 self.client.upsert( collectionself.collection, idsids, vectorsvectors, metadatametadata ) def search(self, query_vector, top_k10, filtersNone): # 检索时带上过滤条件 results self.client.query( collectionself.collection, vectorquery_vector, top_ktop_k, filterfilters ) return results def delete(self, ids): # 按 id 删除记忆 self.client.delete(collectionself.collection, idsids)这里有一个特别容易踩的坑创建集合时指定的 dimension 必须和 embedding 模型实际输出的维度完全一致。如果你先拿 768 维的模型建了集合后来换了 1024 维的模型写入时就会直接报错而且这个错误在查询阶段很难定位往往是你发现“怎么查什么都查不到”之后才反应过来。还有一点关于检索指标。COS Vectors 支持多种距离度量包括欧氏距离、内积、余弦相似度。我做记忆召回时始终用 cosine因为记忆内容的长短差异大cosine 对向量模长不敏感更适合文本语义相似度这种场景。3.3 记忆抽取与写入流程下面是我实际运行的抽取函数简化过后保留核心逻辑def extract_memories(dialogue_turns): prompt build_extract_prompt(dialogue_turns) result llm.chat(prompt, response_format{type: json_object}) items json.loads(result) memories [] for item in items[memories]: memories.append({ content: item[content], type: item[type], importance: float(item[importance]) }) return memories注意我在 Prompt 里要求模型必须输出 JSON并且用 response_format 参数约束了输出格式。这一步很关键如果让模型自由发挥回来的文本会让你解析到崩溃。抽取完成之后写入流程如下def save_memories(user_id, session_id, memories): for mem in memories: # 先检查是否存在高度相似的记忆 query_vec embedding_model.encode(mem[content]) similar vector_store.search( query_vec, top_k1, filters{user_id: user_id} ) if similar and similar[0][score] 0.9: # 高度相似更新原有记忆 merge_and_update(similar[0][id], mem) else: # 新增记忆 vector_store.insert( vectors[query_vec], metadata[{ user_id: user_id, session_id: session_id, content: mem[content], type: mem[type], importance: mem[importance], created_at: datetime.utcnow().isoformat() }], ids[generate_memory_id()] )merge_and_update 的实现我用的是 LLM 来做合并。比如旧记忆是“用户不吃香菜”新记忆是“用户讨厌香菜味的东西”让模型把两条合并为一条更完整的表述。这里如果用简单字符串拼接很快会积累出一堆又长又啰嗦的重复记忆而且后续检索时 token 占比还大。3.4 读取与 Prompt 注入读取端的函数相对简单核心是按用户过滤和按阈值过滤两步def get_relevant_memories(user_id, query, top_k20): query_vec embedding_model.encode(query) results vector_store.search( query_vec, top_ktop_k, filters{user_id: user_id} ) # 阈值过滤保留相关度足够的记忆 filtered [r for r in results if r[score] 0.35] # 按重要度排序取前 8 条 filtered.sort(keylambda x: x[metadata][importance], reverseTrue) return filtered[:8]这里有两个数值值得展开说说。top_k 我设为 20是想让召回阶段尽量多捞一些候选宁可错杀也不要漏掉真正相关的记忆。但真正进入 prompt 的只有前 8 条排序标准不是相似度而是重要度。这和直觉不太一样但我实测下来效果更好相似度高不代表值得记一个寒暄话题的相似度可能很高但用户随口一句灰心话远不如“用户对花生过敏”这种信息重要。阈值 0.35是基于分数分布调出来的经验值。这个我们放到下一节展开。最终拼进 prompt 的代码def build_memory_section(memories): if not memories: return lines [f- {m[metadata][content]} for m in memories] return memory\n \n.join(lines) \n/memory特别提醒一句单条记忆不要存太长。我在实现里限制了 content 不超过 50 个中文字符超出部分会被截断。因为一条记忆是给模型看的关键事实不是一篇小作文。存得越长检索时 token 浪费越多而且太冗长的记忆会让模型抓不住重点。3.5 定期清理与更新记忆不是只增不减的。如果你让它无限膨胀三个月后集合里会有几万条记录检索速度下降还在其次更容易出现“大量低相关记忆被误召回”的问题。我每周跑一次清理任务逻辑有两个维度过期清理expires_at 字段非空且已经过期的直接删掉。比如“用户下周去出差”这种临时状态任务结束后就没意义了。低重要度合并把同样用户下面 importance 低于 0.3 的记忆按主题聚类合并。比如用户在不同时间说过“最近有点累”“这几天睡眠不好”可以合并成一条“用户近期睡眠质量不佳容易疲劳”。清理任务的代码粗略如下def cleanup_expired_memories(): expired vector_store.list( filter{ expires_at: {$lt: datetime.utcnow().isoformat()} } ) for item in expired: vector_store.delete(ids[item[id]])这个任务不要跑得太频繁每天或者每周一次就够了。频繁跑只会增加负载对数据质量帮助不大。4. 踩过的坑和排查实录4.1 向量维度不一致导致写入失败这件事发生在我换 embedding 模型的那天印象太深了。项目一开始用的是社区里的一个老模型768 维后来发现它在专业领域的语义理解不够好换成了一个更强的模型。兴致勃勃地配置完一跑写入接口直接报错。翻日志发现是维度对不上新模型输出 1024 维集合索引还是按 768 维建的。解决办法有两种一个是重建集合再全量导入另一个是新建一个集合用不同前缀区分。我选了后者因为重建集合要跑全量数据迁移太折腾了。这个坑的教训是每次更换 embedding 模型前必须确认新模型的输出维度并提前规划好集合的拆分或重建方案。4.2 召回率低Agent 还是像失忆记忆层接上以后我满心期待用户再也不用重复自己的偏好了。但跑了两天发现效果并没有想象中好用户还是会在对话里抱怨“我不是说过了吗”。排查方式也很笨把检索到的记忆和对应的相似度分数全部打印出来。打印结果让我很意外检索到的记忆其实挺多的但分数普遍很低很多只有 0.2 左右。我一开始设置的阈值是 0.6结果就是几乎所有记忆都被过滤掉了等于 Agent 根本没有拿到任何记忆。后来我统计了一周的数据发现大多数有效记忆的分数集中在 0.35 到 0.5 之间。低于 0.3 的基本是无关噪声高于 0.6 的少之又少。于是我把阈值调到 0.35召回率明显上升用户体感也上来了。这里给一个经验阈值不是拍脑袋定的一定要先跑一段时间看分数分布。每家的数据不一样最靠谱的办法就是收集一批真实用户问题离线计算分数分布再决定阈值砍在哪里。4.3 记忆重复膨胀跑了两周以后我有一个清晨打开向量集合看了一眼差点没被吓到同样是“用户不吃香菜”这条信息集合里躺着六七条写法分别是“用户不喜欢香菜”“用户不吃香菜”“用户对香菜过敏”“别放香菜”等等。原因是写入链路里的相似度检查阈值 0.9 设得太高。这些句子语义明明是一回事但向量相似度只有 0.75 到 0.85根本达不到阈值于是每条都变成了新记忆。解决方式是把阈值降到 0.8同时增加一个 LLM 判断当相似度在 0.7 到 0.8 之间时让模型判断这两条是否在说同一件事。这个方法能覆盖更多边界情况。顺便说一句重复记忆不是小问题。它不仅浪费存储还会在召回时重复占用 prompt 的 token 额度。本来能塞 8 条不同维度的记忆结果 4 条都在说同一个小事召回质量自然拉胯。4.4 Embedding 成本控制如果你用 API 方式的 embedding一定要做好成本控制。我第一版很随意每条消息都调 embedding后来看账单才意识到这个模块的消耗远超预期。我做了三个优化对话级缓存同一个用户十分钟内的消息若内容高度相似直接复用之前的 query 向量不再重新计算。批量处理记忆抽取和写入放在异步队列里攒够一批再统一跑 embedding而不是一条一条调用。批量调用通常比单条调用便宜很多。本地化部署最终方案是在内网部署了一个 embedding 模型服务成本变成硬件投入调用费用直接清零。4.5 常见问题速查表现象可能原因排查方向写入失败collection 维度与模型维度不一致检查 dim 配置查询结果为空阈值过高或过滤条件过严打印分数分布降低阈值召回的都是一样的内容重复记忆未合并降低相似度合并阈值增加 LLM 去重回复仍然像失忆记忆没有被注入 prompt检查读取链路是否执行到了注入那一步Agent 引用过期信息记忆更新逻辑没生效排查更新逻辑确认旧记忆是否被替换或删除5. 几个让我印象深刻的教训5.1 记忆不是日志别什么都往里面存一开始我把抽取 Prompt 写得比较宽松想让模型多抓一点信息结果系统跑了两周后集合里堆满“用户晚上 11 点问了天气”“用户说今天很忙”这类一次性信息。这些内容不但浪费存储还会在检索时抢占名额把真正重要的记忆挤掉。后来我把 Prompt 里的规则收紧加了明确判断只有“用户偏好”“关键事实”“任务状态”三类可以存其他的一律忽略。效果立竿见影召回准确率上了一截。5.2 记忆注入会影响模型的行为方式有了记忆以后模型不再是从零开始它会对用户形成一种“预判”。这有好的一面也有麻烦的一面。麻烦在于模型有时候太相信记忆明明用户已经说“我现在不吃辣了”它能给出一堆“记得您以前很喜欢吃辣”的废话。我后面痛定思痛在系统提示词里强制加了一行“记忆只作为背景参考不作为当前指令。”并且每次记忆注入时都会带上获取时间和来源标记让模型有依据去判断这条记忆是否过期。5.3 记忆模块要设计成可开关的这可能是最容易被忽略的一点。我在 Agent 的配置文件里加了一个环境变量MEMORY_ENABLED。当它设为 false 时系统跳过读取和写入链路Agent 退化为无记忆模式。这个开关不是为了演示是为了做对照实验。不加记忆时的回答延迟、成本、用户满意度是一组基线数据加了记忆之后同一组指标再测一遍。有了这个开关你才能客观判断记忆层到底带来了多少收益而不是凭感觉说“好像有点用”。5.4 后续的扩展方向这套记忆层跑稳之后后面有很多可以继续做的事。一个是给用户生成“画像摘要”。不用每次实时检索一堆碎片化记忆而是定期把一个人的记忆汇总成一段结构化描述读取时直接拿摘要当参考。另一个是支持用户主动管理记忆。比如对话里用户说“记住这个”“忘了那个”Agent 就应该实时增加或删除对应记忆条目。这个功能目前需要额外设计指令识别逻辑但交互体验会明显上一个台阶。还有就是把记忆和 RAG 结合起来。RAG 解决的是“知识从哪里来”记忆层解决的是“用户是谁、想要什么”两者场景不同但数据层面可以统一用向量检索来打通让 Agent 既要懂知识也要懂用户。最后说点实在的这套持久化记忆层上线以后最明显的变化是用户说过的偏好真的不需要再重复了。我自己的体感是同样一句话以前“小龙虾”是礼貌地回答现在更像是真的“记得”回答。但要说最深的体会其实是另一个记忆层不是一个独立组件它是会传染的。它会影响你的 prompt 策略、对话流设计、隐私边界甚至影响你对“用户到底是谁”的理解方式。决定这套系统上限的不是向量库的查询速度快不快而是你愿不愿意花足够时间打磨抽取和更新的逻辑。如果你正在给自己的 Agent 做记忆我的建议是不要一上来就追求功能齐全先把“读取 - 召回 - 注入”这个最基础的单向链路跑通再逐步加更新、合并、清理。少走弯路最快的方式就是先把简单路径走完。