大模型记忆管理实战：Paperclip记忆夹设计与实现

1. 为什么我会捣鼓一个叫 paperclip 的记忆夹最近 paperclip 这个词在开发者圈子里又热了起来。大多数人对它的第一反应是办公桌上那种弯弯的铁丝回形针但做 AI 应用的人看到这个词脑子里蹦出来的多半是另一件事——怎么把大模型聊着聊着就丢掉的上下文像回形针一样重新夹住。我去年在做一个客服问答机器人项目时被同一个问题反复折磨模型转过头就忘了用户两分钟前说的关键信息。用户明明在第三轮对话里说过我家的猫三岁对花生过敏到第七轮问那它能吃这个零食吗模型已经完全不记得猫的存在一本正经地开始推荐含花生的产品。多轮对话根本聊不下去。后来我们内部搭了一个叫 paperclip 的记忆夹服务把散落在各轮对话里的信息抽出来、归档、按需调回整个系统的可用性才算立住了。这个项目不大核心代码不到两千行但它解决的是一个非常要命的问题大模型没有长期记忆而绝大多数真实业务场景都需要记忆。这篇文章就把这套东西从设计、实现到踩坑的完整过程拆开聊聊。适合谁看正在做大模型应用、聊天机器人、Agent 工作流的开发者或者被AI 聊着聊着就失忆问题折磨过的产品经理。你不用照着我的代码抄但里面的设计思路、参数取舍、坑点是通用的。先给个底线paperclip 解决的是记忆管理不是模型能力。它不会让模型变聪明只是让模型记得住。2. 设计核心记忆夹的三个腔室动手写代码之前我先把记忆这个词拆开了。很多人一上来就想搞个大而全的向量数据库结果做出来的东西又重又难用召回还全是噪声。我的做法相反先把记忆分成了三个层次每个层次各司其职。2.1 短期便签层会话窗口内的即时记忆这一层最快也最容易被忽略。它对应的是当前这次会话里用户刚刚说过什么。不用上数据库一个内存里的 dict 就够。键是会话 ID值是一个结构化的状态对象记录当前对话的上下文槽位比如user_name、pet_type、allergy_info、unfinished_order_id。# 短期便签层示意 short_term_memory: dict[str, dict] {} def update_short_memory(session_id: str, slot: str, value: str): short_term_memory.setdefault(session_id, {})[slot] value为什么不用 Redis早期我用过 Redis后来发现这个场景根本不需要跨进程共享单机内存足够而且速度快到可以忽略。如果以后要水平扩展再换 Redis 也就改一个类的事。短期层的关键是写入要快、读取要即时它服务于当前对话的连贯性不需要持久化。2.2 长期档案层跨会话的持久化记忆这一层才是 paperclip 的核心。用户隔三天再回来你得还记得他家的猫叫什么名字。长期层我做了一张memory_items表字段很简单字段类型说明idTEXT记忆条目唯一 IDsession_idTEXT来源会话user_idTEXT归属用户contentTEXT记忆内容已结构化memory_typeTEXT实体 / 偏好 / 事件 / 任务importanceREAL重要度 0-1expires_atDATETIME过期时间NULL 表示永久created_atDATETIME写入时间last_accessed_atDATETIME最近被召回的时间存储引擎用的是 SQLite。有人可能觉得寒酸但实际项目里 SQLite 配合 WAL 模式扛住一个小团队的业务量绰绰有余。等真到了每天几千万次读写的量级把存储层换成 PostgreSQL 或者专用的向量库接口不变只是换个实现。2.3 检索夹层怎么把对的记忆捞回来记忆存进去不是终点能捞回来才是。检索层是整个 paperclip 里我迭代次数最多的地方。最初的版本只有一个关键词匹配效果惨不忍睹。用户说上次那个过敏的事儿关键词匹配完全找不到花生这条记忆。后来改成两段式先粗筛再精排。粗筛用 BM25 关键词匹配 时间衰减过滤候选集压到 50 条以内。精排把候选集和当前对话做语义相似度打分embedding取 top 5。粗筛用关键词是因为快、便宜、可解释精排用语义是因为能兜住同义表达。过敏的事儿和花生过敏字面上没有任何共同词但语义上确实相关。3. 从零装配一次最小可用原型设计讲完直接上代码。我会带你过一遍 paperclip 最小可用版本的全过程大约三百行代码足以处理记住用户偏好 跨会话找回这个核心场景。3.1 环境与依赖其实只需要一张表项目依赖我压到了最少sqlite3Python 内置、openai或任意兼容 SDK做 embedding 和结构化抽取、fastapiuvicorn做服务暴露。不需要上 LangChain不需要上 ChromaDB那些都是后话。数据库初始化import sqlite3 DB_PATH paperclip.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS memory_items ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL, user_id TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT NOT NULL, importance REAL DEFAULT 0.5, expires_at DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_accessed_at DATETIME ) ) conn.execute(PRAGMA journal_modeWAL) conn.commit() conn.close()需要注意PRAGMA journal_modeWAL这行在并发读写场景下能避免很多 database is locked 的报错代价是多两个临时文件无所谓。3.2 记忆写入把对话压缩成可存放的条目写入流程的核心是抽取。你不能把每轮对话原文都存进去一是 token 成本太高二是检索时噪声太大。我让一个轻量模型做结构化抽取只提取值得记的东西。from openai import OpenAI import json, uuid client OpenAI() EXTRACT_PROMPT 你是记忆抽取器。从对话中提取值得长期记住的信息输出 JSON 数组。 每条信息包含 - content: 一句话描述该记忆 - memory_type: 只能是 entity / preference / event / task - importance: 0 到 1 的数字越重要越大 重点关注 - 用户的个人信息名字、宠物、家人 - 用户明确表达的偏好喜欢、不喜欢、过敏、禁忌 - 尚未完成的承诺或任务 - 对后续对话有价值的背景事实 忽略寒暄、临时情绪、与用户无关的信息。 对话内容 {conversation} def extract_memories(conversation: str): resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: EXTRACT_PROMPT.format(conversationconversation)}], temperature0, ) raw resp.choices[0].message.content # 防御模型可能输出 markdown 代码块 raw raw.strip() if raw.startswith(): raw raw.split(\n, 1)[-1].rsplit(, 1)[0] return json.loads(raw) def write_memory(session_id: str, user_id: str, conversation: str): memories extract_memories(conversation) conn sqlite3.connect(DB_PATH) for m in memories: mid str(uuid.uuid4()) conn.execute( INSERT INTO memory_items (id, session_id, user_id, content, memory_type, importance) VALUES (?,?,?,?,?,?), (mid, session_id, user_id, m[content], m[memory_type], m[importance]), ) conn.commit() conn.close()这里有三个容易被忽略的细节。第一temperature 必须设 0。抽取任务要确定性不要创造性否则同一段对话两次抽取的结果可能完全不同。第二防御模型输出 markdown 代码块。很多模型喜欢把 JSON 包在json ...里直接json.loads会炸。这个防御看起来笨但实测能省掉大量报错。第三抽取只发生在关键节点不是每轮对话都调用。我一般在用户结束一轮任务、或对话累积了 5 条以上消息时触发一次抽取。每轮都抽token 烧得太快。3.3 记忆召回两段式筛选的工程取舍召回的输入是当前对话的最后几句 当前会话上下文输出是最相关的 5 条记忆。import sqlite3 def recall_memories(user_id: str, query: str, top_k: int 5): conn sqlite3.connect(DB_PATH) # 阶段一关键词粗筛 时间过滤 keywords [w for w in query.split() if len(w) 1][:10] candidates [] for kw in keywords: rows conn.execute( SELECT * FROM memory_items WHERE user_id ? AND (expires_at IS NULL OR expires_at CURRENT_TIMESTAMP) AND content LIKE ? ORDER BY importance DESC LIMIT 50 , (user_id, f%{kw}%), ).fetchall() candidates.extend(rows) conn.close() if not candidates: return [] # 阶段二语义精排 texts [c[4] for c in candidates] # content 列 query_vec client.embeddings.create(modeltext-embedding-3-small, inputquery).data[0].embedding scores [] for i, t in enumerate(texts): t_vec client.embeddings.create(modeltext-embedding-3-small, inputt).data[0].embedding scores.append((cosine_sim(query_vec, t_vec), candidates[i])) scores.sort(keylambda x: x[0], reverseTrue) return [c for s, c in scores[:top_k]]cosine_sim就是个手写的向量夹角余弦代码略。这个结构看起来简单但有一个工程点要强调embedding 是每次召回现算的没做缓存。如果会话量大了给每一条候选记忆单独调 embedding 接口延迟和成本都会失控。更好的做法是写入时就提前算好content_embedding存到表里召回时只算 query 的向量然后逐条比余弦。我在生产版里就是这么干的SQLite 加一列embedding BLOB就好。3.4 接入对话主流程中间件式挂载paperclip 不是一个独立运行的服务它是嵌在对话流程里的一个中间件。我用 FastAPI 写了个薄薄的封装核心逻辑是每次对话前召回、每次对话后写入。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): session_id: str user_id: str message: str history: list[str] app.post(/chat) def chat(req: ChatRequest): # 1. 召回记忆拼进 system prompt memories recall_memories(req.user_id, req.message) memory_block \n.join(f- ({m[5]}) {m[4]} for m in memories) # 2. 构造带记忆的对话 system f以下是关于用户的历史记忆请参考但不盲从\n{memory_block} # 3. 调用模型 resp model_chat(system, req.history, req.message) # 4. 更新短期记忆必要时触发长期抽取 update_short_memory(req.session_id, last_query, req.message) if len(req.history) % 5 0: write_memory(req.session_id, req.user_id, \n.join(req.history[-10:])) return {reply: resp}最关键的一步是第 1 步换来的第 2 步把召回的记忆拼进 system prompt。这里有个技巧给每条记忆标注类型和置信度。比如- (preference) 用户对花生过敏模型就知道这是偏好信息不是当前指令参考即可不必严格服从。这个细节让记忆的利用率大幅提升因为模型能分清事实背景和行动指令。4. 实测中踩过的坑上下文漂移与召回噪声原型能跑通只是开始。真正让它变得可用的是后面几轮地狱般的调试。我把踩过的坑按症状列出来你大概率也会遇到。4.1 症状一模型忘记了十分钟前说过的话表现记忆已经成功写入数据库召回也返回了但模型给出的回答还是像没看见一样。排查下来问题出在 system prompt 的组织方式上。我最初把记忆块放在 system prompt 的开头后面跟了很长一段任务说明。模型在生成时注意力会集中在靠近末尾的指令上开头的历史记忆被稀释了。解法把记忆块从 system prompt 挪到 user message 末尾紧贴着当前问题。改造后的结构用户历史记忆仅供参考 - 用户对花生过敏 - 家里有一只三岁的猫 当前问题那它能吃这个零食吗实测改动前后记忆利用率提升了大概 30%。原因是模型对紧跟在自己问题后面的信息关注度天然更高。这个发现很反直觉但效果极其明显。4.2 症状二检索回来的记忆是对的废话表现召回结果语义相似度很高但内容毫无用处。用户问上次那个宠物零食买了吗召回回来的全是用户家里有猫用户喜欢买罐头之类的大路货真正关键的上次下单后因过敏退货这条反而没召回。问题出在重要度权重没有参与排序。纯语义相关性和信息增益是两回事。针对这个问题我把排序分数改成了加权公式final_score 0.7 * 语义相似度 0.3 * importance同时在召回阶段加了一条硬规则importance 0.8的记忆无条件进入精排候选集。这样那些虽然后续对话没直接提、但信息增益极高的关键记忆不会被粗筛滤掉。调完这个参数问答质量立刻上了一个台阶。这也让我意识到语义相关性不等于有用性用户真正需要的是此刻欠缺的关键信息不是泛泛相关的背景。4.3 症状三token 莫名翻倍的真相表现接入 paperclip 一周后账单出来吓一跳token 消耗比预期多了将近两倍。逐项排查后发现隐藏的两处开销每轮对话都触发召回每次召回对候选记忆现算 embedding每次调用都是 token 消耗。抽取逻辑只在len(req.history) % 5 0时才触发但实际生产里用户经常连续发几十条消息每满 5 条就抽一次没有会话级去重。解法做了三件事。第一embedding 结果全部缓存写入时就算好存库。第二抽取加了冷却时间同一会话 30 分钟内最多触发一次。第三召回频率降级为每轮对话的后处理在后台异步执行主流程只在关键节点比如用户疑似提出新需求时才同步召回。优化后 token 消耗回落到接近裸调模型时的水平但回答质量没下降反而因为记忆更精准了少了许多来回追问的对话轮次。4.4 症状四并发写库的脏数据表现压测时开了 20 个线程模拟并发用户SQLite 频繁报database is locked还有几条记忆的id主键冲突。SQLite 的并发写能力确实有限但我的场景远没到它的极限问题出在连接管理上。最初代码每次操作都新建连接写完就关这在并发场景下会产生大量锁竞争。解法改成单例连接 显式事务写操作走队列串行化。import queue, threading _write_queue queue.Queue() def async_write_memory(session_id, user_id, conversation): _write_queue.put((session_id, user_id, conversation)) def _write_worker(): while True: item _write_queue.get() if item is None: break session_id, user_id, conversation item try: write_memory(session_id, user_id, conversation) except Exception as e: log(fwrite memory failed: {e}) finally: _write_queue.task_done() threading.Thread(target_write_worker, daemonTrue).start()写入全部走单一 worker 线程读取可以继续走多线程连接。这样既保住了 SQLite 的简单性又不会出现锁竞争。核心原则是写串行、读并行。5. 把 paperclip 调到顺手参数与策略每个人的业务场景不一样这些参数没有绝对正确的取值但有相对合理的区间。我把自己在实战中调校的结果列出来你可以当初始参考值再根据你的业务做修正。5.1 分块粒度太小记不住太大记不全记忆条目如果太碎——用户养了一只猫猫三岁猫叫咪咪——召回来的信息东一块西一块模型拼不出完整画面。如果太粗——用户家里有一只三岁的名叫咪咪的猫此外还有一条狗狗对鸡肉过敏——又导致检索时噪音增大不够聚焦。我的经验是一条记忆控制在 20~50 字一个条目只承载一个完整事实。多条相关事实允许重复出现关键实体方便各自的检索命中。宁可用五条精确的小事实去描述一个复杂的背景也不要糊成一条大段子。5.2 时效衰减与记忆权重记忆是有保质期的。用户上个月说我最近在减肥和五年前说我最近在减肥后者大概率已经失效了。我在召回排序里加了一个时间衰减因子time_decay 1 / (1 0.01 * days_since_created)也就是说记忆创建 100 天后它的权重乘子降到 0.5200 天后降到 0.33。同时配合expires_at某些明显的临时性信息比如用户这周在出差我直接设置 7 天过期。身边有同事觉得时间衰减麻烦直接按 importance 排序。但实测衰减策略能显著减少模型根据过期偏好做决策的翻车率尤其是餐饮、健康、购物这类偏好变化快的领域。5.3 路由策略什么时候该翻档案什么时候不该翻这是最容易忽视的一块。不是每轮对话都要去翻长期记忆很多寒暄和短问答翻档案纯属浪费。我加了一个简单的路由判断查询里包含明确的实体词人名、物品名、时间词或者出现了上次、之前、还记得吗这类回溯词才触发深度召回。否则只走短期便签层。BACKREF_KEYWORDS {上次, 之前, 还记得, 后来, 那个, 当时} def should_deep_recall(message: str) - bool: return any(k in message for k in BACKREF_KEYWORDS)这个策略让召回调用量直接降了 60%系统响应平均快了 150 毫秒而且效果没受影响——因为该翻档案的用户自然会说出回溯词自然而然触发。5.4 top_k 的取舍top_k 我试过从 3 到 10 的各个档位。3 条太少有时候漏关键信息10 条太多模型会被大量记忆干扰反而把背景当成了用户的最新需求。最终稳在 5 条。如果你用的是上下文窗口较大的模型比如 128k可以放宽到 8 条窗口小的模型建议压到 3 条否则记忆会挤占太多的对话空间。给模型太多记忆和给得太少效果都会打折中间那个甜点区间需要你用自己的业务数据实测出来。6. 它还能怎么长从单聊到多智能体协作paperclip 的当前版本已经稳定跑了好几个月。但我知道它还能往前走一步这一步在于把一个人的记忆夹变成一群 Agent 的共享便签夹。我在做的扩展方向主要有三个。第一个是共享记忆总线。当系统里有多个 Agent 协作时每个 Agent 各干各的活但都需要了解用户的整体上下文。paperclip 的多用户共享模式本质是加一张memory_permissions表控制哪类记忆对哪个 Agent 可见。用户对客服说最近在备孕这个信息客服 Agent 应该知道但推荐广告的 Agent 就不该看到。记忆的边界控制比记忆本身更重要。第二个是主动遗忘机制。大多数记忆系统都在做加法但真实用户的记忆是有取舍的。我做了一个定时任务扫描expires_at已过期的条目直接物理删除对last_accessed_at超过 180 天且 importance 低于 0.3 的记忆标记为待清理。定期做减法既控制存储成本也避免过期记忆污染新决策。第三个是记忆的冲突消解。用户上个月说我爱吃辣这个月说最近胃不好戒辣了。两条记忆同时存在模型不知道该信哪条。我的计划是在写入时做一次冲突检测如果新记忆和旧记忆指向同一实体且语义相反就把旧记忆标记为 superseded已被取代排序权重直接清零。事实会被时间更新记忆系统也必须能接受用户变了。这几个方向的探索让我对记忆系统的理解深了一层记忆夹从来不是简单的存取而是建模一个人或一个组织的变化轨迹。回形针看着小夹住的却是时间。我自己用下来的最大感受是记忆系统做减法比做加法难得多。判断一条信息值得记只需要一个抽取模型但判断这条记忆何时过时、何时该让位给新事实却需要仔细设计冲突消解和遗忘策略。这也是 paperclip 后续最值得投入精力的地方。如果你的系统也在被上下文丢失折磨不妨先按这套思路搭一个最小版本跑上两周你会对记忆的复杂度有一个全新的认识。