039、Agent的记忆持久化：Redis与SQLite
那天下午我差点把服务器砸了。
客户那边报了个诡异的问题：Agent跟用户聊了二十分钟，一切正常。但只要服务一重启，Agent就像失忆了一样，用户刚才报的工单号、车牌号、甚至自己刚才说过的话，全都不记得了。用户当场炸毛：“这破AI是金鱼吗？”
我上去一看代码，发现对话历史存在一个全局Map<String, List<Message>>里。进程活着的时候一切美好，进程一死，记忆跟着GC一起烟消云散。这事儿不怪Agent，怪我没给Agent装上“永久记忆”——也就是所谓的记忆持久化。
今天这期博客，就聊聊我做Agent记忆持久化时趟过的两条路：Redis 和 SQLite。别指望我给你列个表对比完事，我会直接把调试现场和踩坑记录摊开给你看。
先说Agent的记忆到底长什么样。一个正经Agent，它的记忆分两层：短期记忆是当前会话的上下文窗口，比如最近几轮对话、临时状态；长期记忆是跨会话的持久信息，比如用户偏好、历史事实、业务数据。这两层需求完全不一样，短期要高吞吐低延迟，长期要稳定可靠不丢失。于是很多人犯了和我一样的错：什么都往内存里塞，或者什么都往数据库里写。
我后来把短期记忆放在Redis里，把长期记忆丢给SQLite。先说Redis。
用Redis做Agent的短期记忆，核心价值在一个“快”字。对话过程中每一轮消息都要读写，用户还在等响应，你搞个SQL查询动辄几十毫秒，体验直接崩了。Redis 是纯内存操作，延迟通常在亚毫秒级，刚好满足这个场景。我用的是redis-py，典型的结构是这样：
importredisimportjson r=redis.Redis(host='localhost',port=6379,db=0)# 把一整轮对话存成一条hash记录，key是session_id# 别用string硬拼，后面取字段会哭的session_key=f"agent:session:{session_id}"r.hset(session_key,mapping={"round":round_num,"role":role,"content":content,"time":timestamp})# 同时设置过期时间，比如30分钟无操作就自动清空r.expire(session_key,1800)这里有个坑，我差点没哭出来。Redis的key如果设置了TTL，过期之后整个会话就没了。我当时给所有会话key都配了60分钟过期，结果用户聊到第59分钟时打了个长电话，回来继续聊，Agent直接失忆。更尴尬的是，因为Redis里的历史没了，Agent还跟用户说“我们第一次聊天吧”。你说气不气。解决方式很简单：每次有新消息写入时都要重新刷新过期时间，或者干脆在业务逻辑里做成滑动窗口——每次读写都重新expire。
再说说为什么别把所有历史一股脑全塞进Redis。内存不是无限的，而且Redis本身是缓存定位，你把所有用户几十万条聊天记录都塞进去，等机器内存报警，就等着被运维请喝茶吧。我的建议是：Redis里只保存当前会话的近期上下文，比如最近10轮对话。更早的，打包丢到SQLite里去存档。这个“导出+压缩”的动作，我用的是异步任务，不影响主流程。代码大概长这样：
# 注意：这里别同步做，会卡住对话响应defarchive_memory(session_id,cutoff_time):# 从Redis取出这个session下早于cutoff_time的所有消息# 批量写入SQLite对应的表中# 然后从Redis里删掉这些消息（或者只保留最近N轮）pass接下来是SQLite。一开始我有点瞧不上这玩意儿，觉得就是个小文件数据库，能行吗？后来发现行得很。Agent的长期记忆恰恰不需要每秒几千次的写入，它要的是可靠和结构化。SQLite 单文件、零运维、事务支持好，特别适合在单机环境里存长期记忆。我建了一张表记录用户长期事实，比如姓名、车辆信息、服务偏好，还有一张表存历史会话的归档消息。每次Agent启动时，先查SQLite把该用户的事实加载到自己的长期知识库里。
SQLite 的写入锁是个真坑。刚开始我用多线程写SQLite，结果一有并发写，直接报database is locked。后来才明白SQLite同一时刻只允许一个写事务，写多读少的场景还好，如果你边聊天边写历史，用户正在疯狂发消息，多个线程同时写，就会阻塞。解决方案有两个：一是把所有SQLite写操作丢到同一个单线程队列里；二是开启 WAL 模式，它能让读和写并发执行。我建议两个都上：
importsqlite3 conn=sqlite3.connect('agent_memory.db',check_same_thread=False)# 开启WAL模式，这里亲自试过，并发读写靠谱多了conn.execute("PRAGMA journal_mode=WAL;")# 顺便把busy_timeout设长一点，别一锁就报错conn.execute("PRAGMA busy_timeout=5000;")写入的时候一定要用参数化查询，别拼字符串。有次我把用户输入直接拼进SQL，结果用户发了句“are you sure”，单引号差点没把库给炸了。这个教训和Python的SQL注入警告一样深刻。
长期记忆的表结构，我建议别搞太死板。因为Agent的记忆类型会变，今天记的是用户生日，明天可能记的是用户关注的股票代码。这时候最好用“实体-属性-值”的方式，或者干脆存JSON。SQLite 支持JSON字段，配合Python的字典操作，舒服得很。比如：
# 别用text字段硬存，你迟早会想查某个属性的conn.execute(""" CREATE TABLE IF NOT EXISTS long_term_memory ( user_id TEXT, memory_key TEXT, memory_value TEXT, updated_at TIMESTAMP, PRIMARY KEY (user_id, memory_key) ) """)# 存的时候序列化memory_value=json.dumps({"stock":["AAPL","TSLA"],"preference":"low_risk"})conn.execute("INSERT OR REPLACE INTO long_term_memory (user_id, memory_key, memory_value, updated_at) VALUES (?, ?, ?, ?)",(user_id,"investment_preferences",memory_value,datetime.now()))读出来的时候再json.loads一下，搞定。
再聊聊两个方案怎么配合。我的习惯是：用户进来先查SQLite，把长期事实加载到Agent的上下文里；对话过程中，每轮消息先写Redis的短期窗口，保证当前逻辑能快速取到上下文；当一轮对话结束或者达到归档阈值，就把Redis中的本轮记录搬进SQLite的历史表，同时把Redis里对应的key清掉或者缩短TTL。这样Agent既有短期的高频响应能力，又有长期的稳定记忆。
还有一个血泪教训：记忆持久化一定要加版本号或者时间戳。Agent的记忆不是静态的，用户今天说喜欢运动风，明天可能就改成商务风了。如果你只存一个值，没有时间维度，Agent就会把昨天的旧信息和今天的新信息混在一起，产生“记忆混乱”。我后来在SQLite表里加了个version字段，每次用户确认新信息都会递增版本，读取的时候取最高版本。虽然有点丑，但确实有效。
顺便提一句，Redis的持久化配置也别忽视。虽然Redis是内存数据库，但你至少可以开启AOF或RDB来做落盘。我遇到过一台机器断电重启，Redis所有会话数据全没的情况。后来我开了AOF，但代价是写入性能略有下降。取个平衡：如果你只是把Redis当短期缓存，丢了也不致命，那就别开AOF，省点事；但如果Redis里存了用户当下没归档的重要上下文，那就开AOF，而且注意设置appendfsync everysec，别每写一次都同步磁盘，慢。
那到底该选Redis还是SQLite？说实话，不冲突。有一类Agent只用Redis做短期，长期直接扔给后端MySQL，SQLite反而多余。但如果你做的是本地运行、单机部署、边缘设备上的Agent，SQLite就是最佳选择，因为它不需要额外服务，存在本地文件里，随拿随走。反过来，如果你的Agent是云端多实例部署，需要多个进程共享同一个短期记忆，那SQLite的文件锁就不合适了，必须上Redis这种独立的KV服务。
我见过有的方案把SQLite放在网络盘上，然后多实例共用。别这么干，SQLite的锁机制在网络文件系统上会出各种怪异问题，比你想象中难缠得多。
再聊聊序列化的事。Agent记忆里不光有文本，还可能包含状态机、对象图、甚至函数调用的缓存结果。你在存Redis或者SQLite之前，一定要设计好序列化格式。我现在统一用JSON，但记住，JSON只能保存基本数据类型。如果你的记忆里有datetime、Decimal、bytes这类对象，得先转换成字符串或整型。不然json.dumps直接报TypeError。这里有个小技巧：自定义一个JSON编码器，把常见类型都转好，免得每个地方都写转换逻辑。
importjsonfromdatetimeimportdatetimeclassMemoryEncoder(json.JSONEncoder):defdefault(self,o):ifisinstance(o,(datetime,)):returno.isoformat()ifisinstance(o,bytes):returno.decode('utf-8',errors='ignore')# 还有别的类型，自己扩展returnstr(o)然后json.dumps(memory_obj, cls=MemoryEncoder)，省心。
最后说下测试。我建议你写一套模拟“重启后记忆恢复”的集成测试。别等到线上用户骂娘才来后悔。做法很简单：启动Agent，跟它聊几轮，存一些关键信息，然后模拟进程崩溃强制退出（别客气，直接kill -9），再重启Agent，看它能不能正确回忆出刚才聊的内容。这个测试跑通了，你的持久化策略才算基本合格。
我自己的经验是：记忆持久化不要追求一步到位。先做最简单的方案，跑起来，再根据真实使用场景慢慢优化。如果你刚开始就把Redis和SQLite玩出花来，反而容易被复杂逻辑带偏。毕竟Agent的核心是智能，记忆只是它的地基。地基塌了，楼再漂亮也没用。
下次你的Agent再“失忆”，先查一下它到底有没有长期记忆的落库，别只怪模型不行。
