对话模型长期记忆服务claude-mem的设计与实现
最近好几周我都在调一个基于对话模型的服务端应用,最头疼的就是“记忆”。模型本身状态能力有限,每次对话只能喂进去固定长度的上下文,聊长了就记不住开头说过的事,用户在应用里反复解释同一件事,体验非常糟糕。正好那段时间我在整理一个叫 claude-mem 的小项目,初衷就是把长期对话记忆从模型外面剥离出来,单独做一层记忆服务。这篇文章我会把它拆开讲清楚,包括为什么这样设计、核心模块怎么建、实际接入代码长什么样,以及我在运行中踩过的坑。适合正在做对话应用、想让模型具备长期记忆能力,或者好奇上下文管理底层逻辑的开发者阅读。
1. 项目定位与整体设计思路
1.1 先聊痛点:对话模型为什么记不住事
做过对话应用的人都会发现一个共性规律:模型本身是无状态的。我调用一次接口,传一摞历史消息,模型就基于这摞消息做一次续写,然后接口结束,状态清零。也就是说,模型只拥有“请求内记忆”,没有“请求间记忆”。这跟人脑的工作记忆很像——你手头能记住的电话号码就那么几个,打完就忘,真正长年累月记住的事靠的是外部笔记本或者大脑皮层固化后的长期记忆。
麻烦的地方在于,现实里的对话应用几乎都需要跨请求记忆。用户昨天说“我小孩今年五岁,对恐龙特别着迷”,今天再来,正常的助手应该记得这个背景,而不是再次问“您有孩子吗”。如果我要自己实现这个效果,通常只有两条路:把整个聊天记录每次全量塞进上下文窗口,或者自己维护一个精简档案。全量塞法的劣势非常明显。我对接的Claude接口有上下文长度上限,窗口越撑越满,几轮之后就可以准备截断;而且每多塞一倍的tokens,成本就翻倍,几轮深聊之后账单相当可观。更麻烦的是,截断策略不好做,直接砍掉最旧的几条消息,往往会把关键约束砍没了。
后来我尝试过自己写简单的SQLite表,把每轮助手回复额外存一分摘要,但摘要怎么触发、怎么合并、怎么在合适的时机重新注入上下文,这些问题越搞越复杂。走到这一步,我意识到这类需求天然应该收敛成一个独立模块,而不是散落在业务代码里到处补丁。claude-mem 就是在这个背景下成型的。它是一个轻量的记忆服务层,专门负责会话持久化、摘要生成、上下文重组这三个事,业务侧只需要调用几个方法,就能让模型“记得”之前发生过什么。
1.2 claude-mem 的定位:不是模型插件,是记忆服务
一开始我把 claude-mem 想复杂了,以为要在模型侧做东西,后来发现没必要。它本质上是位于应用和Claude API之间的一个中间层,模型依然是无状态的,但每次调用API前,应用从 claude-mem 取一段精选上下文,拼进提示词里;调用完之后,再把整轮对话写回记忆库。模型本身什么都没改,可表现出来的行为就变成了“它记得我”。
这个定位有几个明显好处。第一,它不绑定特定推理引擎,今天接Claude,明天想换成别的模型,记忆层不用动。第二,它只依赖常规的文件系统或数据库组件,没有很重的服务依赖,部署成本低。第三,它可以把记忆能力拆出来独立测试,我甚至用纯文本模拟API回包来做单元测试,速度很快。用一句话概括定位:claude-mem 负责“把对话历史变成有用的上下文”,让模型只专注于理解和推理。
1.3 架构设计:三层分离
我把 claude-mem 拆成三个核心层:Session层、Storage层、Summarize层。
Session层管“哪一场对话”。每一场对话有独立session_id,所有读写都按这个ID隔离,相当于笔记本上的一页。Storage层管“数据放哪”。默认实现用SQLite做持久化,聊天记录逐条存表,同时维护一张记忆索引表;如果做了向量检索,还会把向量信息落在同一份数据库里,避免多组件部署。Summarize层是灵魂,它决定什么时候把旧内容压缩成摘要、摘要里保留哪些信息、摘要什么时候被重新注入提示词。
数据在 claude-mem 里是这么流动的:用户消息进来,先写入Storage;应用向Claude要回复前,调 claude-mem 的get_context(session_id),它会按当前策略拼出一段精简上下文;模型回复之后,再调add_message(session_id, content)把回复写入。整个协作关系非常直接,不依赖外部消息队列,也不需要反向回调。我用表格概括一下三层职责:
| 层级 | 主要职责 | 关键接口 |
|---|---|---|
| Session层 | 会话隔离、消息顺序、快照定位 | create_session, delete_session |
| Storage层 | 消息持久化、摘要持久化、向量索引 | add_message, get_history, save_memory |
| Summarize层 | 摘要触发、摘要合并、上下文重组 | get_context, summarize_if_needed |
设计成三层,主要是为了让排查问题时有清晰的边界。上下文重组不对,就去Summarize层查;数据丢了,去Storage层查;消息串了,去Session层查。工程上的可维护性比“一个函数搞定一切”好太多。
2. 记忆存储与上下文重组机制
2.1 消息事件模型:让存储更通用
我在设计存储层时没有直接存“用户说”“助手说”这样的裸文本,而是先抽象成一个统一的消息事件模型。每条事件至少包含四个字段:session_id、role、content、ts。role可以是user、assistant,也可以扩展到system或tool。ts是毫秒级时间戳,用于排序和后续做时间衰减。
之所以要事务化地逐条存,而不是整段存一个JSON快照,是因为后续需要精细控制到底取哪些消息。如果存的是JSON快照,想取第5条到第10条就得自己解析;想定位某一天之后的消息还得再维护索引。用事件表,这些都是标准SQL查询。另外,如果未来接入工具调用,每条工具结果也能作为独立事件写入,摘要时能区分出“这是用户说的”还是“这是系统返回的”,避免把工具日志误当用户指令。
2.2 三级记忆结构:窗口、摘要、事实库
这是我的设计里比较核心的部分。我不想把所有历史都塞进上下文,也不想只留一句干巴巴的总结。我把它分成三级:
第一级是短期窗口,存最近几轮完整原文。一般取最近5到10轮,这部分消息质量最高,有完整细节,适合让模型感知当前对话风格。
第二级是滚动摘要,存更早对话的高度压缩版本。一旦累计tokens超过某个阈值,Summarize层就把最老的一部分原文喂给Claude,让它生成一段第三人称摘要,再和旧的滚动摘要做合并。这个摘要会保留用户偏好、明确约束、待办事项和关键实体。
第三级是长期事实库,专门存从对话里抽取出来的“结论型信息”,比如“用户有一个5岁孩子,喜欢恐龙”“用户工作单位在浦东”这类事实。它们会被组织成键值对或短句,放在持久化表里。事实库的好处是,即使对话摘要被进一步压缩,这些刚性事实依然可以单独被检索并注入上下文,不容易丢。
很多记忆类项目只做其中一级,但实际用了会发现不够。只做窗口,聊长了照样爆;只做摘要,模型容易失去对细节的把握;只做事实库,又不知道怎么处理那些“不太好归纳成事实”的对话情绪和过程。三级配合是实践中比较稳的组合。
2.3 上下文重组:把记忆拼回提示词里
每次调用Claude API前,我会用get_context把三级内容按顺序拼成一段上下文放进system prompt。组合顺序基本是:长期事实库(如果有)优先,然后滚动摘要,最后短期窗口消息。这样模型最先看到的是稳定信息,然后是压缩历史,接着是细粒度原文,逻辑上从高到低,过渡自然。
拼的时候必须考虑token预算。我维护一个context_budget参数,比如4000 tokens。先放事实库和摘要,每个都预估token,如果预算够,再放短期窗口;如果不够,就削减短期窗口的轮数。这里不能省摘要的token,因为摘要压缩过一次,性价比远高于原文。整个计算类似背包问题,但没必要用复杂优化,贪心策略在工程上已经够用。
3. 实操:从零搭一个可用的 claude-mem 模块
3.1 环境准备与依赖选择
我在模拟项目X里用的是Python 3.10,依赖只有三个必须项:SQLite3(内置)、一个HTTP客户端用来调Claude API、一个轻量token估算工具。SQLite3不需要额外安装,这对我这种不喜欢引重型依赖的人来说非常友好。向量检索则是后加的,先把基础跑通再加,否则一开始就上向量库容易让调试复杂度失控。
Python版本最好3.10以上,因为后面用了dataclass和新一点的类型注解。调Claude API我用的是官方SDK,但也可以直接用requests发HTTP请求,两种我都在不同版本里试过。官方SDK的好处是自带流式解析,坏处是封装层级多,出了问题要挖进去看;requests则更透明,但需要自己处理请求结构。我的建议是,项目初期用SDK图省事,调记忆阶段用requests方便打日志。
3.2 核心存储与上下文读取实现
先展示最基础的存储层。核心是一个MemStore类,内部维护SQLite连接,提供add_message和get_recent方法。实际代码里我还会加时间戳和索引,这里保留骨架让逻辑更直白。
import sqlite3 import time import json class MemStore: def __init__(self, db_path="claude_mem.db"): self.conn = sqlite3.connect(db_path) self.conn.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, ts INTEGER NOT NULL ) """) self.conn.execute(""" CREATE INDEX IF NOT EXISTS idx_session_ts ON messages(session_id, ts) """) self.conn.commit() def add_message(self, session_id, role, content): self.conn.execute( "INSERT INTO messages(session_id, role, content, ts) VALUES(?,?,?,?)", (session_id, role, content, int(time.time() * 1000)) ) self.conn.commit() def get_recent(self, session_id, limit=10): rows = self.conn.execute( "SELECT role, content FROM messages WHERE session_id=? ORDER BY ts DESC LIMIT ?", (session_id, limit) ).fetchall() rows.reverse() return rowsget_recent取的是最近N条,并且倒序后再翻转,保证返回顺序和对话原文一致。这个细节最开始我没注意,直接按DESC返回,结果注入提示词后对话顺序整个反了,模型上下文混乱得不行。
接下来是Summarize层。它负责在需要时触发摘要,并把摘要存进一张独立的summaries表。摘要生成文本我直接把它当作一次Claude API调用,但提示词固定为“请把下面的对话压缩成要点”。“要点”比“摘要”这个词好用,因为模型会倾向于保留行动项和约束,而不是泛泛复述。
class Summarizer: def __init__(self, store, client): self.store = store self.client = client def summarize_old_messages(self, session_id, older_messages): prompt = "请把以下对话压缩为要点,保留用户偏好、明确约束、实体和待办事项。\n\n" + \ json.dumps(older_messages, ensure_ascii=False) response = self.client.chat(prompt) self.store.save_summary(session_id, response) return response这里有一个很容易踩的坑:不能在新对话开始前只调用一次摘要,然后就再也不管。对话是动态的,用户可能在新消息里纠正过去的偏好。所以我设了一个策略,每新增10条消息或者短期窗口token超过预算的80%时,就对最老的一半消息重新摘要,并把新摘要与旧摘要再合并一次。合并操作本身也是调用Claude,但量小,成本可控。
3.3 接入Claude API的完整示例
把上面两个模块接起来,核心就是一个chat_with_memory函数。它先从记忆库取上下文,再带上用户的新消息调用Claude,最后把用户消息和助手回复都写回记忆库。
def chat_with_memory(session_id, user_message, memo_client, store, summarizer, max_context_tokens=3000): # 1. 读取历史上下文 facts = store.get_facts(session_id) summary = store.get_summary(session_id) recent = store.get_recent(session_id, limit=8) # 2. 按预算重组上下文 context_parts = [] if facts: context_parts.append("已知事实:" + "; ".join(facts)) if summary: context_parts.append("历史摘要:" + summary) if recent: context_parts.append("近期对话:") for role, content in recent: context_parts.append(f"{role}: {content}") system_prompt = "\n".join(context_parts[:max_context_tokens]) # 3. 调用对话模型接口 response = memo_client.chat(system_prompt=system_prompt, user_message=user_message) # 4. 写回记忆 store.add_message(session_id, "user", user_message) store.add_message(session_id, "assistant", response) summarizer.summarize_if_needed(session_id) return response这个流程虽然简单,却包含了我认为最核心的记忆闭环:读取、重组、调用、写回。summarize_if_needed内部再判断是否需要做摘要合并,不需要每次调用都做。生产环境中还要把session_id换成带业务属性的标识,比如用户ID加时间戳或用途标签,防止串会话。
3.4 参数调优:三个关键数字
真正调节奏的是三个参数:context_budget、summary_trigger_tokens、recent_window_size。
context_budget是每次API请求允许注入的上下文tokens上限。我建议从3000开始,然后根据自己业务复杂度上下调整。千万不能用满模型的上下文窗口,因为模型回复本身也会占用token空间,窗口满了会导致请求被拒。
summary_trigger_tokens控制“旧消息达到多长就开始做摘要”。我习惯设为窗口总预算的70%。举例,如果预算3000,触发值就是约2100。达到触发值后,我会把最近窗口之外的历史消息整体丢给摘要器,压缩后再继续。
recent_window_size是保留完整原文的轮数。一般设置在5到8轮。超过这个轮数的消息,即使之前没被摘要,在组装上下文时也会被强制裁掉。这样既保证即时语气连贯,又不会让原文无限膨胀。
4. 常见问题与排查技巧实录
4.1 会话串号:记忆互相污染
最典型的问题是不同用户的记忆串场。有一次我在日志里看到,用户A的上下文里出现了用户B提到的“家里养了两只猫”,明显是session_id 处理出了漏洞。排查后发现是前端在切换账号时没有重新创建session_id,导致同一浏览器还在用旧ID。后来我规定session_id必须由服务端生成,格式为业务前缀 + 用户ID + 时间戳,并加了一层namespace参数作为一级隔离。在存储层查询时,所有会话引用都必须带namespace,从源头杜绝串号。
4.2 摘要越缩越干:事实细节丢失
摘要合并如果做得太激进,会发现模型越到后面越不懂用户。问题是摘要器在多次递归压缩时,把具体名词给丢了。比如用户说的是“我想在六月底前上线”,压缩两次后可能就变成“用户有计划”,执行时间没了。我的解决办法是在摘要提示词里强制增加一个“关键数据不得遗漏”清单,并要求输出必须包含时间、人物、数量、地点四类信息。每次合并时,提示词里都会带上上一次的摘要原文,并补充一句“请保留上文中提到的所有时间点和数字”。
4.3 Token估算不准:请求被截断
早期我用字符数乘一个固定系数来估算token,后来发现不同类型内容偏差很大。中文对话里,这个系数波动很大,标点符号、列表格式都会影响实际token数。最后我改成直接调用模型服务自带的tokenizer接口来做估算,并且在代码里统一封装成estimate_tokens函数。如果临时没有tokenizer服务,也可以留出10%的冗余预算,宁可比上限少塞一点,也不要赌精确值。
4.4 向量检索召回噪音:记忆相关但未必有用
加上向量检索之后,新问题又来了:检索出的“相关记忆”有时候只是一句话里出现了相同关键词,但对该上下文根本没用。比如用户之前说“火山地貌”,后来说“电影里的火山场景”,语义不同但向量距离很远。我开始用相似度分数阈值和时效衰减来解决。相似度低于0.7的不注入,超过30天的记忆额外打七折;再配合一个重排规则,优先选实体匹配程度高的记忆。最终效果是准确率提升明显,但召回率略有下降,换成更好的向量模型可以弥合。
5. 避坑清单与进阶扩展方向
5.1 可以不用向量库的轻量替代方案
很多朋友一听到记忆增强,就条件反射想上向量数据库。其实很多场景根本不需要。根据我的经验,如果会话数量在十万条以内,SQLite内置的FTS5全文搜索加上简单的关键词权重排序,效果已经够用。我的第一版claude-mem就是纯SQLite,没有引入任何向量组件。搜索“恐龙”这种词,FTS5能直接找到,速度非常快。只有当需要语义相似度匹配时,再考虑引入向量索引。
5.2 SQLite并发写与连接管理
项目并发量上来以后,SQLite会偶尔报“database is locked”。这是因为多个连接同时写会冲突。我做了三件事:开启WAL模式、短事务快速提交、按session_id做写锁。
self.conn.execute("PRAGMA journal_mode=WAL") self.conn.execute("PRAGMA busy_timeout=5000")WAL模式允许多线程并行读,写操作串行排队。把busy_timeout设为5000毫秒,能减少业务高峰期的偶发冲突。如果线上并发还是高,可以把claude-mem的存储层替换成PostgreSQL,接口定义保持不变,影响面很小。
5.3 记忆可观测性:给记忆装一个回放面板
记忆系统最怕的是“黑盒”。模型回复不对,你不知道是它理解错了,还是记忆注入错了。所以我在claude-mem里加了一个可选的debug_trace文件,每次请求完成后,把实际注入的system_prompt和传给模型的消息原文记下来。这样一旦回答异常,我直接看日志就能判断是不是记忆侧的问题。这个功能开发成本不高,却大大减少定位问题的耗时,我强烈建议保留。
5.4 后续扩展方向
现在的claude-mem已经能稳定处理我的项目需要,但我还在尝试几个扩展方向。第一是支持对话修复,让用户在侧边栏直接编辑或删除某条记忆,并把这个动作作为新事件写回系统,实现“遗忘权利”。第二是自动摘要分主题,比如把工作相关、生活相关分开存储,避免混合摘要稀释信息。第三是做敏感信息脱敏,让摘要生成器自动识别证件号码、银行卡号,并将其替换为掩码,避免敏感数据长期驻留。这些扩展都不会改变核心架构,只是在Summarize层和Storage层加策略。
我在实际运行中最大的体会是,记忆系统不能“过度记忆”。想尽可能多地塞进上下文,反而会让模型失去重点。claude-mem 现在每次只露三层记忆的最关键部分,剩下都留在SQLite里,等需要时再取。这个“克制”的设计,才是它效果稳定的主要原因。如果你也在做类似的对话应用,不妨从最简单的事件表和滚动摘要开始,先把闭环跑通,再逐步加上向量检索和事实抽取。记忆功能从来不是越重越好,而是越准确越好。