☰
给Claude API加上外挂记忆:基于向量检索的长期记忆方案实战
2026/10/8 21:21:43 网站建设 项目流程

最近在做一个基于 Claude API 的客服机器人,最让我难受的并不是模型能力,而是它真的没有记忆。用户昨天刚报过的订单号,今天换了个会话再问,它就跟失忆了一样,一切又得从头教。翻遍各种方案之后,我看到了一个叫 claude-mem 的项目思路,它不修改模型,而是在 Claude 外面加一层“外挂记忆”,把对话里值得记住的信息抽取、存储,下次对话时再自动召回注入。这个思路解决了我一直头疼的长期记忆问题,而且不需要微调模型,只用 API 就能落地。这篇文章我想把这个项目的核心设计拆开来讲,并且带大家手写一个最小可用的版本,顺便把我踩过的坑都记录下来,希望对同样在折腾 Claude 应用的人有帮助。

1. 先聊聊为什么 Claude 需要一块“外挂记忆”

1.1 原生对话的“金鱼记忆”尴尬

很多人刚开始用 Claude API 时会犯一个错觉:它上下文窗口那么大,是不是把聊天记录全塞进去就相当于有记忆了?实际测下来远不是这么回事。API 本身是无状态的,每次请求都需要你把历史消息重新传一遍,否则模型面前就是一张白纸。就算你勉强把历史全带上,对话一旦跨天、跨会话,这些记录就完全断掉了。

我做过一个内部知识库问答机器人,最典型的情况是用户第一轮问“我们公司报销流程是什么”,第二轮问“那差旅住宿的上限呢”。单看第二轮,其实它依赖第一轮里的上下文,这还没什么。真正崩溃的是用户隔了一周回来说“还是上次那个报销的问题”,但系统里完全没有“上次”的概念,模型只能继续问“您上次问的是什么?”这种体验放在真实场景里根本没法用。

另一个问题是 token 成本。假设一段对话有 200 轮,每轮平均 500 token,光历史就 10 万 token 起步。如果每天都这么堆,窗口再大也会爆,而且大部分历史内容跟当前问题无关,纯粹是浪费。所以我们需要一种更聪明的记忆方式:只记住有用的、可以跨会话复用的信息,而不是把所有流水账都囤着。

1.2 claude-mem 给了一种什么解决思路

claude-mem 最核心的思想,是把“记忆”从模型会话里拿出来,放到一个独立的存储层。它不做模型微调,也不依赖 Claude 本身的能力,而是作为应用中一个额外的模块存在。你可以把它想象成一个贴身秘书:每次你跟 Claude 聊完,这个秘书会把聊天里提到的关键信息记在小本子上;下次聊天开始前,秘书翻出小本子里跟当前话题相关的部分,悄悄递给 Claude,让它看起来像是一直记得你。

具体来说,这条链路包括三步:抽取、存储、召回。抽取环节用 Claude 自己来做对话摘要,把用户偏好、事实信息、待办事项等结构化地提炼出来;存储环节把这些结构化内容落到本地数据库,同时为每条记忆生成向量索引;召回环节则是在用户每次发消息前,用当前问题去检索最相关的若干条记忆,注入到 system prompt 里。

这个思路好在哪里?首先是省 token,你不需要把全部历史都搬进来,只搬最相关的记忆;其次是跨会话,记忆写进持久化存储,用户换个设备、隔个几天再回来,照样能召回;最后是可控,你可以随时查看、编辑甚至删除某条记忆,隐私和合规方面也比黑盒式的上下文缓存要踏实。

我实际用它重构了客服机器人之后,明显感觉到对话的连续性上了一个台阶,至少用户不用每次重复报信息了。这才是我理想中的“长期记忆”形态。

2. 核心拆解:记忆的写入、存储和召回

2.1 记忆从哪里来:对话摘要与关键信息抽取

记忆不是把整段对话往数据库里一扔就完事了。那样既浪费存储空间,检索时也会带回大量噪音,反而干扰模型。claude-mem 的做法是先用 Claude 对每轮或每组对话做一次摘要抽取,要求模型输出一段固定结构的 JSON,把对话里的长期有价值信息提炼出来。

我在实战中是这样设计抽取 Prompt 的:

你是一个对话记忆提取器。请阅读以下对话记录,抽取需要被长期记住的信息。 只提取客观事实、用户偏好、明确承诺的任务,不要提取临时性寒暄。 输出格式必须是 JSON,字段如下: { "facts": ["用户ID是U12345", "用户住在上海"], "preferences": ["用户偏好简洁回答", "用户喜欢用表格"], "in_progress": ["用户正在申请退款,需要跟进"] } 如果是多轮对话,先去重合并相同主体,再输出最终结果。

关键在于强制模型输出 schema 而非自由文本。自由摘要在存储和检索时很难统一字段,而结构化 JSON 则可以直接拆成多条记录,每条记忆都有明确的类型和来源。

抽取频率上我也做过调整。如果每一轮对话都调用一次 Claude 做抽取,成本和延迟都扛不住。比较合理的做法是当对话达到一定轮数(比如 5 轮)或用户主动结束一个主题时,再对这一段历史做一次批量抽取。这样既能覆盖关键信息,又不会抽得太碎片化。

2.2 记忆存在哪里:SQLite + 向量索引的混合方案

记忆抽取完之后,下一步就是存储。这里我用的是“SQLite + 向量索引”的混合方案,而不是单纯上一个重型向量数据库。原因很简单:个人项目或者中小型应用,不需要为记忆上整套 Milvus 或者 Qdrant,SQLite 足够稳,维护成本也低。

我会建一张memories表,核心字段如下:

字段类型说明
idINTEGER PRIMARY KEY记忆 ID
user_idTEXT用户标识,用于多用户隔离
contentTEXT记忆内容文本
memory_typeTEXTfacts / preferences / in_progress
embeddingBLOB向量的二进制存储
source_session_idTEXT来源会话 ID
created_atDATETIME创建时间
last_accessed_atDATETIME最近召回时间
hit_countINTEGER召回命中次数

为什么还需要向量?因为用户表述往往不是字面精确匹配。比如记忆里存的是“用户养了一只英短”,用户下次问“我家猫最近不爱吃东西怎么办”,单纯用关键词检索“英短”是搜不到的,但用向量相似度就能关联上“猫”和“英短”的语义关系。所以我在每条记忆写入时都会生成一个 embedding,存到同一个表里。

对于向量计算,我建议先用现成的 Embedding API,比如 OpenAI 的text-embedding-3-small或者本地的bge-small-zh。如果不想再引入一个外部依赖,也可以先用 TF-IDF 或者 BM25 顶一版,但效果差距明显。初次实现我强烈建议直接上向量,后面你会少走很多弯路。

2.3 记忆怎么被想起来:相似度检索与主动注入

存储只是基础,真正决定体验的是召回策略。claude-mem 的召回不是单纯把最新记忆丢给模型,而是根据用户当前问题做相关性判断,只挑最靠谱的几条。

具体流程是:用户发来一条新消息,我先把它转换成一个 embedding 向量,然后在 SQLite 里把所有记忆的 embedding 取出来做余弦相似度计算,按得分倒序取 TopK。这个 K 值我一般取 5~10,太少可能漏掉重要记忆,太多则容易噪音淹没。

召回结果不能直接塞给模型,要格式化一下。我的做法是把记忆拼成一个只读的“记忆快照”,放到 system prompt 里:

以下是用户与系统此前交互中提炼出的长期记忆,供参考: - [fact] 用户ID是U12345,常住上海 - [preference] 用户偏好简洁回答,喜欢用表格 - [in_progress] 用户正在申请退款,需要跟进 请根据这些记忆回答用户问题。如果记忆与当前问题无关或明显过时,请忽略。

加最后那句“无关或过时请忽略”非常重要。模型有时候会过度解读记忆,把不相关的旧信息硬扯进答案里,这句话能有效抑制幻觉式联想。

另一个关键点是召回时机。我测试过两种方案:一种是只在会话开始时召回一次,之后整场对话都固定用那批记忆;另一种是每轮用户发消息前都动态召回。实测下来,会话开始时召回一次 + 每轮动态增补增量记忆比较合理。因为用户聊着聊着可能会开启新话题,固定记忆会漏掉新出现的相关性,而全部动态召回又会让模型读到的记忆不断变化,反而可能造成前后不一致。

3. 手把手搭一个最小可用的 claude-mem

3.1 环境准备与依赖

这部分我按最简单的方式来做,目的是让你先跑通整个链路,再根据实际场景优化。需要 Python 3.10 以上,然后安装几个包:

pip install anthropic openai numpy

anthropic用来调用 Claude,openai库在这里只是用来调用 Embedding API(也可以直接用 httpx 手动请求,不过用现成库省事),numpy用来做向量余弦相似度计算。如果不想用外部 Embedding API,也可以换成sentence-transformers跑本地模型,但首次运行要下载几百 MB 模型,比较慢。

我个人的建议:先把外部 Embedding API 流程跑通,确认记忆召回效果符合预期,再考虑换成本地模型做隐私优化。没必要一开始就陷入自托管的坑。

3.2 编写记忆管理器

记忆管理器是整个 claude-mem 的心脏。我会写一个MemoryManager类,包含写入和检索两个核心方法。

先看写入逻辑。会话中抽取出结构化记忆文本后,调用这个方法:

import sqlite3 import numpy as np import json class MemoryManager: def __init__(self, db_path="memories.db", embed_fn=None): self.conn = sqlite3.connect(db_path) self.embed_fn = embed_fn self._init_db() def _init_db(self): self.conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT, content TEXT, memory_type TEXT, embedding BLOB, source_session_id TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_accessed_at DATETIME DEFAULT CURRENT_TIMESTAMP, hit_count INTEGER DEFAULT 0 ) """) self.conn.commit() def add_memory(self, user_id, content, memory_type, source_session_id=""): embedding = self.embed_fn(content) # 返回 list[float] self.conn.execute( """INSERT INTO memories (user_id, content, memory_type, embedding, source_session_id) VALUES (?, ?, ?, ?, ?)""", (user_id, content, memory_type, np.array(embedding, dtype=np.float32).tobytes(), source_session_id), ) self.conn.commit()

写入时需要注意,embedding 需要通过二进制序列化后再存入 BLOB,否则 SQLite 没法直接处理 list。取出时可以反过来np.frombuffer(...)恢复成向量。

接下来是检索逻辑。我把当前用户消息转换成向量,然后与库里所有该用户的记忆向量做余弦相似度计算,返回 TopK:

def retrieve(self, user_id, query, top_k=5): query_vec = np.array(self.embed_fn(query), dtype=np.float32) rows = self.conn.execute( "SELECT id, content, memory_type, embedding, hit_count, last_accessed_at FROM memories WHERE user_id = ?", (user_id,) ).fetchall() scored = [] for row in rows: mem_id, content, mem_type, emb_blob, hit_count, last_at = row emb = np.frombuffer(emb_blob, dtype=np.float32) score = float(np.dot(query_vec, emb) / (np.linalg.norm(query_vec) * np.linalg.norm(emb) + 1e-8)) scored.append((score, mem_id, content, mem_type, hit_count, last_at)) scored.sort(key=lambda x: x[0], reverse=True) results = scored[:top_k] # 更新命中次数和最近访问时间 for _, mem_id, _, _, _, _ in results: self.conn.execute( "UPDATE memories SET hit_count = hit_count + 1, last_accessed_at = CURRENT_TIMESTAMP WHERE id = ?", (mem_id,) ) self.conn.commit() return [ {"content": content, "type": mem_type, "score": score} for score, _, content, mem_type, _, _ in results ]

这个实现非常朴素,也没有做分页或真·向量索引,但对千万级以下的数据量来说是够用的。如果你后续数据量增长,推荐把向量索引迁移到sqlite-vec或者专门的向量数据库。前期别过度设计。

3.3 集成到 Claude API 的流畅对话中

记忆管理器写完,接下来就是把记忆注入到 Claude 的调用链里。我会封装一个带记忆的聊天函数,核心思路是:先检索记忆,再拼 system prompt,最后调 Claude。

from anthropic import Anthropic client = Anthropic(api_key="your-api-key") mm = MemoryManager(db_path="memories.db", embed_fn=embed_query) def chat_with_memory(user_id, user_message, history=None): # 1. 召回相关记忆 memories = mm.retrieve(user_id, user_message, top_k=5) memory_text = "\n".join( f"[{item['type']}] {item['content']}" for item in memories ) system_prompt = f"""你是一个乐于助人的智能助手。 以下是此前交互中提炼出的用户长期记忆,供参考: {memory_text} 请根据记忆回答用户问题。如果记忆与当前问题无关或明显过时,请忽略。""" # 2. 组装消息 messages = [] if history: # history 是 [(role, content), ...] messages.extend( {"role": role, "content": content} for role, content in history ) messages.append({"role": "user", "content": user_message}) # 3. 调用 Claude resp = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, system=system_prompt, messages=messages, ) reply = resp.content[0].text # 4. 这里可以做一个简化版记忆抽取:只抽取本轮明显的用户偏好 # 更完整做法是积累多轮后批量抽取 if len(user_message) > 20: facts = extract_memory_simple(user_message) for fact in facts: mm.add_memory(user_id, fact, "fact", source_session_id=current_session_id) return reply

需要注意几点:

第一,history不要无限增长。我通常会只保留最近 10 轮对话,更早的消息既然已经抽成记忆了,就没必要继续占用上下文窗口。第二,记忆抽取我上面写的是一个简化占位函数,真实用例应该是像 2.1 节里那样用 Claude 批量抽取,否则一条用户消息可能会抽出很多无效信息。第三,system_prompt里的记忆文本要控制在合理长度内,我会在拼装前先做一次 token 估算,超过预算就裁剪到 TopK 更小。

这一套跑通之后,你会发现对话像是“长出”了记忆。用户第一次说“我喜欢简洁的回答”,第二次再问技术问题时,Claude 会自动给出更精炼的答复。这个体验提升非常直观。

4. 踩坑实录与调优心得

4.1 记忆污染比没记忆更可怕

集成完第一版后我遇到一个很恼火的问题:用户随口说了一句“今天天气不错”,系统就把它当成长期偏好记下来了。第二天用户提问时,模型开始一本正经地结合“天气不错”来回答,完全跑偏。

这就是典型的记忆污染。不是所有对话内容都值得长期记住,抽取时必须做过滤。我后来在抽取 Prompt 里明确加上“不抽取天气寒暄、临时情绪、一次性事件”,但效果仍然不够稳。最终解决办法是给每条记忆加一个“置信度”字段,置信度低的记忆进入待确认状态,如果之后多次在对话中被重复提及或验证,才提升为长期记忆。比如用户第一次说喜欢表格,你存一个低置信偏好;三次对话里他都要求用表格,再把它标记为高置信。

另一个策略是定期做记忆整理。每周把低命中率、长时间未访问的记忆交给 Claude 做一次合并和清理,把冗余内容合并成一条,把过时内容删除。这个动作类似人脑的睡眠巩固,能有效抑制记忆库膨胀和噪音累积。

4.2 上下文窗口和 token 预算控制

记忆注入不是免费的。假设每条记忆 50 token,TopK 取 10 就是 500 token,看着不多,但如果历史对话还有 20 轮,每轮 400 token,加起来就接近 8500 token,再叠加 system prompt 和工具定义,一个小窗口就被塞满了。

我的建议是给记忆注入设置硬性预算。比如总窗口是 200K,那么记忆最多占用 8K,历史最多占用 32K,剩下留给生成和动态内容。如果检索出的记忆超过预算,按相似度分数从高到低截取,而不是硬塞 TopK。简单公式如下:

记忆预算 tokens = min( int(0.04 * total_window_tokens), len(retrieved_memories) * avg_memory_tokens )

实测下来,记忆占比控制在 4%~5% 以内时,Claude 的表现最稳定。太少了记忆起不到作用,太多了模型容易把记忆当作事实来源,甚至出现“编造记忆”的情况。调试的时候我会打印出每次实际注入的记忆内容,肉眼检查相关性。

4.3 隐私数据与多用户隔离

如果你是单机自用,用户 ID 隔离可能没那么重要。但只要你的应用会服务多个用户,就一定要在存储层做硬隔离。我在第一版就犯过漏加 user_id 过滤的错:检索时只查了 TopK 向量相似度,结果有次把 A 用户的订单号带给了 B 用户,虽然只是测试环境,也吓出一身冷汗。

安全做法参考我上面的retrieve方法:SQL 查询强制WHERE user_id = ?,从源头保证只检索当前用户的数据。另外,记忆文本中如果包含地址、手机号等敏感字段,建议先做脱敏再存储。比如把“联系手机 138xxxx”存成“用户手机尾号 1234”,能降低泄露风险。

还要提供“遗忘”能力。GDPR 这类合规要求先不提,单从用户体验出发,用户应该有权利说“忘掉我的所有记忆”。我会在管理后台提供一个一键清空按钮,调用DELETE FROM memories WHERE user_id = ?,同时把 embedding 缓存也一并清掉。这个操作虽然简单,但能让你后续面对隐私合规审核时更有底气。

多用户场景还有一个容易忽视的坑:session 过期后,用户重启应用,旧记忆仍然会被召回。如果用户已经明确登出并切换账号,必须先切换 user_id,而不是沿用旧会话标识。我在实现时会把 user_id 和会话绑定,每次请求都从认证上下文里取,而不是从客户端传入的参数里取,防止被篡改。

4.4 记忆召回时机与多轮对话的协同

最后再说一个调参层面的心得:记忆召回并不是每轮都必须执行。对于简单问候、寒暄或与长期记忆无关的操作指令(比如“帮我打开设置”),召回反而可能拖慢响应。我后来加了一个前置判断:先判断当前消息里是否包含可检索的实体或意图,如果明显是通用交谈,就跳过召回,直接走正常的无记忆路径。

而且多轮对话中,历史消息本身已经包含了上下文,这时候重复注入同样的记忆会显得冗余。我的策略是:会话第一轮必须注入相关记忆;之后只有当新消息中出现了新的实体、关键词或话题转移信号时,才做增量召回。这么做能明显降低 API 调用次数和 token 消耗,同时保持记忆的连续性。

我实际跑客服机器人项目时就是这样做的:第一轮注入长期记忆,之后每一轮把上一轮的对话压缩成短摘要塞给模型,而不是无限堆原始消息。整套流程跑了一个月,记忆库 2000 多条,响应速度和准确率都还在可接受范围内。

最后分享一个小技巧:我在每条记忆上都留了source_session_id字段,本意只是为了追踪来源,后来排查问题时发现它太有用了。当模型给出的答案明显依赖某条记忆,且那条记忆其实是错的,我可以顺着 session 回到原始对话,快速定位是抽取错了还是存储错了。这比在黑盒里猜要高效得多。

做完这套改造,我的一个深切体会是:给 Claude 加记忆,最难的不是技术,而是怎么设计记忆的边界。记忆不是越多越好,而是越精准越好。少而精的记忆库配上合理的召回策略,反应到用户体验上是质的提升。如果你也在被 AI 的“金鱼记忆”折磨,我非常推荐尝试一下这个思路,从一个小型记忆管理器开始,逐步迭代到完整的 claude-mem 方案,你会发现长期记忆这件事,真的没那么神秘。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询