给大模型Agent装上记忆系统:从SQLite到向量检索的完整实现
“刚告诉它的事情转头就忘”——这种体验但凡玩过AI助手的人都懂。让Agent真正“过目不忘”核心就是给Agent装上一套大模型记忆系统。最近我在调一个客服类Agent时也踩了同样的坑用户明明上一轮说过“我在南方家里湿度常年70%以上”下一轮它居然还在问“您所在地区气候如何”。问题不在模型笨而是这个Agent压根没有跨对话的长期记忆。这篇文章我会把记忆系统的原理、架构、代码实现一次讲透用最简单的方式带你自己搭一个能“记住你”的Agent记忆模块。不管你只写过几行Python还是已经在做AI应用开发按着步骤来都能跑通。1. Agent记忆是什么先搞懂“失忆”是怎么发生的很多人的第一反应是大模型本身不是挺能记的吗你跟它聊一天它都能接上话。但那是假象它记住的只是当前这轮对话里的上下文一旦会话关闭或者token长度超出限制前面说过的话就会被“挤”出去什么都不剩。了解这一点才能明白为什么我们要单独做一套记忆系统。1.1 大模型为什么天生“记不住事”大模型本质上是一个“无状态函数”。它每次接收你给的输入然后预测输出处理完这轮就结束了。你上一次问过它什么它其实“不记得”只是因为你把聊天记录原封不动地传给了它它才“看起来记得”。一旦消息长度超出上下文窗口限制最早的对话就会被截断丢弃。这里有个很现实的例子一个客服Agent每天接待几十个用户每个会话都有几十轮消息。如果不做任何持久化处理过几天用户回来说“我之前问过退款的事情”Agent只能一脸茫然。更麻烦的是如果用户之前明确说过“我不喜欢推荐太多东西”系统也没记住下次照样疯狂给用户推广告。用户会觉得这个助手“不聪明”实际上它就是“没有记忆”。所以要治“失忆”不能只靠模型本身要靠外围的记忆系统替它把重要信息存下来在需要的时候再喂回去。这也是Agent应用和单纯“调API”最大的分水岭——有没有记忆体验天差地别。1.2 三种记忆各司其职短期、长期、工作记忆很多人一上来就做“记忆系统”但没想清楚到底要做哪种记忆。按我的经验Agent的记忆至少可以拆成三层记忆类型生活类比作用范围典型实现短期记忆便利贴当前对话上下文消息列表、上下文窗口长期记忆归档文件夹跨会话持久保存数据库、向量存储工作记忆办公桌面处理当前任务临时组装的信息临时组装并注入Prompt的内容短期记忆就是普通对话里的上下文这个大模型自带不用我们操心。长期记忆才是真正需要自己动手的部分——把重要的用户信息、历史结论、偏好设置存到数据库里下次对话时再取出来。工作记忆则是一个临时集合比如用户问“帮我对比一下上周提到的两种方案”你需要同时检索长期记忆里的“上周方案A、方案B”相关内容再加上当前的问题拼成一段完整的输入喂给模型。这三层记忆不是互斥的而是层层叠起来的。我的习惯是短期记忆交给模型框架自带的会话管理工作记忆在每次请求时现场组装长期记忆用外部存储自己维护。这套分层方式能让代码结构清晰很多排查问题也方便。2. 记忆系统三件套写入、存储、检索怎么设计搞清楚记忆分几层之后再来看底层逻辑。任何一个记忆系统无论做得简单还是复杂都绕不开三件事什么时候记、记在哪里、怎么想起来。这就是写入、存储、检索三件套。别小看这三点每一步都有不少隐藏的坑。2.1 写入策略哪些记忆值得记下来这是整个记忆系统里最容易被忽略的一环。很多人一开始的做法很简单粗暴把所有对话原文全部存下来。结果存了一个月数据库里全是垃圾检索的时候什么都搜不出来系统还会因为上下文太冗长而变慢变笨。所以第一步要定义“什么值得记”。我常用的写入策略分为两类显式写入和隐式抽取。显式写入比较机械适合硬性规则。比如用户在对话里直接说“我叫李明”“我的收货地址是……”这种信息很明确可以直接用规则或正则匹配提取。再比如业务上要求“凡涉及退款申请必须记录工单号”这种也是显式规则。隐式抽取则需要动用模型的理解能力。比如用户说“我平时上班比较忙一般晚上八点后才有空”这句话里没有明确说“偏好”但模型可以推断出“该用户晚间空闲适合安排晚间提醒”。我的做法是每轮对话结束后把最近几轮消息发送给大模型让它判断“这段对话里是否有值得长期保存的用户特征、偏好或结论”如果有就输出一段结构化的记忆描述没有就返回空。这里有个关键点一定要给抽取加一个“允许保存”的过滤条件。如果不加约束模型会把“用户今天心情不好”这种一次性情绪也存进长期记忆下次对话时反而干扰判断。我自己实践下来建议在抽取指令里明确写只保存跨会话仍然有效、且对未来交互有影响的信息一次性事件不存情绪化表达不存。2.2 存储方案从SQLite到向量数据库怎么选存储层决定了记忆系统的规模上限。前期完全不需要上来就上重型组件一个SQLite数据库足够应付绝大多数场景了。SQLite的好处是零部署、单文件、读写快特别适合个人项目或中小型应用。你可以把记忆存成一张表字段包括用户ID、记忆内容、类型、重要程度、创建时间。查询的时候按时间排序或者按类型过滤非常简单。如果你还在探索阶段或者不想折腾运维就直接用SQLite别犹豫。向量数据库是在需要“语义检索”的时候才逐步引入的。什么是语义检索简单说用户的问题和记忆内容不一定有共同的关键词。比如用户问“我上次说想换台不吵的电脑”记忆里存的是“用户对现役笔记本风扇噪音很不满”——两者没有一字重合但语义相关。关键词匹配搜不出来必须用向量检索。向量数据库方案也多像某款本地轻量向量库、某开源库、某云端托管服务各有各的适应场景。我的建议是分阶段走先用SQLite存原始文本同时用embedding接口把文本转成向量存到内存或本地文件里测试阶段完全够用等数据量到了十万条以上再迁移到真正的向量数据库。这样不会一上来就被技术选型绑死。2.3 检索策略怎么把记忆准确“想”起来“写入”做好只是第一步“想起来”才是决定体验的关键。很多人把记忆存进去了但到用的时候不管三七二十一全部塞给模型结果上下文被一堆无关内容撑爆该记住的反而被淹没了。检索策略的精髓是“少而准”。常用的检索方式有三种按时间近因检索直接取最近N条记忆。适合“用户刚说过的话”这类场景。按关键词精匹配检索适合工单号、订单号、具体人名等强标识信息。按语义向量检索适合模糊回忆、意图匹配。用户说“上次那个方案”你得靠语义去找“上次”到底指哪个方案。我一般在实现时把三种方式混合用先用关键词精确匹配锁死强标识字段再用向量检索召回语义相关的内容最后按“创建时间新鲜度相关度”做一个简单加权排序取前几条注入上下文。这样既不会漏也不会太多。还有一点要注意注入Prompt的记忆需要和当前问题的时效性对齐。比如用户上个月说“我下个月要去出差”现在这个月已经过完了这条记忆就不该再被注入否则模型会一本正经地引用过期信息闹出笑话。3. 实战用Python从零搭一个Agent记忆模块理论说了一堆下面上代码。我会用最轻量的方式实现一个可用的记忆管理模块SQLite负责持久化加一个简单的抽取和注入流程。你把这套代码跑起来再接入自己常用的模型服务就能得到一个“跨对话记得住事”的Agent。3.1 环境准备与数据表设计开始之前需要准备好Python环境。我用的是Python 3.10以上版本不需要装任何重量级框架只要标准库里的sqlite3再加上requests用于调用模型接口就行。如果你已经装了其他请求库比如httpx也完全可以用但为了减少依赖下面示例统一用requests。记忆表的设计很关键直接影响后面的查询灵活性。我习惯用下面这样的结构CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, category TEXT DEFAULT general, importance REAL DEFAULT 0.5, created_at REAL DEFAULT 0 );字段含义很好理解user_id标识是哪个用户的记忆content是记忆内容本身category用来分类比如preference、fact、todoimportance是重要程度created_at是时间戳。有了user_id这个字段以后做多用户隔离、按用户清空记忆都非常方便。另外我会额外建一个索引按user_id和created_at查询的频率最高加上索引能避免数据量大了之后查询变慢CREATE INDEX IF NOT EXISTS idx_user_time ON memories (user_id, created_at);3.2 MemoryManager记忆的增删改查直接用标准库sqlite3实现一个MemoryManager类方法尽量精简能加、能查、能删就行。代码如下import sqlite3 import time from typing import List, Dict class MemoryManager: def __init__(self, db_path: str agent_memory.db): self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, category TEXT DEFAULT general, importance REAL DEFAULT 0.5, created_at REAL DEFAULT 0 ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_user_time ON memories (user_id, created_at) ) self.conn.commit() def add_memory(self, user_id: str, content: str, category: str general, importance: float 0.5): self.conn.execute( INSERT INTO memories (user_id, content, category, importance, created_at) VALUES (?, ?, ?, ?, ?), (user_id, content, category, importance, time.time()) ) self.conn.commit() def get_recent(self, user_id: str, limit: int 10) - List[str]: cursor self.conn.execute( SELECT content FROM memories WHERE user_id ? ORDER BY created_at DESC LIMIT ?, (user_id, limit) ) return [row[0] for row in cursor.fetchall()] def get_by_category(self, user_id: str, category: str, limit: int 10) - List[str]: cursor self.conn.execute( SELECT content FROM memories WHERE user_id ? AND category ? ORDER BY created_at DESC LIMIT ?, (user_id, category, limit) ) return [row[0] for row in cursor.fetchall()] def delete_memory(self, user_id: str, memory_id: int): self.conn.execute( DELETE FROM memories WHERE id ? AND user_id ?, (memory_id, user_id) ) self.conn.commit() def clear_user(self, user_id: str): self.conn.execute(DELETE FROM memories WHERE user_id ?, (user_id,)) self.conn.commit()这套接口足够覆盖80%的基础场景。日常使用时对话结束之后调用add_memory把抽取到的记忆存进去下次对话开始前调用get_recent把历史记忆取出来组装进系统提示词里。delete_memory和clear_user是给“遗忘”功能留的接口——记忆系统不能只进不出用户如果要求“忘掉我”你要能一键清空。3.3 把记忆模块接进Agent对话流程存储类写好之后接下来是重头戏让记忆真正参与对话。完整流程是这样一个循环接收用户的当前消息。根据用户ID从记忆库里检索相关历史记忆。把历史记忆组装进system prompt。调用大模型接口得到回复。从这段对话中抽取值得记下来的新信息写入记忆库。下面是一段完整示例代码我写了详细注释import requests # 这里使用你常用的大模型服务商提供的接口地址、模型名和自己的密钥 LLM_API_URL https://你的模型服务商地址/v1/chat/completions LLM_MODEL 一个主流的对话模型标识 LLM_API_KEY 你的密钥 def call_llm(messages: list) - str: resp requests.post( LLM_API_URL, headers{Authorization: fBearer {LLM_API_KEY}}, json{model: LLM_MODEL, messages: messages} ) resp.raise_for_status() return resp.json()[choices][0][message][content] def extract_memory(user_message: str, assistant_reply: str) - str: prompt f 你负责从一段对话中抽取值得长期保存的用户信息。 以下是用户的消息和助手的回复 用户{user_message} 助手{assistant_reply} 请判断这段对话中是否存在跨会话仍然有效的用户偏好、个人事实或重要结论。 如果存在请用一句简洁的话输出如果不存在请只输出无。 只输出内容不要解释。 result call_llm([ {role: system, content: 你是记忆抽取助手只输出抽取结果。}, {role: user, content: prompt} ]).strip() return if result 无 else result def chat_loop(message: str, user_id: str, memory: MemoryManager) - str: # 1. 检索历史记忆取最近5条 history memory.get_recent(user_id, limit5) system_prompt 你是智能助手请结合用户的历史信息回答问题。\n用户历史信息如下\n for item in history: system_prompt f- {item}\n if not history: system_prompt - 暂无历史信息\n # 2. 组装消息并调用大模型 messages [ {role: system, content: system_prompt}, {role: user, content: message} ] reply call_llm(messages) # 3. 抽取新记忆并入库存 new_memory extract_memory(message, reply) if new_memory: memory.add_memory(user_id, new_memory, categorypreference, importance0.7) return reply运行这套代码之后你会看到一个非常直观的变化第一次对话告诉Agent“我平时晚上十点才下班”然后关掉程序第二次重新启动再用同样的user_id问“我适合几点听播客”它能准确回答“晚上十点之后”。因为上一条记忆已经被持久化并且注入到了system prompt里。这就是“过目不忘”的最小可行实现。4. 升级记忆向量检索让Agent“想起来”而不是“翻出来”SQLite版本能解决“记忆的持久化”但解决不了“记忆的召回”。想象一下用户问“上次你推荐的降噪方案是什么”而记忆里存的是“用户对办公环境噪音敏感建议更换静音键盘”——两个句子没有公共关键词简单SQL查询完全命不中。这时就需要向量检索上场。4.1 关键词匹配的局限为什么需要向量关键词匹配像在图书馆里按照书名逐字查找书名里没有那个词书就找不到了。向量检索则像把每本书都提炼成一个“意思摘要”当你用含义相近的词去描述需求时它照样能找出那本内容相关的书。原理上模型会把文本映射成一组高维数字语义相近的文本在这些数字上距离很近。判断是否需要向量检索有个很简单的标准如果用户的提问方式和记忆内容之间经常存在“换一种说法”的情况就需要。比如记忆存的是“用户提到喜欢极简风格”用户问的是“有没有设计简单点的方案”——这中间就需要语义理解。而如果记忆只是“订单号12345”“用户叫张三”这类强标识信息关键词匹配更快更准没必要上向量。我现在的做法是两套并行强标识字段用SQL精准查语义模糊回忆用向量检索。两者互为补充效果最好。4.2 极简向量记忆组件实现不引入第三方向量数据库也能做向量检索。思路很简单调用大模型服务商提供的embedding接口把文本转成向量然后用numpy计算余弦相似度。数据量几万条以内这个方案完全跑得动。代码如下import numpy as np import requests EMB_API_URL https://你的模型服务商地址/v1/embeddings EMB_MODEL 一个embedding模型标识 class VectorMemory: def __init__(self, api_key: str): self.api_key api_key self.items: list[str] [] self.vectors: list[np.ndarray] [] def _embed(self, text: str) - np.ndarray: resp requests.post( EMB_API_URL, headers{Authorization: fBearer {self.api_key}}, json{model: EMB_MODEL, input: text} ) resp.raise_for_status() vec resp.json()[data][0][embedding] return np.array(vec, dtypenp.float32) def add(self, text: str): vec self._embed(text) self.items.append(text) self.vectors.append(vec) def search(self, query: str, top_k: int 3) - list: q_vec self._embed(query) scores [] for vec in self.vectors: # 余弦相似度值越接近1表示语义越接近 score np.dot(q_vec, vec) / (np.linalg.norm(q_vec) * np.linalg.norm(vec)) scores.append(float(score)) top_indices np.argsort(scores)[::-1][:top_k] return [(self.items[i], scores[i]) for i in top_indices]这套代码虽然简单但胜在无依赖、逻辑直白。使用时把历史记忆一条条add进去然后search出当前问题最相关的几条注入Prompt。和SQLite方案不同的是VectorMemory不需要表结构也不需要维护复杂的查询条件一切交给向量距离说话。4.3 向量记忆和关键词记忆怎么搭配用很多初学者在这里会纠结到底用哪种方案我的建议是别二选一而是叠起来用。比如用户说“帮我查一下上次那个订单的售后进度”其中“订单”可能对应记忆里的“用户反映某订单延迟发货”。这里需要向量检索召回这条记忆。但如果用户直接说“订单号A123456”SQL精确匹配一次到位根本不用向量。顺序上我建议先做SQL精准匹配消耗小、速度快如果命中就直接用没命中再走向量检索做语义召回。最后把两路召回的记得合并去重按照“重要程度时间新鲜度相似度”加权排序取前3到5条。这样做的好处是精准信息不丢模糊信息也能想起来。尤其在做客服、答疑类Agent时整体回答质量会有非常明显的提升。5. 常见问题与避坑实录代码能跑通只是第一步真正让人头秃的是那些在文档里查不到的问题。我把自己实操中踩过的坑整理出来希望能帮你少走弯路。5.1 高频问题速查表现象直接原因解决办法记忆存了但好像没生效注入Prompt时排序/截断逻辑有问题先直接用get_recent取最近几条检查是否注入到system prompt最前方模型把临时对话内容也存成了长期记忆抽取规则没限制“一次性信息”在抽取Prompt里明确要求不保存情绪化、临时性内容上下文太长模型越答越糊涂记忆注入过多挤占上下文窗口设置注入条数上限建议3到5条加token预算控制检索出来的记忆完全不对只用关键词匹配遇不到同义改写引入向量检索或增加分类过滤用户要求“忘掉我”但数据删不干净没有提供清空接口加clear_user方法并且在界面上提供“清除记忆”按钮不同用户之间记忆串了查询时没按user_id过滤所有检索方法强制带user_id禁止裸查全表5.2 我在实战里踩过的三个坑第一个坑是“记忆污染”。早期我做抽取时直接让模型从对话里“自由发挥”结果它把“用户当前问的是XX”这种一次性上下文也当作长期记忆写进去了。那之后我加了一道过滤抽取出来的内容要再经过一次系统检查确保它是一个“脱离当前对话仍然成立的事实陈述”。简单说如果这句话放到三天后的语境里依然有意义才允许保存。第二个坑是“记忆超载”。我一度认为记忆越多越聪明于是把检索到的所有记忆全部塞进上下文。结果模型因为信息太杂开始答非所问甚至把几条互相矛盾的旧记忆同时引用出来。后来我强制加上限单次最多注入5条每条记忆前标注时间让模型清楚哪些是旧信息、哪些是最近信息。效果立竿见影。第三个坑是“隐私边界”。做记忆系统很容易陷入“什么都想记”的冲动但用户的敏感信息比如身份证号、银行卡号、家庭详细地址一旦入库就需要承担很大的安全责任。我现在的原则是默认不记录敏感字段非必要不记录如果业务确实需要则加密存储并且提供一键删除入口。这不是技术能力的问题而是做产品的基本底线。最后说句实在话记忆系统不是越复杂越好。刚开始做一个SQLite加一个embedding接口已经能解决绝大部分“失忆”问题。跑通之后你会发现真正麻烦的不是技术而是怎么定义“哪些该记、哪些该忘”。我个人的习惯是先把规则写死在抽取指令里跑一段时间再慢慢放开让模型自己判断。有了记忆的Agent才算真正开始“认识”你的用户。