1. 为什么 Agent 的记忆总在“搬家”做过 Agent 项目的人大概都有过这种体验今天用某个框架搭了一个能记住用户偏好的助手明天想换个编排引擎或者换一家模型服务结果发现之前攒下来的对话历史、用户画像、任务上下文全都带不走。要么导出成一份谁都不认识的 JSON要么干脆锁死在某个平台的数据库里迁移一次等于重做一次。标题里说的“记忆不跟着工具搬家”戳的就是这个痛点。我自己从最早用 LangChain 的ConversationBufferMemory到后来自己写 Redis 存对话再到尝试各种带长期记忆的 Agent 框架前后踩了不下十次“记忆迁移”的坑。最夸张的一次是给一个客服 Agent 换了底层编排框架结果三万多条历史对话的向量索引全部作废因为新旧框架对 embedding 的存储结构定义完全不一样。那次之后我就下定决心记忆层必须和工具层解耦。这篇内容适合两类人看一类是正在做 Agent 开发、被记忆管理搞得焦头烂额的工程师另一类是想入门 AI Agent、但还没意识到“记忆”这件事有多麻烦的初学者。我会把记忆分层的思路、存储选型、迁移方案、并发处理、以及实际落地时那些文档里不会写的坑全部摊开讲一遍。核心观点就一句话Agent 的记忆应该是一份独立于任何框架的资产工具可以换记忆不能丢。2. Agent 记忆到底分几层别再混着存了2.1 短期记忆和长期记忆的本质区别很多人一上来就把所有对话往一个数据库里塞这是最典型的错误。短期记忆和长期记忆在生命周期、访问频率、存储介质上的要求完全不同混在一起存查询会越来越慢成本会越来越高。短期记忆指的是当前会话窗口内的上下文它的特点是生命周期短、读写频繁、容量有限。比如用户问“帮我订明天去上海的机票”Agent 需要记住前面几轮提到的出发城市、时间偏好、舱位要求。这些信息在会话结束后基本就没用了或者只需要保留一个摘要。短期记忆通常放在内存或者带 TTL 的缓存里比如 Redis 设置 30 分钟过期或者直接放在进程内存里。长期记忆则是跨会话、跨时间持续存在的信息特点是生命周期长、写入频率低、需要检索。比如用户说“我对花生过敏”这条信息应该在三个月后的对话里依然生效。长期记忆需要持久化存储并且通常需要支持语义检索也就是用向量数据库来做相似度匹配。我见过太多项目把这两者塞进同一张 MySQL 表结果就是每次对话都要全表扫描历史记录QPS 一上来直接崩。正确的做法是分层存储短期用 Redis 或内存长期用向量库加关系库的组合。2.2 双网络记忆模型的启发热词里提到的“双网络记忆模型”和“记忆score时间半衰期”其实指向同一个思路记忆不是平等对待的每条记忆都应该有一个权重这个权重由重要性和时效性共同决定。我参考这个思路设计了一套打分机制。每条长期记忆在写入时计算一个基础分数来源包括用户显式声明的信息比如“我住在北京”给高分Agent 推断出来的信息比如从对话中推测用户可能喜欢咖啡给低分。然后引入时间半衰期比如设定半衰期为 30 天那么一条 30 天前的高分记忆现在的有效分数会衰减到原来的一半。具体公式可以简化为effective_score base_score * (0.5 ** (days_elapsed / half_life))检索的时候按 effective_score 排序低于阈值的记忆直接不返回。这样做的好处是过时的、不重要的记忆会自动沉底不需要手动清理也不会干扰当前对话。2.3 记忆编码从 1 到 100 的粒度控制“1到100记忆编码大全”这个热词听起来夸张但背后是一个真实需求记忆的粒度需要可控。粗粒度的记忆比如“用户是一个程序员”细粒度的记忆比如“用户上周三提到他在用 Python 3.11 写一个爬虫项目”。我的做法是把记忆分成几个层级事实层、偏好层、事件层、摘要层。事实层存不变的属性偏好层存喜好事件层存具体发生过的事摘要层存对话的压缩总结。每一层用不同的编码方式事实层和偏好层用结构化字段事件层和摘要层用向量加元数据。这样设计之后迁移的时候只需要导出这四层的数据换任何框架都能重新加载因为结构是自定义的不依赖任何框架的内部格式。3. 记忆与工具解耦的架构设计3.1 为什么要把记忆层独立出来工具会换框架会换模型会换但记忆是跟着业务走的。把记忆层独立成一个服务好处有三个第一迁移成本低换框架只需要改调用接口第二可以复用多个 Agent 共享同一份用户记忆第三便于调试记忆的读写有独立的日志和监控。我现在的项目里记忆层是一个独立的 Python 服务对外暴露 REST 和 gRPC 两种接口。Agent 框架通过 HTTP 调用它不直接碰数据库。这样即使我把编排框架从 LangChain 换成别的记忆层完全不用动。3.2 存储选型向量库加关系库的组合拳存储选型上我试过纯向量库、纯关系库、以及两者组合。结论是组合最优。关系库我用的是 PostgreSQL存结构化的事实和偏好比如用户 ID、属性名、属性值、创建时间、基础分数。向量库我用的是 Milvus也试过 Qdrant存事件和摘要的 embedding附带元数据指向关系库里的记录。为什么不只用向量库因为事实类查询需要精确匹配比如“查用户 ID 为 123 的所有偏好”向量库做这个很别扭。为什么不只用关系库因为语义检索需要向量相似度关系库做不了。两者通过一个统一的 memory_id 关联查询的时候先走关系库拿结构化数据再走向量库拿语义相关的事件最后合并排序。3.3 接口设计让记忆层对框架无感知接口设计的关键是不暴露任何框架特有的概念。我定义了四个核心接口write_memory(user_id, content, memory_type, metadata)写入一条记忆read_memory(user_id, query, top_k, memory_types)检索记忆update_memory(memory_id, content, metadata)更新记忆delete_memory(memory_id)删除记忆所有参数都是通用类型content 是字符串metadata 是字典memory_type 是枚举。框架调用的时候只需要把对话内容传进来不需要关心底层是 Redis 还是 PostgreSQL。这样设计之后我甚至可以用同一个记忆层同时服务两个不同的 Agent 框架一个用 LangChain一个用自己写的编排逻辑互不干扰。4. 实操从零搭一个可迁移的记忆层4.1 环境准备与依赖安装先列一下我用的技术栈和版本避免版本不一致导致踩坑Python 3.11PostgreSQL 15Milvus 2.3单机版用 Docker 起Redis 7短期记忆用FastAPI 0.104记忆层服务安装依赖pip install fastapi uvicorn psycopg2-binary pymilvus redis sentence-transformersembedding 模型我用的是all-MiniLM-L6-v2体积小、速度快适合本地部署。如果对精度要求高可以换成更大的模型但推理成本会上升。4.2 数据库表结构设计PostgreSQL 里建两张表一张存记忆主体一张存记忆的元数据索引CREATE TABLE memories ( memory_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id VARCHAR(64) NOT NULL, memory_type VARCHAR(32) NOT NULL, content TEXT NOT NULL, base_score FLOAT DEFAULT 1.0, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_user_type ON memories(user_id, memory_type);Milvus 里建一个 collection字段包括 memory_id、embedding、user_id、memory_typefrom pymilvus import CollectionSchema, FieldSchema, DataType fields [ FieldSchema(namememory_id, dtypeDataType.VARCHAR, max_length64, is_primaryTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim384), FieldSchema(nameuser_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(namememory_type, dtypeDataType.VARCHAR, max_length32), ] schema CollectionSchema(fields)dim384 是因为 MiniLM 的输出维度是 384换模型的话这个值要跟着改。4.3 写入与检索的核心逻辑写入的时候先算 embedding然后同时写 PostgreSQL 和 Milvus。这里要注意事务问题两个库没法做分布式事务我的做法是先写 PostgreSQL拿到 memory_id 之后再写 Milvus如果 Milvus 写失败就回滚 PostgreSQL 的记录。检索的时候分两步先根据 query 算 embedding去 Milvus 查 top_k 个相似记忆拿到 memory_id 列表然后去 PostgreSQL 查这些 memory_id 的详细信息和分数最后按 effective_score 排序返回。effective_score 的计算import math from datetime import datetime def effective_score(base_score, created_at, half_life_days30): days_elapsed (datetime.now() - created_at).days decay 0.5 ** (days_elapsed / half_life_days) return base_score * decay这个函数在检索时对每条记忆调用一次排序后取前 N 条。4.4 短期记忆的缓存策略短期记忆用 Redis 的 List 结构存key 是session:{session_id}每条消息是一个 JSON 字符串。设置 TTL 为 1800 秒也就是 30 分钟无活动自动过期。写入短期记忆import redis, json r redis.Redis(hostlocalhost, port6379, db0) def write_short_term(session_id, role, content): key fsession:{session_id} message json.dumps({role: role, content: content, ts: time.time()}) r.rpush(key, message) r.expire(key, 1800)读取的时候用lrange拿最近 N 条拼成上下文传给模型。会话结束时把短期记忆压缩成摘要写入长期记忆。5. 并发场景下记忆层怎么扛住压力5.1 并发写入的冲突问题“AI Agent 怎么扛并发”这个热词问到了点子上。记忆层的并发压力主要来自两个方面一是多个会话同时写入二是同一个用户的多轮对话快速写入。PostgreSQL 本身能扛住一定的并发但 Milvus 的写入在高并发下会有延迟。我的做法是引入一个写入队列用 Redis 的 Stream 做缓冲后台起一个消费者进程批量写入 Milvus。这样前端写入请求只需要写 PostgreSQL 和 Redis Stream响应时间稳定在 10ms 以内。批量写入的消费者逻辑def consume_and_write(): while True: messages r.xreadgroup(memory_group, consumer1, {memory_stream: }, count100, block1000) if messages: batch [] for _, msgs in messages: for msg_id, data in msgs: batch.append(data) # 批量算 embedding 并写入 Milvus write_batch_to_milvus(batch) r.xack(memory_stream, memory_group, msg_id)5.2 读取缓存的命中率优化检索是读多写少的场景缓存能大幅降低延迟。我在记忆层前面加了一层 Redis 缓存key 是query:{user_id}:{hash(query)}value 是检索结果的 JSON。TTL 设 300 秒。缓存命中率的关键是 query 的归一化。用户问“我喜欢什么”和“我的偏好是什么”语义相近但字符串不同直接做 key 会命中不了。我的做法是先用 embedding 算相似度如果和缓存里的 query 相似度超过 0.95就直接返回缓存结果。实测下来在客服场景下缓存命中率能到 60% 以上平均检索延迟从 80ms 降到 30ms。5.3 记忆更新的幂等性保证同一个用户可能在不同会话里重复说同一件事比如“我住在北京”说了三次。如果每次都写一条新记忆长期下来会有大量重复。我的做法是在写入前先做一次相似度检查如果已有记忆的 embedding 和待写入内容的相似度超过 0.9就更新已有记忆的 base_score 和 updated_at而不是新增。这样既避免了重复又能让反复提到的信息获得更高的权重。更新逻辑def write_with_dedup(user_id, content, memory_type): embedding encode(content) similar milvus_search(embedding, user_id, top_k1) if similar and similar[0].score 0.9: update_memory_score(similar[0].memory_id, boost0.1) else: insert_new_memory(user_id, content, memory_type, embedding)6. 迁移实战换框架不丢记忆6.1 导出与导入的标准化格式迁移的核心是格式标准化。我定义了一个 JSON Schema所有记忆导出都遵循这个格式{ version: 1.0, exported_at: 2024-01-15T10:00:00Z, memories: [ { memory_id: uuid, user_id: user123, memory_type: fact, content: 用户住在北京, base_score: 1.0, created_at: 2024-01-01T00:00:00Z, embedding: [0.1, 0.2, ...] } ] }embedding 也一起导出这样导入新系统时不需要重新算节省大量时间。如果新系统用的 embedding 模型不同那就需要重新编码但至少原始内容还在。6.2 从旧框架迁移的实操步骤假设要从一个用 LangChain 内置记忆的项目迁移到自建记忆层步骤是这样的第一步写一个脚本遍历旧框架的记忆存储把对话历史导出来。LangChain 的ConversationBufferMemory可以通过load_memory_variables拿到全部历史。第二步把历史对话按轮次切分每一轮用户输入和 Agent 回复合并成一条事件记忆调用记忆层的write_memory接口写入。第三步对重要的用户声明做提取比如用正则或小模型识别“我是...”“我喜欢...”“我不要...”这类句式单独写成事实记忆或偏好记忆。第四步验证迁移结果随机抽几条旧对话在新系统里检索看能否召回相关记忆。整个过程我实测下来一万条对话的迁移大概需要 20 分钟主要时间花在 embedding 计算上。如果导出时带了 embedding时间能压缩到 5 分钟以内。6.3 迁移后的验证清单迁移完不能直接上线要过一遍验证清单检查项验证方法通过标准记忆总数对比新旧系统 count误差小于 1%检索召回抽样 100 条 query召回率大于 90%分数衰减检查 30 天前记忆的分数符合半衰期公式并发写入压测 100 QPS无丢失、无重复缓存命中监控 1 小时命中率大于 50%这个清单我每次迁移都会跑一遍能提前发现大部分问题。7. 常见问题与排查技巧实录7.1 记忆检索不准的排查思路检索不准通常有三个原因embedding 模型不适合当前语言、相似度阈值设得不对、记忆粒度太粗。先检查 embedding 模型。如果用的是英文模型处理中文内容效果会很差。换成多语言模型或者中文模型召回率能提升一大截。再检查阈值。Milvus 默认返回 top_k 个结果但不带阈值过滤。如果 top_k 设成 10可能后 5 个都是不相关的。我的做法是加一个相似度阈值比如 0.7低于这个值的不返回。最后检查粒度。如果一条记忆里塞了太多信息比如“用户喜欢咖啡、住在北京、是程序员”检索“用户住哪”的时候可能召回这条但信息不精确。拆成三条独立记忆检索准确率会高很多。7.2 记忆膨胀导致成本失控长期运行的系统记忆会越来越多存储成本和检索延迟都会上升。我遇到过一个月涨了 50 万条记忆的情况Milvus 查询明显变慢。解决办法是定期做记忆压缩。把低分的、过时的记忆归档到冷存储或者合并成摘要。我写了一个定时任务每周跑一次把 effective_score 低于 0.1 的记忆标记为归档从 Milvus 里删除但保留在 PostgreSQL 里备查。归档逻辑def archive_low_score_memories(threshold0.1): memories pg_query(SELECT memory_id, base_score, created_at FROM memories) to_archive [] for m in memories: if effective_score(m.base_score, m.created_at) threshold: to_archive.append(m.memory_id) milvus_delete(to_archive) pg_update(UPDATE memories SET archived TRUE WHERE memory_id ANY(%s), (to_archive,))这样 Milvus 里只保留活跃记忆查询速度快成本也可控。7.3 多用户记忆隔离的坑多用户场景下记忆隔离是必须的。我见过一个项目因为没做隔离A 用户的偏好被 B 用户检索到了直接导致线上事故。隔离要在三个层面做数据库查询必须带 user_id 条件Milvus 检索必须带 user_id 过滤缓存 key 必须包含 user_id。任何一层漏了都会出问题。Milvus 的过滤表达式search_params {metric_type: IP, params: {nprobe: 10}} results collection.search( data[embedding], anns_fieldembedding, paramsearch_params, limit10, exprfuser_id {user_id} )这个 expr 一定要加不加就是全库检索既慢又不安全。7.4 记忆写入失败的兜底方案写入失败不能影响主流程。我的做法是写入操作全部异步化前端只负责把消息丢进队列后台慢慢处理。如果队列积压就降级为只写 PostgreSQLMilvus 的写入延后。降级逻辑def write_memory_safe(user_id, content, memory_type): try: write_to_pg(user_id, content, memory_type) write_to_stream(user_id, content, memory_type) except Exception as e: log_error(e) # 降级只写 PG标记待补 write_to_pg(user_id, content, memory_type, pending_embeddingTrue)后台有一个补偿任务定期扫描 pending_embedding 为 true 的记录补算 embedding 并写入 Milvus。8. 几个我踩过的坑和对应技巧第一个坑是 embedding 模型升级导致的历史记忆失效。我一开始用 MiniLM后来想换成效果更好的 BGE 模型结果发现两个模型的向量空间不兼容旧记忆的 embedding 在新模型下检索完全不准。解决办法是保留旧模型的编码器迁移时用旧模型重新编码所有历史记忆或者干脆双模型并行一段时间逐步切换。第二个坑是 Redis 短期记忆的序列化问题。我一开始用 pickle 序列化消息后来发现不同 Python 版本之间不兼容迁移环境后反序列化失败。换成 JSON 之后就没这个问题了虽然体积大一点但兼容性好。第三个坑是 Milvus 的 collection 重建。有一次我改了 schema想直接删了重建结果忘了先导出数据几万条记忆直接没了。从那以后我养成了习惯任何 schema 变更前先跑一次全量导出确认备份文件没问题再动手。第四个坑是时间半衰期的参数选择。半衰期设太短记忆衰减太快用户上周说的话这周就忘了设太长过时信息一直干扰检索。我试过 7 天、30 天、90 天最后在客服场景下选了 30 天在个人助手场景下选了 90 天。这个参数没有标准答案要根据业务场景调。第五个坑是并发写入时的重复记忆。两个请求同时判断“没有相似记忆”然后同时插入结果出现两条一样的。解决办法是在 PostgreSQL 里对 user_id 加 content 的哈希做唯一索引插入冲突时改为更新。CREATE UNIQUE INDEX idx_unique_memory ON memories(user_id, md5(content));这样即使并发插入数据库层面也能保证不重复。9. 记忆层的未来扩展方向这套架构跑了大半年稳定性没问题但我还在持续优化。一个方向是引入记忆的重要性自动评估用一个小模型对每条新记忆打分而不是靠规则。另一个方向是记忆的跨用户共享比如同一个团队的用户可以共享某些公共记忆这在企业场景下很有用。还有一个我比较感兴趣的方向是记忆的可解释性。现在检索出来一条记忆只能看到内容和分数但为什么这条记忆被召回、它对当前对话有什么影响是不透明的。如果能给每条记忆附上一个“召回理由”调试起来会方便很多。这些扩展都不影响核心架构因为记忆层和工具层已经解耦了上层怎么变底层的数据结构和接口都不用动。这也是我一开始坚持解耦的原因工具会过时框架会淘汰但一份结构清晰、格式标准的记忆资产可以一直用下去。
Agent记忆不跟工具搬家：分层存储与可迁移架构实战
1. 为什么 Agent 的记忆总在“搬家”做过 Agent 项目的人大概都有过这种体验今天用某个框架搭了一个能记住用户偏好的助手明天想换个编排引擎或者换一家模型服务结果发现之前攒下来的对话历史、用户画像、任务上下文全都带不走。要么导出成一份谁都不认识的 JSON要么干脆锁死在某个平台的数据库里迁移一次等于重做一次。标题里说的“记忆不跟着工具搬家”戳的就是这个痛点。我自己从最早用 LangChain 的ConversationBufferMemory到后来自己写 Redis 存对话再到尝试各种带长期记忆的 Agent 框架前后踩了不下十次“记忆迁移”的坑。最夸张的一次是给一个客服 Agent 换了底层编排框架结果三万多条历史对话的向量索引全部作废因为新旧框架对 embedding 的存储结构定义完全不一样。那次之后我就下定决心记忆层必须和工具层解耦。这篇内容适合两类人看一类是正在做 Agent 开发、被记忆管理搞得焦头烂额的工程师另一类是想入门 AI Agent、但还没意识到“记忆”这件事有多麻烦的初学者。我会把记忆分层的思路、存储选型、迁移方案、并发处理、以及实际落地时那些文档里不会写的坑全部摊开讲一遍。核心观点就一句话Agent 的记忆应该是一份独立于任何框架的资产工具可以换记忆不能丢。2. Agent 记忆到底分几层别再混着存了2.1 短期记忆和长期记忆的本质区别很多人一上来就把所有对话往一个数据库里塞这是最典型的错误。短期记忆和长期记忆在生命周期、访问频率、存储介质上的要求完全不同混在一起存查询会越来越慢成本会越来越高。短期记忆指的是当前会话窗口内的上下文它的特点是生命周期短、读写频繁、容量有限。比如用户问“帮我订明天去上海的机票”Agent 需要记住前面几轮提到的出发城市、时间偏好、舱位要求。这些信息在会话结束后基本就没用了或者只需要保留一个摘要。短期记忆通常放在内存或者带 TTL 的缓存里比如 Redis 设置 30 分钟过期或者直接放在进程内存里。长期记忆则是跨会话、跨时间持续存在的信息特点是生命周期长、写入频率低、需要检索。比如用户说“我对花生过敏”这条信息应该在三个月后的对话里依然生效。长期记忆需要持久化存储并且通常需要支持语义检索也就是用向量数据库来做相似度匹配。我见过太多项目把这两者塞进同一张 MySQL 表结果就是每次对话都要全表扫描历史记录QPS 一上来直接崩。正确的做法是分层存储短期用 Redis 或内存长期用向量库加关系库的组合。2.2 双网络记忆模型的启发热词里提到的“双网络记忆模型”和“记忆score时间半衰期”其实指向同一个思路记忆不是平等对待的每条记忆都应该有一个权重这个权重由重要性和时效性共同决定。我参考这个思路设计了一套打分机制。每条长期记忆在写入时计算一个基础分数来源包括用户显式声明的信息比如“我住在北京”给高分Agent 推断出来的信息比如从对话中推测用户可能喜欢咖啡给低分。然后引入时间半衰期比如设定半衰期为 30 天那么一条 30 天前的高分记忆现在的有效分数会衰减到原来的一半。具体公式可以简化为effective_score base_score * (0.5 ** (days_elapsed / half_life))检索的时候按 effective_score 排序低于阈值的记忆直接不返回。这样做的好处是过时的、不重要的记忆会自动沉底不需要手动清理也不会干扰当前对话。2.3 记忆编码从 1 到 100 的粒度控制“1到100记忆编码大全”这个热词听起来夸张但背后是一个真实需求记忆的粒度需要可控。粗粒度的记忆比如“用户是一个程序员”细粒度的记忆比如“用户上周三提到他在用 Python 3.11 写一个爬虫项目”。我的做法是把记忆分成几个层级事实层、偏好层、事件层、摘要层。事实层存不变的属性偏好层存喜好事件层存具体发生过的事摘要层存对话的压缩总结。每一层用不同的编码方式事实层和偏好层用结构化字段事件层和摘要层用向量加元数据。这样设计之后迁移的时候只需要导出这四层的数据换任何框架都能重新加载因为结构是自定义的不依赖任何框架的内部格式。3. 记忆与工具解耦的架构设计3.1 为什么要把记忆层独立出来工具会换框架会换模型会换但记忆是跟着业务走的。把记忆层独立成一个服务好处有三个第一迁移成本低换框架只需要改调用接口第二可以复用多个 Agent 共享同一份用户记忆第三便于调试记忆的读写有独立的日志和监控。我现在的项目里记忆层是一个独立的 Python 服务对外暴露 REST 和 gRPC 两种接口。Agent 框架通过 HTTP 调用它不直接碰数据库。这样即使我把编排框架从 LangChain 换成别的记忆层完全不用动。3.2 存储选型向量库加关系库的组合拳存储选型上我试过纯向量库、纯关系库、以及两者组合。结论是组合最优。关系库我用的是 PostgreSQL存结构化的事实和偏好比如用户 ID、属性名、属性值、创建时间、基础分数。向量库我用的是 Milvus也试过 Qdrant存事件和摘要的 embedding附带元数据指向关系库里的记录。为什么不只用向量库因为事实类查询需要精确匹配比如“查用户 ID 为 123 的所有偏好”向量库做这个很别扭。为什么不只用关系库因为语义检索需要向量相似度关系库做不了。两者通过一个统一的 memory_id 关联查询的时候先走关系库拿结构化数据再走向量库拿语义相关的事件最后合并排序。3.3 接口设计让记忆层对框架无感知接口设计的关键是不暴露任何框架特有的概念。我定义了四个核心接口write_memory(user_id, content, memory_type, metadata)写入一条记忆read_memory(user_id, query, top_k, memory_types)检索记忆update_memory(memory_id, content, metadata)更新记忆delete_memory(memory_id)删除记忆所有参数都是通用类型content 是字符串metadata 是字典memory_type 是枚举。框架调用的时候只需要把对话内容传进来不需要关心底层是 Redis 还是 PostgreSQL。这样设计之后我甚至可以用同一个记忆层同时服务两个不同的 Agent 框架一个用 LangChain一个用自己写的编排逻辑互不干扰。4. 实操从零搭一个可迁移的记忆层4.1 环境准备与依赖安装先列一下我用的技术栈和版本避免版本不一致导致踩坑Python 3.11PostgreSQL 15Milvus 2.3单机版用 Docker 起Redis 7短期记忆用FastAPI 0.104记忆层服务安装依赖pip install fastapi uvicorn psycopg2-binary pymilvus redis sentence-transformersembedding 模型我用的是all-MiniLM-L6-v2体积小、速度快适合本地部署。如果对精度要求高可以换成更大的模型但推理成本会上升。4.2 数据库表结构设计PostgreSQL 里建两张表一张存记忆主体一张存记忆的元数据索引CREATE TABLE memories ( memory_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id VARCHAR(64) NOT NULL, memory_type VARCHAR(32) NOT NULL, content TEXT NOT NULL, base_score FLOAT DEFAULT 1.0, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_user_type ON memories(user_id, memory_type);Milvus 里建一个 collection字段包括 memory_id、embedding、user_id、memory_typefrom pymilvus import CollectionSchema, FieldSchema, DataType fields [ FieldSchema(namememory_id, dtypeDataType.VARCHAR, max_length64, is_primaryTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim384), FieldSchema(nameuser_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(namememory_type, dtypeDataType.VARCHAR, max_length32), ] schema CollectionSchema(fields)dim384 是因为 MiniLM 的输出维度是 384换模型的话这个值要跟着改。4.3 写入与检索的核心逻辑写入的时候先算 embedding然后同时写 PostgreSQL 和 Milvus。这里要注意事务问题两个库没法做分布式事务我的做法是先写 PostgreSQL拿到 memory_id 之后再写 Milvus如果 Milvus 写失败就回滚 PostgreSQL 的记录。检索的时候分两步先根据 query 算 embedding去 Milvus 查 top_k 个相似记忆拿到 memory_id 列表然后去 PostgreSQL 查这些 memory_id 的详细信息和分数最后按 effective_score 排序返回。effective_score 的计算import math from datetime import datetime def effective_score(base_score, created_at, half_life_days30): days_elapsed (datetime.now() - created_at).days decay 0.5 ** (days_elapsed / half_life_days) return base_score * decay这个函数在检索时对每条记忆调用一次排序后取前 N 条。4.4 短期记忆的缓存策略短期记忆用 Redis 的 List 结构存key 是session:{session_id}每条消息是一个 JSON 字符串。设置 TTL 为 1800 秒也就是 30 分钟无活动自动过期。写入短期记忆import redis, json r redis.Redis(hostlocalhost, port6379, db0) def write_short_term(session_id, role, content): key fsession:{session_id} message json.dumps({role: role, content: content, ts: time.time()}) r.rpush(key, message) r.expire(key, 1800)读取的时候用lrange拿最近 N 条拼成上下文传给模型。会话结束时把短期记忆压缩成摘要写入长期记忆。5. 并发场景下记忆层怎么扛住压力5.1 并发写入的冲突问题“AI Agent 怎么扛并发”这个热词问到了点子上。记忆层的并发压力主要来自两个方面一是多个会话同时写入二是同一个用户的多轮对话快速写入。PostgreSQL 本身能扛住一定的并发但 Milvus 的写入在高并发下会有延迟。我的做法是引入一个写入队列用 Redis 的 Stream 做缓冲后台起一个消费者进程批量写入 Milvus。这样前端写入请求只需要写 PostgreSQL 和 Redis Stream响应时间稳定在 10ms 以内。批量写入的消费者逻辑def consume_and_write(): while True: messages r.xreadgroup(memory_group, consumer1, {memory_stream: }, count100, block1000) if messages: batch [] for _, msgs in messages: for msg_id, data in msgs: batch.append(data) # 批量算 embedding 并写入 Milvus write_batch_to_milvus(batch) r.xack(memory_stream, memory_group, msg_id)5.2 读取缓存的命中率优化检索是读多写少的场景缓存能大幅降低延迟。我在记忆层前面加了一层 Redis 缓存key 是query:{user_id}:{hash(query)}value 是检索结果的 JSON。TTL 设 300 秒。缓存命中率的关键是 query 的归一化。用户问“我喜欢什么”和“我的偏好是什么”语义相近但字符串不同直接做 key 会命中不了。我的做法是先用 embedding 算相似度如果和缓存里的 query 相似度超过 0.95就直接返回缓存结果。实测下来在客服场景下缓存命中率能到 60% 以上平均检索延迟从 80ms 降到 30ms。5.3 记忆更新的幂等性保证同一个用户可能在不同会话里重复说同一件事比如“我住在北京”说了三次。如果每次都写一条新记忆长期下来会有大量重复。我的做法是在写入前先做一次相似度检查如果已有记忆的 embedding 和待写入内容的相似度超过 0.9就更新已有记忆的 base_score 和 updated_at而不是新增。这样既避免了重复又能让反复提到的信息获得更高的权重。更新逻辑def write_with_dedup(user_id, content, memory_type): embedding encode(content) similar milvus_search(embedding, user_id, top_k1) if similar and similar[0].score 0.9: update_memory_score(similar[0].memory_id, boost0.1) else: insert_new_memory(user_id, content, memory_type, embedding)6. 迁移实战换框架不丢记忆6.1 导出与导入的标准化格式迁移的核心是格式标准化。我定义了一个 JSON Schema所有记忆导出都遵循这个格式{ version: 1.0, exported_at: 2024-01-15T10:00:00Z, memories: [ { memory_id: uuid, user_id: user123, memory_type: fact, content: 用户住在北京, base_score: 1.0, created_at: 2024-01-01T00:00:00Z, embedding: [0.1, 0.2, ...] } ] }embedding 也一起导出这样导入新系统时不需要重新算节省大量时间。如果新系统用的 embedding 模型不同那就需要重新编码但至少原始内容还在。6.2 从旧框架迁移的实操步骤假设要从一个用 LangChain 内置记忆的项目迁移到自建记忆层步骤是这样的第一步写一个脚本遍历旧框架的记忆存储把对话历史导出来。LangChain 的ConversationBufferMemory可以通过load_memory_variables拿到全部历史。第二步把历史对话按轮次切分每一轮用户输入和 Agent 回复合并成一条事件记忆调用记忆层的write_memory接口写入。第三步对重要的用户声明做提取比如用正则或小模型识别“我是...”“我喜欢...”“我不要...”这类句式单独写成事实记忆或偏好记忆。第四步验证迁移结果随机抽几条旧对话在新系统里检索看能否召回相关记忆。整个过程我实测下来一万条对话的迁移大概需要 20 分钟主要时间花在 embedding 计算上。如果导出时带了 embedding时间能压缩到 5 分钟以内。6.3 迁移后的验证清单迁移完不能直接上线要过一遍验证清单检查项验证方法通过标准记忆总数对比新旧系统 count误差小于 1%检索召回抽样 100 条 query召回率大于 90%分数衰减检查 30 天前记忆的分数符合半衰期公式并发写入压测 100 QPS无丢失、无重复缓存命中监控 1 小时命中率大于 50%这个清单我每次迁移都会跑一遍能提前发现大部分问题。7. 常见问题与排查技巧实录7.1 记忆检索不准的排查思路检索不准通常有三个原因embedding 模型不适合当前语言、相似度阈值设得不对、记忆粒度太粗。先检查 embedding 模型。如果用的是英文模型处理中文内容效果会很差。换成多语言模型或者中文模型召回率能提升一大截。再检查阈值。Milvus 默认返回 top_k 个结果但不带阈值过滤。如果 top_k 设成 10可能后 5 个都是不相关的。我的做法是加一个相似度阈值比如 0.7低于这个值的不返回。最后检查粒度。如果一条记忆里塞了太多信息比如“用户喜欢咖啡、住在北京、是程序员”检索“用户住哪”的时候可能召回这条但信息不精确。拆成三条独立记忆检索准确率会高很多。7.2 记忆膨胀导致成本失控长期运行的系统记忆会越来越多存储成本和检索延迟都会上升。我遇到过一个月涨了 50 万条记忆的情况Milvus 查询明显变慢。解决办法是定期做记忆压缩。把低分的、过时的记忆归档到冷存储或者合并成摘要。我写了一个定时任务每周跑一次把 effective_score 低于 0.1 的记忆标记为归档从 Milvus 里删除但保留在 PostgreSQL 里备查。归档逻辑def archive_low_score_memories(threshold0.1): memories pg_query(SELECT memory_id, base_score, created_at FROM memories) to_archive [] for m in memories: if effective_score(m.base_score, m.created_at) threshold: to_archive.append(m.memory_id) milvus_delete(to_archive) pg_update(UPDATE memories SET archived TRUE WHERE memory_id ANY(%s), (to_archive,))这样 Milvus 里只保留活跃记忆查询速度快成本也可控。7.3 多用户记忆隔离的坑多用户场景下记忆隔离是必须的。我见过一个项目因为没做隔离A 用户的偏好被 B 用户检索到了直接导致线上事故。隔离要在三个层面做数据库查询必须带 user_id 条件Milvus 检索必须带 user_id 过滤缓存 key 必须包含 user_id。任何一层漏了都会出问题。Milvus 的过滤表达式search_params {metric_type: IP, params: {nprobe: 10}} results collection.search( data[embedding], anns_fieldembedding, paramsearch_params, limit10, exprfuser_id {user_id} )这个 expr 一定要加不加就是全库检索既慢又不安全。7.4 记忆写入失败的兜底方案写入失败不能影响主流程。我的做法是写入操作全部异步化前端只负责把消息丢进队列后台慢慢处理。如果队列积压就降级为只写 PostgreSQLMilvus 的写入延后。降级逻辑def write_memory_safe(user_id, content, memory_type): try: write_to_pg(user_id, content, memory_type) write_to_stream(user_id, content, memory_type) except Exception as e: log_error(e) # 降级只写 PG标记待补 write_to_pg(user_id, content, memory_type, pending_embeddingTrue)后台有一个补偿任务定期扫描 pending_embedding 为 true 的记录补算 embedding 并写入 Milvus。8. 几个我踩过的坑和对应技巧第一个坑是 embedding 模型升级导致的历史记忆失效。我一开始用 MiniLM后来想换成效果更好的 BGE 模型结果发现两个模型的向量空间不兼容旧记忆的 embedding 在新模型下检索完全不准。解决办法是保留旧模型的编码器迁移时用旧模型重新编码所有历史记忆或者干脆双模型并行一段时间逐步切换。第二个坑是 Redis 短期记忆的序列化问题。我一开始用 pickle 序列化消息后来发现不同 Python 版本之间不兼容迁移环境后反序列化失败。换成 JSON 之后就没这个问题了虽然体积大一点但兼容性好。第三个坑是 Milvus 的 collection 重建。有一次我改了 schema想直接删了重建结果忘了先导出数据几万条记忆直接没了。从那以后我养成了习惯任何 schema 变更前先跑一次全量导出确认备份文件没问题再动手。第四个坑是时间半衰期的参数选择。半衰期设太短记忆衰减太快用户上周说的话这周就忘了设太长过时信息一直干扰检索。我试过 7 天、30 天、90 天最后在客服场景下选了 30 天在个人助手场景下选了 90 天。这个参数没有标准答案要根据业务场景调。第五个坑是并发写入时的重复记忆。两个请求同时判断“没有相似记忆”然后同时插入结果出现两条一样的。解决办法是在 PostgreSQL 里对 user_id 加 content 的哈希做唯一索引插入冲突时改为更新。CREATE UNIQUE INDEX idx_unique_memory ON memories(user_id, md5(content));这样即使并发插入数据库层面也能保证不重复。9. 记忆层的未来扩展方向这套架构跑了大半年稳定性没问题但我还在持续优化。一个方向是引入记忆的重要性自动评估用一个小模型对每条新记忆打分而不是靠规则。另一个方向是记忆的跨用户共享比如同一个团队的用户可以共享某些公共记忆这在企业场景下很有用。还有一个我比较感兴趣的方向是记忆的可解释性。现在检索出来一条记忆只能看到内容和分数但为什么这条记忆被召回、它对当前对话有什么影响是不透明的。如果能给每条记忆附上一个“召回理由”调试起来会方便很多。这些扩展都不影响核心架构因为记忆层和工具层已经解耦了上层怎么变底层的数据结构和接口都不用动。这也是我一开始坚持解耦的原因工具会过时框架会淘汰但一份结构清晰、格式标准的记忆资产可以一直用下去。
