这两年帮不少LLM应用做过“接脑子”的活，绕不开的核心词就是 ai-memory。你大概也遇到过一模一样的问题：上下文窗口明明越开越大，模型能“看到”的内容越来越多，但只要换一个Session，或者隔几天再回来聊，它就完全不记得你是谁、关心什么、上次做到哪一步了。本质上，大模型没有“记住”，只有“看到”。ai-memory要做的，就是给AI补上真正的跨会话长期记忆：把值得留存的对话沉淀下来、按需召回、适度遗忘。这篇文章我会把完整的落地过程摆出来——从记忆分级、数据建模、向量检索到遗忘策略，再到你大概率会踩的坑。适合正在给AI应用加记忆能力的开发者、纠结工具选型的架构师，以及想搞明白记忆系统和普通RAG有什么区别的朋友。
1. 先别急着写代码：AI记忆到底记什么、忘什么
1.1 大模型的“失忆”是结构性的
很多人以为模型记不住是因为Prompt写得不对，其实问题出在根上。Transformer的注意力机制是在一个有限的上下文窗口里做计算的，一旦输入超出窗口上限，最前面的内容会被硬截断；就算窗口足够大，离当前回答较远的信息也会被注意力分布稀释，表现就像是“看是看到了，但想不起来”。
更关键的是，绝大多数应用在做接口调用时是Session隔离的。上一轮对话的内容如果不手动传给下一次请求，模型就是无状态的。它擅长的是推理和生成，不是回忆。这个“失忆”不是缺陷，而是架构的默认行为。
所以做AI记忆系统，本质上不是在模型上打补丁，而是在应用层加一个“外部笔记本”：把重要信息在对话结束后写下来，下次对话开始前，再把相关片段翻出来放到Prompt里。模型不需要真的“记得”，它只需要在关键时候“看到”我们递上去的笔记。
1.2 四种记忆别混着存
我一开始把所有聊天记录无脑塞进向量库，结果就是召回出一堆“昨天午饭吃了什么”级别的噪声。后来把记忆分了层，效果立刻不一样。现在我的划分是这样：
- 工作记忆：当前会话最近几轮对话，直接放在Prompt里，不需要持久化，清掉就清掉。
- 情景记忆：跨会话的历史片段，比如“上周讨论过项目A的接口方案”，需要支持语义召回。
- 语义记忆：用户的长期偏好和固定事实，比如“用户是后端工程师，偏好Python，喜欢简洁方案”。变化慢、复用率高。
- 程序记忆：Agent会调用哪些工具、每个工具的参数约束。这部分多数场景放在系统配置里，不需要做成向量记忆。
这个分法的意义在于存储成本和召回策略不同。工作记忆追求低延迟，语义记忆要长期稳定，情景记忆需要时间衰减。如果全部混在一个集合里，召回时优先级根本无法区分。
1.3 先判断：你的应用真需要长期记忆吗
不是所有项目都该上记忆系统。我有个比较快的自测清单：
- 用户是否会说“上次我们聊到的那个事情”？
- 用户是否期待跨Session记住设置、偏好、进度？
- 应用是否需要在多轮交互中持续维护一个复杂目标？
三个答案如果都是“否”，就不要给自己加戏。如果都是“是”，再想清楚你要的是“助理型记忆”还是“知识库型记忆”。前者记用户状态和上下文，后者记文档和事实。ai-memory通常偏向前者，但核心实现都会落到同一套向量检索体系上，所以这篇文章的方法对两边都适用。
2. 记忆系统的四阶段链路：写入、存储、召回、遗忘
2.1 写入端：从对话里抽出“值得记”的东西
把原始聊天记录全部存进去是最省事但也最蠢的做法。对话里有大量寒暄、确认、重复信息，直接入库会带来三个问题：存储膨胀、召回噪声变大、隐私风险失控。
我的方案是用LLM做结构化抽取，而不是规则匹配。先给一个抽取Schema，让模型只输出有价值的信息：
{ "memory_text": "规范后的记忆描述", "memory_type": "preference | fact | progress | task", "importance": "1-5", "tags": ["领域标签"] }抽取Prompt大概长这样：
你是一个记忆抽取器。请从以下对话中提取值得长期记住的信息，只输出JSON数组。抽取原则：用户明确表露的偏好、确定的结论、个人事实、项目里程碑才需要记住；寒暄、临时指令、隐私字段一律不抽。隐私字段包括手机号、密码、验证码、银行卡号。
写入触发条件我一般设三条：用户明确说“以后都按这个来”；对话中产出了确定结论；出现了个人信息或项目进度。满足任意一条就进入抽取流程。实操中还有一个双保险：先用正则把手机号、密钥这类搞成脱敏标记，再交给LLM抽取，避免模型把敏感信息原样存进记忆库。
2.2 存储端：为什么向量数据库是最佳载体
回忆通常是模糊的。用户不会说“请精确查询ID为xx的记录”，而是说“我之前好像提过一个优化思路”。SQL那种精确匹配在这里完全用不上，必须有语义检索能力。
语义检索的基本流程是：先把文本用Embedding模型转成向量，让语义相近的文本在向量空间里靠在一起；查询时也转成向量，用余弦相似度或点积找到最近的几个邻居。
常用存储方案对比：
| 方案 | 优点 | 缺点 | 适合场景 |
|---|---|---|---|
| FAISS | 轻量、速度快 | 不擅长payload过滤，持久化要自己管 | 单机、小规模 |
| Chroma | 上手快、零成本 | 生产稳定性一般 | 原型验证 |
| Qdrant | 过滤能力强、Rust性能好 | 需要独立服务 | 生产主力，我目前选它 |
| Milvus | 分布式、海量 | 部署运维复杂 | 大规模平台 |
| pgvector | 复用已有PostgreSQL | 数据量大后性能弱 | 不想引入新组件 |
我的建议是先用Chroma把逻辑跑通，再迁移到Qdrant或者pgvector。不要一上来就上Milvus，运维成本会让你在还没看到效果时就先疯掉。
2.3 召回端：怎么把“过去”带回当前Prompt
召回链路我固定成四步：
- Query改写：用户问“通知功能怎么样了”，原始问题太口语化，先扩展成“通知功能 进度 最近更新”再检索。
- 向量化：用和写入时完全相同的Embedding模型，维度必须一致。
- 召回候选集：按相似度倒序取TopK，一般在20条左右。
- 重排与筛选：过滤掉低于阈值的，再按重要度和时间加权，最后只保留3到8条。
参数我给一个起步值：TopK取20，最终保留5条；相似度阈值0.6到0.75之间，具体要看你用的Embedding模型。这里有个特别重要的认知：召回不是越多越好。旧记忆太密会稀释当前任务，所以宁可少给，也别把Prompt塞成一锅粥。
2.4 遗忘端：记忆不能只进不出
没有遗忘机制的记忆系统迟早变成垃圾场。我采用的策略比较朴素但很有效：
- 时间衰减：每条记忆越久没被访问，参与召回的权重就越低。
- 访问强化：被成功召回一次，就把access_count加一，近期权重提升，模拟人的“回看”行为。
- 冲突淘汰：新旧记忆矛盾时，旧记忆标记为superseded，不再参与正常召回。
- 定期归档：超过90天没有被访问的记忆，从在线库搬到冷备库，不参与检索，只保留可回溯能力。
这四件事我用一个后台任务每6小时跑一次。刚开始我也担心“遗忘掉用户重要信息”会被投诉，实际上用户问一次就得触发一次召回，召回后访问时间被刷新，重要记忆不会被误杀，真正被杀的都是陈年废料。
3. 手写一个可落地的ai-memory模块
3.1 技术栈与选型逻辑
我用的是一套很朴素的组合，刻意绕开厚重的框架：
- Python 3.10+
- Embedding：中文场景用BGE-M3，量大之后可以蒸馏成小模型
- 向量库：Qdrant，Docker单节点起步
- 元数据：SQLite，只存结构化信息
- LLM抽取和改写：任意主流模型即可，gpt-4o-mini这个级别足够
为什么不直接用LangChain的Memory组件？封装太厚，出问题你根本不知道是向量索引的问题还是Prompt拼接的问题。自己写一遍全链路也就几百行代码，顺手还能把数据模型彻底吃透，后面接任何框架都是降维打击。
先装依赖：
pip install qdrant-client sentence-transformers openai pydantic启动Qdrant最省事的方式：
docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant3.2 记忆数据结构
在Qdrant里建一个名为memory_records的Collection。向量维度要和你选的Embedding模型严格对应，BGE-M3输出1024维，OpenAI的text-embedding-3-small是1536维。
每条记忆的Payload我固定这组字段：
@dataclass class MemoryRecord: user_id: str session_id: str content: str memory_type: str # preference / fact / progress / task importance: float # 1-5 tags: list[str] created_at: datetime last_access_time: datetime access_count: int status: str # active / superseded / archived索引方面，Qdrant里要把user_id、status、memory_type都建为keyword索引。别小看这个步骤，没有索引的时候全表扫描能把延迟从10ms拖到300ms。
3.3 写入流程实现
写入的核心是“抽取—脱敏—向量化—入库”，我封装成两个函数：
from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, VectorParams, Distance client = QdrantClient(host="localhost", port=6333) def embed_texts(texts: list[str], model) -> list[list[float]]: return model.encode(texts, normalize_embeddings=True) def save_memory(user_id: str, record: MemoryRecord, embedding: list[float]): client.upsert( collection_name="memory_records", points=[ PointStruct( id=hash(record.session_id + record.content[:50]), vector=embedding, payload={ **dataclasses.asdict(record), "user_id": user_id, "status": "active", }, ) ], )写入触发放在对话流之外，用一个异步队列去跑，不能让用户等记忆写完才看到回复。抽取步骤交给LLM，但我会限制输出条数，一次最多抽5条，避免模型把几句闲聊拆成20条“记忆”。
这里有个心得：写入前先做一次相似度检测，如果当前要写的内容和已存在的记忆相似度超过0.9，就跳过不写，相当于去重。否则同一个设置被重复写十几次，后续召回全部是冗余。
3.4 召回流程实现
召回是整个模块的核心，我一步步说：
def recall_memories(user_id: str, query: str, top_k: int = 20) -> list[MemoryRecord]: # 1. 改写query，扩写关键实体 rewritten_query = rewrite_query(query) # 2. 向量化 query_vector = embed_texts([rewritten_query], embed_model)[0] # 3. 向量检索，强制带上user_id和status过滤 hits = client.search( collection_name="memory_records", query_vector=query_vector, query_filter=Filter( must=[ FieldCondition(key="user_id", match=MatchValue(value=user_id)), FieldCondition(key="status", match=MatchValue(value="active")), ], ), limit=top_k, with_payload=True, ) # 4. 重排：相似度阈值过滤 + 时间权重调整 ranked = [] for hit in hits: if hit.score < 0.65: continue age_days = (datetime.now() - hit.payload["last_access_time"]).days time_boost = 1.0 / (1.0 + 0.05 * age_days) final_score = hit.score * 0.7 + hit.payload["importance"] * 0.1 + time_boost * 0.2 ranked.append((final_score, hit)) ranked.sort(reverse=True) return [hit.payload for _, hit in ranked[:5]]这里有个非常容易踩的坑：user_id过滤必须同时出现在向量检索的Filter里，而不是在拿到结果之后再过滤。否则用户A的查询会把用户B的记忆捞出来，哪怕后面你手动过滤掉了，也已经把不该看的向量内容暴露在了检索链路里。
召回结果拼进System Prompt时，我习惯带上一个免责声明：
以下是关于用户的历史记忆，仅作参考。如果记忆与当前对话矛盾，以当前对话为准。
不然模型容易把一个过期状态当成事实来回答。
3.5 遗忘与更新落地
冲突处理的逻辑是这样：用户把通知时间从早上9点改成11点。写入新记忆前，先用“通知时间 设置”做一次检索，找到旧记忆后将其status标记为superseded，再写入新记忆。
def mark_superseded(user_id: str, old_memory_id: str): client.set_payload( collection_name="memory_records", payload={"status": "superseded"}, points=[old_memory_id], )后台定时的权重刷新，我直接写一个Python脚本用cron跑，每隔6小时扫描一次active记忆：
- 超过90天未访问：改为archived
- 超过30天未访问：importance减0.5
- 访问次数大于10：importance加0.2，模拟“用户反复回看，说明重要”
这套策略跑了两周后，在线库的记忆条数不再快速增长，召回结果的相关性明显变好。
4. 工具选型：哪些轮子值得自己造，哪些坑别踩
4.1 自研 vs 开源Memory框架
现在市面上有不少现成的AI Memory方案，比如Mem0、Zep、LangChain里的一堆Memory组件。我的态度是：先用自研跑通，再决定要不要换框架。
| 方案 | 核心能力 | 适合场景 | 主要顾虑 |
|---|---|---|---|
| Mem0 | 自动抽取和更新记忆，API简洁 | 快速集成到已有应用 | 定制性差，内部逻辑黑盒 |
| Zep | 基于时间图谱，能表达复杂关系 | 需要记忆关联推理的项目 | 部署重，运维复杂度高 |
| LangChain Memory | 封装简单，多种存储后端 | Demo和个人项目 | 状态管理弱，缺乏生产级遗忘机制 |
| 自研 | 可控性最高，链路清晰 | 核心用户长期使用、数据隔离要求高 | 需要自己写抽取和遗忘逻辑 |
如果你做的是内部工具，直接上Mem0没毛病。如果是面向大量用户的SaaS产品，我强烈建议自研或者至少把框架源码读透再做二次开发。记忆系统直接决定对话质量，出了问题黑盒状态下你连排查的入口都找不到。
4.2 Embedding模型是召回效果的第一变量
召回效果70%由Embedding模型决定，20%由检索参数决定，只有10%在重排。我把几个主流方案摆出来：
| 模型 | 向量维度 | 中文效果 | 成本 | 场景适配 |
|---|---|---|---|---|
| BGE-M3 | 1024 | 优秀，长句理解强 | 免费本地部署 | 中文业务，生产常用 |
| text-embedding-3-small | 1536 | 良好 | 按量付费 | 不想维护模型服务 |
| text-embedding-3-large | 3072 | 更强，但贵 | 高 | 预算充足 |
| bge-large-zh | 1024 | 中文专精 | 免费 | 纯中文场景 |
换Embedding模型之后有个典型坑：之前标定好的相似度阈值全部失效。原因不难理解，不同模型产出的向量空间分布完全不同，0.7的阈值在text-embedding-3-small上能正常召回，换到BGE-M3可能就变成什么都召不出来。所以每次换模型，都要重新拿一批真实对话测一遍阈值。
5. 常见问题与排查实录
5.1 召回结果驴头不对马嘴
我遇到最多的症状是：明明库里有一条很相关的记忆，搜索结果却完全没影子。排查顺序按下面这张表来：
| 症状 | 可能原因 | 排查方向 |
|---|---|---|
| 完全搜不到任何记忆 | 向量维度不一致、集合名错 | 打印embedding shape，确认集合存在 |
| 能搜到但相关度极低 | Embedding模型换了、阈值过高 | 重新标定相似度阈值 |
| 其他用户的记忆被召回 | 检索Filter忘了加user_id | 检查Filter，写跨用户隔离测试 |
| 旧记忆频繁被召回 | superseded状态没过滤 | 确认检索条件里带了status=active |
排查时最有效的手段是“手动单条验证”：单独取出一条已知记忆，和query算一次相似度，看分值在什么量级。如果单条相似度就低得可怜，那问题一定在Embedding或query改写上，而不是向量库。
5.2 Prompt里的旧记忆越堆越多
有人一开始设计得“豪爽”，每轮把召回的全部历史记忆都塞进Prompt，结果上下文窗口很快被打爆，而且模型开始分不清哪条记忆重要。
我的做法是给召回结果设硬上限：最多5条，每条记忆在Prompt里最多用两句话表达。如果一条记忆超过100个字，就得在写入时就压缩成摘要。这样即使命中10条，注入总量也能控制在600字以内。
5.3 新旧记忆互相打架
用户先说“通知改到早上9点”，隔两天又说“还是11点吧”。如果不处理，模型会在一次对话里同时看到两条矛盾记忆，最后给出一个四不像的回答。
解决思路就是前面说的superseded机制：写入新“设置类”记忆前，先做一次同类检索，发现可能是冲突记忆就交给LLM做一句话判断，确认矛盾后把旧记忆标记失效。这个方法不复杂，但能让用户觉得AI“真的很懂”，因为它知道自己改过主意。
5.4 跨用户隐私边界
这是我在早期版本踩过最重的一个坑。当时图省事，检索时只按相似度召回，没加user_id过滤，结果一个用户问“我的服务器配置”，回答里混进了另一个用户的服务器信息。这类问题一旦出现就是事故级。
防御措施有两条：所有写入和查询的代码路径必须携带user_id；测试用例里固定写一条“跨用户隔离”用例，每次发版前跑一遍，确保用户A永远搜不到用户B的记忆。另外，密码、验证码、银行卡这类信息从产品安全角度就不该进入记忆库，哪怕脱敏了也别存。能用规则过滤掉的先用规则挡掉，别把希望全寄托在LLM判断上。
5.5 数据量涨上来之后的性能问题
在十万条记忆以内，向量库默认配置不会出问题。但到了几十万甚至上百万条，就需要调索引参数了。我用的Qdrant配HNSW索引，起步参数可以这样设：
VectorParams( size=1024, distance=Distance.COSINE, hnsw_config=HnswConfig( m=16, ef_construction=200, ef_search=100, ), )我实测的感觉是，在普通开发机上，20万条向量单次检索压到15到30毫秒完全没问题。如果某次召回变慢，先看生效的索引是不是用对了，再看有没有在Filter里用了未建索引的字段。
最后说一点体感最深的经验：做ai-memory最难的不是代码，而是“什么该记住、什么该忘掉”的判断。我前前后后调了一个多月，发现最有效的优化不是换更好的Embedding，而是把user_id过滤和冲突淘汰做对。如果你也在做类似的东西，建议先把端到端场景跑通：用户在第一天的会话里说了一个偏好，第二天的会话里用完全不同的措辞问出来，AI能准确对上，那你的记忆系统就算立住了。
