做Agent开发的朋友应该都有过这种体会:聊得好好的,换一个话题,它就把三句话前你刚交代的事情忘得一干二净。上一阶段我们打通了基础对话链路,能接工具、能调上下文,可这些能力全悬在内存里,进程一重启、对话一刷新,全部归零。用户得把刚才说过的话原样再喂一遍,初稿问“我叫XX”,改稿又问“你叫什么”,体验直接回到石器时代。这个阶段要解决的问题,就是给Agent装上“海马体”——把短期的工作记忆固化成长时记忆,让它在会话中断、进程重启之后,依然认得你是谁、你们聊到哪、你还欠着什么没做完。
这篇文章会从记忆体系的分层设计讲起,落到一个用SQLite就能撑住的完整实现,再到断点续聊的恢复流程、上下文组装策略,最后是我实际开发中踩过的坑和调优记录。整套方案不依赖重型框架,能直接抄进自己的项目里,适合已经做完基础对话能力、正在给Agent补记忆能力的开发者参考。
1. 记忆体系设计:从“金鱼脑”到“海马体”
1.1 Agent为什么需要持久化记忆
先看一个现象。在纯对话阶段,Agent的每一轮交互都是无状态的,它只看到当前请求带进来的上下文。很多人觉得这没什么:把整个对话历史塞进Prompt不就行了?确实,短期内可以这么做,但很快会遇到三个问题。
第一是信息超载。对话越长,塞进去的token越多,模型的注意力被大量无关历史稀释,回答质量反而下降,成本也在线性上涨。第二是跨会话的断裂。用户今天问你A项目怎么规划,三天后回来问你A项目的进展,历史记录如果按会话隔离,Agent就是一脸茫然。第三是身份和偏好的丢失。用户说过“我平时用Python写后端”“我周六才有时间改稿”,这些信息一旦出了当前上下文窗口就没用了,每次都要用户重新交代一遍。
持久化记忆要解决的,本质上是这三类问题。它不只是“把聊天记录存下来”,而是要建立一套信息分层、有取舍、能检索的存储机制,让Agent具备真正的时间连续性。这就引出了下面的分层架构。
1.2 分层记忆架构选型
人脑的记忆并不是一个单一仓库,而是分层的。Agent的记忆系统也可以借鉴这个模型,我把它分成三层。
工作记忆对应的是当前正在处理的对话上下文,它短暂、易失,活在每次请求的Prompt里。情景记忆对应的是完整的对话历史,它按事件顺序记录“发生了什么”,是断点续聊的基础。语义记忆则是从对话中提炼出来的通用事实,比如用户的偏好、项目背景、长期目标,它不依赖具体某次对话,而是跨会话长期有效的知识。
这个分层的核心逻辑在于:不是所有历史都值得永久保存,也不是所有事实都值得每条对话逐字存储。情景记忆可以全量落库,但它进Prompt时要经过裁剪和摘要;语义记忆需要单独提取、存储和检索,因为它才是跨会话迁移的关键。至于程序性记忆——Agent掌握的技能和工作流——我倾向不放进存储层,直接用代码和配置文件固化下来,用版本管理管好就行,没有必要在数据库里重复维护一份。
提示:刚开始不要一上来就上向量数据库。绝大多数场景里,SQLite加上一套简单的关键词检索和规则提取,效果已经足够。
1.3 存储介质选型:为什么优先考虑SQLite
做持久化记忆,第一步是选存储介质。我见过有人直接用JSON文件存,也有人一上来就接一个向量数据库。我的建议是:个人项目或中小型应用,优先考虑SQLite。
纯内存存数据,进程一退出就没了,谈不上持久化。JSON文件存数据,读写是整个文件级别的,会话一多IO会变成瓶颈,更麻烦的是并发写容易出问题。向量数据库适合相似度检索,但部署、运维、数据备份都有额外成本,在数据量还没到几十万条之前,属于“杀鸡用牛刀”。
SQLite的优势很实在:单文件、零部署、支持真正的SQL查询、事务处理可靠,读写性能对Agent的记忆场景来说绰绰有余。一台云服务器、一个数据文件,不需要额外进程,也不需要考虑网络连接,备份只需要复制这个文件。后续如果数据量真的大到SQLite扛不住了,再平滑迁移到PostgreSQL或者独立的向量库,记忆表结构基本不用大改。
1.4 表结构与数据模型设计
我的记忆库一共三张表,各司其职。sessions表记录会话元信息,比如创建时间、更新时间、自定义的元数据JSON字段;messages表存原始对话记录,相当于情景记忆的全量存档;memory_items表存提取出来的语义记忆,这是跨会话复用的核心。
先看建表语句,我直接贴能跑的代码。
CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, meta TEXT DEFAULT '{}' ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS memory_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, importance REAL DEFAULT 0.5, created_at TEXT NOT NULL, last_access_at TEXT NOT NULL, access_count INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, id); CREATE INDEX IF NOT EXISTS idx_memory_session ON memory_items(session_id, memory_type);memory_items表里的importance字段是我想重点说明的。它是记忆的重要度评分,0到1之间,决定了这条记忆在检索排序时的权重。不是所有记忆都同等重要,用户随口说“今天午饭吃了面”和认真交代“我在做跨平台数据迁移项目,预算有限”,后者的长期价值明显更高。importance可以由规则或LLM提取时自动赋值,它是后续检索策略的基础。
2. 存储层实现:搭一个靠得住的记忆库
2.1 基础读写封装
选定SQLite之后,我习惯先写一个极简的存储层封装,把连接管理、建表、消息写入这些基础操作统一起来。
import sqlite3 import json from datetime import datetime, timezone DB_PATH = "agent_memory.db" def get_conn(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_conn() conn.executescript(""" CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, meta TEXT DEFAULT '{}' ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS memory_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, importance REAL DEFAULT 0.5, created_at TEXT NOT NULL, last_access_at TEXT NOT NULL, access_count INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, id); CREATE INDEX IF NOT EXISTS idx_memory_session ON memory_items(session_id, memory_type); """) conn.commit() conn.close() def create_session(session_id, meta=None): conn = get_conn() now = datetime.now(timezone.utc).isoformat() conn.execute( "INSERT OR IGNORE INTO sessions (session_id, created_at, updated_at, meta) VALUES (?, ?, ?, ?)", (session_id, now, now, json.dumps(meta or {}, ensure_ascii=False)) ) conn.commit() conn.close() def save_message(session_id, role, content): conn = get_conn() now = datetime.now(timezone.utc).isoformat() conn.execute( "INSERT INTO messages (session_id, role, content, created_at) VALUES (?, ?, ?, ?)", (session_id, role, content, now) ) conn.execute( "UPDATE sessions SET updated_at = ? WHERE session_id = ?", (now, session_id) ) conn.commit() conn.close()这段代码看起来简单,但有几个细节值得注意。一是所有时间字段统一用ISO 8601字符串存储,UTC时区,读取的时候再转换成本地时间。这么做的好处是排序和比较不会因为时区问题错乱。二是row_factory设为sqlite3.Row,这样读取时可以直接按字段名访问,方便后面构造上下文。三是每张表都有明确索引,消息表按session_id和id建复合索引,记忆表按session_id和memory_type建索引,实际数据量上来之后,查询速度差距明显。
session_id的生成也值得设计一下。我一般用UUID,由应用层生成后传进来,不在数据库里依赖自增主键。因为自增主键在不同环境迁移时容易冲突,UUID天然适合分布式场景,而且用户端可以把session_id存到本地,下次回来直接拿它恢复会话。
2.2 记忆提取管线
存储层只解决“放得下”的问题,真正让记忆有用的是“提得出来”。记忆提取是记忆系统里最像“海马体固化功能”的部分——从对话里筛选值得长期保留的信息,写入memory_items表。
我推荐用规则和LLM混合的方案。规则负责低成本地捕捞高频的明确信息,比如用户说“我叫XX”“我在XX公司”“我喜欢Python”,这些有相对固定的句式和关键词,用正则或者关键词匹配就能捕捉。但要处理“其实我之前做过的那个项目,用的方案后来想想不太合适”这种信息,规则就无能为力了,这时候需要让LLM在每轮对话结束后做一个提取和总结。
完整方案是在每轮User消息进来后,除了保存原始消息,还提取一次候选记忆。
import re def extract_memories_by_rules(session_id, role, content): candidates = [] # 常见偏好句式 patterns = { "preference": [ r"我(?:喜欢|不喜欢|偏爱|更爱|愿意|倾向于)[^。!?]{2,30}", r"(?:记住|记得|以后)[^。!?]{2,50}", ], "fact": [ r"我(?:是|在|有|住在|毕业于|就职于)[^。!?]{2,40}", r"我的(?:名字|职业|职位|公司|项目)是[^。!?]{2,40}", ], } for mtype, regexes in patterns.items(): for regex in regexes: for match in re.finditer(regex, content): phrase = match.group(0).strip() if 4 < len(phrase) <= 60: candidates.append((mtype, phrase, 0.4)) return candidates规则提取出来的只是候选,写入前我会再经过一道去重和重要度更新。去重的逻辑很简单:如果memory_items表里已经存在一条content相似度很高的记录,就不新建,而是更新原记录的last_access_at和access_count。这样避免“我叫XX”这句话在每轮对话里被反复提取,产生一堆内容重复的垃圾记忆。
LLM提取可以做一个定时批处理,比如每轮对话结束后,把最近几轮的消息打包发给模型,让它输出结构化的事实列表。输出格式我用JSON,要求模型只输出符合格式的事实条目,不生成冗余解释。这个提取不需要实时,可以放到后台队列里,避免阻塞主对话链路。
2.3 记忆去重与更新策略
记忆这种东西,很容易越存越乱。用户过几天可能修正自己说过的话,也可能重复提同样的事。如果系统只是机械地插入,memory_items表很快就会膨胀成一座垃圾山。
我维护了几条更新策略。一是上文提到的近似去重,插入前先查一下同session、同类型、内容相近(可以比较编辑距离或重叠词比例)的记录,如果有就直接更新时间戳和重要度。二是纠错优先级,如果新提取的记忆与旧记忆冲突,以新记忆为准,旧的标记为invalid。三是重要度衰减,长期没有被命中的记忆,重要度逐步下降,降到阈值以下就可以考虑归档或删除。
这些策略看起来简单,但能显著减少记忆污染。我见过不少Agent产品早期“记忆混乱”的案例,根因往往不是检索算法不行,而是写入端没有做质量控制,什么脏数据都进了记忆库。写入管线的质量,直接决定检索管线的上限。
3. 记忆检索与上下文组装实战
3.1 检索策略:不能只靠一个滑动窗口
记忆存好了,下一个问题是怎么让它发挥作用。最常见的错误做法是把所有历史消息一股脑塞回Prompt,Token爆炸,效果还差。正确的思路是:按需取用,分层组装。
我把检索拆成三路并行。第一路是最近窗口,取当前会话最近N轮消息,保留对话的即时连续性。第二路是滚动摘要,把窗口之前的消息压缩成一段摘要,让Agent对“更早发生了什么”保持大概认知。第三路是语义检索,从memory_items表里捞出和当前用户问题相关的长期记忆。三路结果合并成一个结构化的上下文块。
窗口大小需要根据模型上下文长度动态调整。我在实现里用了一个很简单的办法:根据当前请求携带的token预算,把窗口从3轮到20轮之间动态伸缩。预算宽裕就多给原始消息,预算紧张就优先保语义检索的结果。
3.2 相关度评分与时间衰减
语义检索这一路,核心是给记忆条目标相关度。数据量小的时候,我先用关键词重叠度做粗筛,再用一个轻量评分函数排序。数据量大了再平滑切换成向量检索,评分函数可以直接复用到向量召回的排序阶段。
评分函数我实现了两个维度:文本相关度和时间新鲜度。文本相关度看用户当前问题和记忆内容的重叠关键词比例;时间新鲜度则用衰减函数计算,越久远的记忆权重越低。
import math from datetime import datetime, timezone def time_decay(created_at_str, half_life_days=7): created = datetime.fromisoformat(created_at_str) age_days = (datetime.now(timezone.utc) - created).total_seconds() / 86400 return math.pow(0.5, age_days / half_life_days) def keyword_overlap(query, memory_content): query_tokens = set(query.lower().split()) memory_tokens = set(memory_content.lower().split()) if not query_tokens or not memory_tokens: return 0.0 return len(query_tokens & memory_tokens) / len(query_tokens) def relevance_score(query, memory_row): overlap = keyword_overlap(query, memory_row["content"]) freshness = time_decay(memory_row["created_at"], half_life_days=7) importance = memory_row["importance"] return 0.5 * overlap + 0.3 * freshness + 0.2 * importance这里有个经验性结论:half_life_days设置成7天,对通用Agent项目比较合理。如果应用场景本身就是短期密集交互,比如一场活动策划,可以调短到3天;如果是长期知识库伴学类应用,可以调到30天。这个参数值得反复测试调优,它是时间敏感性和记忆覆盖面的平衡点。
召回数量我也固定做了上限。默认从memory_items里召回5条,最多不超过10条。因为记忆条目是给模型做参考用的,不是让模型写论文的,条数太多反而会分散注意力。把重要的筛出来,比把相关的全塞进去效果好得多。
3.3 上下文组装模板
检索结果需要组装成一个结构清晰的上下文块,拼进系统Prompt。我用的模板长这样。
def build_context(session_id, user_query, recent_window=10): recent_messages = load_recent_messages(session_id, recent_window) memories = retrieve_relevant_memories(session_id, user_query, limit=5) parts = [] if memories: parts.append("## 与该用户相关的长期记忆") for mem in memories: parts.append(f"- [{mem['memory_type']}] {mem['content']}") if recent_messages: parts.append("## 最近对话记录") for msg in recent_messages: parts.append(f"{msg['role']}: {msg['content']}") return "\n".join(parts)load_recent_messages和retrieve_relevant_memories的SQL实现不复杂,按时间倒序取N条,再反转成时间正序就行。关键点是上下文块的排版顺序:长期记忆放在靠前的位置,最近对话放在后面,因为后者与当前问题的相关性天然更高,放在接近新用户消息的位置,模型理解起来更自然。
提示:Prompt里的“长期记忆”部分要让模型理解为参考信息,而不是强制指令。加上一句“以上记忆信息可能存在偏差,以用户当前说法为准”很有必要。
4. 断点续聊:把现场完整搬回来
4.1 会话恢复流程
断点续聊听起来高大上,本质就一句话:给定一个session_id,把之前存下来的现场重新载入。但载入的不只是聊天记录,还有Agent的内部状态。
我的恢复流程分四步。第一步,根据session_id查sessions表,确认会话存在,拿到meta字段里的自定义状态。第二步,查messages表,把全部历史消息按id正序拉出来,灌回上下文的滚动窗口。第三步,根据本次用户问题的关键词,从memory_items表检索相关长期记忆。第四步,把这三部分拼成完整的初始上下文,交给模型继续生成。
这个流程写成一个restore_session函数,每次对话请求进来时先调用它。
def restore_session(session_id): conn = get_conn() session = conn.execute( "SELECT * FROM sessions WHERE session_id = ?", (session_id,) ).fetchone() if not session: return None messages = conn.execute( "SELECT role, content FROM messages WHERE session_id = ? ORDER BY id", (session_id,) ).fetchall() meta = json.loads(session["meta"] or "{}") return { "session_id": session_id, "meta": meta, "messages": messages, }恢复会话时遇到session不存在的情况,不能直接报错,要当新会话处理。很多用户是换了设备、清了缓存,把session_id丢了,服务器端的会话却还在。优雅的做法是打印一条提示,告诉用户“这是一次新的对话,如果你之前聊过某个话题,可以简单提一下”,而不是甩一个冷冰冰的异常。
4.2 Agent内部状态恢复
断点续聊最容易忽视的是内部状态。聊天记录恢复只是表象,Agent做到一半的事、算到一半的参数、选型选到一半的开关,这些才是真正决定“续得上聊”的关键。
比如用户让Agent做一个跨平台数据迁移方案,Agent已经分析完源数据格式,正准备生成映射规则,这时候断线了。回来之后如果只恢复对话历史,Agent看到的是“分析源数据格式”这半截消息,它不知道分析结果是什么。内部状态恢复要解决的就是这个问题:把agent运行过程中的中间产物、当前任务进度、下一步计划,存进sessions表的meta字段。
我在save_state接口里维护了一个状态槽,Agent在每个关键步骤结束时把状态快照写进去。
def save_state(session_id, state_dict): conn = get_conn() session = conn.execute( "SELECT meta FROM sessions WHERE session_id = ?", (session_id,) ).fetchone() meta = json.loads(session["meta"]) if session and session["meta"] else {} meta["running_state"] = state_dict conn.execute( "UPDATE sessions SET meta = ?, updated_at = ? WHERE session_id = ?", (json.dumps(meta, ensure_ascii=False), datetime.now(timezone.utc).isoformat(), session_id) ) conn.commit() conn.close()恢复的时候,把这个running_state从meta里取出来,拼进上下文,让Agent知道“你在断线前已经完成到哪一步、接下来该做什么”。具体格式可以是一段纯文本,也可以是结构化JSON。我实际测试下来,给模型一小段“当前任务状态描述”比给一堆零散变量更有效,模型理解起来更轻松。
4.3 恢复过程中的上下文一致性处理
恢复上下文时,有一个隐蔽的坑:模型在生成长回复的过程中,可能引用了一些旧上下文里的细节,但这些细节和最新的记忆有冲突。比如用户第一天说“项目用Java”,第五天改口说“已经切到Go了”,如果恢复时把第一天和第二天的记忆都保留,模型就会困惑。
我的处理方案是记忆冲突检测。当新提取的记忆与旧记忆冲突时,旧记忆不是被物理删除,而是被打上过期标记。检索时优先取最新有效记忆,过期记忆只有在最新记忆缺失时才作为兜底候选。具体实现给memory_items表加一个is_valid字段就行,查询时默认过滤掉is_valid=0的记录。
另外,恢复后的第一轮回复,我建议让Agent先做一次简短的“状态确认”,而不是直接埋头干活。这个确认不是说废话,而是帮模型校准上下文,比如“根据之前的对话,你在做XX项目的数据迁移,目前已完成源数据分析,接下来要生成映射规则,是否继续?”。这既能验证恢复的上下文是否正确,也让用户有机会纠正偏差,避免Agent带着错误的记忆跑偏。
5. 实战踩坑与调优记录
5.1 常见问题排查速查表
这一路开发下来,我把自己和身边朋友踩过的坑整理成了一个速查表,按症状、原因、解决方案三列对照。
| 症状 | 原因 | 解决方案 |
|---|---|---|
| 多会话之间记忆串味 | 检索时没有按session_id过滤 | 所有SQL查询强制带上session_id条件,并在检索函数里断言隔离 |
| 恢复后回答文不对题 | 内部状态没有持久化,只恢复了历史消息 | 实现save_state接口,把任务进度写入sessions.meta |
| 记忆条目爆炸,检索慢 | 没有做去重和清理,垃圾记忆堆积 | 提取前查重,重要度低于阈值自动归档 |
| 模型引用过期信息 | 冲突记忆没有覆盖或标记失效 | 新记忆写入时标记旧记忆为invalid |
| 上下文中历史消息过多 | 无脑加载全部历史消息进Prompt | 用滚动窗口+滚动摘要替代全量历史 |
| 断线恢复后第一句就胡说 | 缺少状态确认机制 | 增加“基于恢复状态做一次输出”的确认步骤 |
5.2 Token预算与记忆策略自适应
Token预算永远是Agent开发的紧箍咒。很多人刚开始兴致勃勃地设计复杂记忆系统,结果上线后发现一次请求的token消耗翻了三倍,成本压不住,只能砍功能。
我的经验是做一个自适应的预算分配器。在组装上下文前,先确认本次请求的token上限,然后按优先级分配:用户当前消息和系统Prompt必须保证,最近窗口的原始消息次之,长期记忆再次之,滚动摘要最后。预算不足时,优先压缩最近窗口的轮数,而不是扔掉长期记忆,因为长期记忆往往承载了跨会话的核心信息。
还可以按session做细粒度调优。高频活跃会话给更多预算,低频冷启动会话给更少预算。这个策略通过sessions表的meta字段记录历史会话频率,定期更新。
5.3 记忆管理的产品化思考
最后说个容易被技术人忽视的点:记忆系统做到后面,产品化比技术更重要。用户应该能看得到Agent“记得什么”,也应该能手动删掉某些记忆——没有这个能力,记忆系统做得再精巧,用户也只会觉得Agent是个固执的偏执狂。
我在项目里加了一个简单的记忆管理接口:允许用户查询当前会话保存了哪些长期记忆,支持按条目删除。实现起来很简单,一个SELECT加上按id的DELETE。但这个功能带来的信任提升非常明显,用户发现自己能掌控Agent的记忆,而不是被一个看不见的系统记录一切,反而更愿意长期使用。
另外,我还做了一个“记忆智能化补全”的小功能。当用户问了Agent一个问题,Agent发现与某条记忆高度相关但记忆又不够具体时,可以在回复末尾自然地问一句“你之前提到过在这个方向的一些约束,现在还需要保留吗”。这种主动澄清比盲目猜测可靠得多,也是记忆系统能持续保持高质量的关键。测试下来,用户对这种追问的接受度远高于我最初的预期。
这个阶段做完,最大的体感是Agent终于“像个人了”。它不再是每次对话都从零开始的无状态接口,而是对你有一个模糊但连续的整体认识。下一步迭代,我准备往记忆提取里加入更细粒度的事件关联和情绪标签,让记忆系统不止记住事实,还能理解用户当时为什么这么说。如果你正在给Agent补记忆能力,我建议从这套分层模型和SQLite方案起步,跑通闭环后再逐步接入向量检索,这条路最稳。最后再分享一个小技巧:在整个记忆系统的开发阶段,记得给memory_items表加一个“来源消息”的字段,指向它来自哪条历史消息。排查问题时,这个字段能帮你快速定位一条记忆是怎么来的、为什么会被检索出来,省下的调试时间远超加字段的那几分钟。