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表，字段很简单：
| 字段 | 类型 | 说明 |
|---|---|---|
| id | TEXT | 记忆条目唯一 ID |
| session_id | TEXT | 来源会话 |
| user_id | TEXT | 归属用户 |
| content | TEXT | 记忆内容（已结构化） |
| memory_type | TEXT | 实体 / 偏好 / 事件 / 任务 |
| importance | REAL | 重要度 0-1 |
| expires_at | DATETIME | 过期时间，NULL 表示永久 |
| created_at | DATETIME | 写入时间 |
| last_accessed_at | DATETIME | 最近被召回的时间 |
存储引擎用的是 SQLite。有人可能觉得寒酸，但实际项目里 SQLite 配合 WAL 模式，扛住一个小团队的业务量绰绰有余。等真到了每天几千万次读写的量级，把存储层换成 PostgreSQL 或者专用的向量库，接口不变，只是换个实现。
2.3 检索夹层：怎么把对的记忆捞回来
记忆存进去不是终点，能捞回来才是。检索层是整个 paperclip 里我迭代次数最多的地方。
最初的版本只有一个关键词匹配，效果惨不忍睹。用户说"上次那个过敏的事儿"，关键词匹配完全找不到"花生"这条记忆。后来改成两段式：先粗筛，再精排。
- 粗筛：用 BM25 关键词匹配 + 时间衰减过滤，候选集压到 50 条以内。
- 精排：把候选集和当前对话做语义相似度打分（embedding），取 top 5。
粗筛用关键词，是因为快、便宜、可解释；精排用语义，是因为能兜住同义表达。"过敏的事儿"和"花生过敏"字面上没有任何共同词，但语义上确实相关。
3. 从零装配：一次最小可用原型
设计讲完，直接上代码。我会带你过一遍 paperclip 最小可用版本的全过程，大约三百行代码，足以处理"记住用户偏好 + 跨会话找回"这个核心场景。
3.1 环境与依赖：其实只需要一张表
项目依赖我压到了最少：sqlite3（Python 内置）、openai或任意兼容 SDK（做 embedding 和结构化抽取）、fastapi+uvicorn（做服务暴露）。不需要上 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_mode=WAL") conn.commit() conn.close()需要注意PRAGMA journal_mode=WAL，这行在并发读写场景下能避免很多 "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( model="gpt-4o-mini", messages=[{"role": "user", "content": EXTRACT_PROMPT.format(conversation=conversation)}], temperature=0, ) 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(model="text-embedding-3-small", input=query).data[0].embedding scores = [] for i, t in enumerate(texts): t_vec = client.embeddings.create(model="text-embedding-3-small", input=t).data[0].embedding scores.append((cosine_sim(query_vec, t_vec), candidates[i])) scores.sort(key=lambda x: x[0], reverse=True) 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(f"write memory failed: {e}") finally: _write_queue.task_done() threading.Thread(target=_write_worker, daemon=True).start()写入全部走单一 worker 线程，读取可以继续走多线程连接。这样既保住了 SQLite 的简单性，又不会出现锁竞争。核心原则是：写串行、读并行。
5. 把 paperclip 调到顺手：参数与策略
每个人的业务场景不一样，这些参数没有绝对正确的取值，但有相对合理的区间。我把自己在实战中调校的结果列出来，你可以当初始参考值，再根据你的业务做修正。
5.1 分块粒度：太小记不住，太大记不全
记忆条目如果太碎——"用户养了一只猫""猫三岁""猫叫咪咪"——召回来的信息东一块西一块，模型拼不出完整画面。如果太粗——"用户家里有一只三岁的名叫咪咪的猫，此外还有一条狗，狗对鸡肉过敏"——又导致检索时噪音增大，不够聚焦。
我的经验是一条记忆控制在 20~50 字，一个条目只承载一个完整事实。多条相关事实允许重复出现关键实体，方便各自的检索命中。宁可用五条精确的小事实去描述一个复杂的背景，也不要糊成一条大段子。
5.2 时效衰减与记忆权重
记忆是有保质期的。用户上个月说"我最近在减肥"和五年前说"我最近在减肥"，后者大概率已经失效了。我在召回排序里加了一个时间衰减因子：
time_decay = 1 / (1 + 0.01 * days_since_created)也就是说，记忆创建 100 天后，它的权重乘子降到 0.5；200 天后降到 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 后续最值得投入精力的地方。如果你的系统也在被"上下文丢失"折磨，不妨先按这套思路搭一个最小版本，跑上两周，你会对记忆的复杂度有一个全新的认识。
