“ai-memory”这个词最近在项目群里出现的频率高得吓人。上个月我们给一个智能客服系统做用户回访反馈最集中的一条不是“回答得不对”而是“它根本不记得我之前说过什么”。用户把手机号、收货地址、上周投诉的完整经过重新复述一遍之后耐心基本就耗尽了。AI记忆或者说 ai-memory本质上就是解决这个“说完就忘”的问题——让系统记住用户是谁、聊过什么、偏好什么并在下一次交互时把这些信息真正用起来。这篇文章不聊空中楼阁的概念直接讲我是怎么用一周时间把一个带记忆的对话机器人从零跑起来的以及过程中踩过的坑和最终沉淀下来的方案希望能给正在做类似事情的人省点时间。1. 先搞清楚“AI记忆”到底该记什么1.1 用户说“你上次不是知道吗”的时候问题出在哪大部分对话系统的架构本质上是“无状态”的。也就是说每一次用户发来的消息对系统来说都像是第一次见面。模型能记住的只有当前这一轮对话窗口里塞进去的那几千个 token一旦超出窗口长度或者会话结束前面的内容就全被冲掉了。这带来一个很尴尬的体验用户上周刚在客服里报修了一台设备这周再打开对话又得从头描述一遍故障现象。更糟的是用户在某次沟通里明确说了“我喜欢邮件通知别打电话”下次系统依然用默认方式去联系他。这种“失忆”不是模型笨而是架构上压根没有给模型留一块“长期记事本”。所以做 ai-memory 的第一步不是急着上技术而是想清楚一个问题到底哪些信息值得被记住。如果什么都记记忆库很快就会变成一堆垃圾信息如果只记对话记录那和日志系统没有区别。我最终把记忆拆成了四类后面所有的存储设计、召回策略都是围绕这个分类展开的。1.2 把记忆拆成四类设计才不混乱这四类分别是事实记忆、会话记忆、偏好记忆和行为记忆它们的来源、更新频率和用途都完全不同。记忆类型说明典型示例更新频率事实记忆用户明确说出的客观信息手机号、收货地址、家庭成员低极少变化会话记忆当前或历史对话中的具体事件上周报修过冰箱、出差时间中随事件发生偏好记忆用户表达的喜好与习惯喜欢图文回复、夜间安静低但需要长期维护行为记忆系统观察到的用户行为模式每周一早上查物流、常在地铁阅读持续更新不同记忆类型的“保质期”也不一样。事实记忆和偏好记忆是长期的核心资产必须精细维护会话记忆则偏向短期过一段时间就该归档或弱化行为记忆是通过统计归纳出来的可能只是阶段性规律需要定期重新计算。这个分类直接影响后面的数据表设计和向量库索引。我当时踩的第一个坑就是把所有类型的记忆统一塞进一张表里结果召回时经常把“用户昨天想吃火锅”这种一次性会话记忆和“用户对花生过敏”这种终身事实混为一谈回复质量反而下降。拆开之后每条记忆被赋予不同的衰减策略和重要度权重问题立刻清爽很多。2. 核心细节记忆的存储结构与召回机制2.1 一份能直接跑通的记忆数据结构想清楚记忆类型之后下一步就是设计存储结构。我的方案是用一张主表存记忆内容用向量索引支撑语义检索字段设计如下面的 JSON 所示。这个结构看起来简单但每一个字段都是我测试过之后才留下的。{ memory_id: mem_8f3a9d2c1e, user_id: user_10086, type: fact, // fact | session | preference | behavior content: 用户李明的收货地址是杭州市西湖区文三路138号, entities: [李明, 杭州, 文三路138号], importance: 0.85, // 0~1越高越不容易被遗忘 access_count: 7, // 被召回次数用于统计活跃度 created_at: 2025-01-10T10:22:00Z, last_accessed_at: 2025-01-12T09:30:00Z, expires_at: null, // 过期时间null 表示长期有效 source_message_id: msg_2233, // 回溯依据 embedding: [0.012, -0.045, ...] // 768维向量 }importance和expires_at是这套结构里的两个灵魂字段。重要性决定了记忆在召回时能拿到多少加权分过期时间则让系统能优雅地处理那些“记了也没用”的信息。比如“用户今天午饭想吃面”这条会话记忆expires_at设为当天24点过期之后就不再参与召回避免它在一个月后莫名其妙地影响推荐结果。实体抽取那块我建议单独拆出来做不要在写入记忆的主流程里写得很复杂。刚开始我为了省事直接用大模型做一步式的“内容实体”生成结果速度和成本都压不住。后来改成先用规则抽时间、数字、地点这类硬实体再让模型补泛化实体效果好很多。2.2 为什么必须上向量检索关键词匹配差在哪很多人第一次做记忆召回第一反应是“用 LIKE 查询不就行了吗”。我试过效果确实不行。原因是用户表达同一个意思的方式太多了关键词匹配根本接不住。举个例子用户在 3 月 10 日说过“我下周要飞北京出差”。一周之后系统需要召回相关记忆时用户可能会说“帮我查一下去首都机场的航班”。“北京”和“首都机场”在字面上完全没有重合但如果把两段话分别转成向量它们的语义距离是非常近的向量检索就能把这条记忆捞出来。这就是生活里“换了个说法但意思没变”的场景关键词抓不住向量能抓住。向量检索的本质是把文本转换成一串数字让语义相近的句子在数字空间里距离更近。实际操作中我直接用向量数据库具体方案见第 3 节来存储和检索每一条记忆在写入时用嵌入模型生成一个 embedding检索时把当前用户消息也转成向量再去库里找距离最近的 Top-K 条记忆。2.3 召回时的排序相似度、时间衰减与重要度向量检索只是拿到“候选记忆”真正决定用户体验的是排序逻辑。只按相似度排序会出现两类问题一是老旧的记忆不断被重复召回挤掉了近期更相关信息二是那些完全不相干但字面上相似的信息混进来。我的排序公式很简单但非常管用score 0.55 * semantic_similarity 0.30 * importance - 0.15 * time_decaysemantic_similarity是向量距离换算出的相似度importance是记忆自带的重要度time_decay是根据last_accessed_at计算出的衰减值越久没被用到衰减越大。三个权重是我做了几轮人工评估调出来的先满足“当前问题相关”这个第一优先级再让重要事实稳定浮现最后让冷门记忆慢慢沉底。一个值得注意的细节importance不是写死的。当一条记忆被成功召回多次并且用户在后续交互中主动使用了其中的信息比如报出了自己的订单号系统就应该给它加一点分。我给每条记忆设了一个阈值低于 0.2 的长期未命中记忆会被自动清理这能有效防止记忆库越堆越臃肿。3. 实操从零给对话机器人装上记忆的完整流程3.1 选型嵌入模型、向量库、对话框架怎么配我落地这个方案时主要关注的是三个环节的选型嵌入模型、向量数据库和对话框架的缝合方式。嵌入模型我选了通用场景表现稳定、中文支持好的那一类开源模型。选择时关键看两个指标一是向量维度太高的维度会显著推高存储和检索成本二是推理速度因为每条用户消息进来都要现场生成一个 query 向量如果模型太慢用户感知到的延迟会非常明显。建议先用本地小模型跑通全流程再根据线上延迟决定是否换更大的模型或加缓存。向量数据库的选择直接取决于你的部署环境。我对比了三种典型方案方案部署方式适合场景缺点Chroma嵌入式随应用启动小规模、本地 Demo、单机测试数据量大时性能一般Milvus独立服务可分布式生产环境、数据量大、高并发部署运维成本较高pgvectorPostgreSQL 插件已有 PG 体系的团队检索性能不如专业向量库我自己的测试环境用 Chroma 起步生产环境迁移到了支持分布式部署的向量服务。迁移过程中没有改业务代码因为我把向量操作封装成了一个统一接口换后端只动配置不动逻辑。这一点强烈建议在一开始就做好。对话框架不用刻意追求某个特定平台。实际上只要模型 API 支持 system prompt 注入记忆就能塞进去。我的原则是尽量少改原有对话链路把记忆模块做成一个独立的中间层负责写入、召回、排序最后把结果拼到 system prompt 里。3.2 记忆写入管线消息进来自动拆成记忆完整的记忆写入管线分四步接收新消息、判断是否值得记忆、抽取核心信息、写入存储并计算向量。下面是我简化后的核心代码基本可以直接拿去改改跑通。import json import uuid from datetime import datetime, timedelta from typing import Dict, List import chromadb from sentence_transformers import SentenceTransformer class MemoryWriter: def __init__(self, embed_model_name: str BAAI/bge-m3): self.encoder SentenceTransformer(embed_model_name) self.client chromadb.Client() self.collection self.client.get_or_create_collection(user_memories) # 简单规则包含时间、地点、联系方式等关键词的消息才进入记忆判断 self.trigger_keywords [电话, 地址, 住, 下周, 喜欢, 讨厌, 过敏, 订单号, 邮箱] def should_extract(self, message: str) - bool: # 先用规则做一次粗筛避免所有闲聊都触发记忆写入 return any(kw in message for kw in self.trigger_keywords) def extract_memory(self, message: str, user_id: str) - Dict: # 实际项目中这里会调用大模型做结构化抽取这里简化演示 # 返回的 dict 至少包含 type/content/importance/expires_at return { type: fact, content: message, importance: 0.8, expires_at: None } def write(self, message: str, user_id: str, source_message_id: str): if not self.should_extract(message): return None mem self.extract_memory(message, user_id) memory_id mem_ uuid.uuid4().hex[:12] embedding self.encoder.encode(mem[content]).tolist() self.collection.upsert( ids[memory_id], documents[mem[content]], metadatas[{ user_id: user_id, type: mem[type], importance: mem[importance], expires_at: mem[expires_at] or , source_message_id: source_message_id, created_at: datetime.utcnow().isoformat() }], embeddings[embedding] ) return memory_id这套代码里最值得琢磨的是should_extract这一步。最初我天真地以为每条消息都值得记忆结果客服系统上线半天就存了几万条“你好”“谢谢”之类的垃圾记忆。加了一道规则粗筛之后写入量降到了原来的十分之一而且后续召回的准确率明显提升。3.3 记忆召回与 Prompt 注入让模型“想起来”写入做好了下一步是召回。每次用户发来新消息系统把当前消息转成向量从库里捞回 Top-K 条记忆再拼到 system prompt 里。拼的时候一定不要一股脑把原始记忆全塞进去而是先做一层“摘要化”。class MemoryRetriever: def __init__(self): self.encoder SentenceTransformer(BAAI/bge-m3) self.client chromadb.Client() self.collection self.client.get_collection(user_memories) def recall(self, user_id: str, query: str, top_k: int 5) - List[str]: # 先按 user_id 过滤防止串记忆 query_embedding self.encoder.encode(query).tolist() results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, where{user_id: user_id} # 关键强制隔离用户 ) return results[documents][0] if results[documents] else [] def build_system_prompt(self, base_prompt: str, memories: List[str]) - str: if not memories: return base_prompt memory_block \n.join([f- {m} for m in memories]) return base_prompt \n\n以下是关于用户的长期记忆回答时请优先参考\n memory_block这个where{user_id: user_id}的过滤条件是防止“记忆串台”的关键防线。如果不加这个过滤向量库会在全量数据里找最相似的记忆一旦有相似用户就会出现 A 的地址被 B 的对话给调用的严重事故。关于串台的问题后面第 4 节还会详细讲。召回数量top_k同样需要调参。我试过 3、5、8、10 这四档最后发现 5 条左右效果最好。太少时关键信息经常漏掉太多时 Prompt 被无关记忆占满模型反而变得“思前想后”回答问题犹豫不决。如果你的对话窗口较大、模型上下文较长可以适当上调到 7~8 条但一定要留足空间给真正的用户输入。3.4 记忆更新的两个时机冲突覆盖与用户纠正记忆不是写完就一劳永逸的。用户可能会改地址、换手机号也会否定之前说过的话。如果只写不改旧记忆就会和新信息打架。我总结了两个必须触发更新操作的时机。第一个时机是检测到冲突覆盖。当新收到的信息与某条旧记忆的核心字段重叠时应当把旧记忆标记为无效然后写入新记忆。比如用户说“我搬家了新地址是浦东新区”旧记忆里的“家住浦西”如果不失效下次召回时就可能给出两个矛盾结论。我的做法是抽取实体之后先查同用户、同实体的旧记忆如果存在且内容不同就把旧记忆的expires_at设置为当前时间再写入新记忆。第二个时机是用户明确纠正。当用户说“我上次说的那个不对”“其实不是这样”这类反馈时模型应该主动触发一条记忆修正流程。这块不能光靠规则必须在对话主流程里接一个“意图分类”的钩子识别出纠正意图后调用记忆更新接口。项目初期我没有做这个结果用户纠正了三次系统还在坚持错误的旧记忆用户体验非常糟糕。def update_memory_on_conflict(user_id: str, old_content: str, new_content: str): # 简化示意通过内容匹配旧记忆进行失效处理 client chromadb.Client() collection client.get_collection(user_memories) existing collection.query( query_texts[old_content], n_results1, where{user_id: user_id} ) if existing[ids]: mem_id existing[ids][0][0] collection.update( ids[mem_id], metadatas[{expires_at: datetime.utcnow().isoformat()}] ) # 再写入新记忆 writer MemoryWriter() writer.write(new_content, user_id, source_message_idcorrection)4. 常见问题与排查实录4.1 记忆串台A 用户的信息跑到 B 用户那里去了这是最严重、最要命的问题。我测试环境里出现过一次用户甲问“我的收货地址是什么”机器人答出了用户乙的地址。当时第一反应是向量库查询条件写错了仔细查完发现问题出在我最初设计时没有在索引层面强隔离用户。排查和修复的关键有两点。第一写入和查询都要带上user_id过滤这个在前面的代码里已经体现了第二向量库的元数据过滤如果没生效会在集合层面直接污染结果。我建议在测试阶段写一个“串台自检脚本”批量插入几百对相似用户的数据然后随机抽取查询断言返回的结果必须属于同一个user_id。把这个脚本挂到 CI 里比人肉测试可靠得多。4.2 记忆膨胀存得太多回复反而变差记忆库越积越多之后会出现一种反直觉的现象召回到的记忆越多模型回复质量反而下降。原因是 Top-K 里挤满了细节性记忆真正关键的核心事实被挤掉了。比如用户问“我要退货”系统召回 5 条记忆里 3 条都是“用户上周某天在某店铺买过某个颜色的小电器”这种低价值记录。解决办法要双管齐下。一方面在写入时提高门槛把should_extract的触发条件收得再严一些另一方面建立定期压缩机制对低重要度且长期未命中的记忆做合并或删除。我现在每周跑一次离线清理任务把importance 0.3且last_accessed_at超过 30 天的记忆批量删除。经过一轮清理召回准确率能从低谷回升一成以上这个数字相当可观。4.3 隐私安全用户说“你怎么知道我的手机号”记忆功能天然会触碰用户的隐私边界。用户说出一句“你怎么知道我的手机号”说明系统在“不该说出信息的时候”把记忆暴露了。这种问题非常敏感必须在设计阶段就做出克制。我最终的策略是给记忆加上“可见性”等级。核心的联系方式、家庭住址等只允许在用户明确询问时被召回一般的偏好和习惯可以在上下文需要时使用浏览行为类记忆则要经过用户授权才能参与推荐。实际实现时就是在元数据里加一个visibility字段召回排序之后再过一道白名单校验。这个策略牺牲了一部分“智能感”但换来了用户的信任感长期看是值得的。4.4 记忆存进去了但始终没有被召回这是另一个高频问题记忆明明写进去了内容也正确但对话时模型完全没反应。排查这类问题我建议按下面的表格逐个环节查比自己瞎猜快得多。现象可能原因排查方法召回结果为空user_id过滤条件写反或者索引没有建到对应分区打印召回 SQL检查 where 条件召回了但排在后面相似度分数低被其他记忆挤掉临时调大 TopK 查看候选集看目标记忆是否在里面模型没使用召回记忆Prompt 里记忆被放在很靠后的位置被系统截断检查 token 用量确保 system prompt 完整注入记忆未更新写入管线没有触发should_extract加日志打印每条消息的判断结果最后一个问题经常被忽略如果你的对话框架有上下文截断策略记忆虽然在 Prompt 中但处在被截断的位置模型实际上看不到。我建议把记忆块放在 system prompt 的最前面确保它优先被处理而不是附在末尾。5. 从 Demo 到生产三个容易被忽视的工程问题5.1 记忆的可解释与可删除Demo 阶段你可以不管记忆的可解释性但生产环境必须让用户能看到“你记住了我的什么”。我给系统加了一个“记忆管理面板”用户可以查看所有关于自己的记忆条目可以手动删除某一条也可以一键清空。这个功能对合规和信任都有巨大帮助。技术上这就意味着每条记忆必须保留source_message_id和created_at方便回溯来源。当用户删除一条记忆时数据库里不能只是逻辑删除——向量库里的 embedding 也要同步删除否则召回时还会捞出来。我为此写了一个同步删除的钩子在主存储和向量索引之间保持一致性。5.2 多智能体与跨端记忆同步如果你的系统将来要拆成多个智能体比如客服机器人、导购机器人、闲聊机器人记忆模块必须做成服务化的而不是塞在某个机器人内部。我们当时把记忆服务独立成一个带 API 的微服务所有智能体通过 HTTP 调用读写同一份记忆。这样用户在客服端留下的地址在导购端也能用得上。跨端同步的难点是用户的唯一标识。同一个用户在小程序端和 Web 端登录拿到的会话标识可能不同需要先做一层统一身份映射。这块我采用的是账号绑定手机号之后用平台级用户 ID 作为主键各端传上来的临时 ID 全部映射到主键上。注意这一步如果做不好跨端同步会变成新的“串台”隐患。5.3 成本控制嵌入、存储与调用怎么省钱记忆系统有三个明显的成本点嵌入计算的算力、向量库的存储空间、大模型因为 Prompt 变长多出来的 token 费用。嵌入成本可以用缓存来压。同一个user_id的相似消息短时间内重复调用嵌入模型是没有意义的我用了一个简单的近似去重先算消息的哈希命中缓存就直接复用向量。存储成本则靠第 4 节说的定期清理机制来控制同时把向量索引量化等级从全精度降到半精度对召回准确率的影响不到 1%但索引体积能减少接近一半。Prompt 变长带来的 token 费用增长往往被忽视。每轮对话都注入 5 条记忆看起来不多但如果每条记忆平均 80 个 token一天百万次对话成本就很可观。我的做法是加了一道“记忆压缩层”召回 5 条记忆之后再让一个小模型把它们压缩成 3 条摘要式的句子只把摘要注入系统 Prompt。测试下来对答复质量影响很小但长期 token 消耗能压掉三成以上。最后再分享一个小技巧。这套记忆方案我从测试到线上跑了三个多月最大的体会是记忆不是存得越多越好而是该记的记、该忘的忘。早期我执着于把用户每一句话都变成记忆结果系统越来越“啰嗦”。把“遗忘”当成一个一等公民来设计之后整个系统的可用性反而上了一个台阶。如果你准备在自己的项目里落地 ai-memory我建议先只接一个场景比如收货地址记忆或者商品偏好跑通全链路再逐步扩大范围这样排查问题的时候至少知道该从哪里入手。
