ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

用SQLite给Agent做记忆层:轻量方案与向量数据库选型对比

用SQLite给Agent做记忆层:轻量方案与向量数据库选型对比 前阵子帮团队一个小伙伴排查 Agent 的记忆模块他一上来就问我“用哪个向量数据库比较好”。我看了眼他的场景无非是记住用户偏好的技术栈、记一下上次任务执行到哪一步、存几个关键配置项。我直接跟他说你这需求一个 SQLite 文件就够了。这不是抬杠。我自己的几个 Agent 项目记忆层就是基于 SQLite 实现的稳定跑了小半年没有专门的向量数据库服务更没有额外的基础设施开销。启动快、备份方便召回速度在某些场景反而比向量检索还顺。这篇博文就把这套设计思路、表结构、完整代码和踩过的坑全部放出来给正准备给 Agent 加记忆层的朋友一个可落地的参考。1. 先搞清楚Agent 的“记忆”到底要解决什么问题很多人在设计记忆层之前根本没想清楚一个基础问题你的 Agent 需要的是哪种记忆这决定了后面一套技术选型的方向方向错了后面再堆什么技术都别扭。1.1 对话记忆和长期记忆是两码事对话记忆解决的是“同一个会话里上下文别丢”。这件事现在主流做法是靠大模型的 context window把之前的对话记录、中间结果不断拼接进去。它的特点是生命周期短会话一结束这些东西基本就没用了属于短期记忆。但 Agent 的长期记忆完全是另外一回事。它要解决的是“跨会话、跨任务把重要的信息沉淀下来”。比如你一个月前告诉 Agent“我是前端开发代码里变量命名用驼峰”这个信息不应该随着对话窗口的关闭而消失Agent 下次给你生成代码的时候应该主动把这条约束捡回来。我见过不少项目把长期记忆和对话历史混在一个存储模型里结果就是对话历史越堆越大真正有用的用户偏好反而淹没在一堆闲聊里召回的准确率越来越差。记忆层的设计第一步就是把这两者分开对话历史交给模型窗口或者短期缓存长期记忆交给独立的持久化存储。1.2 长期记忆里究竟存了些什么数据按我自己的落地经验Agent 的长期记忆可以归成三类。事实型记忆比如用户的姓名、公司、技术栈、接口地址、部署环境这类信息特征是键值关系明确适合精确查询。偏好型记忆比如用户喜欢简洁的回答风格、代码里要求写中文注释、默认部署到测试环境这类信息带主观性会影响 Agent 后续的行为决策。状态型记忆比如上次任务进行到哪一步、某个任务的进度是百分之多少、某个流程中已经确认过哪些条件这类记忆用于恢复上下文保证 Agent 在多次调用之间能续上进度。这三种记忆有一个共同特点它们本质上都是“结构化数据”。事实型就是键值偏好型就是条件约束状态型就是一个记录。这类数据用 SQL 查询、精确匹配完全没有问题压根不需要向量语义检索。1.3 判断标准你的记忆是“查出来”还是“想起来”我一般会给开发一个非常直觉的判断题当 Agent 需要回忆一条信息时用户心里的预期是一个明确的关键词还是一段模糊的语义描述。如果用户会说“上次我们讨论的那个项目你帮我找找当时的方案”这就是一段模糊语义需要联想想起来那确实是向量检索的强项。但如果用户会说“帮我查一下数据库地址”、“我的偏好是简洁风格”、“刚才那个任务卡在第二步”这些都是“查出来”的场景SQL 精确命中就完了。坦白讲目前市面上绝大多数 Agent 的记忆需求都属于“查出来”这一类。非要给这类需求配上 embedding 流程和向量存储属于典型的杀鸡用牛刀。先把数据模型设计清楚比先选数据库要重要得多。2. 为什么我没有一上来就上向量数据库不是说向量数据库不好而是很多场景根本用不上。我选 SQLite 做记忆层是在对比了多套方案之后做的决定核心原因其实就一句话架构要匹配需求不要为了技术而技术。2.1 向量数据库在 Agent 记忆场景的三大门槛先说说向量数据库在个人项目、中小团队项目里常遇到的实际情况。第一是部署和运维成本。一个正经的向量数据库要么是独立服务要么至少是一个单独的进程你得考虑连接管理、容量规划、备份恢复、升级兼容。对个人开发者和 3 到 5 人的小团队来说这些成本不是写几行代码能抹平的。第二是 embedding 流程。数据入库前要做向量化意味着你要额外调用 embedding 接口这是一个持续的费用和延迟开销。记忆条目少的时候感受不明显一旦每天有上千条记忆写入embedding 的调用成本就会变成一笔看得见的账。第三是召回结果的不可控性。向量检索返回的是“相似”不是“精确”。你以为用户问“数据库地址”能精确召回那条 addressxxx但向量检索可能因为语义相近召回一堆 mysql、postgresql、redis 之类的内容。记忆场景里精确匹配经常比相似匹配更实用。2.2 我选择 SQLite 的五个实际理由用表格来对比一下 SQLite 和专职向量数据库在记忆层场景下的表现对比维度SQLite向量数据库部署方式零部署嵌入式一个文件独立服务或至少独立进程写入延迟本地文件毫秒级网络 IO依赖服务状态查询方式SQL 精确查询、范围查询、全文检索向量相似度、混合检索备份迁移拷贝一个文件即可需要导出/导入或快照操作稳定性单文件天然适合单机场景服务挂了影响全部 Agent适用数据模型结构化记忆、键值、标签长文本、语义匹配坦白讲后面几项优势针对性很强Agent 的记忆层数据量通常不会大到单文件处理不了Agent 运行环境经常是用户本地、边缘节点或者一台小服务器嵌入式数据库有天然优势备份和迁移这一个文件凑齐了这种行为对开发者来说太顺手了。2.3 SQL 和向量在记忆里的实际分工我目前的状态是混合使用但主力是 SQLite。对于 Agent 的记忆上下文涉及模糊语义联想的内容我会单独划一个可选的向量索引层配合 embedding 接口做增强。但记住这是增强不是基座。默认情况下所有记忆先落到 SQLite。记忆的写入是结构化的元数据是明确的召回以精确匹配和关键词匹配为主。只有当精确匹配拿不出结果、语义召回又确实是用户需要的时候才触发向量检索。这么分层的好处是90% 的常用路径走 SQLite成本低、响应快、可控性强剩下 10% 的模糊场景才引入向量能力把资源花在刀刃上。3. 记忆层表结构设计一张表撑起一套记忆系统设计记忆层的表结构其实是在回答一个问题一条记忆最少需要哪些字段才能支撑起存储、检索、遗忘整个生命周期我把这套设计拆得非常细。3.1 先想清楚记忆本质上是什么把各种记忆抽象一下你会发现它们的共同结构是一样的一段内容(content)它是什么类型的(memory_type)它有多重要(importance)它在什么时间被创建、什么时间被访问以及它关联到哪个 Agent、哪个会话。这套抽象和向量数据库的 collection 思路本质上是一样的。向量库里的每条记录也是“文本内容 metadata 元数据 embedding 向量”你只是把 embedding 向量换成了重要性分数和访问统计。结构对了后面怎么检索都是水到渠成的事。3.2 核心建表 SQL我实际的表结构是这样CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL DEFAULT default, session_id TEXT, memory_type TEXT NOT NULL DEFAULT fact, content TEXT NOT NULL, metadata TEXT, importance REAL NOT NULL DEFAULT 5.0, access_count INTEGER NOT NULL DEFAULT 0, last_accessed_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_memories_agent_type ON memories(agent_id, memory_type); CREATE INDEX IF NOT EXISTS idx_memories_agent_accessed ON memories(agent_id, last_accessed_at); CREATE INDEX IF NOT EXISTS idx_memories_agent_updated ON memories(agent_id, updated_at);字段解释一下。agent_id 用于区分多个 Agent 实例避免不同 Agent 的记忆互相污染。session_id 可以为空用于标记这条记忆是从哪个会话里提取的方便回溯来源。memory_type 是记忆分类我习惯用 fact、preference、state、skill 四种。content 是记忆正文。metadata 用 JSON 文本存储扩展属性比如来源消息 ID、当时的模型温度参数、关联任务 ID 等。importance 是我这套设计的核心字段代表这条记忆的初始重要程度。access_count 和 last_accessed_at 记录访问情况是遗忘策略的数据基础。3.3 metadata 里到底塞什么很多初学者会忽略 metadata 的价值其实这是记忆层扩展性的关键。举几个实际例子一条事实型记忆“用户的部署环境是 staging”。content 里存这句话metadata 里可以存{field: deploy_env, value: staging, source: user_input, confirmed: true}。一条状态型记忆“任务 #42 已完成 3/5 个步骤”。metadata 里可以存{task_id: 42, step: 3, total_steps: 5}。有了这批结构化元数据你就可以在召回时做更加精准的过滤。比如只召回 sourceuser_input 且 confirmedtrue 的记忆这种粒度在纯向量检索里很难做到。SQLite 从 3.38 版本开始内置了 JSON 函数(json_extract、json_set、json_type 等)可以直接在 SQL 里操作 metadata 字段。如果你用的是新版本甚至可以针对 metadata 里的某个键建立表达式索引比如CREATE INDEX idx_memories_deploy_env ON memories(json_extract(metadata, $.field));这种灵活性是纯向量数据库很难给你的。3.4 索引策略加三个就够别过度设计我只加了三个单列索引覆盖了最常见的召回路径按 Agent 和类型过滤按最后访问时间排序按更新时间排序。实际使用下来几千条到几万条记忆的量级查询响应时间都在毫秒级别完全不需要更复杂的索引。如果你硬要给 content 建普通 B-tree 索引其实对 LIKE 前缀查询有一点帮助但 LIKE 的模糊匹配基本用不上索引。真正提升全文搜索体验的是 FTS5这个我在下一节专门讲。4. 核心实现写入、召回、遗忘一个小类搞定表结构定了核心逻辑就是一个 Python 类的事。我把记忆层封装成一个 SQLiteMemoryLayer支持记忆写入、召回、访问更新和遗忘清理。4.1 MemoryLayer 类的骨架import sqlite3 import json import math import time from datetime import datetime class SQLiteMemoryLayer: def __init__(self, db_pathagent_memory.db): self.conn sqlite3.connect(db_path) self.conn.row_factory sqlite3.Row self._init_schema() def _init_schema(self): cur self.conn.cursor() cur.executescript( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL DEFAULT default, session_id TEXT, memory_type TEXT NOT NULL DEFAULT fact, content TEXT NOT NULL, metadata TEXT, importance REAL NOT NULL DEFAULT 5.0, access_count INTEGER NOT NULL DEFAULT 0, last_accessed_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_memories_agent_type ON memories(agent_id, memory_type); CREATE INDEX IF NOT EXISTS idx_memories_agent_accessed ON memories(agent_id, last_accessed_at); CREATE INDEX IF NOT EXISTS idx_memories_agent_updated ON memories(agent_id, updated_at); ) self.conn.commit()这个骨架很简单连接数据库、初始化表结构。关键是要在项目入口只初始化一次不要每个 Agent 调用都重新建表连接复用能省很多开销。4.2 写入一条记忆过滤、打分、去重写入不是简单 insert我会做三层校验。第一层是长度过滤。content 太短的信息没有存储价值比如“好的”、“知道了”这种直接丢弃超过长度上限的超长文本需要截断或摘要防止记忆库被无效信息淹没。第二层是去重。对近似重复的记忆做检查比如用户已经说过“我喜欢简洁回答”下次又说“回答简洁点”系统应该更新原记忆的 importance而不是再插一条新记录。第三层是重要度打分。写入时如果没有显式指定 importance我会根据这条记忆的类型来默认打分用户主动声明的内容给高分会话中自动提取的上下文给中等分数。def remember(self, content, memory_typefact, agent_iddefault, session_idNone, metadataNone, importanceNone): content (content or ).strip() if not content or len(content) 4: return None now datetime.utcnow().isoformat() # 去重看看有没有语义上高度一样的已有记忆 dup self.conn.execute( SELECT id, importance FROM memories WHERE agent_id? AND content? AND memory_type? LIMIT 1, (agent_id, content, memory_type) ).fetchone() if dup: new_imp min(10.0, dup[importance] 0.5) self.conn.execute( UPDATE memories SET importance?, updated_at? WHERE id?, (new_imp, now, dup[id]) ) self.conn.commit() return dup[id] if importance is None: importance self._default_importance(memory_type) metadata_json json.dumps(metadata, ensure_asciiFalse) if metadata else {} cur self.conn.execute( INSERT INTO memories (agent_id, session_id, memory_type, content, metadata, importance, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?), (agent_id, session_id, memory_type, content, metadata_json, importance, now, now) ) self.conn.commit() return cur.lastrowid这里有个细节值得展开去重逻辑我用的是完全匹配 content因为自己维护的记忆层里同一个 Agent 主动存储的偏好文本通常比较规整。如果你担心两个说法不同但意思相同的重复记忆可以额外存一个 keywords 字段或者专门的标签字段用标签去做去重判断这是低成本又可控的办法。4.3 召回多个过滤条件叠加召回函数支持按 agent_id、memory_type、关键词、最小重要度来过滤。核心是尽量把过滤条件下推到 SQL 里减少上层的数据处理量。一个简化但实用的版本def recall(self, agent_iddefault, memory_typeNone, keywordNone, min_importance0.0, limit10): sql SELECT id, content, memory_type, metadata, importance FROM memories WHERE agent_id? params [agent_id] if memory_type: sql AND memory_type? params.append(memory_type) if min_importance 0: sql AND importance? params.append(min_importance) if keyword: sql AND content LIKE ? params.append(f%{keyword}%) sql ORDER BY importance DESC, last_accessed_at DESC LIMIT ? params.append(limit) rows self.conn.execute(sql, params).fetchall() return [dict(row) for row in rows]这个版本能覆盖大部分需求但 LIKE 模糊匹配在数据量上去之后性能会下降这是逃不掉的。更优的替代方案是把 keyword 相关的检索交给 FTS5 全文搜索下一小节专门讲。4.4 全文搜索FTS5 让 SQLite 也能做关键字联想SQLite 内置的 FTS5 扩展是一个虚拟表模块专门做全文检索支持 BM25 相关性排序。这就像一个迷你版搜索引擎索引建在 content 或者其他文本字段上查询的时候用 MATCH 语法匹配比 LIKE 效率高一个量级。给 memories 表加上 FTS5 同步的完整做法CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5( content, contentmemories, content_rowidid ); CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN INSERT INTO memories_fts(rowid, content) VALUES (new.id, new.content); END; CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content) VALUES (delete, old.id, old.content); END; CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content) VALUES (delete, old.id, old.content); INSERT INTO memories_fts(rowid, content) VALUES (new.id, new.content); END;有了这个全文索引原来的 keyword 搜索可以升级成 FTS 版def recall_by_search(self, query, agent_iddefault, limit10): sql SELECT m.id, m.content, m.memory_type, m.metadata, m.importance, bm25(memories_fts) AS rank FROM memories m JOIN memories_fts ON memories_fts.rowid m.id WHERE memories_fts.content MATCH ? AND m.agent_id ? ORDER BY rank ASC LIMIT ? rows self.conn.execute(sql, (query, agent_id, limit)).fetchall() return [dict(row) for row in rows]这里有个需要注意的点FTS5 默认的 tokenizer 是 unicode61对英文分词效果不错但对中文只能做到按字切分所以“用户偏好注释风格”会被切成多个单字来匹配。对于 Agent 记忆这种关键词粒度偏细的场景这个表现其实够用如果你需要真正的词级中文搜索可以在写入时用 jieba 做分词把分词结果存到额外的一列再把 FTS5 索引建在那列上。这个方案我在后面踩坑部分细说。4.5 记忆权重更新与遗忘机制记忆不是写了就完事定期做衰减和清理才能保证记忆层不会变成垃圾堆。我用的是一套简单的权重更新模型每次被成功召回时access_count 加 1更新 last_accessed_at。遗忘有两个触发条件一是超过 N 天没有被访问并且重要度低于阈值这条记忆就进入候选删除列表二是重要度本身就极低存在时间又长比如超过 180 天直接删除。def touch(self, memory_id): now datetime.utcnow().isoformat() self.conn.execute( UPDATE memories SET access_countaccess_count1, last_accessed_at? WHERE id?, (now, memory_id) ) self.conn.commit() def forget(self, max_age_days180, min_importance3.0): cursor self.conn.execute( SELECT id FROM memories WHERE importance ? AND datetime(last_accessed_at) datetime(now, ?), (min_importance, f-{max_age_days} days) ) stale_ids [row[id] for row in cursor.fetchall()] for mid in stale_ids: self.conn.execute(DELETE FROM memories WHERE id?, (mid,)) self.conn.commit() return len(stale_ids)关于权重计算还有一个更精细的版本把重要度、访问频率、时间衰减结合起来算出一个“活跃分数”按分数排序决定谁排在召回结果最前面。公式大概是score importance * (1 log1p(access_count)) / (1 days_since_last_access / 30)这个公式的思路是基础重要度越高、被访问越多、最近被访问这条记忆的排名就越高。对应到 SQL 里可以结合 strftime 做排序代码量不大但效果比纯 importance 排序好不少。4.6 一个完整调用示例把前面的方法拼起来就是一套完整的 Agent 记忆交互memory SQLiteMemoryLayer(agent_memory.db) # Agent 提取到一条用户偏好 memory.remember( content用户偏好代码注释用中文函数头部要 docstring, memory_typepreference, agent_iddev-assistant, metadata{source: user_input, confirmed: True}, importance8.0 ) # Agent 下次启动召回相关偏好 hits memory.recall_by_search(注释 偏好, agent_iddev-assistant, limit3) for hit in hits: print(hit[content], hit[importance]) memory.touch(hit[id])这就是完整闭环写入、检索、访问回收。整个过程没有请出任何外部服务就是一个 SQLite 文件在工作。5. 什么时候该换回向量数据库诚实版前面说了 SQLite 的诸多好处但我不打算把它吹成万能药。有些场景SQLite 确实顶不住必须引入向量数据库或者至少是向量索引能力。5.1 当记忆库开始变成“知识库”时如果你要处理的不是“用户偏好”这种短小的记忆而是大量长文档型的知识内容比如让 Agent 基于你的团队文档库回答问题那你面对的不再是记忆层而是一个知识库。记忆层的特征是条目小而多、生命周期动态变化、强调准确召回知识库的特征是文档长而大、结构相对稳定、强调语义相关性。知识库场景下embedding 和向量检索是标配SQLite 本身的全文检索能做到关键词匹配但做不到“意思相近但用词不同”的召回。5.2 当精确查询覆盖率太低于时如果实践中发现Agent 的提问有相当大比例无法用精确匹配和关键词匹配命中需要跨主题联想去“回忆”这个信号说明你的记忆层已经超出了 SQLite 的舒适区。比如 Agent 被问到“上次用户提到过的一个跟缓存有关的方案”但记忆里那条原文可能是“之前聊到用 Redis 存热点数据说可以减少 DB 压力”。关键词“缓存”和“方案”都不在原文里SQLite 的 FTS5 未必能召回。这时候向量检索就有不可替代的优势。5.3 数据量到底多大才需要考虑升级很多人担心 SQLite 撑不住大数据量。说实话单表几百万行的普通查询SQLite 完全能扛加了合适的索引和 FTS 全文索引之后日常召回在几万到几十万条这个量级响应时间依然是毫秒级。我认为真正的瓶颈不在行数而在检索模式如果你每天要跑几万个向量相似度查询同时还有高并发写SQLite 的库级写锁会成为瓶颈这时候该考虑更专门的数据库。但纯 Agent 记忆场景单用户或者几十个 Agent 实例通常是看不出这个瓶颈的。5.4 一套简单的选型决策清单我给自己定的决策清单是这样的也分享给你记忆内容平均长度小于 100 字选 SQLite。记忆需要按类型、来源、标签做多维度过滤选 SQLite。记忆需要跨会话精确还原选 SQLite。记忆是长文档、手册、规范需要按语义相似检索考虑向量。记忆检索结果必须支持“相似内容联想”考虑向量。每天记忆写入量超过万级且并发写很多考虑其他数据库。这套清单不严谨但足够实用。它能帮你在五分钟内对整个项目做出初步判断。6. 踩坑记录与调优心得最后这部分是我真正想写的因为全套方案在落地过程中总有一些文档里不会写的问题。6.1 并发写别被“单文件”骗了SQLite 的一个常见坑是写锁是数据库级别的。两个进程同时写总会有一个要等待甚至报 database is locked。Agent 场景里如果你有多个 Agent 实例在后台同时跑都往同一个 SQLite 文件里写记忆就可能撞车。我的解决办法有两个。一是开启 WAL 模式self.conn.execute(PRAGMA journal_modeWAL) self.conn.execute(PRAGMA busy_timeout5000)WAL 模式让读写并行busy_timeout 让写冲突时等待而不是立刻报错。二是如果 Agent 实例数很多可以考虑每个 Agent 一个独立数据库文件用 agent_id 做文件名区分。备份迁移虽然多几个文件但并发冲突从根上消失了。6.2 JSON 函数在旧版本不可用我在一个老项目里吃过亏代码里写了 json_extract本地跑得好好的部署到客户的服务器上直接报错。一查发现那台机器的 SQLite 版本还是 3.31JSON 函数从 3.38 才开始内置。如果你的代码依赖 SQLite 的 JSON 函数要么在启动时检查版本if sqlite3.sqlite_version_info (3, 38, 0): raise RuntimeError(SQLite 3.38 required for JSON functions)要么干脆避免在 SQL 里用 JSON 函数把 metadata 的过滤逻辑放到应用层用 Python 的 json 模块解析。前一种方式干净后一种方式兼容性更好看你的客户端环境情况。6.3 FTS5 与中文分词的简单解法FTS5 的默认分词器对英文很友好对中文就比较粗糙。实测下来“用户偏好注释风格”这种句子默认会被切成单个字比如“注”“释”“风”“格”查询“注释”时能匹配到但相关性排序不准确有时候单字匹配会把不相关的结果捞进来。一个低成本改进是在写入时用 jieba 分词把分词结果用一个空格拼接存到一个 content_segmented 字段然后把这个字段也纳入 FTS5 索引。这样检索“注释风格”时能精确匹配到“注释 风格”这个分词序列准确率提升明显。代价是你多一个依赖但对中文 Agent 场景来说这个投入非常值得。6.4 记忆污染比想象中更严重这是我最想强调的一点。记忆层做得再好如果写入端不做质量控制后面全白搭。Agent 在日常对话中会产生大量噪声信息比如“嗯嗯”、“好的”、临时无意义的闲聊如果不加过滤这些垃圾会占据记忆库的存储空间还会在召回时制造干扰。我在 remember 函数里坚持做三层校验长度过滤、去重、重要度分级。默认低于 2 分的记忆根本不允许入库。这个策略维护起来很简单但能长期保证记忆库的干净度。6.5 只拷贝文件备份时WAL 会坑你在 WAL 模式下数据并不都在主 .db 文件里一部分在 -wal 和 -shm 文件里。如果你直接复制 .db 文件做备份可能把最新的几条记忆落在外面。正确做法是两个要么备份前执行一次 checkpointself.conn.execute(PRAGMA wal_checkpoint(FULL))要么干脆把-wal、-shm文件一起备份。我自己习惯用 checkpoint 的方式因为备份后拿着单个 .db 文件到处跑很方便。至于日常稳定性SQLite 本身非常可靠我跑了小半年没遇到过数据损坏。只要你保持这些细节这套记忆层方案在 Agent 项目里是能长期放心用的。
返回列表