做对话应用的人应该都遇到过这个场景用户上午刚聊完的需求下午换个会话再问模型完全不记得甚至还要重新自我介绍一遍。这类问题的根源在于对话模型本身是无状态的每一次调用都是独立的唯一的“记忆”就是你塞进上下文里的那些字。claude-mem 这类项目想要解决的正是这个最让人头疼的短板——给模型配一个真正能用的外置记忆。我个人的定位是它更适合正在做智能助手、客服机器人、个人知识管家或者任何需要“记得用户是谁”的应用开发者。这篇文章会从需求拆解开始把 claude-mem 的核心设计讲清楚然后给出一套可以直接跑的 Python 实现最后把我踩过的坑和排查经验整理出来。不光是能用重点是让你知道每个设计决策背后的理由。1. 先拆需求对话模型为什么需要记忆动手写代码之前得先想明白我们要解决的问题到底是什么。很多人一上来就想着塞向量库、搞 embedding结果做出来一个“什么都能搜到但什么都没记住”的玩具。这就需要把记忆这件事拆开来看。1.1 无状态这个老毛病用 API 调用对话模型的时候每一个请求都是独立的。你可以把它想象成一个重度失忆的临时工他每次上班前都会把前一晚的记忆格式化你交给他的所有背景信息都得写在当天的“工作便签”上这个便签就是 Prompt。你说得越多便签越长能写下的信息也越有限。这带来两个直接问题。第一是每次都要重复信息用户体验极差。用户上次已经告诉你他住南方城市、家里有两个小孩、讨厌吃辣这次再聊做饭建议又得重新说一遍。第二是成本问题上下文越长每次调 API 的费用和延迟都在涨。如果一个人积累了大量的长期偏好靠“全文塞进去”的方式根本不现实。所以记忆模块要做的事情很清晰把值得记住的信息抽出来放在模型的外部存储里下次对话的时候再把最相关的那部分取回来注入到上下文中。存储和读写都在外部完成模型本身还是无状态的但在用户的视角里这个应用终于“记住了”他。1.2 记忆到底要存什么不是所有对话内容都值得存。我一开始犯过的错就是什么都记最后向量库里堆了一堆“今天天气不错”之类的废话真正要用的偏好反而被淹没。做记忆抽取的时候至少要把信息分成这么几类事实型记忆用户的名字、职业、所在城市、家庭情况、有猫、开什么车。这类信息相对稳定是后续所有对话的基础。偏好型记忆喜欢简洁的回答、讨厌玩笑、偏好代码示例、不吃香菜。这类信息影响生成风格和内容排序。任务与进度当前正在做的项目、已经完成的步骤、下一步计划。这类信息有助于多轮长期协作。情景型记忆最近一次聊到的具体事件或场景。比如“上周提到的搬家计划”这类信息时效性强过期价值不大。不同类型对应不同的存储策略。事实和偏好应该进入长期存储任务进度需要做状态更新情景记忆则可以设置有效期过期就清理。许多记忆模块的失败恰恰是没有做这个区分。1.3 记忆系统的三层架构在实际实现里我更倾向于把记忆分成三层短时记忆当前会话内最近几轮对话的原文直接注入上下文无需特殊存储。工作记忆当前会话的摘要用于跨轮保持主线。每轮结束后更新一次或者每 N 轮更新一次。长期记忆抽取出来的稳定偏好、事实、任务进度存进向量存储。这三个层次的读写路径完全不同。短时记忆是直接拼字符串工作记忆是更新一份摘要文本长期记忆则是“抽取—向量化—检索—注入”。很多 claude-mem 的初版实现其实只需要认真做后两层就能覆盖大多数使用场景。真正的难点在于三层之间的优先级和 token 配额怎么分配这一点我会在第三节的实际实现里展开。2. 核心设计claude-mem 的骨架现在把需求转成设计。一个完整可用的记忆模块在我看来至少要包含存储层、写入流程、召回策略、上下文注入四个部分。每个部分都有不少细节值得聊。2.1 数据结构与存储选型存储层是所有记忆操作的基础。我的建议是小规模场景直接上 SQLite不要一上来就引入重型的向量数据库。一张表存会话一张表存记忆条目这个组合最简洁。会话表记录会话 ID 和开始时间、更新时间用于管理上下文的生命周期。记忆条目表的核心字段是id、用户 ID、记忆类型、记忆内容、向量、来源会话 ID、重要度、置信度、创建时间、更新时间。向量字段可以存在同一张表的二进制列里也可以单独做一张向量关联表。检索的时候纯 Python 读出来做余弦相似度排序。这个方案在几千条记忆的规模下完全够用查询时间也就几十毫秒。等到数据量真的到了十万条以上再迁移到专门的向量数据库也不迟。思路是先用最简单的方案把业务逻辑跑通而不是一上来就引入一堆依赖。选 SQLite 还有一个额外的好处所有数据都是本地文件备份、迁移、隐私清理都极其简单。后面讲隐私问题的时候会再提到这个特性省了我很多事。2.2 记忆写入流程让模型帮你做信息抽取记忆不是把对话原文塞进数据库而是经过一道“提炼”的工序。理想的写入流程是这样的对话结束后把这一轮新增的对话交给大模型做结构化抽取。我习惯让模型返回 JSON格式大致是“记忆类型、内容、重要度、置信度”。抽取完成之后把内容向量化然后先在现有记忆里做一次相似度检查。这一步非常关键。如果新抽取出来的记忆和已有的某条记忆相似度超过阈值就不要新增而是更新已有那条。比如用户昨天说“公司在城西”今天说“公司搬到了城东”如果不做去重更新库里就会出现两条矛盾的信息。相似度检查可以解决大部分这种冲突。最后再根据记忆类型和重要度决定存储策略。重要度高的常驻重要度低的按需召回带时效性的记忆设置过期时间。写入流程中信息抽取和调用主对话的模型可以是同一个但建议用一个更便宜的模型来完成抽取控制成本。2.3 召回策略不让有用记忆淹没在无关结果里存储做得再好召回方式不对也白搭。我把召回分成两路并行处理后再合并。第一路是基于用户当前输入的语义检索。把当前问题向量化在记忆库里算余弦相似度取 top K。这一路负责找出和当前话题最相关的长期事实和偏好。第二路是近期记忆的加权召回按照记忆的更新时间倒序取最近几条。这一路解决的是语义检索容易忽略的“最近刚聊过的事”比如“刚才说的那个 API 密钥叫什么来着”。两条结果合并之后我会再做一次按重要度的重排。实际的打分可以参考这个公式score 0.6 * 语义相似度 0.3 * 时效权重 0.1 * 重要度权重其中时效权重可以通过指数衰减函数计算比如exp(-(now - last_update) / half_life)半衰期根据记忆类型设置。事实型记忆半衰期可以很长情景记忆则设短一些。这套打分方式比单纯按余弦相似度排序要稳得多因为它考虑了“是不是用户正在聊的”和“这条记忆本身有多重要”这两个维度。召回结果不是全部塞进上下文。我通常只保留得分最高的 3 到 5 条记忆并忽略得分低于阈值的条目。阈值需要反复调设置太高会漏设置太低会注入垃圾。2.4 上下文注入如何把记忆优雅地放进 Prompt有了召回结果最后一步是把它注入到模型输入里。注入的位置建议放在 System 部分内容用明确的边界标识包起来。注入模板我个人习惯这样组织[记忆区开始] 以下是关于用户的部分已知信息仅作参考若与当前对话冲突以当前对话为准。 - 类型事实 | 内容用户居住于南方城市家中两个孩子 - 类型偏好 | 内容用户喜欢简洁直接的回复需要代码时用 Python 示例 - 类型进度 | 内容正在搭建一个用户画像模块已完成存储层设计 [记忆区结束]放在 System 而不是直接拼在用户消息之前是为了保持用户消息的干净。同时要明确告诉模型记忆只是参考如果和当前对话冲突以当前对话为准。这个细节能避免模型被过期记忆带偏在记忆冲突场景下尤其重要。Token 预算必须提前规划。我习惯给记忆区设置一个硬上限比如 800 到 1200 token超出就按得分截断。宁可少带几条高质量记忆也不要让记忆区挤掉真正的对话上下文空间。有些实现甚至会给“常驻记忆”单独留出配额把用户最核心的身份信息始终放进去这部分不参与检索直接注入。这个方案适合“必须记住我是谁”的场景。3. 从零实现一个最小可用的记忆模块设计归设计真正跑起来才是硬道理。这一节我会给出一套完整的 Python 实现代码不算长但覆盖了上面说的所有核心流程。你在跑通之后再根据自己对接的模型平台和业务场景替换掉对应的部分即可。3.1 环境准备与依赖选择先说明一下我用的是一套相对通用的工具链刻意没有绑定特定平台。你只需要准备三类东西Python 3.10 以上环境这是基础。一个 SQLite 驱动。Python 自带的sqlite3就够了不需要额外装 ORM。一个 embedding 接口。你可以用某个大模型平台的 embedding API也可以用本地开源的向量模型。我个人测试下来中文场景用常见的中文向量模型效果不错英文场景各类主流 embedding 接口也都能胜任。关键是把你自己的 embedding 调用封装成一个函数。代码里我用了一个embed_text()占位函数你需要替换成自己的实现。这一点后面会再强调。3.2 存储层实现先用 SQLite 把逻辑跑通存储层我写了一个MemoryStore类处理建表和基础读写。import sqlite3 import numpy as np import time import uuid class MemoryStore: def __init__(self, db_pathclaude_mem.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS conversations ( id TEXT PRIMARY KEY, created_at REAL, updated_at REAL, summary TEXT ) ) cursor.execute( CREATE TABLE IF NOT EXISTS memory_items ( id TEXT PRIMARY KEY, user_id TEXT, mem_type TEXT, -- fact / preference / progress / episode content TEXT, embedding BLOB, source_conversation_id TEXT, importance REAL DEFAULT 0.5, confidence REAL DEFAULT 1.0, created_at REAL, updated_at REAL, expires_at REAL NULL ) ) cursor.execute( CREATE INDEX IF NOT EXISTS idx_memory_user ON memory_items(user_id) ) self.conn.commit()关于向量存储我选择直接把 numpy 数组序列化存进 BLOB 字段检索时全表扫描算相似度。这个方案在千条量级完全没问题还省掉了引入额外组件的成本。向量维度取决于你用的 embedding 模型常见的有 384 维、768 维、1024 维等存储时固定维度即可。3.3 写入与检索整个记忆模块的核心逻辑写入一条记忆的时候需要做归一化和去重。def add_memory(self, user_id, mem_type, content, embedding, importance0.5, confidence1.0, source_conversation_idNone, expires_atNone): # 先去重如果存在内容非常相似的记忆就更新那一条 similar self.search_memory(user_id, content, top_k1) if similar and similar[0][1] 0.9: existing_id similar[0][0] now time.time() self.conn.execute( UPDATE memory_items SET content ?, embedding ?, importance ?, confidence ?, updated_at ?, expires_at ?, source_conversation_id ? WHERE id ? , (content, embedding, importance, confidence, now, expires_at, source_conversation_id, existing_id)) self.conn.commit() return existing_id mem_id str(uuid.uuid4()) now time.time() self.conn.execute( INSERT INTO memory_items (id, user_id, mem_type, content, embedding, source_conversation_id, importance, confidence, created_at, updated_at, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , (mem_id, user_id, mem_type, content, embedding, source_conversation_id, importance, confidence, now, now, expires_at)) self.conn.commit() return mem_id检索的时候把存储的向量全部读出来做余弦相似度。几百上千条记录的性能完全够用。def search_memory(self, user_id, query_embedding, top_k5): cursor self.conn.execute( SELECT * FROM memory_items WHERE user_id ?, (user_id,) ) items cursor.fetchall() if not items: return [] scores [] for row in items: mem_id, _, _, content, emb_blob, _, importance, confidence, _, updated_at, expires_at row if expires_at and time.time() expires_at: continue emb np.frombuffer(emb_blob, dtypenp.float32) score float(np.dot(emb, query_embedding) / (np.linalg.norm(emb) * np.linalg.norm(query_embedding) 1e-8)) scores.append((mem_id, score, content, importance, confidence, updated_at)) scores.sort(keylambda x: x[1], reverseTrue) return scores[:top_k]配合refine_score把时效权重加进来这一步才是真正提升召回质量的关键。代码不复杂就是在排序前对分数做修正。3.4 整合对话流程让记忆真正参与每一轮对话核心类MemoryAwareAssistant的业务流程是构建带记忆的 System Prompt调用对话模型把返回内容展示给用户最后异步抽取并写入新记忆。class MemoryAwareAssistant: def __init__(self, store: MemoryStore, user_id: str, embed_func): self.store store self.user_id user_id self.embed_func embed_func self.conversation_history [] def build_system_prompt(self, user_input): query_emb self.embed_func(user_input) memories self.store.search_memory(self.user_id, query_emb, top_k5) if not memories: return 你是用户的 AI 助手。 lines [] for _, score, content, importance, confidence, updated_at in memories: lines.append(f- [置信度 {confidence:.2f}] {content}) memory_block \n.join(lines) return f [记忆区开始] 以下是关于用户的已知信息仅作参考若与当前对话冲突以当前对话为准。 {memory_block} [记忆区结束] 你是用户的 AI 助手请在回答时参考以上信息。 .strip() def chat(self, user_input): system_prompt self.build_system_prompt(user_input) messages ( [{role: system, content: system_prompt}] self.conversation_history [{role: user, content: user_input}] ) # 这里调用你自己的模型接口比如官方 SDK 的对话补全 # response your_llm_client.chat(messagesmessages) response placeholder_call_llm(messages) self.conversation_history.append({role: user, content: user_input}) self.conversation_history.append({role: assistant, content: response}) # 异步执行记忆抽取避免阻塞主回复 self._async_extract_and_store(user_input, response) return response def _async_extract_and_store(self, user_input, response): # 把这段新对话交给模型抽取结构化记忆然后调用 store.add_memory memories extract_memories_with_llm(user_input, response) for mem in memories: emb self.embed_func(mem[content]) self.store.add_memory( self.user_id, mem[mem_type], mem[content], emb, importancemem.get(importance, 0.5), confidencemem.get(confidence, 1.0), )extract_memories_with_llm是这个模块的灵魂。我给它设计了一个抽模型 prompt 模板让模型返回 JSON 数组每项包含mem_type、content、importance、confidence。你只需要接好你自己的模型调用函数即可。3.5 关键参数配置表跑通后你会面临一个参数调优阶段。这里放一个我常用的参数表作为起点非常合适参数推荐初值作用top_k5召回记忆条数上限。太低会漏太高上下文会被挤爆。min_score0.35余弦相似度阈值。低于这个值的记忆不注入。recency_half_life7 天时效权重半衰期。时间敏感性内容设短一些。importance_weight0.3重要度在最终打分的占比。semantic_weight0.6语义相似度在最终打分的占比。mem_token_limit1000记忆区注入的 token 上限。溢出部分截断。dedup_threshold0.9记忆去重阈值。相似度高于此值则更新而非新增。这些参数是我自己调出来的经验值不代表所有场景通用。跑一段时间后建议打印召回日志看看哪些记忆被注入但没起作用哪些该召回的却被漏掉针对性微调即可。4. 踩坑记录与排错速查这套东西看着不难真正落地时问题层出不穷。我把最常见的几类问题按症状、原因、解决方案整理出来。这些经验是我在调试过程中踩过坑之后换来的希望你不用全经历一遍。4.1 症状一该记住的根本没记住召回结果一团糟有人反馈模型和用户聊了两天结果再问道“我的猫叫什么名字”模型还是张嘴就来我不知道。这个问题往往不是模型的问题而是抽取环节根本没把猫的名字写入记忆库。第一个要查的是抽取 prompt。很多抽取模型觉得“猫叫咪咪”这种信息不重要就过滤掉了。我后来在 prompt 里明确加了“用户明确提供的信息即使看起来不重要也需保留”情况立刻改善。第二个要查的是 embedding 质量。用户用中文提问“我家猫叫什么”而记忆里存的是“我的猫名字叫咪咪”如果 embedding 模型对短文本语义理解能力差这两句话的相似度可能很低导致召回失败。建议测试时打印召回日志看看到底召回了什么。4.2 症状二上下文被记忆区挤爆主对话反而被压缩记忆注入不是越多越好。有些人把 top_k 调到了 15又把 min_score 调得很低结果每次对话都塞进几千字的记忆摘要正经对话还没开始token 就快用完了。我的经验是记忆区严格控制在总上下文的三分之一以内核心对话内容永远优先。出现超限时不要简单地从头截断而应该按评分排序后丢弃低分记忆。此外把特别重要的几条记忆设为常驻不参与检索直接注入其他记忆走检索和打分。这样既保证了核心信息稳定存在又不会让记忆区无限膨胀。4.3 症状三记忆自相矛盾模型被旧信息带偏用户换了城市、换了工作、改了偏好这些信息变动很常见。如果不做冲突处理库里就会出现“用户在上海”和“用户在北京”并存模型偶尔被旧的带偏。解决方案有三道防线。第一道是写入时的去重更新相似度高于阈值的直接覆盖旧条目。第二道是注入时的提示词约束明确告诉模型“记忆区信息与当前对话冲突时以当前对话为准”。第三道是为每条记忆保留 updated_at 和 confidence在检索打分时高置信度和新写入的记忆应该有更高的权重。4.4 症状四隐私和数据安全没底这个我不展开讲太多规范性的东西只谈实际操作。既然记忆模块存了用户的偏好和事实信息理应用户有权知道自己被存储了什么、能要求删除。我在项目里单独写了一个管理命令可以查看某个 user_id 的全部记忆也支持一键清空。这个功能在开发和测试阶段特别有用排查问题时分分钟就能看到库里的记忆是不是符合预期。另外建议敏感信息在写入记忆库之前做脱敏处理。身份证号、完整地址、银行账号这类信息最好在抽取阶段就明确禁止写入或者做部分遮蔽。真需要存储也应该加密落盘。SQLite 的本地文件简单但也别高估它的安全性——拿走文件就等于拿走了全部记忆。最后的实际操作心得跑 claude-mem 这类记忆模块一段时间之后我最大的体会是记忆质量远比记忆数量重要。与其把每一轮对话都塞进去不如只抽取那些真正稳定、真正影响后续交流的信息。我发现一个用户积累到两三百条高质量记忆之后再往上加对对话体验的提升已经很有限。另一个注意点是不要把记忆模块当成一次搞定的功能。它会随着用户和系统的长期交互慢慢“生长”所以日志和可观测性一定要从一开始就做好。我把每次检索命中的记忆以及打分信息都写进日志出问题的时候回看这些日志很快就能定位是抽取的问题、存储的问题还是召回的问题。没有日志排查只能靠猜。如果你也想做一个类似的记忆模块我的建议是从最小场景切入只做到“记住用户名字和几个核心偏好”这一层先把整套流程跑通再逐步增加事实、任务进度、时效处理这些更复杂的能力。记忆系统看起来简单真正打磨好要靠很长时间的真实交互来验证和迭代。