先说实话给AI加上记忆之前我心里是有点抵触的。模型本身能在上下文里记住很多信息可只要会话一关它忘得比金鱼还快。做聊天机器人和Agent应用的人迟早都会撞上这堵墙你说过的需求它秒懂第二天再问一脸茫然。这个问题的本质就是AI的记忆系统缺失。所谓的ai-memory在我这里指的不是模型参数里固化的知识而是给大模型外挂的一套长期记忆系统从对话里提取有价值的信息用向量库和结构化存储把它保存下来下次对话时再把相关记忆检索出来、注入上下文让AI真正做到“越聊越懂你”。这篇文章我就把自己搭建这套系统的完整思路和代码原原本本拆给你包括记忆怎么采集、怎么存储、怎么召回、怎么防污染以及我在实操里踩过的坑。适合谁看所有在做ChatBot、Agent、私人助理类应用的开发者或者想给现有AI应用加记忆能力的产品、算法同学。下面这套方案不绑定任何特定平台我用Python实现你可以轻松换到自己熟悉的语言上。1. 先搞清楚AI记忆到底在解决什么痛点1.1 大模型自带的“工作记忆”有多脆弱很多人觉得大模型上下文窗口越来越大动辄128K、200K token那不就不需要什么记忆系统了吗这个想法我一开始也有。等真上线跑过一段时间就明白了上下文窗口是“工作记忆”不是“长期记忆”。工作记忆的特点就是容量有限、随聊随丢。一个128K窗口的模型你把十轮对话历史、参考资料、系统提示都怼进去很快就占满了。更麻烦的是研究里提到过一个现象模型对长上下文中处在中间位置的信息关注度明显下降也就是所谓的“lost in the middle”。窗口是大了但大量信息淹没在长文里模型根本抓不住重点。我实测过一个200K窗口的模型塞进去几十页资料后让它回忆最早对话里提过的一句关键需求它经常胡编。这说明什么单纯拉大窗口解决不了记忆问题只会让成本和噪声一起涨。你要的是“记得住”不是“装得下”。1.2 三种记忆的划分没有长期记忆的AI约等于金鱼在设计记忆系统之前我先把“记忆”这个概念拆成三层这也成了后来整个架构的基石工作记忆Working Memory当前对话上下文窗口内的全部信息相当于人思考时脑内短暂保留的信息。容量小、易丢失。短期记忆Short-term Memory单次会话内、尚未提炼的历史信息比如一整段多轮对话的原始文本。可以被摘要化但跨会话后基本失效。长期记忆Long-term Memory跨会话持久保存的信息经过提取、结构化、权重标注比如用户的偏好、习惯、目标、项目状态。没有记忆系统的时候AI只有工作记忆撑死加上短期记忆。会话一关全没了。这就好比金鱼——每次见面都是陌生人。有了长期记忆AI 才能像真人助理一样“我记得你上次说过……”。后面我详细讲这三层怎么联动。1.3 记忆系统带来的实际价值可能有人问记忆系统到底能带来什么我总结成三点。第一是个性化体验。同样是旅游推荐AI知道你是素食主义者、预算有限、偏好避开人群推荐结果完全不一样。没有记忆AI每次都在问“你吃什么”“你预算多少”用户早烦了。第二是任务连续性。做Agent、做自动化工作流时任务往往分散在多轮对话里。比如你在做一个市场调研Agent昨天确定了调研框架今天希望它接着执行。没有记忆它得让你重新交代一遍甚至需要重新粘贴所有资料。第三是数据资产沉淀。用户每一次对话里都隐含大量有价值的信息记忆系统能自动提炼成可复用的结构化数据。这些数据反过来可以做用户画像、做推荐优化、做产品洞察。也就是说记忆不只是功能还是数据资产。2. 记忆系统的架构设计从采集到召回2.1 记忆采集从对话里提炼什么构建记忆系统的第一个问题记忆是什么是原始的对话记录吗不是。对话记录是原材料不是记忆。十几轮闲聊里有用信息可能就两三句话。如果原样存下来检索的时候噪声极大还白白烧token。所以采集环节的关键动作是信息提炼。我用LLM做“记忆提取器”每轮或每几轮对话后让模型按固定格式输出需要长期记住的信息。提取维度通常分四类记忆类型典型内容重要度初始值事实类用户职业、年龄、城市、家庭成员3~4偏好类口味、风格、阅读偏好、沟通方式4~5项目状态当前项目进度、下一步动作4临时信息当下具体需求与长期无关1~2提取提示词我反复迭代过多次最终版大概是你是记忆提取器。从下面这段对话中提取值得长期记住的用户信息。 只输出JSON数组不要输出其他内容。每个元素包含 - content一句完整的话描述一个事实或偏好 - categoryfact / preference / project / none - importance1~5的整数5代表非常重要 对话 {conversation_text}这里有个小心思我让模型把不重要的信息也输出出来categorynone但后续写入时不会放入长期记忆而是丢弃或归档。好处是模型“认真看过”所有对话不会因为“只提取重要的”而误丢关键信息。采集时机也很讲究。我建议异步触发用户说完话、模型回复完整个回合结束后在后台调用提取器。不要每来一条消息就同步提取那样延迟高、成本高还会打断主流程。一般可以累积3~5轮或每隔1分钟批量提取一次。实测下来这个频次提取效果最稳上下文聚在一起信息更完整。2.2 记忆存储向量库和关系表各管哪一段采集到记忆后存哪里这里有一个常见的误区很多教程让你把记忆一股脑塞进向量数据库。但实际做下来向量库适合做语义检索不适合做精确查询和管理。我最终用的是“关系表 向量库”的双存储方案SQLite/PostgreSQL 作为主存储每条记忆有 id、content、category、importance、created_at、status 等字段负责精确查询、更新、统计和生命周期管理。向量库存 embedding 向量负责语义召回用 doc_id 关联主存储里的记录。为什么要这么拆向量库的元数据过滤能力虽然能用但复杂条件查询、更新、统计非常别扭。而记忆系统日常操作里有大量“把旧记忆更新掉”“查一下用户有没有说过某件事”这类精确需求。让关系表当主存储维护起来省心得多。向量库这边我用的是ChromaDB。不是因为它最好——论规模它不如Milvus论性能它不如FAISS但它的核心优势是开发体验极好pip install chromadb一行代码创建持久化客户端自带embedding接口非常适合项目原型和中小规模应用。等到数据量大了可以更换底层向量库关系表那套完全不用动。嵌入模型的选择也说一下嵌入模型特点适用场景OpenAI text-embedding-3-small质量稳定、API简单快速上线可接受外部APIBGE-small-zh中文效果好、可本地部署中文应用、隐私敏感M3E中文语义向量友好中文应用、轻量场景Ollama embedding本地运行离线/低算力环境这个选择直接影响中文检索的效果。我有一次用通用英文嵌入模型跑中文记忆检索出来的记忆驴唇不对马嘴换了BGE系列之后效果立刻好了。中文场景就别省这一步了。2.3 记忆召回怎么把对的记忆捞回来存储不是目的用起来才是。召回环节我按“输入查询 → 向量检索 → 重排序 → 上下文注入”四步来做。向量检索这一步特别容易出现“看起来召回了几条实际没用”的情况。我最早直接查Top-5发现精确度很低用户问“附近有什么素食餐厅”时向量库可能把“用户不喜欢辣”这样相关的记忆召回也可能把“用户之前去过杭州”这样无关的拉出来。向量相关的判断是语义距离不是语义“正确性”。为此我加了两个处理手段。第一是混合检索对记忆内容同时做向量相似度和关键词匹配BM25或SQL的LIKE都行两者分数按比例融合。关键词匹配能确保“素食”“不吃辣”这类颗粒度信息不漏掉向量检索能兜住语义变化。第二是重排序召回Top-20后用LLM根据当前查询给每条记忆打分取分数最高的少数几条进入上下文。这一步成本很低但对回答质量的提升非常明显。召回后的上下文注入位置很讲究。我通常把记忆放在系统提示词的末尾用明确的分隔段落标出来比如【历史记忆】 - 用户偏好清淡饮食不吃辣 - 用户正在策划毕业旅行预算5000元以内 - 用户对民宿感兴趣不喜欢连锁酒店放在这个位置的用意是让它作为“可供参考的额外信息”而不是改写了模型的主人格。如果放在对话历史开头容易被模型当成“用户的当下发言”出现角色混淆。我见过不少项目就是被这个细节坑了。3. 实操用Python给大模型装上记忆模块3.1 选型嵌入模型与向量库的组合先交代一下我最终采用的技术组合方便你直接抄作业Python 3.10ChromaDB做向量库SQLite做关系表OpenAI的API同时承担对话和记忆提取两个角色。如果你是本地部署党可以把对话模型换成Ollama里的qwen或者llama嵌入模型换成BGE整体思路一模一样。为什么坚持用ChromaDB而不是别的上面提到过开发体验好。还有一个重要原因它支持PersistentClient数据直接落盘不需要额外起服务进程。对于个人项目和中小型应用少一个依赖就少一个故障点。将来数据量上来了平滑切到Qdrant或pgvectorDAO层接口换掉就行业务代码不用动。嵌入模型这里我提醒一句不要选太大太强的模型来做记忆嵌入。记忆的句子普遍很短一两句话用大模型嵌入和小模型嵌入的效果差距没想象中那么大但成本和延迟是实打实的。我用的是OpenAI text-embedding-3-small需要的话剪到1024维召回效果已经足够。3.2 核心代码MemoryManager的完整实现下面这段代码就是我的记忆管理类核心我尽量保留主干逻辑方便你直接套用import json import uuid import sqlite3 from datetime import datetime import chromadb from chromadb.utils import embedding_functions class MemoryManager: def __init__(self, db_path./memory.db, persist_dir./vector_store): 初始化连接SQLite做主存储ChromaDB做向量索引 self.db_path db_path self._init_sqlite() self.client chromadb.PersistentClient(pathpersist_dir) self.collection self.client.get_or_create_collection( nameuser_memory, embedding_functionembedding_functions.OpenAIEmbeddingFunction( api_key你的API_KEY, model_nametext-embedding-3-small ) ) def _init_sqlite(self): self.conn sqlite3.connect(self.db_path, check_same_threadFalse) self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT, category TEXT, importance INTEGER, status TEXT DEFAULT active, created_at TEXT, updated_at TEXT ) ) self.conn.commit() def add_memory(self, content: str, category: str fact, importance: int 3, memory_id: str None): 写入一条记忆先存SQLite再写向量库 mid memory_id or str(uuid.uuid4()) now datetime.now().isoformat(timespecseconds) self.conn.execute( INSERT INTO memories VALUES (?,?,?,?,?,?,?), (mid, content, category, importance, active, now, now) ) self.conn.commit() self.collection.upsert( ids[mid], documents[content], metadatas[{ category: category, importance: importance, created_at: now }] ) return mid def search_memory(self, query: str, top_k: int 5, min_importance: int 0): 语义检索记忆支持重要度过滤 results self.collection.query( query_texts[query], n_resultstop_k, where{importance: {$gte: min_importance}} ) memories [] for idx, mid in enumerate(results[ids][0]): row self.conn.execute( SELECT * FROM memories WHERE id?, (mid,) ).fetchone() if row and row[5] active: memories.append({ content: row[1], category: row[2], importance: row[3], score: 1 - results[distances][0][idx] }) return memories def delete_memory(self, memory_id: str): 软删除SQLite置状态向量库也删掉 self.conn.execute( UPDATE memories SET statusdeleted, updated_at? WHERE id?, (datetime.now().isoformat(timespecseconds), memory_id) ) self.conn.commit() try: self.collection.delete(ids[memory_id]) except Exception: pass def list_all(self, limit: int 100): cur self.conn.execute( SELECT * FROM memories WHERE statusactive ORDER BY importance DESC, updated_at DESC LIMIT ?, (limit,) ) return [dict(zip([id, content, category, importance, status, created_at, updated_at], row)) for row in cur.fetchall()]这段代码里有几个细节值得多说一句。我用upsert而不是add来写向量库目的是支持更新记忆内容时保留同一个id。比如用户之前说过“我喜欢喝茶”过段时间改口“我开始喝咖啡了”你不能把两条冲突的记忆都留着而是应该用新内容覆盖旧内容。另外我做了软删除而不是把SQLite记录物理删除。原因是记忆可能被其他模块的日志或关联数据引用物理删了会出问题。业务上标记成deleted就够向量库里的向量同步删掉查询时自然找不到了。3.3 接入LLM让记忆真正参与对话光有MemoryManager还不够得让它和对话接口联动起来。我封装了一个conversation_with_memory函数def conversation_with_memory(user_input: str, history: list) - str: # 1. 召回相关记忆 memories memory_manager.search_memory(user_input, top_k5) # 2. 把记忆拼进system prompt memory_block \n.join([f- {m[content]} for m in memories]) system_prompt BASE_SYSTEM_PROMPT \n\n【历史记忆】\n memory_block # 3. 调用对话模型 response chat_with_gpt(system_prompt, history [{role: user, content: user_input}]) # 4. 异步提取新记忆 extract_memories_from_dialogue(history [{role: user, content: user_input}, {role: assistant, content: response}]) return response时序上要注意先召回再拼上下文最后调用模型。如果顺序反了等模型都回复了才去查记忆那这一次对话完全用不上记忆体验就很割裂。第4步提取是异步的不要在用户等待响应的时候同步跑延迟会明显拉长。你可能注意到我用了BASE_SYSTEM_PROMPT常量。对于有记忆注入的场景系统提示词建议写成固定的角色设定把变化的部分专门放在“历史记忆”区块。这样模型行为一致性更好也不容易被记忆内容带偏。4. 检索策略与记忆生命周期管理4.1 检索打分语义相似度之外还有时间与重要性前面我用ChromaDB的query直接返回了距离分数但在实际项目里这个分数只能算是初筛。真正决定“该把哪条记忆放给模型看”的是一个综合打分函数。我自己用的打分公式final_score similarity_score * 0.6 (importance / 5) * 0.25 recency_score * 0.15其中similarity_score是归一化后的余弦相似度ChromaDB的distance需要转成similarity一般用1 - distanceimportance / 5是把1~5的重要度归一化到0~1recency_score采用时间衰减函数recency_score exp(-lambda * days_since_update)lambda一般取0.03~0.05。取0.05的话20天前的记忆衰减到37%左右100天前基本趋近于零。这个衰减速度适配大多数对话场景如果业务是低频长期项目可以把lambda调小一点比如0.02。这个公式的意义在于单纯语义相关不等于应该被使用。用户半年前随口说的“我对XX一般般”和一周前明确表达的偏好即使语义相似度接近后者显然更能影响当前回答。重要度和时效性必须和语义相关度一起参与排序。4.2 记忆写入的防污染机制记忆系统最大的风险不是“记不住”而是记错、记住垃圾、记住冲突。这类问题我在线上踩过不少次总结出几项防污染措施第一设置记忆提取的“闸门”。不是每一条模型输出都要进记忆库。我给提取器加了一条硬规则只有用户明确表达的信息才写入长期记忆模型自己的推测和“正确的废话”不进库。这个靠提示词约束我试下来效果不错。第二冲突检测与覆盖。新记忆写入前先做一次相似度查询。如果发现已有记忆和新记忆语义相关但内容相反比如“用户不吃辣” vs “用户喜欢吃辣火锅”要判断哪个更新、哪个可信度更高。我的做法是新记忆的updated_at和importance若都更高就覆盖旧记忆否则保留旧记忆把新记忆降级为“临时信息”。第三敏感信息隔离。不是所有记忆都适合进长期记忆库。手机号、身份证、住址这类隐私信息强烈建议单独存到加密表并且不进向量库——因为向量库无法做到字段级的“忘记”。隐私合规这块后面专门展开说。4.3 遗忘与整理记忆库不是储物间记忆库如果只加不减用三个月就会变成垃圾场。每条记忆不管多重要都会被新记忆淹没检索精准度直线下降。所以我系统里有一套定期整理机制每隔几天跑一次离线任务合并相似记忆按向量相似度聚类把“用户是北京人”“用户住在北京海淀区”这类等价信息合并成一条保留信息最全的那条。降权过期记忆把超过90天且importance低于3的记忆自动标记为“archive”不再进入常规检索只保留在历史表里。人工复核队列部分高重要性但来源不可靠的记忆比如模型从对话里推断出来的结论推送到后台列表运营或用户本人可以删除确认。遗忘机制听起来像是让系统“变笨”实际上恰好相反——给记忆做减法的目的是让每次召回留下的都是最有价值的记忆。这跟整理通讯录一个道理存了几百个联系人真找人的时候还得靠搜索和分组。5. 实战中遇到的问题与排查心得5.1 检索结果难用先查这几个地方不少朋友跑完上面那套代码后回来问我为什么AI记忆系统召回的都是一堆没用的东西我总结了几个最高频的原因直接给你排查顺序现象常见原因解决办法召回结果语义不相关嵌入模型不适合当前语言中文场景换BGE/M3E系列召回结果泛泛而谈重要度权重太低调大importance系数或过滤min_importance旧记忆被持续召出缺少时间衰减把recency_score纳入最终打分相关记忆就是召不出来向量相似度阈值过低降低阈值同时增加混合检索兜底召出内容互相矛盾没做冲突检测参照上面4.2的覆盖规则排查的时候我最建议先把“原始相似度分数打出来看一眼”。很多问题是嵌入模型输出的分数区间比较特殊默认阈值不适合导致大量低质量匹配混进来。先看分数分布再设定阈值基本能解决80%的问题。5.2 记忆导致模型“答非所问”的处理记忆注入不是越多越好。我刚开始时把Top-10记忆全塞进去结果模型被一堆历史信息带偏回答里掺杂大量无关旧信息。最夸张的一次用户问“今天天气怎么样”模型先回了“根据您之前对历史文化的兴趣……”。后来我做了三个调整每次对话只注入3~5条记忆超过5条信息量反而下降。提示词里明确写“以下信息是用户历史背景仅供回答参考。如果与当前问题无关请忽略。”如果检索到的记忆平均置信度低于阈值比如相似度低于0.3就不注入记忆当成普通对话处理。这三个调整落地后答非所问的情况几乎绝迹。这说明记忆系统的价值高低不在于塞得多少而在于相关性准不准。5.3 成本控制与隐私保护的取舍最后聊两个容易被忽视但很重要的点。成本方面记忆提取需要频繁调用LLM如果每轮对话都做费用会非常可观。我的优化方式把多轮对话聚合成一段再提取减少调用次数。用更便宜的小模型比如gpt-4o-mini做提取对话主模型用更强的模型。对提取结果做去重和diff检查重复内容不再调用LLM直接沿用旧记录。隐私方面现在的个人信息保护要求越来越高AI记忆系统恰好是重灾区。我的架构里强制做了几条规矩记忆库默认不存真实姓名、手机号等标识真实身份信息用id_hash关联而不是明文存储用户随时可以“清除全部记忆”这个接口必须开放而且要能在30秒内真正执行完。做AI产品的人坚持“记忆可删除”这个原则既是合规底线也是用户信任的基础。最后再分享一点我的体会给AI做记忆系统最忌“贪多求全”。技术架构可以复杂业务上一定要克制——每次注入的必须是“与当前问题高度相关的、最新的、实打实有用的”那几条。做产品的时候用户不会因为你的记忆库大而感动只会因为在某个瞬间发现“这个AI居然记得我说过的话”而惊艳。这个度要靠时间去打磨也正是这个项目最有意思的地方。 给大模型装上长期记忆：AI记忆系统架构设计与Python实战
先说实话给AI加上记忆之前我心里是有点抵触的。模型本身能在上下文里记住很多信息可只要会话一关它忘得比金鱼还快。做聊天机器人和Agent应用的人迟早都会撞上这堵墙你说过的需求它秒懂第二天再问一脸茫然。这个问题的本质就是AI的记忆系统缺失。所谓的ai-memory在我这里指的不是模型参数里固化的知识而是给大模型外挂的一套长期记忆系统从对话里提取有价值的信息用向量库和结构化存储把它保存下来下次对话时再把相关记忆检索出来、注入上下文让AI真正做到“越聊越懂你”。这篇文章我就把自己搭建这套系统的完整思路和代码原原本本拆给你包括记忆怎么采集、怎么存储、怎么召回、怎么防污染以及我在实操里踩过的坑。适合谁看所有在做ChatBot、Agent、私人助理类应用的开发者或者想给现有AI应用加记忆能力的产品、算法同学。下面这套方案不绑定任何特定平台我用Python实现你可以轻松换到自己熟悉的语言上。1. 先搞清楚AI记忆到底在解决什么痛点1.1 大模型自带的“工作记忆”有多脆弱很多人觉得大模型上下文窗口越来越大动辄128K、200K token那不就不需要什么记忆系统了吗这个想法我一开始也有。等真上线跑过一段时间就明白了上下文窗口是“工作记忆”不是“长期记忆”。工作记忆的特点就是容量有限、随聊随丢。一个128K窗口的模型你把十轮对话历史、参考资料、系统提示都怼进去很快就占满了。更麻烦的是研究里提到过一个现象模型对长上下文中处在中间位置的信息关注度明显下降也就是所谓的“lost in the middle”。窗口是大了但大量信息淹没在长文里模型根本抓不住重点。我实测过一个200K窗口的模型塞进去几十页资料后让它回忆最早对话里提过的一句关键需求它经常胡编。这说明什么单纯拉大窗口解决不了记忆问题只会让成本和噪声一起涨。你要的是“记得住”不是“装得下”。1.2 三种记忆的划分没有长期记忆的AI约等于金鱼在设计记忆系统之前我先把“记忆”这个概念拆成三层这也成了后来整个架构的基石工作记忆Working Memory当前对话上下文窗口内的全部信息相当于人思考时脑内短暂保留的信息。容量小、易丢失。短期记忆Short-term Memory单次会话内、尚未提炼的历史信息比如一整段多轮对话的原始文本。可以被摘要化但跨会话后基本失效。长期记忆Long-term Memory跨会话持久保存的信息经过提取、结构化、权重标注比如用户的偏好、习惯、目标、项目状态。没有记忆系统的时候AI只有工作记忆撑死加上短期记忆。会话一关全没了。这就好比金鱼——每次见面都是陌生人。有了长期记忆AI 才能像真人助理一样“我记得你上次说过……”。后面我详细讲这三层怎么联动。1.3 记忆系统带来的实际价值可能有人问记忆系统到底能带来什么我总结成三点。第一是个性化体验。同样是旅游推荐AI知道你是素食主义者、预算有限、偏好避开人群推荐结果完全不一样。没有记忆AI每次都在问“你吃什么”“你预算多少”用户早烦了。第二是任务连续性。做Agent、做自动化工作流时任务往往分散在多轮对话里。比如你在做一个市场调研Agent昨天确定了调研框架今天希望它接着执行。没有记忆它得让你重新交代一遍甚至需要重新粘贴所有资料。第三是数据资产沉淀。用户每一次对话里都隐含大量有价值的信息记忆系统能自动提炼成可复用的结构化数据。这些数据反过来可以做用户画像、做推荐优化、做产品洞察。也就是说记忆不只是功能还是数据资产。2. 记忆系统的架构设计从采集到召回2.1 记忆采集从对话里提炼什么构建记忆系统的第一个问题记忆是什么是原始的对话记录吗不是。对话记录是原材料不是记忆。十几轮闲聊里有用信息可能就两三句话。如果原样存下来检索的时候噪声极大还白白烧token。所以采集环节的关键动作是信息提炼。我用LLM做“记忆提取器”每轮或每几轮对话后让模型按固定格式输出需要长期记住的信息。提取维度通常分四类记忆类型典型内容重要度初始值事实类用户职业、年龄、城市、家庭成员3~4偏好类口味、风格、阅读偏好、沟通方式4~5项目状态当前项目进度、下一步动作4临时信息当下具体需求与长期无关1~2提取提示词我反复迭代过多次最终版大概是你是记忆提取器。从下面这段对话中提取值得长期记住的用户信息。 只输出JSON数组不要输出其他内容。每个元素包含 - content一句完整的话描述一个事实或偏好 - categoryfact / preference / project / none - importance1~5的整数5代表非常重要 对话 {conversation_text}这里有个小心思我让模型把不重要的信息也输出出来categorynone但后续写入时不会放入长期记忆而是丢弃或归档。好处是模型“认真看过”所有对话不会因为“只提取重要的”而误丢关键信息。采集时机也很讲究。我建议异步触发用户说完话、模型回复完整个回合结束后在后台调用提取器。不要每来一条消息就同步提取那样延迟高、成本高还会打断主流程。一般可以累积3~5轮或每隔1分钟批量提取一次。实测下来这个频次提取效果最稳上下文聚在一起信息更完整。2.2 记忆存储向量库和关系表各管哪一段采集到记忆后存哪里这里有一个常见的误区很多教程让你把记忆一股脑塞进向量数据库。但实际做下来向量库适合做语义检索不适合做精确查询和管理。我最终用的是“关系表 向量库”的双存储方案SQLite/PostgreSQL 作为主存储每条记忆有 id、content、category、importance、created_at、status 等字段负责精确查询、更新、统计和生命周期管理。向量库存 embedding 向量负责语义召回用 doc_id 关联主存储里的记录。为什么要这么拆向量库的元数据过滤能力虽然能用但复杂条件查询、更新、统计非常别扭。而记忆系统日常操作里有大量“把旧记忆更新掉”“查一下用户有没有说过某件事”这类精确需求。让关系表当主存储维护起来省心得多。向量库这边我用的是ChromaDB。不是因为它最好——论规模它不如Milvus论性能它不如FAISS但它的核心优势是开发体验极好pip install chromadb一行代码创建持久化客户端自带embedding接口非常适合项目原型和中小规模应用。等到数据量大了可以更换底层向量库关系表那套完全不用动。嵌入模型的选择也说一下嵌入模型特点适用场景OpenAI text-embedding-3-small质量稳定、API简单快速上线可接受外部APIBGE-small-zh中文效果好、可本地部署中文应用、隐私敏感M3E中文语义向量友好中文应用、轻量场景Ollama embedding本地运行离线/低算力环境这个选择直接影响中文检索的效果。我有一次用通用英文嵌入模型跑中文记忆检索出来的记忆驴唇不对马嘴换了BGE系列之后效果立刻好了。中文场景就别省这一步了。2.3 记忆召回怎么把对的记忆捞回来存储不是目的用起来才是。召回环节我按“输入查询 → 向量检索 → 重排序 → 上下文注入”四步来做。向量检索这一步特别容易出现“看起来召回了几条实际没用”的情况。我最早直接查Top-5发现精确度很低用户问“附近有什么素食餐厅”时向量库可能把“用户不喜欢辣”这样相关的记忆召回也可能把“用户之前去过杭州”这样无关的拉出来。向量相关的判断是语义距离不是语义“正确性”。为此我加了两个处理手段。第一是混合检索对记忆内容同时做向量相似度和关键词匹配BM25或SQL的LIKE都行两者分数按比例融合。关键词匹配能确保“素食”“不吃辣”这类颗粒度信息不漏掉向量检索能兜住语义变化。第二是重排序召回Top-20后用LLM根据当前查询给每条记忆打分取分数最高的少数几条进入上下文。这一步成本很低但对回答质量的提升非常明显。召回后的上下文注入位置很讲究。我通常把记忆放在系统提示词的末尾用明确的分隔段落标出来比如【历史记忆】 - 用户偏好清淡饮食不吃辣 - 用户正在策划毕业旅行预算5000元以内 - 用户对民宿感兴趣不喜欢连锁酒店放在这个位置的用意是让它作为“可供参考的额外信息”而不是改写了模型的主人格。如果放在对话历史开头容易被模型当成“用户的当下发言”出现角色混淆。我见过不少项目就是被这个细节坑了。3. 实操用Python给大模型装上记忆模块3.1 选型嵌入模型与向量库的组合先交代一下我最终采用的技术组合方便你直接抄作业Python 3.10ChromaDB做向量库SQLite做关系表OpenAI的API同时承担对话和记忆提取两个角色。如果你是本地部署党可以把对话模型换成Ollama里的qwen或者llama嵌入模型换成BGE整体思路一模一样。为什么坚持用ChromaDB而不是别的上面提到过开发体验好。还有一个重要原因它支持PersistentClient数据直接落盘不需要额外起服务进程。对于个人项目和中小型应用少一个依赖就少一个故障点。将来数据量上来了平滑切到Qdrant或pgvectorDAO层接口换掉就行业务代码不用动。嵌入模型这里我提醒一句不要选太大太强的模型来做记忆嵌入。记忆的句子普遍很短一两句话用大模型嵌入和小模型嵌入的效果差距没想象中那么大但成本和延迟是实打实的。我用的是OpenAI text-embedding-3-small需要的话剪到1024维召回效果已经足够。3.2 核心代码MemoryManager的完整实现下面这段代码就是我的记忆管理类核心我尽量保留主干逻辑方便你直接套用import json import uuid import sqlite3 from datetime import datetime import chromadb from chromadb.utils import embedding_functions class MemoryManager: def __init__(self, db_path./memory.db, persist_dir./vector_store): 初始化连接SQLite做主存储ChromaDB做向量索引 self.db_path db_path self._init_sqlite() self.client chromadb.PersistentClient(pathpersist_dir) self.collection self.client.get_or_create_collection( nameuser_memory, embedding_functionembedding_functions.OpenAIEmbeddingFunction( api_key你的API_KEY, model_nametext-embedding-3-small ) ) def _init_sqlite(self): self.conn sqlite3.connect(self.db_path, check_same_threadFalse) self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT, category TEXT, importance INTEGER, status TEXT DEFAULT active, created_at TEXT, updated_at TEXT ) ) self.conn.commit() def add_memory(self, content: str, category: str fact, importance: int 3, memory_id: str None): 写入一条记忆先存SQLite再写向量库 mid memory_id or str(uuid.uuid4()) now datetime.now().isoformat(timespecseconds) self.conn.execute( INSERT INTO memories VALUES (?,?,?,?,?,?,?), (mid, content, category, importance, active, now, now) ) self.conn.commit() self.collection.upsert( ids[mid], documents[content], metadatas[{ category: category, importance: importance, created_at: now }] ) return mid def search_memory(self, query: str, top_k: int 5, min_importance: int 0): 语义检索记忆支持重要度过滤 results self.collection.query( query_texts[query], n_resultstop_k, where{importance: {$gte: min_importance}} ) memories [] for idx, mid in enumerate(results[ids][0]): row self.conn.execute( SELECT * FROM memories WHERE id?, (mid,) ).fetchone() if row and row[5] active: memories.append({ content: row[1], category: row[2], importance: row[3], score: 1 - results[distances][0][idx] }) return memories def delete_memory(self, memory_id: str): 软删除SQLite置状态向量库也删掉 self.conn.execute( UPDATE memories SET statusdeleted, updated_at? WHERE id?, (datetime.now().isoformat(timespecseconds), memory_id) ) self.conn.commit() try: self.collection.delete(ids[memory_id]) except Exception: pass def list_all(self, limit: int 100): cur self.conn.execute( SELECT * FROM memories WHERE statusactive ORDER BY importance DESC, updated_at DESC LIMIT ?, (limit,) ) return [dict(zip([id, content, category, importance, status, created_at, updated_at], row)) for row in cur.fetchall()]这段代码里有几个细节值得多说一句。我用upsert而不是add来写向量库目的是支持更新记忆内容时保留同一个id。比如用户之前说过“我喜欢喝茶”过段时间改口“我开始喝咖啡了”你不能把两条冲突的记忆都留着而是应该用新内容覆盖旧内容。另外我做了软删除而不是把SQLite记录物理删除。原因是记忆可能被其他模块的日志或关联数据引用物理删了会出问题。业务上标记成deleted就够向量库里的向量同步删掉查询时自然找不到了。3.3 接入LLM让记忆真正参与对话光有MemoryManager还不够得让它和对话接口联动起来。我封装了一个conversation_with_memory函数def conversation_with_memory(user_input: str, history: list) - str: # 1. 召回相关记忆 memories memory_manager.search_memory(user_input, top_k5) # 2. 把记忆拼进system prompt memory_block \n.join([f- {m[content]} for m in memories]) system_prompt BASE_SYSTEM_PROMPT \n\n【历史记忆】\n memory_block # 3. 调用对话模型 response chat_with_gpt(system_prompt, history [{role: user, content: user_input}]) # 4. 异步提取新记忆 extract_memories_from_dialogue(history [{role: user, content: user_input}, {role: assistant, content: response}]) return response时序上要注意先召回再拼上下文最后调用模型。如果顺序反了等模型都回复了才去查记忆那这一次对话完全用不上记忆体验就很割裂。第4步提取是异步的不要在用户等待响应的时候同步跑延迟会明显拉长。你可能注意到我用了BASE_SYSTEM_PROMPT常量。对于有记忆注入的场景系统提示词建议写成固定的角色设定把变化的部分专门放在“历史记忆”区块。这样模型行为一致性更好也不容易被记忆内容带偏。4. 检索策略与记忆生命周期管理4.1 检索打分语义相似度之外还有时间与重要性前面我用ChromaDB的query直接返回了距离分数但在实际项目里这个分数只能算是初筛。真正决定“该把哪条记忆放给模型看”的是一个综合打分函数。我自己用的打分公式final_score similarity_score * 0.6 (importance / 5) * 0.25 recency_score * 0.15其中similarity_score是归一化后的余弦相似度ChromaDB的distance需要转成similarity一般用1 - distanceimportance / 5是把1~5的重要度归一化到0~1recency_score采用时间衰减函数recency_score exp(-lambda * days_since_update)lambda一般取0.03~0.05。取0.05的话20天前的记忆衰减到37%左右100天前基本趋近于零。这个衰减速度适配大多数对话场景如果业务是低频长期项目可以把lambda调小一点比如0.02。这个公式的意义在于单纯语义相关不等于应该被使用。用户半年前随口说的“我对XX一般般”和一周前明确表达的偏好即使语义相似度接近后者显然更能影响当前回答。重要度和时效性必须和语义相关度一起参与排序。4.2 记忆写入的防污染机制记忆系统最大的风险不是“记不住”而是记错、记住垃圾、记住冲突。这类问题我在线上踩过不少次总结出几项防污染措施第一设置记忆提取的“闸门”。不是每一条模型输出都要进记忆库。我给提取器加了一条硬规则只有用户明确表达的信息才写入长期记忆模型自己的推测和“正确的废话”不进库。这个靠提示词约束我试下来效果不错。第二冲突检测与覆盖。新记忆写入前先做一次相似度查询。如果发现已有记忆和新记忆语义相关但内容相反比如“用户不吃辣” vs “用户喜欢吃辣火锅”要判断哪个更新、哪个可信度更高。我的做法是新记忆的updated_at和importance若都更高就覆盖旧记忆否则保留旧记忆把新记忆降级为“临时信息”。第三敏感信息隔离。不是所有记忆都适合进长期记忆库。手机号、身份证、住址这类隐私信息强烈建议单独存到加密表并且不进向量库——因为向量库无法做到字段级的“忘记”。隐私合规这块后面专门展开说。4.3 遗忘与整理记忆库不是储物间记忆库如果只加不减用三个月就会变成垃圾场。每条记忆不管多重要都会被新记忆淹没检索精准度直线下降。所以我系统里有一套定期整理机制每隔几天跑一次离线任务合并相似记忆按向量相似度聚类把“用户是北京人”“用户住在北京海淀区”这类等价信息合并成一条保留信息最全的那条。降权过期记忆把超过90天且importance低于3的记忆自动标记为“archive”不再进入常规检索只保留在历史表里。人工复核队列部分高重要性但来源不可靠的记忆比如模型从对话里推断出来的结论推送到后台列表运营或用户本人可以删除确认。遗忘机制听起来像是让系统“变笨”实际上恰好相反——给记忆做减法的目的是让每次召回留下的都是最有价值的记忆。这跟整理通讯录一个道理存了几百个联系人真找人的时候还得靠搜索和分组。5. 实战中遇到的问题与排查心得5.1 检索结果难用先查这几个地方不少朋友跑完上面那套代码后回来问我为什么AI记忆系统召回的都是一堆没用的东西我总结了几个最高频的原因直接给你排查顺序现象常见原因解决办法召回结果语义不相关嵌入模型不适合当前语言中文场景换BGE/M3E系列召回结果泛泛而谈重要度权重太低调大importance系数或过滤min_importance旧记忆被持续召出缺少时间衰减把recency_score纳入最终打分相关记忆就是召不出来向量相似度阈值过低降低阈值同时增加混合检索兜底召出内容互相矛盾没做冲突检测参照上面4.2的覆盖规则排查的时候我最建议先把“原始相似度分数打出来看一眼”。很多问题是嵌入模型输出的分数区间比较特殊默认阈值不适合导致大量低质量匹配混进来。先看分数分布再设定阈值基本能解决80%的问题。5.2 记忆导致模型“答非所问”的处理记忆注入不是越多越好。我刚开始时把Top-10记忆全塞进去结果模型被一堆历史信息带偏回答里掺杂大量无关旧信息。最夸张的一次用户问“今天天气怎么样”模型先回了“根据您之前对历史文化的兴趣……”。后来我做了三个调整每次对话只注入3~5条记忆超过5条信息量反而下降。提示词里明确写“以下信息是用户历史背景仅供回答参考。如果与当前问题无关请忽略。”如果检索到的记忆平均置信度低于阈值比如相似度低于0.3就不注入记忆当成普通对话处理。这三个调整落地后答非所问的情况几乎绝迹。这说明记忆系统的价值高低不在于塞得多少而在于相关性准不准。5.3 成本控制与隐私保护的取舍最后聊两个容易被忽视但很重要的点。成本方面记忆提取需要频繁调用LLM如果每轮对话都做费用会非常可观。我的优化方式把多轮对话聚合成一段再提取减少调用次数。用更便宜的小模型比如gpt-4o-mini做提取对话主模型用更强的模型。对提取结果做去重和diff检查重复内容不再调用LLM直接沿用旧记录。隐私方面现在的个人信息保护要求越来越高AI记忆系统恰好是重灾区。我的架构里强制做了几条规矩记忆库默认不存真实姓名、手机号等标识真实身份信息用id_hash关联而不是明文存储用户随时可以“清除全部记忆”这个接口必须开放而且要能在30秒内真正执行完。做AI产品的人坚持“记忆可删除”这个原则既是合规底线也是用户信任的基础。最后再分享一点我的体会给AI做记忆系统最忌“贪多求全”。技术架构可以复杂业务上一定要克制——每次注入的必须是“与当前问题高度相关的、最新的、实打实有用的”那几条。做产品的时候用户不会因为你的记忆库大而感动只会因为在某个瞬间发现“这个AI居然记得我说过的话”而惊艳。这个度要靠时间去打磨也正是这个项目最有意思的地方。