claude-mem：大模型记忆管理架构设计与工程实践

1. 项目概述与核心定位1.1 这个项目到底在解决什么问题claude-mem这个名字第一次看到的时候我愣了一下。claude大家都熟mem显然是 memory 的缩写。合在一起指向一个很明确的方向给 Claude 这类大语言模型做记忆管理。做过 AI 应用开发的人都知道一个痛点——大模型的上下文窗口是有限的。你跟它聊了 50 轮前面 40 轮的内容早就被挤出去了。用户下次再问一个跟之前强相关的问题模型一脸茫然因为它根本不记得你们聊过什么。这不是模型笨是它的“工作记忆”被物理截断了。claude-mem要干的事情就是在这个截断发生之前把关键信息捞出来、存好、在需要的时候再塞回去。听起来简单做起来全是细节。哪些信息值得存存成什么格式什么时候取取多少怎么保证取出来的信息不把上下文窗口撑爆这些问题每一个都够写一篇长文。我最初接触这个方向是因为一个客服机器人的项目。用户跟机器人聊了半小时中间提到了自己的订单号、偏好、之前投诉过的问题。结果聊到后面机器人开始答非所问因为它已经把前面的关键信息“忘”了。用户体验断崖式下跌。那时候我就意识到记忆管理不是一个锦上添花的功能而是决定 AI 应用能不能真正“像人一样对话”的基础设施。claude-mem这个项目标题背后我看到的是一整套围绕“记忆”构建的技术方案。它可能是一个库、一个中间件、一套 API 设计也可能是一个完整的记忆层架构。不管具体形态如何核心目标是一致的让 Claude 在长对话、多轮交互、跨会话场景下依然能保持对关键信息的“记忆”。1.2 适合哪些人深入关注这个项目不是给普通用户看的。如果你只是用 Claude 聊聊天、写写文案那claude-mem对你来说可能有点“底层”。但如果你在构建以下类型的应用它就非常值得花时间研究多轮对话系统客服机器人、心理咨询助手、教育辅导工具这些场景天然需要记住用户说过的话。个性化 AI 助手用户希望 AI 记住自己的偏好、习惯、历史决策下次不用重复交代。长文档分析与问答一份几百页的合同、论文、技术手册不可能一次性塞进上下文需要分段处理并维护一个“记忆索引”。Agent 类应用自主 Agent 在执行任务时会产生大量中间状态这些状态需要被有效管理否则 Agent 会“迷失”在自己的历史里。我见过太多团队在这个问题上踩坑。有人简单粗暴地把所有历史对话拼接到 prompt 里结果 token 消耗爆炸成本失控。有人完全不做记忆用户体验稀碎。claude-mem这类项目的价值就是给出一条经过验证的中间路线。1.3 核心关键词拆解围绕claude-mem有几个关键词是绕不开的上下文窗口管理Claude 的上下文窗口虽然大但终究有限。如何在这个限制下最大化信息密度是核心命题。记忆压缩把冗长的对话历史压缩成简洁的摘要或结构化数据减少 token 占用。记忆检索在需要的时候从存储的记忆中快速找到最相关的片段。记忆持久化跨会话保存记忆让 AI 在下次对话时还能“记得”上次的内容。向量化存储用 embedding 把文本转成向量通过相似度检索来实现记忆的快速定位。这些关键词构成了claude-mem的技术骨架。后面我会逐一展开把每个环节的实操细节和踩坑经验都讲清楚。2. 记忆管理的整体架构设计2.1 为什么不能简单粗暴地拼接历史先说说最常见的错误做法。很多开发者一开始的想法很直接把用户和 AI 的所有历史对话拼成一个长字符串每次请求都带上。简单、粗暴、有效——直到 token 账单把你吓死。我算过一笔账。假设一轮对话平均 200 个 token聊了 50 轮就是 10000 个 token。如果每次请求都带上全部历史第 50 轮请求的输入 token 就是 10000。而如果只带最近 5 轮输入 token 只有 1000。成本差了 10 倍。更别说 Claude 的定价是按 token 计费的长对话场景下这个差距会迅速放大。除了成本还有性能问题。上下文越长模型的响应速度越慢而且过长的上下文会导致“注意力稀释”——模型在大量无关信息中找不到重点回答质量反而下降。这就像你让一个人记住一整本书的内容再回答问题他可能还不如只记住关键章节的人答得好。所以claude-mem的核心思路一定是不是记住所有东西而是记住该记住的东西。这就引出了记忆管理的三个核心环节写入、存储、读取。2.2 三层记忆架构的设计思路参考人类记忆的工作方式我给claude-mem设计了一套三层记忆架构。这个架构不是凭空想出来的而是在多个项目中反复迭代后的结果。第一层短期记忆工作记忆这一层对应的是当前对话的最近几轮交互。通常保留最近 3 到 5 轮对话的完整内容不做任何压缩。为什么是 3 到 5 轮因为大多数对话的上下文依赖关系不会超过这个范围。用户说“它”指的是上一句提到的某个东西用户说“再改一下”指的是刚才讨论的那个方案。这些指代和省略需要完整的最近对话才能理解。这一层的实现最简单就是一个固定长度的队列。新消息进来旧消息出去。但这里有个细节不能机械地按轮数截断而应该按“对话回合”截断。一个回合包括用户消息和 AI 回复如果只截了用户消息没截 AI 回复上下文就断了。第二层中期记忆摘要记忆当对话超过一定轮数后把较早的对话内容压缩成摘要。这个摘要不是简单的截断而是用模型自己来生成。具体做法是当短期记忆队列满了把最旧的一轮对话拿出来跟已有的摘要合并生成新的摘要。这里的关键是摘要的粒度。太粗了丢失关键信息太细了摘要本身就很长失去了压缩的意义。我的经验是把摘要控制在 200 到 500 个 token 之间。这个长度足够保留关键事实人名、数字、决策、偏好又不会占用太多上下文空间。摘要的生成 prompt 也很讲究。不能简单地说“总结以下对话”而应该给出明确的指令“提取以下对话中的关键事实、用户偏好、已做出的决策和未解决的问题。用简洁的要点形式输出忽略寒暄和无关内容。”这样生成的摘要信息密度更高。第三层长期记忆持久化记忆这一层解决的是跨会话记忆的问题。用户今天聊完明天再来AI 应该还记得上次的关键信息。长期记忆通常存储在外部数据库中可以是关系型数据库也可以是向量数据库。存储的内容不是原始对话而是经过提取的结构化信息。比如用户说“我住在北京喜欢喝美式咖啡对花生过敏”长期记忆里存的就是三个键值对location: 北京、coffee_preference: 美式、allergy: 花生。下次对话时根据当前话题的相关性把这些信息检索出来注入到 system prompt 或上下文开头。三层架构的核心逻辑是短期记忆保证对话连贯中期记忆保证长对话不丢失关键信息长期记忆保证跨会话的个性化体验。每一层解决不同的问题合在一起构成完整的记忆系统。2.3 记忆写入的触发时机什么时候把信息写入记忆这个问题比看起来复杂。如果每轮对话都写入存储压力大而且很多信息是冗余的。如果等到对话结束再写入又可能丢失中间的重要信息。我的做法是采用事件驱动 定期扫描的混合策略事件驱动当检测到特定类型的信息时立即写入。比如用户明确说“记住我喜欢...”、“以后都...”、“我的...是...”这些是显式的记忆指令必须立即处理。还有一些隐式信号比如用户纠正 AI 的错误“不对应该是...”这说明之前的理解有误需要更新记忆。定期扫描每隔 N 轮对话我通常设为 5 轮触发一次记忆提取。把这几轮对话喂给模型让它判断哪些信息值得长期保留。这个判断的 prompt 需要精心设计否则模型会把所有东西都标为“重要”。这里有个坑我踩过早期版本里我让模型自由判断什么值得记住结果它把用户的每一句话都提取成了记忆点存储量爆炸检索时噪音极大。后来我加了一个约束“只提取在未来对话中可能被再次引用的信息。如果一条信息只在当前对话中有意义不要提取。”这个约束显著提升了记忆质量。2.4 记忆检索的相关性排序存了记忆怎么取这是另一个核心问题。最简单的做法是把所有记忆都塞回上下文但这又回到了 token 爆炸的老路。所以必须做相关性排序只取最相关的几条。相关性排序通常用向量相似度来实现。把当前用户的 query 转成 embedding跟所有记忆的 embedding 做余弦相似度计算取 top-K。K 一般设为 3 到 5太多会引入噪音太少可能漏掉关键信息。但纯向量检索有个问题它擅长语义相似不擅长精确匹配。用户问“我的订单号是多少”向量检索可能找出一堆跟“订单”相关的记忆但真正包含订单号的那条可能排在后面。所以我在向量检索之外加了一层关键词匹配作为补充。如果 query 中包含数字、专有名词、特定标识符优先用关键词匹配来检索。还有一个细节记忆的时间衰减。越久远的记忆相关性权重应该越低。但不是线性衰减而是指数衰减。具体公式是weight base_score * exp(-lambda * days_since_access)其中 lambda 控制衰减速度。我一般设 lambda 为 0.01意味着大约 70 天后权重降到一半。这个参数可以根据业务场景调整高频交互的场景衰减可以慢一些。3. 核心细节解析与实操要点3.1 记忆提取的 Prompt 设计记忆提取的质量90% 取决于 prompt 的设计。我试过几十个版本最终稳定下来的 prompt 结构是这样的你是一个记忆提取助手。请从以下对话中提取值得长期记住的信息。 提取规则 1. 只提取用户明确陈述的事实、偏好、决策和约束条件 2. 忽略寒暄、闲聊、AI 的回复内容 3. 每条记忆用“主语 谓语 宾语”的简洁句式表达 4. 如果信息与已有记忆冲突标注为“更新”并说明旧值 5. 最多提取 5 条按重要性排序 已有记忆 {existing_memories} 对话内容 {conversation} 请以 JSON 格式输出 { memories: [ {content: ..., type: fact|preference|decision|constraint, importance: 1-5} ] }这个 prompt 有几个关键设计点。第一明确限定提取范围只提取用户的信息不提取 AI 的回复。因为 AI 的回复是基于用户信息生成的存了也是冗余。第二要求用简洁句式避免存一整段话。第三加入冲突检测当用户改变偏好时能更新旧记忆。第四限制数量强制模型做优先级排序。实测下来这个 prompt 的提取准确率在 85% 以上。剩下的 15% 主要是边界情况比如用户用反问句表达偏好“难道我不喜欢美式吗”或者信息分散在多轮对话中需要跨轮推理。这些情况需要额外的后处理逻辑来兜底。3.2 向量化存储的选型与参数记忆的向量化存储我对比过几个方案方案优势劣势适用场景内存向量库零依赖启动快重启丢失容量有限开发测试、小规模应用SQLite 向量扩展单文件易部署并发性能一般中小规模生产环境专用向量数据库高性能可扩展运维复杂成本高大规模生产环境云服务向量存储免运维弹性扩展数据隐私顾虑持续计费快速上线、无运维团队我个人偏好 SQLite 向量扩展的方案。原因很简单记忆数据量通常不会特别大一个用户几百到几千条记忆SQLite 完全扛得住。而且单文件部署极其方便备份就是复制一个文件。对于大多数中小规模应用来说这是性价比最高的选择。Embedding 模型的选择也很关键。我试过几种通用文本 embedding维度高通常 768 或 1536语义表达能力强但计算成本高。轻量级 embedding维度低128 或 256速度快适合对延迟敏感的场景。多语言 embedding如果应用需要支持中文必须选多语言版本否则中文语义相似度计算会很不准。我的建议是如果记忆条目不多 10000 条用通用 embedding精度优先。如果条目很多或者对延迟要求极高用轻量级 embedding但要做好召回率下降的准备。3.3 上下文注入的策略检索出相关记忆后怎么注入到当前对话的上下文中这看似简单其实有很多讲究。最直接的做法是把记忆拼成一个字符串放在 system prompt 里。比如你是一个有记忆的助手。以下是你记得的关于用户的信息 - 用户住在北京 - 用户喜欢喝美式咖啡 - 用户对花生过敏 - 用户上次提到正在做一个电商项目 请根据这些信息来回答用户的问题。但这种做法有个问题如果记忆条目很多system prompt 会变得很长挤占对话空间。而且所有记忆都同等对待没有优先级。我的优化方案是分层注入高重要性记忆importance 4直接放在 system prompt 开头用明确的标签标注比如“【重要】用户对花生过敏”。中等重要性记忆importance 2-3放在 system prompt 末尾用简洁列表呈现。低重要性记忆importance 1只在相关时才注入不主动放在 system prompt 里。另外注入的记忆要跟当前话题相关。如果用户正在问技术问题把“喜欢喝美式咖啡”这种记忆注入进去就是噪音。所以注入前要做一次相关性过滤只保留跟当前 query 相似度超过阈值的记忆。3.4 记忆去重与冲突处理记忆系统运行一段时间后必然会出现重复和冲突。比如用户第一次说“我喜欢美式”第二次说“我一般喝美式”这两条记忆本质上是同一件事。如果不做去重检索时会返回两条几乎一样的记忆浪费上下文空间。去重的策略有两种写入时去重在存储新记忆之前先跟已有记忆做相似度比较。如果相似度超过 0.9就不存新的或者合并两条记忆。读取时去重检索出 top-K 记忆后再做一次内部去重把相似度高的记忆合并成一条。我倾向于写入时去重因为这样存储更干净检索效率更高。但写入时去重会增加写入延迟需要权衡。我的做法是写入时做粗粒度去重相似度 0.95 才合并读取时做细粒度去重相似度 0.85 就合并。冲突处理更复杂。用户说“我搬到上海了”而记忆里存的是“用户住在北京”。这时候不能简单地去重而要识别出这是对旧信息的更新。我的做法是在记忆提取的 prompt 里加入冲突检测指令让模型判断新信息是否与旧记忆冲突。如果冲突生成一条“更新”类型的记忆同时把旧记忆标记为“已过期”。这里有个坑不要直接删除旧记忆。因为有些冲突可能是模型误判保留旧记忆作为历史记录在检索时优先返回新记忆即可。万一新记忆是错的还能回滚。4. 实操过程与核心环节实现4.1 环境准备与依赖安装假设我们用 Python 来实现这套记忆系统基础环境需要以下依赖pip install anthropic # Claude API 客户端 pip install numpy # 向量计算 pip install sqlite-vec # SQLite 向量扩展 pip install sentence-transformers # 本地 embedding 模型如果你打算用云服务的 embedding API可以把sentence-transformers换成对应的 SDK。但我建议至少准备一个本地 embedding 方案作为 fallback避免网络问题导致整个记忆系统不可用。数据库初始化很简单import sqlite3 import sqlite_vec db sqlite3.connect(claude_mem.db) db.enable_load_extension(True) sqlite_vec.load(db) db.enable_load_extension(False) db.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, content TEXT NOT NULL, memory_type TEXT DEFAULT fact, importance INTEGER DEFAULT 3, embedding BLOB, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_accessed TIMESTAMP, access_count INTEGER DEFAULT 0, status TEXT DEFAULT active ) ) db.execute( CREATE VIRTUAL TABLE IF NOT EXISTS memory_vectors USING vec0(embedding float[384]) )这里 embedding 维度设为 384对应的是all-MiniLM-L6-v2这个轻量级模型。如果你用更大的模型维度要相应调整。注意memory_vectors是虚拟表用来做向量检索memories是主表存原始数据。4.2 记忆写入的完整流程写入流程分为四步提取、去重、向量化、存储。第一步提取记忆def extract_memories(conversation, existing_memories): prompt f你是一个记忆提取助手。请从以下对话中提取值得长期记住的信息。 提取规则 1. 只提取用户明确陈述的事实、偏好、决策和约束条件 2. 忽略寒暄、闲聊、AI 的回复内容 3. 每条记忆用“主语 谓语 宾语”的简洁句式表达 4. 如果信息与已有记忆冲突标注为“更新”并说明旧值 5. 最多提取 5 条按重要性排序 已有记忆 {existing_memories} 对话内容 {conversation} 请以 JSON 格式输出 response claude_client.messages.create( modelclaude-sonnet-4-20250514, max_tokens1000, messages[{role: user, content: prompt}] ) return parse_json_response(response.content[0].text)这里用 Claude 自己来做记忆提取效果比用规则引擎好得多。因为 Claude 能理解语义能判断什么信息在未来可能被再次引用。第二步去重检查def is_duplicate(new_memory, existing_memories, threshold0.95): new_embedding get_embedding(new_memory[content]) for mem in existing_memories: similarity cosine_similarity(new_embedding, mem[embedding]) if similarity threshold: return True, mem[id] return False, None去重阈值设为 0.95 是经过调参的。太低会误合并不同记忆太高会漏掉重复。0.95 在我的测试集上表现最均衡。第三步向量化from sentence_transformers import SentenceTransformer model SentenceTransformer(all-MiniLM-L6-v2) def get_embedding(text): return model.encode(text, normalize_embeddingsTrue)normalize_embeddingsTrue很重要它把向量归一化到单位长度这样余弦相似度计算就简化为点积速度更快。第四步存储def store_memory(content, memory_type, importance, embedding): cursor db.execute( INSERT INTO memories (content, memory_type, importance, embedding) VALUES (?, ?, ?, ?), (content, memory_type, importance, embedding.tobytes()) ) memory_id cursor.lastrowid db.execute( INSERT INTO memory_vectors (rowid, embedding) VALUES (?, ?), (memory_id, embedding.tobytes()) ) db.commit() return memory_id注意memory_vectors表的rowid跟memories表的id是对应的这样检索时可以通过 rowid 关联回主表。4.3 记忆检索的实现细节检索流程分为三步query 向量化、向量检索、重排序。def retrieve_memories(query, top_k5, min_similarity0.3): query_embedding get_embedding(query) # 向量检索 results db.execute( SELECT m.id, m.content, m.memory_type, m.importance, vec_distance_cosine(mv.embedding, ?) AS distance FROM memory_vectors mv JOIN memories m ON m.id mv.rowid WHERE m.status active ORDER BY distance ASC LIMIT ? , (query_embedding.tobytes(), top_k * 2)).fetchall() # 过滤低相似度结果 memories [] for row in results: similarity 1 - row[distance] if similarity min_similarity: memories.append({ id: row[id], content: row[content], type: row[memory_type], importance: row[importance], similarity: similarity }) # 重排序相似度 * 重要性权重 * 时间衰减 for mem in memories: mem[score] ( mem[similarity] * 0.6 (mem[importance] / 5) * 0.3 time_decay(mem[id]) * 0.1 ) memories.sort(keylambda x: x[score], reverseTrue) # 更新访问记录 for mem in memories[:top_k]: db.execute( UPDATE memories SET last_accessed CURRENT_TIMESTAMP, access_count access_count 1 WHERE id ?, (mem[id],) ) db.commit() return memories[:top_k]这里有几个关键点。第一向量检索时取top_k * 2条给重排序留出空间。第二重排序的权重分配是 0.6 相似度 0.3 重要性 0.1 时间衰减。这个比例可以根据业务调整如果更看重时效性可以提高时间衰减的权重。第三每次检索后更新访问记录这样时间衰减函数可以用last_accessed来计算。时间衰减函数import math from datetime import datetime def time_decay(memory_id): row db.execute( SELECT last_accessed, created_at FROM memories WHERE id ?, (memory_id,) ).fetchone() last row[last_accessed] or row[created_at] last_dt datetime.fromisoformat(last) days (datetime.now() - last_dt).days return math.exp(-0.01 * days)4.4 上下文组装与注入检索出记忆后需要把它们组装成一段文本注入到对话上下文中。def build_memory_context(memories): if not memories: return high_priority [m for m in memories if m[importance] 4] normal_priority [m for m in memories if m[importance] 4] parts [] if high_priority: parts.append(【重要记忆】) for m in high_priority: parts.append(f- {m[content]}) if normal_priority: parts.append(\n【相关记忆】) for m in normal_priority: parts.append(f- {m[content]}) return \n.join(parts)然后在构造 Claude API 请求时把这段文本放在 system prompt 里memory_context build_memory_context(retrieved_memories) system_prompt f你是一个有记忆的助手。 {memory_context} 请根据以上记忆信息来回答用户的问题。如果记忆中没有相关信息就正常回答不要编造。 response claude_client.messages.create( modelclaude-sonnet-4-20250514, max_tokens2000, systemsystem_prompt, messagesconversation_history )这里有个细节system prompt 里要明确告诉模型“如果记忆中没有相关信息就正常回答不要编造”。否则模型可能会强行把不相关的记忆扯进来导致回答质量下降。4.5 定期维护与清理记忆系统需要定期维护否则会越来越臃肿。我一般设置三个维护任务任务一过期记忆清理def cleanup_expired_memories(days90): db.execute( UPDATE memories SET status archived WHERE status active AND last_accessed datetime(now, -{} days) AND importance 3 .format(days)) db.commit()超过 90 天没被访问且重要性低于 3 的记忆标记为归档。归档的记忆不参与检索但保留在数据库中以备不时之需。任务二记忆合并def merge_similar_memories(threshold0.9): memories db.execute( SELECT id, content, embedding FROM memories WHERE status active ).fetchall() merged set() for i, m1 in enumerate(memories): if m1[id] in merged: continue for m2 in memories[i1:]: if m2[id] in merged: continue sim cosine_similarity( np.frombuffer(m1[embedding]), np.frombuffer(m2[embedding]) ) if sim threshold: # 保留 importance 更高的那条 keep m1 if m1[importance] m2[importance] else m2 remove m2 if keep m1 else m1 db.execute(UPDATE memories SET status merged WHERE id ?, (remove[id],)) merged.add(remove[id]) db.commit()任务三统计报告def generate_stats(): stats db.execute( SELECT COUNT(*) as total, SUM(CASE WHEN status active THEN 1 ELSE 0 END) as active, SUM(CASE WHEN status archived THEN 1 ELSE 0 END) as archived, AVG(importance) as avg_importance, AVG(access_count) as avg_access FROM memories ).fetchone() return dict(stats)这个统计报告能帮你了解记忆系统的运行状况。如果avg_access很低说明很多记忆存了但从来没被用过可能需要调整提取策略。如果archived占比很高说明记忆更新频率太快可能需要缩短清理周期。5. 常见问题与排查技巧实录5.1 记忆提取不准确怎么办这是最常见的问题。表现是该记的没记住不该记的记了一堆。排查思路先看提取 prompt 是否有问题。把模型的原始输出打出来看看它提取了什么。如果提取的内容明显偏离预期调整 prompt 中的规则描述。我遇到过模型把 AI 的回复也提取成记忆的情况后来在 prompt 里加了“只提取用户陈述的信息”就解决了。如果 prompt 没问题但提取质量还是差可能是模型能力不够。记忆提取是一个需要语义理解的任务小模型往往做不好。建议用 Claude Sonnet 或更高级别的模型来做提取不要为了省钱用 Haiku。还有一个容易被忽略的点对话的格式。如果对话历史是以非结构化文本传入的模型可能分不清哪句是用户说的、哪句是 AI 说的。建议用明确的分隔符比如[用户]和[助手]来标注。速查表问题表现可能原因解决方法提取了 AI 的回复prompt 未限定来源加“只提取用户陈述的信息”提取了寒暄内容未排除闲聊加“忽略寒暄和无关内容”提取数量过多未限制数量加“最多提取 5 条”提取内容太长未要求简洁加“用简洁句式表达”遗漏关键信息模型能力不足换用更高级别的模型5.2 检索结果不相关怎么调检索不相关通常有三种原因embedding 质量差、相似度阈值设置不当、记忆本身质量差。先检查 embedding。把 query 和检索结果的 embedding 相似度打出来如果相似度普遍偏低 0.3说明 embedding 模型不适合你的场景。中文场景一定要用多语言模型用纯英文模型效果会很差。再检查阈值。min_similarity设得太低会返回一堆不相关的记忆设得太高可能什么都返回不了。我的经验值是 0.3 到 0.5 之间具体取决于 embedding 模型的特性。可以先用一批测试 query 跑一下看看相似度分布再定阈值。最后检查记忆质量。如果记忆本身写得含糊不清检索自然不准。比如“用户喜欢喝咖啡”这条记忆用户问“我平时喝什么”能检索到但问“我的饮品偏好”可能就检索不到。所以记忆的表述要尽量完整包含多种可能的表述方式。5.3 上下文窗口被记忆撑爆怎么办这个问题在记忆条目很多的时候特别容易出现。检索出 10 条记忆每条 50 个 token就是 500 个 token。加上 system prompt 和其他内容很容易超过预算。解决方案第一严格控制检索数量。top_k不要超过 5通常 3 条就够了。如果 3 条记忆还不足以回答问题说明记忆提取的粒度太细了应该合并成更概括的记忆。第二对记忆做二次压缩。检索出的记忆如果太长可以用模型再压缩一次。比如把“用户说他住在北京朝阳区平时喜欢去三里屯逛街”压缩成“用户住北京朝阳区常去三里屯”。第三设置 token 预算。在组装上下文时计算记忆部分的总 token 数如果超过预算比如 500 token就按重要性排序截断。def truncate_memories(memories, max_tokens500): total 0 result [] for m in sorted(memories, keylambda x: x[score], reverseTrue): tokens estimate_tokens(m[content]) if total tokens max_tokens: break result.append(m) total tokens return result5.4 跨会话记忆丢失怎么排查跨会话记忆丢失通常是持久化环节出了问题。排查步骤检查数据库连接是否正常。有时候应用重启后数据库路径变了导致读不到之前的记忆。检查记忆的status字段。如果记忆被标记为archived或merged就不会被检索到。检查用户标识。跨会话记忆必须跟用户 ID 绑定如果每次会话生成的用户 ID 不同自然找不到之前的记忆。检查时间衰减。如果记忆太久没被访问时间衰减权重会很低可能导致它排不到 top-K 里。我遇到过一次跨会话记忆丢失排查了半天发现是数据库文件被放在了临时目录应用重启后临时目录被清空了。后来把数据库路径改成持久化目录就解决了。这种低级错误听起来可笑但在实际开发中真的很常见。5.5 性能优化的几个实用技巧记忆系统运行久了性能会下降。几个优化技巧批量操作写入记忆时不要一条一条插入用事务批量提交。检索时如果一次要查多条 query可以批量向量化。索引优化memories表的status和last_accessed字段加索引能显著提升清理和检索的速度。缓存热点记忆把访问频率最高的记忆缓存在内存里减少数据库查询。可以用 LRU 缓存设置合理的容量上限。异步写入记忆写入不需要同步完成可以放到后台队列里异步处理。这样不会阻塞主对话流程用户体验更好。定期 VACUUMSQLite 删除数据后不会自动回收空间定期执行VACUUM命令可以压缩数据库文件提升查询速度。# 批量写入示例 def batch_store_memories(memories): with db: for mem in memories: embedding get_embedding(mem[content]) cursor db.execute( INSERT INTO memories (content, memory_type, importance, embedding) VALUES (?, ?, ?, ?), (mem[content], mem[type], mem[importance], embedding.tobytes()) ) db.execute( INSERT INTO memory_vectors (rowid, embedding) VALUES (?, ?), (cursor.lastrowid, embedding.tobytes()) )用with db:上下文管理器所有操作在一个事务里完成比逐条提交快很多。实测下来批量写入 100 条记忆的时间从 2 秒降到了 0.3 秒。5.6 记忆系统的测试策略记忆系统很难测试因为它的行为依赖于对话内容而对话内容是千变万化的。我总结了一套测试策略单元测试针对每个函数单独测试。比如extract_memories函数准备几组固定的对话输入验证输出是否符合预期。retrieve_memories函数准备一批记忆和 query验证检索结果的相关性。集成测试模拟完整的对话流程。写入记忆、检索记忆、组装上下文、生成回复验证整个链路是否通畅。回归测试每次修改 prompt 或调整参数后跑一遍固定的测试用例确保没有引入退化。我维护了一个包含 50 组对话的测试集覆盖了各种边界情况用户改变偏好、用户提供矛盾信息、用户使用指代和省略、多轮对话中的信息分散等。人工评估自动化测试只能验证功能正确性无法评估记忆质量。定期人工检查记忆提取和检索的结果看看是否有明显的质量问题。我一般每周抽 20 条记忆人工审核一次。这套测试策略帮我避免了很多线上事故。有一次我调整了提取 prompt自动化测试全部通过但人工评估发现模型开始把用户的假设性陈述“如果我搬到上海的话...”当成事实记忆。这种问题自动化测试很难覆盖但人工一眼就能看出来。5.7 隐私与安全注意事项记忆系统存储的是用户的个人信息隐私安全必须重视。几个基本原则最小化存储只存必要的信息不要存敏感数据密码、身份证号、银行卡号等。如果用户不小心说了敏感信息提取时要过滤掉。加密存储数据库文件要加密尤其是部署在共享环境时。SQLite 支持加密扩展或者把数据库放在加密文件系统上。访问控制记忆的读写要有权限控制确保用户只能访问自己的记忆。数据删除提供记忆删除功能用户要求删除时必须彻底清除包括向量索引中的对应条目。审计日志记录记忆的写入和读取操作便于排查问题和合规审计。我在实际项目中遇到过用户要求删除所有记忆的情况。这时候不能只删主表还要删向量索引表否则向量检索时还可能返回已删除的记忆。正确的做法是def delete_all_memories(user_id): memory_ids db.execute( SELECT id FROM memories WHERE user_id ?, (user_id,) ).fetchall() with db: for row in memory_ids: db.execute(DELETE FROM memory_vectors WHERE rowid ?, (row[id],)) db.execute(DELETE FROM memories WHERE user_id ?, (user_id,))先删向量索引再删主表顺序不能反。因为向量索引表通过 rowid 关联主表如果先删主表rowid 就找不到了。这套记忆系统我在三个项目中落地过从客服机器人到个人助理再到文档问答每个场景的侧重点不同但核心架构是一致的。最深的体会是记忆管理不是一个纯技术问题它涉及到产品设计、用户体验、成本控制的平衡。存什么、怎么存、什么时候取这些决策需要结合具体业务场景来定没有一刀切的最优解。我见过有人把记忆系统做得极其复杂结果维护成本高到团队扛不住也见过有人用最简单的方案反而跑得很稳。找到适合自己团队的复杂度比追求技术上的完美更重要。