ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Claude持久记忆系统拆解:从RAG到向量检索的轻量实现

Claude持久记忆系统拆解:从RAG到向量检索的轻量实现

1. 项目概述:claude-mem到底解决什么问题

先说说我为什么会对这个项目感兴趣。做AI应用开发久了,你会发现一个特别拧巴的现象:Claude这类大模型单次对话的上下文窗口越来越大,几十万token看着很够用,可实际跑业务的时候,用户一句话没说完,下次打开会话什么都忘了。每次都要重新解释需求、重新给背景、重新调教语气,体验简直像跟一个天天失忆的同事配合工作。

claude-mem本质上就是在Claude会话外面加了一层持久记忆系统。它不是改模型内部权重,也不是去碰Claude自己那套会话缓存,而是在API调用层做拦截和补充:把对话历史里值得长期记住的信息抽出来,存到外部存储里,下次对话开始前再把相关记忆注入回上下文。核心诉求非常明确——让AI在多次会话之间记住人、记住事、记住偏好,而不是每次都从零开始。

这个项目适合谁?三类人。第一类是深度使用Claude API做私人助理、知识库问答或者自动化流程的开发者,受够了上下文丢失的苦。第二类是研究RAG(检索增强生成)落地实践的人,claude-mem其实就是一个轻量级的记忆版RAG,架构上很值得拆解。第三类是自己折腾智能体(Agent)的玩家,想让多个会话共享同一份用户画像和长期事实,靠手写JSON硬凑太脆弱,需要一个正经的记忆层。读完这篇,你能理解它的核心设计逻辑,也能动手做一个简化版跑通全流程。

2. 核心设计拆解:记忆系统为什么这样搭

2.1 开放式记忆和上下文窗口的关系

要理解claude-mem的架构,先要搞清楚一个关键问题:为什么不能直接把所有历史消息全塞进上下文窗口?

很多人第一反应是,既然Claude支持超长上下文,那我把所有聊天记录拼接起来发过去不就行了吗?实测下来有三座大山。第一是成本,输入token按量计费,历史消息越攒越多,每次对话光背景材料就能吃掉一大笔费用。第二是注意力稀释,模型面对几千条历史消息时,检索关键信息的能力会被大量无关内容干扰,越聊到后面越容易“忘事”。第三是时效性问题,两年前的聊天记录早就过期了,全量塞进去只会让AI基于错误旧信息回答。

claude-mem的解法很干净:不存原始对话全文,而是异步抽取“值得记住的东西”,结构化落库。它管自己叫记忆层(memory layer),我认为本质上是三层漏斗结构:

  • 抽取层:从对话里识别实体、偏好、约定、长期目标这些信息。
  • 存储层:把抽取出的记忆整理成结构化条目,向量化后入库。
  • 检索层:新会话开始前,按相关性把记忆捞出来,注入System Prompt或者上下文开头。

这个设计背后的逻辑其实和人的记忆机制很像。你不会记住和同事的每一句闲聊,但你会记住这个同事是个急性子、负责前端、讨厌周五下午开会——大脑在线做抽取,睡觉时做巩固,用的时候再调取。claude-mem就是把这套机制外置化了。

2.2 为什么选用向量检索而不是只靠关键词

记忆存储有一个绕不开的问题:用户下次提问时往往不会用完全一样的措辞。比如第一次对话里用户说“我养了一只叫豆包的柯基”,下个月再问“豆包现在该打什么疫苗”,关键词“柯基”和“疫苗”完全不重叠,但语义上高度相关。这种场景靠传统的关键词匹配会直接漏掉。

所以必须向量化。claude-mem会把每条记忆转成高维向量,存入支持相似度检索的数据库里。这里有几个关键的工程选型点值得展开讲。

第一,embedding模型的选择直接决定了记忆召回的上限。用OpenAI的text-embedding-3-small还是本地跑bge-m3,这取决于你的部署环境和对隐私的容忍度。我见过很多项目死在embedding这步,不是因为模型不好,而是压根没考虑过中文和英文的混合场景,用了一个只支持英文的模型结果中文记忆全部变成乱码向量。

第二,向量数据库不用迷信重型武器。很多人一听到向量数据库就想着上Milvus或者Weaviate,然后发现为了一个个人助手项目还得先运维一套分布式集群。claude-mem在面对轻量级场景时的思路是对的——用SQLite加扩展,或者直接用支持向量索引的轻量库,数据量在百万条以下根本不需要重型方案。我在实际项目里用sqlite-vec或者chromadb都能跑得很稳。

第三,检索不能只靠向量相似度。一条记忆如果只按向量打分排序,很容易出现“语义上相关但实际没用”的结果。比如用户提到过“我不吃香菜”,这条记忆向量和任何涉及“食物”的对话都沾边,但只有点菜场景才需要调取。所以完整的检索策略应该是:向量相似度打底,加上关键词过滤和记忆优先级加权,这也就是后面要讲的混合检索。

2.3 记忆条目需要生命周期管理

记忆系统最容易被忽略的问题是:存进去之后怎么办?如果只写不删,时间一长库里全是过期的、矛盾的信息。用户上个月说“我最喜欢喝美式”,这周说“最近医生建议我戒咖啡”,库里同时存在两条冲突记忆,检索出来AI就不知道该听谁的。

claude-mem这类系统必须引入记忆生命周期管理,常见做法有三层:

  • 时效性衰减:每条记忆带上时间戳,超过一定时限的弱记忆降低权重,长期没被检索到的记忆可以归档或删除。
  • 冲突解决:新记忆和旧记忆矛盾时,以较新的为准,同时标记旧记忆为失效,而不是直接物理删除,方便追溯。
  • 记忆合并:抽取层发现新信息和已有记忆指向同一实体时,做增量合并。比如今天记住“用户有只柯基叫豆包”,下周记住“豆包三岁了”,应该合并成一条完整档案,而不是新起一条。

这一步做得好不好,直接决定记忆系统用三个月之后是越来越聪明,还是越来越混乱。我见过不少项目demo阶段效果惊艳,跑了一个月后回答质量断崖式下跌,调查下来就是记忆库里堆满了互相打架的碎片条目。

3. 实操落地:从零搭一个claude-mem简化版

3.1 整体架构与依赖清单

我不太喜欢只讲概念,还是直接动手做个能跑通的简化版。目标很明确:每次和Claude对话完,自动抽取记忆入库,新对话开始时自动检索相关记忆注入系统提示词。

技术栈我做了减法,尽量用轻量方案:

  • Python 3.10以上,主流程用FastAPI包一层方便外部调用。
  • Claude API,直接走官方SDK。
  • SQLite存结构化记忆,扩展用sqlite-vec做向量检索。
  • embedding模型先调本地bge-small-zh,避免每次请求都走外部接口,降低延迟。

依赖安装就三行命令,但有个坑要重点说:sqlite-vec目前对Python 3.12以上版本的支持还不算稳定,我实测下来3.10和3.11最省心,别为了尝鲜用最新版Python,不然光编译扩展就能折腾半小时。

# requirements.txt anthropic>=0.25.0 fastapi>=0.110.0 uvicorn[standard] numpy

3.2 记忆抽取模块:怎么把对话变成结构化条目

记忆抽取是整个系统的灵魂。我的做法是让Claude自己当“记忆提炼师”——每轮对话结束后,把原始对话丢给Claude,让它按照预设的Schema提取记忆条目。

这里有一个核心技巧:抽取任务的System Prompt必须写得很死,否则模型会自由发挥。我踩过最大的坑就是让模型“抽取重要信息”,结果它把整段对话都摘要了一遍,入库的全是废话。后来我把Schema限定成几个明确字段:实体、属性/关系、置信度、时效类型。强制模型只输出JSON,不输出任何解释性文字。

import json from anthropic import Anthropic client = Anthropic() EXTRACT_PROMPT = """ 你是记忆提取引擎。从对话中提取需要长期记住的事实性信息。 只输出JSON数组,不要输出任何其他文字。数组每个元素结构如下: { "entity": "实体名称,比如'用户'、'豆包'、'项目名'", "attribute": "关于该实体的具体事实", "relation": "实体之间的关系,没有则为空字符串", "confidence": 0到1之间的数值, "permanence": "temporary"或"permanent" } 提取原则: 1. 只提取事实,不提取情绪和闲聊 2. 偏好、身份、长期计划标记为permanent 3. 临时状态(比如"今天在加班")标记为temporary 4. 如果对话里没有可提取的记忆,输出[] """ def extract_memories(conversation_text): response = client.messages.create( model="claude-sonnet-7-0", max_tokens=2000, system=EXTRACT_PROMPT, messages=[{"role": "user", "content": conversation_text}] ) raw = "".join(block.text for block in response.content) try: return json.loads(raw) except json.JSONDecodeError: # 抽取失败时不阻断主流程,记录日志即可 print(f"[WARN] 记忆抽取返回非法JSON: {raw[:200]}") return []

这个模块有几个细节值得反复调试。第一,Claude返回的JSON有时候会带有Markdown代码块标记,json.loads之前要做一次清洗,把json前缀去掉。第二,max_tokens不能给太短,否则长的对话会被截断导致JSON不完整。第三,抽取失败必须静默降级,绝不能因为记忆模块报错导致主对话流程崩溃。

3.3 存储模块:SQLite加向量的组合拳

存储层我用的是SQLite加sqlite-vec扩展。设计上分两张表,一张存记忆原文和结构字段,一张存向量。分开存的理由是向量检索通常会返回topK候选,拿到候选ID之后还需要回表查原文,加过滤条件的时候也更灵活。

CREATE TABLE memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity TEXT NOT NULL, attribute TEXT NOT NULL, relation TEXT DEFAULT '', confidence REAL DEFAULT 0.5, permanence TEXT DEFAULT 'temporary', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_access_at TIMESTAMP, access_count INTEGER DEFAULT 0 );

另一张向量表用sqlite-vec的虚拟表语法:

CREATE VIRTUAL TABLE memory_vectors USING vec0( memory_id INTEGER PRIMARY KEY, embedding FLOAT[512] );

这里有个选型心得:embedding维度选512是我权衡后的结果。维度太低(比如128)会导致语义区分度不够,维度太高(比如1536)虽然更精准但存储和计算开销都上去了,个人项目没必要。bge-small-zh输出正好是512维,性价比很高。

写入流程就是先把结构化记忆插入memories表拿到自增ID,再做embedding向量化插入memory_vectors表。两条操作需要包在事务里,避免出现有原文没向量或者反向的脏数据。

import sqlite3 import numpy as np DB_PATH = "./claude_mem.db" def _get_connection(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def store_memory(memory_item): conn = _get_connection() try: cursor = conn.cursor() cursor.execute( """INSERT INTO memories (entity, attribute, relation, confidence, permanence) VALUES (?, ?, ?, ?, ?)""", (memory_item["entity"], memory_item["attribute"], memory_item["relation"], memory_item["confidence"], memory_item["permanence"]) ) memory_id = cursor.lastrowid embedding = get_embedding( f"{memory_item['entity']} {memory_item['attribute']}" ) vector_bytes = np.asarray(embedding, dtype=np.float32).tobytes() cursor.execute( "INSERT INTO memory_vectors (memory_id, embedding) VALUES (?, ?)", (memory_id, vector_bytes) ) conn.commit() return memory_id except Exception as e: conn.rollback() print(f"[ERROR] 存储记忆失败: {e}") return None finally: conn.close()

3.4 检索模块:混合召回和重排序

检索模块决定了系统能不能在关键时刻把最恰当的记忆捞出来。我的做法是向量检索和关键词过滤并行跑,最后合并打分。

向量检索自然是用SQLite的KNN查询:

def search_memories(query_text, top_k=5): query_embedding = get_embedding(query_text) query_bytes = np.asarray(query_embedding, dtype=np.float32).tobytes() conn = _get_connection() try: # sqlite-vec 的 KNN 搜索语法 rows = conn.execute( """ SELECT memory_id, distance FROM memory_vectors WHERE embedding MATCH ? ORDER BY distance LIMIT ? """, (query_bytes, top_k * 2) # 多取一些,留给后续过滤 ).fetchall() memory_ids = [row["memory_id"] for row in rows] if not memory_ids: return [] # 回表查原文,同时过滤永久记忆优先 placeholders = ",".join("?" for _ in memory_ids) result_rows = conn.execute( f""" SELECT * FROM memories WHERE id IN ({placeholders}) ORDER BY CASE permanence WHEN 'permanent' THEN 0 ELSE 1 END, confidence DESC """, memory_ids ).fetchall() return [dict(r) for r in result_rows[:top_k]] finally: conn.close()

检索拿到结果后还有一个关键步骤:注入上下文要注意格式。我不建议把记忆条目硬塞进用户消息里,更好的方式是在System Prompt里加一个“对话者档案”段落,让模型把记忆当作背景知识而不是对话内容。

def build_system_prompt(base_prompt, memories): if not memories: return base_prompt memory_lines = [] for m in memories: context = f"{m['entity']}:{m['attribute']}" if m["relation"]: context += f"(与{m['relation']}相关)" memory_lines.append("- " + context) memory_section = "\n".join(memory_lines) return f"""{base_prompt} 以下是长期记忆中与当前对话相关的背景信息,请基于这些背景作答,但不要主动提及记忆的存在: {memory_section}"""

这个Prompt设计有三个精妙之处。一是用“不要主动提及记忆的存在”避免模型每句话都来一句“根据您的长期记忆,我知道你……”,非常出戏。二是记忆放在System层而不是User层,模型会更当回事。三是每条记忆用短横列表呈现,避免长段落压过真实的用户提问。

3.5 用FastAPI包成服务

为了让记忆系统能被主流程调用,我用FastAPI包了一层简单的HTTP接口。三个接口就够用:录入对话、检索记忆、清理记忆。

from fastapi import FastAPI, Request from pydantic import BaseModel app = FastAPI() class ChatRecord(BaseModel): conversation: str # 本轮完整对话文本 class QueryRequest(BaseModel): query: str top_k: int = 5 @app.post("/memorize") def memorize(record: ChatRecord): """接收完整对话,抽取并存储记忆""" memories = extract_memories(record.conversation) saved = [] for item in memories: mid = store_memory(item) if mid is not None: saved.append(mid) return {"saved": len(saved), "memory_ids": saved} @app.post("/recall") def recall(req: QueryRequest): """根据当前问题检索相关记忆""" results = search_memories(req.query, top_k=req.top_k) return {"memories": results} @app.delete("/memories/{memory_id}") def delete_memory(memory_id: int): """单条删除,用于纠正错误记忆""" conn = _get_connection() try: conn.execute("DELETE FROM memories WHERE id = ?", (memory_id,)) conn.execute("DELETE FROM memory_vectors WHERE memory_id = ?", (memory_id,)) conn.commit() return {"deleted": memory_id} finally: conn.close()

跑起来很简单,uvicorn main:app --port 8900就行。我故意没把Claude消息创建流程也包进去,因为实际业务里Claude调用场景太杂,可能是官方SDK、可能是LangChain、可能是内部封装,耦合太深反而不好。保持记忆服务独立,通过HTTP对接是最灵活的方式。

4. 避坑实录:这套方案绕不开的五个大坑

4.1 embedding模型选型的中文暗坑

我在前面提到过bge-small-zh,这里展开说为什么。早期我贪省事直接用了某个英文模型,结果中文人名和口语表达变成了不知所云的向量,检索精度惨不忍睹。后来换成本地部署的bge-small-zh,效果立刻质变。

另一个坑是embedding和Claude本身的tokenizer不统一。embedding模型分词粒度偏粗,Claude对中文按字切分,导致检索出来的记忆文本进到Claude上下文里的实际token数比预估高30%左右。设计注入长度余量时必须把这个因素算进去,否则很容易突然触发长上下文截断。

4.2 碎片化记忆没有合并,越存越蠢

只抽取不合并,是新手做记忆系统最爱犯的错误。我前面讲生命周期管理时说过原理,这里给一个具体反例。用户第一轮说“我在上海工作”,第二轮说“我跳槽去杭州了”,如果不做冲突消解,库里有两条相互矛盾的城市记忆。向量检索时两条都命中,Claude就会困惑到底该说上海还是杭州。

我的做法是抽取后增加一步合并校验:按实体名分组,检查属性是否冲突。冲突时以新记忆为准,把旧记忆的permanence字段改成superseded,同时保留原文方便追溯。这一步可以在抽取模块里同步做,也可以在存储前异步做,看你对实时性的要求。

4.3 记忆注入导致上下文混乱

记忆检索质量不高的时候,盲目注入上下文反而会拖累模型。我之前遇到过一种典型情况:用户问“今天天气怎么样”,系统把“用户有一只叫豆包的柯基”这条记忆召回来了,因为向量相似度勉强沾了点边,结果Claude的回答变成“根据我对你的了解,你可能需要带豆包出门时看天气”,完全跑偏。

解决思路是给检索结果增加一个相似度阈值。低于阈值的记忆直接丢弃,宁可少忆,不可错忆。我一般设定余弦相似度0.45为最低阈值,这个值需要根据你的embedding模型和语料特点调参,不是固定的。另外可以在记忆条目里增加标签字段,检索时先按业务标签粗筛,再算向量相似度,能大幅减少误召。

4.4 检查用户身份边界:多用户隔离

如果你做的不是单机自用项目,而是给多人部署的服务,记忆隔离就是生死线。A用户的花费习惯绝不能注入到B用户的对话里。我把SQLite的表结构里增加了一个user_id字段,存储和检索都强制按用户过滤。

class MemoryItem: def __init__(self, user_id, entity, attribute, relation=""): self.user_id = user_id self.entity = entity self.attribute = attribute self.relation = relation # 存储时WHERE user_id = ? # 检索时WHERE user_id = ? AND id IN (...)

这个改动的成本几乎为零,但能避免极其严重的隐私事故。多用户场景建议直接用PostgreSQL加pgvector,SQLite在多写并发下会出锁定问题,个人项目没事,上了生产会很难受。

4.5 API调用成本失控

每轮对话后做抽取,抽取本身又是一次Claude API调用,这意味着记忆系统的存在会让API调用量翻倍。如果用户每天有大量会话,成本压力不是小数目。

优化方向有三个:第一,不是每轮都抽取,而是对话长度超过阈值才触发抽取,避免短会话浪费。第二,抽取用更小的模型,比如把抽取任务从Claude切到本地跑的小模型,或者用Haiku这类廉价档位。第三,加记忆去重的前置过滤,如果整段对话里没有新实体新事实,直接不调抽取接口。我实践下来第三个优化的效果最明显,能让抽取API调用量下降50%以上。

5. 能力扩展:claude-mem还能往哪走

走到这里,一个可用的claude-mem简化版已经跑通了。但聊几个我最近在思考的扩展方向,很多就是把前面讲过的基础模块重新组合。

第一个方向是给记忆加业务分层。现在很多人做客服机器人和销售助理,需要的不仅仅是用户偏好,还有订单状态、售后进度这些动态业务数据。可以把记忆拆成静态档案和动态状态两层,静态档案走长期存储,动态状态走短时缓存并用业务流程驱动更新。这样检索策略也会更精准:回答价格问题时只查静态价格表,回答配送问题时查动态订单状态。

第二个方向是双向写入。目前记忆都是从对话抽取,属于被动记录。更强的是让AI在对话过程中主动调用记忆写入工具。比如用户明确说“帮我记住每周五下午三点开会”,抽取模块不见得能充分理解这种指令性信息,但如果Claude在对话中能调用一个save_memory工具,把这句话直接结构化入库,准确率会高非常多。这就是从“后处理抽取”进化成“内置记忆工具”。

第三个方向是记忆的可解释性面板。做用户研究或者产品调试的时候,你总要知道系统记住了什么、为什么这个回答会出现某条背景信息。给记忆系统加一个可视化查询界面,按实体浏览记忆图谱、查看每次对话检索命中哪些记忆、手动删除或修正错误条目。这个面板做出来,整个系统的可维护性会上升一个档次。

第四个方向是主动遗忘和隐私合规。我始终觉得记忆系统越强大,越要做克制。给每条记忆设置过期时间,定期清理高频访问但低价值的噪音记忆,提供全量导出和注销即删除的API。这些功能不一定用得上,但一旦用上,就能避免非常被动的隐私事故。

6. 从项目到产品:动手前想清楚这三件事

最后聊点产品层面的体会。如果你看了这篇文章也想去复刻一个claude-mem,请先冷静想清楚三件事,否则很容易做一半失去热情。

第一,记忆抽取的准确性是项目的生命线。第一版就跑通了并不意味着体验好,你得准备好花大量时间打磨抽取Prompt和合并逻辑。我的建议是最初两周集中做50到100条真实对话样本,手工标注出“我认为值得记住的信息”,然后用标注结果一遍遍调优抽取Prompt和Confidence阈值。这个过程很枯燥,但决定成败。

第二,严格守住“记忆宁缺毋滥”原则。一个精心维护的只有50条高质量记忆的系统,远比一个堆了5000条噪音碎片的系统好用。做优先级排序时给自己的代码里写死一条原则:不确定这信息下个月还有没有用,就不存。我看到太多项目死于信息囤积癖。

第三,保持系统的可观测性。不要等到用户说“怎么回答得怪怪的”才发现是记忆注入出了问题。我建议你在上线时就在日志里记录每次检索命中了哪几条记忆、每条的相似度分数、注入到上下文的实际形态。排查问题的时候有日志可查,比对着代码猜状态高效太多了。

claude-mem这个方向的价值在于它重新定义了AI应用的记忆边界。过去我们只能用上下文硬扛,现在的思路给长期记忆开了一条新路。做技术的人最爽的瞬间,就是自己造的东西真的让用户体验发生了肉眼可见的变化——当你的AI助手下一次开口就能叫出用户的名字、记得上次没聊完的方案,那个时刻你会觉得所有踩过的坑都值得。

我自己接下来会继续沿着这个方向做迭代,重点放在上文说的动态业务状态和记忆可视化这两块。希望这篇拆解能给你一些启发,也欢迎你在动手的过程中去踩那些我写在文章里的坑——踩过之后,你才算真正理解记忆系统。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进