大模型什么都好,就是记性太差。十几分钟的连续对话能聊得明明白白,可一旦新建会话,下一秒它就当你是第一次见面。claude-mem 这个项目,就是冲着这个痛点去的:给对话AI加一个可持续累积的长期记忆层。项目的核心思路可以拆成五步——抽取、向量化、存储、检索、注入。适合谁?正在做AI客服、私人助理、知识库问答,或者任何需要跨会话持续了解用户的开发者。我会把从设计到落地的完整方案拆开讲,包括参数怎么定、坑怎么避,尽量做到你照着就能搭出一套可用的记忆系统。
1. 项目定位与核心思路
1.1 会话失忆是硬痛点
做AI应用的人基本都撞过这堵墙:API本身支持多轮对话,但多轮指的是“当前会话内的多轮”,会话窗口一关,之前聊过的内容就全没了。用户上次明确说过“给我发日报用邮件,别用企业微信”,第二天换了个新会话,助手又跑回来问“请问您希望用什么方式接收日报”。这种体验放在真实场景里很难接受,尤其是客服、销售、个人助理这类需要长期关系的应用。
解决跨会话记忆的思路,常见的有三套。第一套是在每次请求前把历史记录全量塞进上下文,简单粗暴,但token成本随天数指数上涨,聊不了几次就撞上上下文窗口。第二套是给每个用户建一张明细表,把偏好做成结构化字段存进数据库,这套适合字段明确、维度固定的场景,可一旦用户描述变成“上次你说那种风格挺好看的”,结构化查询就抓瞎了。第三套就是 claude-mem 走的路线——把对话内容转成向量存起来,用语义相似度来召回。
第三套方案最大的优势在于“不需要提前定义用户画像”。用户说什么,系统就按内容语义记下来,下次问法不一样也能靠向量相似度把相关记忆捞回来。代价是它需要维护一套向量索引,并且对抽取规则和召回质量控制的要求更高。这个项目本质上是把“记什么”“怎么存”“怎么取”“何时注入”四件事串成一条流水线,每一步都不复杂,但串起来之后效果提升非常明显。
1.2 数据流设计
我搭这套东西时,先把整个数据流画成了两端:写入端和读取端。
写入端的流程是:API请求发出并拿到响应之后,系统把用户输入和模型输出一并截获。先按规则判断这一段里有没有值得记忆的东西,比如用户是否表达了新的偏好、是否给出了结论、是否补充了个人信息。有价值的内容就做一次语义向量化,然后连同原文、时间戳、用户ID、会话ID一起写进存储层。写入的动作可以异步执行,不影响主链路。
读取端的流程是:新会话产生一条用户消息,系统先把这条消息向量化,去存储层检索最相关的N条记忆,再把这些记忆连同系统提示词一起拼进模型上下文。这里有个容易被忽略的点——注入的不是“全部记忆”,而是“当前问题相关的记忆”。向量检索的作用就是做语义上的初筛,把关联度低的记忆挡在上下文外面。
这套设计是无侵入式的。不需要微调模型,更不需要改模型权重,只是在请求上下文里做拼装。对于已经跑起来的业务,接入成本最低;对于从零开始的项目,也能把记忆层和其他业务逻辑解耦开。我实际做下来,整个核心模块不到一千行代码就能跑通,真正花时间的是后面的调优和排错。
2. 核心细节与方案选型
2.1 记忆抽取策略
第一版我犯过一个典型错误:把每轮对话都存进去。结果是记忆库三天就变成了垃圾桶,检索出来的内容全是“今天天气不错”“你好请问有什么可以帮你”这类无效文本。后来我才意识到,记忆层的核心瓶颈不是存储容量,而是“抽取得好不好”。
合格的做法是把记忆分成三档。第一档是硬事实,包括用户提过的名字、城市、时间节点、明确说出的偏好和拒绝项。第二档是行为特征,比如用户习惯在凌晨提交请求、喜欢先看结论再看过程、操作路径有明显的模式。第三档是关系型记忆,也就是两个实体之间的关联,例如“A用户负责的项目X和B用户负责的项目Y是上下游关系”。三档的可靠程度逐级下降,写入时要给不同的置信度分数。
抽取实现上,我优先用启发式规则,而不是直接上大模型。规则可以覆盖大部分场景:预设关键词表、数字和日期识别、否定句式识别、人称代词指向。比如用户说出“我不喜欢”“请务必”“以后都”这类短语时,往往意味着一条值得记录的偏好。规则抓不到的内容,再异步调用模型做摘要抽取,但只对单轮对话做,不做全局总结,避免成本和噪声同时失控。
这一条经验很关键:抽取规则宁可多画几层“小筛子”,也不要一上来就搞复杂的意图分类。小筛子出错容易定位,复杂模型出错根本无从查起。我后期加了一条置信度衰减机制——记忆条目在写入时有一个初始分数,之后每次被成功召回并得到用户正向反馈,分数上升;超过30天没有被命中的条目,分数按天递减,低到阈值就自动归档。等于是给记忆库做了一个自然淘汰机制,效果比定时全量清理好得多。
2.2 向量化与存储选型
向量化选择取决于两个条件:你是否接受请求云端嵌入服务,以及你的并发量有多大。如果记下来的内容确实需要私密性,本地轻量级嵌入模型是第一选择,离线可用、无外部调用成本,缺点是语义理解的精度稍低。如果对召回质量要求高,云端嵌入接口的效果确实更好,尤其在中文长句和同义改写场景下,差距比较明显。
存储层的选型我踩了一个来回。刚开始想用重型向量数据库,部署和运维成本都不低,还要额外维护一套集群配置。后来发现单机场景下,普通关系型数据库加向量扩展就完全够用。数据表只需要三张:用户表、会话表、记忆表。记忆表里开一个向量列和几个索引字段,查询时按用户ID过滤再做相似度排序。这种方案的好处是备份、恢复、迁移都简单,出问题时还能直接打开数据库看原始记录。
倒排索引和向量索引的配合也值得说一句。纯靠向量检索,高频实体名容易召回一堆相似但无关的内容;我先建一层基于关键词的倒排索引,把实体名精确命中排在前面,再叠加向量相似度重排,召回质量会明显上一个台阶。索引重建不用太频繁,我通常每周做一次全量重建,增量更新留给当天的写入操作。重建时用双副本切流,避免查询端出现数据空洞。
2.3 注入时机与位置
记忆注入的位置我试过两种:系统提示词、用户消息前缀。系统提示词适合放“用户的长期画像摘要”,比如“该用户偏好邮件沟通、工作日晚间活跃、关注数据准确性”——这类信息是高度浓缩的定性结论。用户消息前缀适合放原始片段,比如“用户上次提到:日报格式需要包含失败原因分析”,这类原始文本能提供更具体的语境。
真正见功力的是上下文预算控制。模型上下文窗口是有限的,记忆注入太多会把正常对话挤出去。我采用的分配比例是:上下文窗口的10%~20%留给记忆层。以窗口8000的模型为例,记忆预算大约是800到1600个token。每条精选记忆文本控制在50到100个token,那么可注入的记忆条数就是8到20条。超过这个量,没必要硬塞,模型记不住也处理不过来。
另一个容易忽略的点是时间衰减。同样是用户说“我喜欢简洁回复”,三个月前说的和昨天说的,置信度应该完全不同。我在每条记忆上打时间戳,检索排序时把时间衰减系数乘进去,越新的记忆权重越高。这样一来,用户最近一次的表态能压过旧记忆,不会出现“用户上个月说不要电话推广,这个月明确改口说可以打”却还按旧规则执行的尴尬。
原始文本片段和总结摘要的比例也需要调整。纯摘要信息密度高,但丢失细节;纯原文保留细节,但占用token多。我的做法是“摘要为主,原文为辅”:Top1到Top3条记忆给完整原文,其余只给摘要。这像记笔记——关键段落摘抄原文,其余写个索引就好。实战里这套组合效果最稳。
3. 实操过程与核心实现
3.1 最小可运行骨架
先给一个可以直接跑的逻辑骨架,我用Python写,语言无关,关键是理解数据流。这个版本去掉了很多工程细节,只保留了记忆层最核心的五步闭环。
import json import sqlite3 import numpy as np # 存储层使用sqlite,向量列按二进制存 DB_PATH = "memory.db" def init_db(): conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT, session_id TEXT, content TEXT, vector BLOB, importance REAL, created_at INTEGER ) """) conn.commit() conn.close() def embed(text: str): # 生产环境请替换成真正的嵌入模型 # 这里用简单字符特征生成固定长度向量,仅演示 vec = np.zeros(64, dtype=np.float32) for i, ch in enumerate(text): vec[i % 64] += ord(ch) norm = np.linalg.norm(vec) if norm > 0: vec = vec / norm return vec def remember(user_id, session_id, content, importance): vec = embed(content) conn = sqlite3.connect(DB_PATH) conn.execute( "INSERT INTO memories (user_id, session_id, content, vector, importance, created_at) VALUES (?,?,?,?,?,?)", (user_id, session_id, content, vec.tobytes(), importance, int(time.time())) ) conn.commit() conn.close() def recall(user_id, query, top_k=5, threshold=0.72): q_vec = embed(query) conn = sqlite3.connect(DB_PATH) rows = conn.execute( "SELECT id, content, vector, importance FROM memories WHERE user_id=?", (user_id,) ).fetchall() results = [] for mem_id, content, vec_bytes, imp in rows: vec = np.frombuffer(vec_bytes, dtype=np.float32) score = float(np.dot(q_vec, vec)) # 时间衰减可以在这里乘一个系数 if score >= threshold: results.append((score, content)) conn.close() results.sort(reverse=True, key=lambda x: x[0]) return results[:top_k] def build_prompt(user_id, user_message, base_prompt): mem_list = recall(user_id, user_message) memory_context = "\n".join([f"- {c}" for _, c in mem_list]) return f"{base_prompt}\n\n关于该用户的历史记忆:\n{memory_context}\n\n当前用户问题: {user_message}"骨架代码里最重要的一行其实是score = float(np.dot(q_vec, vec)),它决定了召回的边界。你可以在本地快速改阈值、改TopK数量,用一小批历史对话直接观察召回结果。很多时候跑完一遍就会发现,问题不在算法本身,而在于入库阶段的抽取质量。所以先用这个骨架把闭环跑通,再慢慢优化每个环节。
代码里用字符特征生成向量只是演示,生产环境必须换成真正的嵌入模型。我自己的做法是先用一个轻量级本地嵌入服务做离线推理,生成结果缓存在本地,既控制成本又保证响应速度。替换嵌入模型时,记得把所有旧向量重新计算一遍,否则向量空间不一致,检索结果会乱掉。
3.2 参数计算与调优
三个参数直接决定记忆系统的性格:similarity threshold(相似度阈值)、top_k(召回条数)、context budget(上下文预算)。它们彼此牵连,不能单独调。
相似度阈值的起步值我建议设在0.72。太高容易漏召回,用户上次说过的事情这次完全想不起来;太低则噪声严重,把不相关的陈年旧事翻出来干扰模型。真实的阈值应该结合你的嵌入模型分布来标定,一个简单办法是跑100条已知相关和不相关的样本,画出相似度分布曲线,取两类分布交叉点附近的值。实测业务里,0.70到0.78这个区间最常见。
top_k的取值依赖上下文预算。计算方法很简单:先估出一条记忆的平均token长度,比如80;如果上下文预算定在1200 token,那top_k就是15。宁可少而精,不要多而滥。我习惯把预算卡到总窗口的15%,因为模型还有系统提示词、工具定义、用户消息、模型历史都要占用空间,记忆层吃太多会挤压正常对话质量。
最后是抽取置信度阈值。每条记忆在写入时都会有一个importance分数,低于0.3的碎片内容我会直接丢弃;0.3到0.7的内容只存摘要;0.7以上存原文。这个分数靠规则模型计算,不追求精确,只需要能区分“闲聊”和“有效信息”。实际操作中,用户主动表达的明确意愿分数最高,模型推测出的内容分数其次,纯粹的寒暄和重复内容分数最低。
3.3 接入与部署
接入现有业务时,有两种形态可以选,我分别都实践过。
Sidecar代理模式:在应用和模型API之间插一层本地代理,所有请求先经过这个代理,由它负责截取对话、写入记忆、注入记忆,业务服务完全无感知。优点是对存量系统友好,几行配置就能接入;缺点是代理需要处理流量复制和异步写入,成为潜在故障点。适合快速试点的场景。
SDK集成模式:把记忆逻辑封装成SDK,在业务代码里显式调用,比如在用户确认某个信息时手动触发remember(),在构造请求前调用build_prompt()。优点是可控性强,知道什么时候写、为什么写;缺点是要改业务代码,而且如果团队成员不理解机制,很容易漏掉关键写入点。
| 对比项 | Sidecar代理模式 | SDK集成模式 |
|---|---|---|
| 接入成本 | 低,配置即用 | 中,需要改造代码 |
| 可观测性 | 较弱,黑盒较多 | 强,日志直观 |
| 故障影响面 | 代理挂了影响全部流量 | 只有调用SDK的模块受影响 |
| 调试效率 | 需要单独排查代理日志 | 单点调试即可 |
| 适合阶段 | 快速验证原型 | 正式产品迭代 |
如果走Sidecar路线,记得把记忆写入放在异步队列里,用失败重试兜底。如果走SDK路线,建议提供一个统一的上下文构建入口,不要散落在多个业务函数里,否则后续想调整记忆注入策略时会非常痛苦。
部署形态上,单机单库足够支撑中小规模场景。客户端量变大后,把向量检索单独拆成服务,存储层换成分库分表,但业务层面无需改动。我见过很多团队一上来就搭分布式存储,最后发现大部分查询都集中在少量热用户身上,根本用不到那么复杂的架构。先单机跑稳,再谈扩展。
4. 常见问题与排查实录
4.1 记忆污染与串味问题
最经常遇到的故障是“记忆串味”:用户随口聊了两句足球,之后每次对话模型都自动聊到足球。排查下来,原因几乎都在入库环节——抽取规则太宽松,把闲聊文本也当成记忆写了进去。
我的处理办法有两步。第一步是在抽取阶段加“对话属性”标签:纯闲聊、任务执行、信息确认、偏好声明。只有后三类允许写入记忆库,纯闲聊直接丢弃。第二步是在检索后加入“相关性复核”,用模型对候选记忆打一个与当前问题是否相关的分数,低于阈值就丢弃。这个步骤会增加一次模型调用,但可以在TopK阶段只对前10条做,成本可控。
如果发现记忆已经被污染了,不要只删单条。把该用户的记忆调出来,按置信度分数排序,批量清理低于阈值的条目。不要用“删除全部重建”的懒办法,会把一些有效信息也洗掉。我后来加了一个管理后台页面,可以按用户查看记忆库、手动标记无效记忆和强制保留重要记忆,运维成本一下子降了下来。
4.2 检索召回效果差
第二个高频问题是“明明存过,但就是搜不出来”。用户说记得自己提过某个要求,系统却完全没召回。这种问题要从三个点排查:
- 嵌入模型是否换过?换模型后旧向量没有重建,向量空间不一致会导致相似度全部异常。
- 查询文本的预处理是否一致?入库和查询如果走了不同的清洗逻辑,比如一边保留标点一边过滤掉,向量差异会非常大。
- 索引更新是否及时?增量写入如果只写数据表没触发索引重建,新记忆查询时就索引不到。
我专门写过一个诊断脚本,输入一条查询文本,输出存储层里Top50的相似度分布。如果分布普遍低,先怀疑嵌入模型不一致;如果分布高但目标记忆还是不出现,再怀疑清洗逻辑不一致。实测下来,最常见的还是第二个问题——两边处理流程漂移了,清洗函数改了没人同步。
另一个容易踩的坑是向量归一化。余弦相似度本身对向量长度不敏感,但如果嵌入模型输出分布已经发生偏移,不经归一化直接算相似度会导致分数虚高,把不相关的内容也捞出来。建议在入库和查询时统一做一次归一化,这个小细节能减少很多偶发错乱。
4.3 成本与性能平衡
给每条消息都做实时向量检索,在高并发场景下撑不住。我压测过一次,峰值请求时检索服务CPU直接打满,主链路延迟从200毫秒飙到1.5秒。优化方法分三层。
第一层是本地缓存。同一个用户在同一时间段内,检索结果变化不大,缓存有效期设为5到10分钟即可,大部分重复查询直接走缓存。第二层是批量预处理。用户不在线的时候,系统定期对他的历史对话做一次摘要合并,把零散记忆聚合成高层画像,检索时优先返回画像摘要,再补少量原始记忆。第三层是写入削峰。记忆写入全部异步化,先写本地队列,消费端批量处理,避免每条对话都触发一次向量化和数据库写入。
成本控制上,最大的开销来自向量化和记忆注入占用的token。向量化可以靠本地模型摊薄成本,token占用则只能靠更精准的抽取。如果一条记忆在注入后没有被用户后续对话“提及”或“复用”,说明它是无效记忆。我用这个标准做月度复盘,把零复用率的记忆类型找出来,反向优化抽取规则。做完这步之后,整体token消耗大概能省下三成。
最后再分享一个经验,也是我认为这个项目最核心的心得:记忆系统的好坏,七成取决于抽取,三成取决于检索。很多人把注意力放在向量数据库、相似度算法这些听起来高级的部分,实际调试时你会发现,真正决定体验的是你敢不敢把那些看似重要、实则没用的内容挡在记忆库外面。先把“不记什么”定清楚,再谈“记住什么”,这套系统的稳定性会超出你的预期。