1. 为什么我要给 Agent 做一套长期记忆系统1.1 从一次尴尬的对话说起前段时间我在调试一个自建的对话 Agent场景很简单让它帮我记录一些项目上的零散想法隔天再问它相关细节。结果第二天打开它一脸茫然前一天聊过的内容全部归零。那一刻我意识到一个没有长期记忆的 Agent本质上就是个金鱼脑——每次对话都是第一次见面用户所有的上下文积累全部白费。这不是个例。我接触过不少做 Agent 的开发者大家在前期的重心几乎都放在提示词工程、工具调用、推理链路上记忆模块往往被放到最后甚至直接用一段拼接的字符串糊弄过去。等到对话轮次一多、会话一长问题就集中爆发上下文窗口塞不下、历史信息检索不到、换个设备记忆全丢、想迁移到另一台机器上还得手动导出导入。所以我决定认真给 Hermes Agent 做一套长期记忆系统。核心目标就三个词可持久化、可迁移、可检索。技术选型上我用了 Qdrant 做向量存储、SQLite 的 WAL 模式做结构化元数据、再包一层 MemoryProvider 抽象层把两者粘起来。这套组合不是拍脑袋定的下面我会把每个选择背后的逻辑掰开讲清楚。1.2 这套记忆系统到底解决什么问题先把需求说透不然选型就是空中楼阁。一个 Agent 的长期记忆我拆成了四类需求持久化进程重启、机器重启、甚至换一台设备记忆不能丢。这就要求数据必须落到磁盘而不是躺在内存或某个临时缓存里。可迁移我想把整个记忆库打包带走在另一台机器上原样恢复。这意味着存储格式要足够简单、自包含不能依赖一堆外部服务。语义检索用户问上次聊的那个关于缓存穿透的方案Agent 得能按语义找到相关片段而不是靠关键词硬匹配。这是向量数据库的活。结构化过滤除了语义我还想按时间、按会话 ID、按标签去筛。比如把上周关于部署的讨论都拉出来纯向量检索做不到精确的时间范围过滤得靠结构化存储。这四类需求一摆出来方案轮廓就清晰了向量库负责语义关系库负责结构抽象层负责统一调度。Qdrant 和 SQLite 各司其职MemoryProvider 做门面。1.3 适合谁来参考这套方案这套东西不是给纯小白看的玩具。如果你满足下面任意一条这篇内容应该对你有用正在做 Agent 或对话系统被上下文管理和记忆持久化卡住想给自己的本地应用加一套能记住事的能力但不想引入重型基础设施对向量数据库和嵌入式数据库的组合使用感兴趣想看看真实项目里怎么落地单纯想理解长期记忆这件事在工程上到底要处理哪些坑。我尽量把每一步的操作意图和踩坑经验都写出来代码和配置可以直接抄但更希望你能理解为什么这么设计。2. 整体架构设计与选型逻辑2.1 三层结构向量层、结构层、抽象层整套记忆系统的架构我分成了三层画成文字就是MemoryProvider统一接口层 ├── VectorStore → Qdrant语义检索 └── MetaStore → SQLite WAL结构化元数据 事务MemoryProvider是对外暴露的唯一入口Agent 只跟它打交道不直接碰底层存储。它负责编排写入时同时写向量和元数据检索时先做向量召回再用元数据过滤删除时两边同步清理。Qdrant存的是记忆片段的向量表示附带少量 payload比如记忆 ID、会话 ID。它的强项是近似最近邻搜索能在海量片段里快速找到语义相近的几条。SQLite存的是记忆的完整元数据原始文本、时间戳、会话 ID、标签、向量 ID 的映射关系。用 WAL 模式是为了兼顾读写并发和崩溃恢复。为什么不让 Qdrant 一个库全包了因为 Qdrant 的 payload 过滤虽然能用但做复杂的时间范围查询、多条件组合、事务性更新还是关系库更顺手。而且 SQLite 是单文件迁移的时候直接拷走就行这一点对可迁移目标太重要了。2.2 为什么选 Qdrant 而不是别的向量库向量库这个赛道选择很多我最终选 Qdrant理由有这么几条第一部署形态灵活。Qdrant 既能以服务形式跑也有本地模式开发阶段我可以直接嵌入式启动不用先起一个 Docker 容器。对于我这种想快速验证的场景省事太多。第二payload 过滤能力强。虽然我把结构化查询主要交给 SQLite但 Qdrant 的过滤能力作为一道预筛还是很有用的。比如检索时先限定session_id能显著缩小搜索范围。第三数据格式自包含。Qdrant 的 collection 可以导出成快照文件配合 SQLite 的单文件特性整个记忆库就是一个快照 一个 db 文件迁移时打包带走即可。第四API 设计干净。Python 客户端用起来很顺手upsert、search、delete 这几个核心操作语义清晰没有太多历史包袱。提示如果你已经在用别的向量库比如本地文件型的方案MemoryProvider 这层抽象的意义就体现出来了——换底层实现时上层 Agent 代码一行不用改。2.3 为什么 SQLite 一定要开 WAL 模式很多人用 SQLite 就是默认配置写的时候一卡一卡的多线程下还容易报database is locked。这是没开 WAL 的典型症状。SQLite 默认用的是 rollback journal 模式写操作会锁住整个数据库文件读也得等写完成。而WALWrite-Ahead Logging模式把写操作先追加到-wal文件读操作可以继续读主库读写不互相阻塞。对记忆系统这种一边写新记忆、一边查旧记忆的场景WAL 几乎是必选项。开启方式很简单连接后执行一句PRAGMA journal_modeWAL;但光开 WAL 还不够我还会顺手配几个参数PRAGMA synchronousNORMAL; -- 兼顾安全与性能WAL 下 NORMAL 已经足够稳 PRAGMA busy_timeout5000; -- 遇到锁时等 5 秒再报错而不是立刻失败 PRAGMA foreign_keysON; -- 如果用了外键约束记得打开synchronousNORMAL是个权衡点。FULL最安全但每次写都 fsync慢NORMAL在 WAL 模式下只在 checkpoint 时 fsync崩溃时最多丢最后一个事务对记忆场景完全可接受。这个取舍我后面在常见问题里还会细说。2.4 可迁移性的关键设计可迁移这个目标听起来简单做起来全是细节。我给自己定了三条硬规则所有路径可配置Qdrant 快照路径、SQLite 文件路径、向量维度全部走配置不硬编码。记忆 ID 全局唯一且稳定用 UUID 做主键向量库和关系库共用同一个 ID迁移后映射关系不变。不依赖绝对路径和机器特征迁移脚本只认相对路径和配置文件换机器直接跑。满足这三条迁移就退化成打包目录 → 拷贝 → 解压 → 改配置 → 启动这么简单。3. 核心模块拆解与实操要点3.1 MemoryProvider 接口怎么设计抽象层的接口设计直接决定了后续好不好用。我反复改了几版最后定下来的核心方法就五个class MemoryProvider: def add(self, text: str, session_id: str, tags: list[str] None) - str: 写入一条记忆返回记忆 ID ... def search(self, query: str, top_k: int 5, session_id: str None, time_range: tuple None) - list[dict]: 语义检索支持会话和时间范围过滤 ... def get(self, memory_id: str) - dict: 按 ID 精确取一条 ... def delete(self, memory_id: str) - bool: 删除一条两边同步 ... def migrate(self, target_dir: str) - None: 导出整个记忆库到目标目录 ...设计上有几个刻意的取舍。add 返回 ID而不是返回布尔值因为调用方经常需要拿到 ID 去做后续关联。search 的过滤参数可选不传就是纯语义检索传了就叠加结构化条件。migrate 放在 Provider 层而不是单独写脚本是为了保证导出逻辑和内部存储结构始终一致不会因为底层改了而迁移脚本失配。注意接口一旦定下来尽量别频繁改。Agent 侧代码会大量依赖这些方法签名改一次要动一片。我第一版把search的过滤参数设计成一个大字典后来发现调用方每次都要拼字典很啰嗦才改成显式的session_id和time_range。3.2 向量化环节模型选择与维度对齐记忆要能被语义检索第一步是把文本转成向量。这里有个容易被忽视的坑向量维度必须和 Qdrant collection 的配置严格一致否则写入直接报错。我用的是一套通用的文本嵌入模型输出 768 维向量。选它的理由是体积适中、推理快、对中英文混合文本表现稳定。如果你追求更高精度可以换更大的模型但维度会变collection 得重建。向量化的代码大概长这样def embed(self, text: str) - list[float]: # 实际项目里这里调用嵌入模型 # 关键点确保输出维度与 collection 配置一致 vector self.embedding_model.encode(text) assert len(vector) self.vector_dim, \ f维度不匹配: 期望 {self.vector_dim}, 实际 {len(vector)} return vector.tolist()那个assert不是多余的。我踩过一次坑换了嵌入模型但忘了改 collection 配置结果写入时 Qdrant 报了个很隐晦的维度错误排查了半小时才定位到。加个断言问题在源头就暴露了。维度对齐的实操建议把vector_dim写进配置文件collection 创建时读它嵌入时也读它保证单一数据源。别在两处分别硬编码。3.3 SQLite 表结构设计元数据表我设计得比较克制核心就一张主表加一张标签关联表CREATE TABLE memories ( id TEXT PRIMARY KEY, -- UUID与向量库共用 text TEXT NOT NULL, -- 原始记忆文本 session_id TEXT NOT NULL, -- 会话标识 created_at INTEGER NOT NULL, -- Unix 时间戳 updated_at INTEGER NOT NULL, vector_id TEXT -- 向量库中的 ID通常等于 id ); CREATE INDEX idx_session ON memories(session_id); CREATE INDEX idx_created ON memories(created_at); CREATE TABLE memory_tags ( memory_id TEXT NOT NULL, tag TEXT NOT NULL, PRIMARY KEY (memory_id, tag), FOREIGN KEY (memory_id) REFERENCES memories(id) ON DELETE CASCADE );几个设计细节值得说。时间戳用 INTEGER 存 Unix 秒而不是 TEXT 存日期字符串因为范围查询时整数比较快得多而且没有时区歧义。标签单独建表而不是塞进主表的一个字段是为了能高效地按标签反查也方便后续扩展标签的元信息。外键带 ON DELETE CASCADE删记忆时标签自动清理省得手动维护。提示vector_id我目前让它等于id看起来冗余但保留这个字段是为了将来可能的一条记忆对应多个向量场景比如长文本分块后每块一个向量。现在留个口子以后扩展不用改表结构。3.4 写入流程两边如何保持一致写入是整套系统里最容易出问题的地方因为要同时操作两个存储。我的流程是生成 UUID 作为记忆 ID先写 SQLite元数据此时向量还没写再写 Qdrant向量 payload如果第 3 步失败回滚第 2 步的写入。为什么先写 SQLite因为 SQLite 有事务回滚成本低Qdrant 的写入相对重放后面。如果反过来先写向量向量写成功但元数据失败就会产生孤儿向量——检索能召回但取不到原文很尴尬。def add(self, text, session_id, tagsNone): memory_id str(uuid.uuid4()) now int(time.time()) vector self.embed(text) # 第一步写元数据事务内 try: with self.sqlite_conn: # 自动事务 self.sqlite_conn.execute( INSERT INTO memories (id, text, session_id, created_at, updated_at, vector_id) VALUES (?, ?, ?, ?, ?, ?), (memory_id, text, session_id, now, now, memory_id) ) if tags: self.sqlite_conn.executemany( INSERT INTO memory_tags (memory_id, tag) VALUES (?, ?), [(memory_id, t) for t in tags] ) except Exception as e: raise RuntimeError(f元数据写入失败: {e}) # 第二步写向量 try: self.qdrant.upsert( collection_nameself.collection, points[PointStruct( idmemory_id, vectorvector, payload{session_id: session_id, created_at: now} )] ) except Exception as e: # 回滚元数据 with self.sqlite_conn: self.sqlite_conn.execute(DELETE FROM memories WHERE id ?, (memory_id,)) raise RuntimeError(f向量写入失败已回滚: {e}) return memory_id这段代码的核心思想是补偿式回滚。SQLite 的事务保证了元数据写入的原子性Qdrant 失败时手动删掉刚写的元数据。虽然理论上存在回滚也失败的极端情况但概率极低而且我加了个后台对账任务定期扫描孤儿数据后面会讲。4. 完整实操流程与关键环节实现4.1 环境准备与依赖安装先把依赖装齐。我用的是 Python 生态核心就两个包pip install qdrant-client # sqlite3 是 Python 标准库不用额外装如果你打算用 Qdrant 的服务模式还需要起一个实例但我更推荐开发阶段用本地模式零依赖启动from qdrant_client import QdrantClient # 本地模式数据存到指定目录无需外部服务 client QdrantClient(path./qdrant_data)本地模式的好处是数据直接落盘到./qdrant_data迁移时连同 SQLite 文件一起打包就行。缺点是同一时间只能一个进程访问生产环境如果要多进程并发还是得切到服务模式。注意本地模式和 Python 进程绑定进程退出后数据是持久的但如果你在多个脚本里同时打开同一个路径会报锁冲突。开发时养成用完关掉的习惯。4.2 初始化 collection 与数据库初始化分两步先建 Qdrant collection再建 SQLite 表。from qdrant_client.models import Distance, VectorParams def init_vector_store(self): collections self.qdrant.get_collections().collections names [c.name for c in collections] if self.collection not in names: self.qdrant.create_collection( collection_nameself.collection, vectors_configVectorParams( sizeself.vector_dim, # 必须与嵌入模型输出一致 distanceDistance.COSINE # 文本语义检索常用余弦距离 ) )距离度量我选的是COSINE余弦距离。文本嵌入向量通常关心方向而非长度余弦距离对向量模长不敏感更适合语义相似度。如果你用的是已经归一化的向量点积和余弦等价但显式写 COSINE 更不容易出错。SQLite 初始化就是执行前面那几段建表语句注意连接后先开 WALdef init_meta_store(self): self.sqlite_conn sqlite3.connect(self.db_path, check_same_threadFalse) self.sqlite_conn.execute(PRAGMA journal_modeWAL;) self.sqlite_conn.execute(PRAGMA synchronousNORMAL;) self.sqlite_conn.execute(PRAGMA busy_timeout5000;) self.sqlite_conn.executescript(SCHEMA_SQL)check_same_threadFalse这个参数要留意。SQLite 默认不允许跨线程使用同一个连接但 Agent 场景经常是多线程的所以关掉这个检查。关掉之后要自己保证线程安全我的做法是给写操作加一把锁读操作走 WAL 的并发读能力。4.3 检索流程向量召回 结构化过滤检索是这套系统最能体现价值的地方。完整流程是把查询文本向量化用向量在 Qdrant 里召回 top_k * 3 条候选多召回一些给过滤留余量拿候选 ID 去 SQLite 查元数据应用时间范围、标签等过滤按向量相似度排序返回 top_k 条。def search(self, query, top_k5, session_idNone, time_rangeNone): query_vector self.embed(query) # 构造 Qdrant 过滤条件 qdrant_filter None if session_id: qdrant_filter Filter( must[FieldCondition(keysession_id, matchMatchValue(valuesession_id))] ) # 多召回给后续过滤留空间 hits self.qdrant.search( collection_nameself.collection, query_vectorquery_vector, limittop_k * 3, query_filterqdrant_filter ) if not hits: return [] ids [h.id for h in hits] score_map {h.id: h.score for h in hits} # 去 SQLite 补全元数据并做结构化过滤 placeholders ,.join(? * len(ids)) sql fSELECT id, text, session_id, created_at FROM memories WHERE id IN ({placeholders}) params list(ids) if time_range: sql AND created_at ? AND created_at ? params.extend(time_range) rows self.sqlite_conn.execute(sql, params).fetchall() results [ {id: r[0], text: r[1], session_id: r[2], created_at: r[3], score: score_map[r[0]]} for r in rows ] results.sort(keylambda x: x[score], reverseTrue) return results[:top_k]这里有个性能上的关键点为什么先向量召回再过滤而不是先过滤再向量检索因为结构化过滤后的集合可能很小但向量检索本身需要遍历索引先过滤反而可能破坏向量索引的效率。先召回再过滤是用向量索引快速缩小范围再用关系库精确筛选的组合拳。top_k * 3这个倍数不是随便定的。如果过滤条件很严格召回 3 倍能保证过滤后还有足够的候选。实测下来 3 倍是个不错的平衡点召回太多会拖慢 SQLite 查询太少又可能过滤后不够数。4.4 迁移实现打包与恢复迁移是我最看重的功能实现上分导出和导入两个动作。导出就是把 Qdrant 快照和 SQLite 文件连同 WAL 文件拷到目标目录再写一份配置清单def migrate(self, target_dir): os.makedirs(target_dir, exist_okTrue) # 1. 导出 Qdrant 快照 snapshot_path self.qdrant.create_snapshot(collection_nameself.collection) shutil.copy(snapshot_path, os.path.join(target_dir, vectors.snapshot)) # 2. 强制 checkpoint把 WAL 内容合并进主库 self.sqlite_conn.execute(PRAGMA wal_checkpoint(TRUNCATE);) shutil.copy(self.db_path, os.path.join(target_dir, memories.db)) # 3. 写配置清单 manifest { vector_dim: self.vector_dim, collection: self.collection, exported_at: int(time.time()), version: 1.0 } with open(os.path.join(target_dir, manifest.json), w) as f: json.dump(manifest, f, indent2)导入则是反过来读配置清单恢复 Qdrant 快照把 db 文件放回指定路径。def restore(self, source_dir): with open(os.path.join(source_dir, manifest.json)) as f: manifest json.load(f) # 校验维度一致性 if manifest[vector_dim] ! self.vector_dim: raise ValueError(向量维度不匹配无法恢复) # 恢复 SQLite shutil.copy(os.path.join(source_dir, memories.db), self.db_path) # 恢复 Qdrant 快照 self.qdrant.recover_snapshot( collection_nameself.collection, locationos.path.join(source_dir, vectors.snapshot) )那个wal_checkpoint(TRUNCATE)是迁移前的关键动作。WAL 模式下最新的写入可能还在-wal文件里没合并进主库如果只拷主库文件会丢数据。checkpoint 把 WAL 内容刷进主库并清空 WAL保证拷贝的是完整数据。提示迁移前最好先停掉所有写入。虽然 checkpoint 能保证一致性但如果在拷贝过程中还有新写入可能拷到一半的数据。生产环境迁移我会先让 Agent 进入只读模式。4.5 后台对账清理孤儿数据前面提到写入可能产生孤儿数据向量写了但元数据没写或反过来。我加了个轻量的对账任务定期跑一次def reconcile(self): # 找出 SQLite 里有、Qdrant 里没有的 sqlite_ids set(r[0] for r in self.sqlite_conn.execute(SELECT id FROM memories).fetchall()) qdrant_ids set() offset None while True: points, offset self.qdrant.scroll( collection_nameself.collection, limit1000, offsetoffset ) qdrant_ids.update(p.id for p in points) if offset is None: break orphan_meta sqlite_ids - qdrant_ids # 元数据孤儿 orphan_vec qdrant_ids - sqlite_ids # 向量孤儿 # 清理策略元数据孤儿直接删向量孤儿也删 for mid in orphan_meta: with self.sqlite_conn: self.sqlite_conn.execute(DELETE FROM memories WHERE id ?, (mid,)) if orphan_vec: self.qdrant.delete( collection_nameself.collection, points_selectorlist(orphan_vec) ) return {meta_orphans: len(orphan_meta), vec_orphans: len(orphan_vec)}对账不用跑太频繁我一般设成每天凌晨跑一次或者手动触发。它的价值在于兜底——正常流程下不该有孤儿但一旦出现比如进程被强杀对账能自动修复不用人工介入。5. 常见问题与排查技巧实录5.1 写入报维度不匹配这是新手最容易撞的墙。症状是 Qdrant 抛出一个关于 vector size 的错误。根因通常是嵌入模型换了但 collection 没重建或者配置里vector_dim和实际模型输出对不上。排查顺序先打印实际嵌入向量的长度再查 collection 的配置维度两者一比就知道。修复方式是删掉 collection 重建或者换回匹配的模型。我前面在embed里加的断言就是为了让这个问题在写入前就暴露。5.2 database is locked 频繁出现WAL 模式下这个错误应该很少见如果还频繁出现检查三点一是busy_timeout是否设了没设的话遇到锁立刻报错二是是否有长事务一直不提交把写锁占着三是check_same_threadFalse之后有没有自己做线程同步多线程同时写同一个连接会出问题。我的做法是给所有写操作套一把threading.Lock读操作不加锁WAL 支持并发读。这样既保证了写安全又不牺牲读性能。5.3 检索结果不相关向量检索召回不准八成是嵌入模型的问题不是 Qdrant 的问题。先确认模型是否适合你的语言和领域——通用模型在专业领域比如医疗、法律表现会打折。其次检查文本预处理如果写入时做了清洗但查询时没做两边不一致会导致语义偏移。还有一个隐蔽的坑距离度量选错。如果向量没归一化却用了点积或者该用余弦却用了欧氏排序会乱。文本场景我统一用 COSINE基本不会错。5.4 迁移后检索不到旧记忆迁移后如果新记忆能查、旧记忆查不到多半是 Qdrant 快照没恢复成功或者恢复到了错误的 collection 名。检查manifest.json里的 collection 名和当前配置是否一致。另一个可能是 SQLite 恢复了但 Qdrant 没恢复导致向量孤儿被对账任务清掉了——所以迁移后先别急着跑对账确认两边数据都在再说。5.5 常见问题速查表问题现象可能原因排查动作解决方式写入报维度错误模型与 collection 维度不一致打印向量长度对比配置重建 collection 或换模型database is locked未设 busy_timeout / 长事务查 PRAGMA 配置和事务设 timeout缩短事务检索结果不相关模型不适配 / 度量选错检查模型和 distance 配置换模型统一用 COSINE迁移后旧记忆丢失快照未恢复 / 对账误删查两边数据量先恢复再对账进程重启记忆丢失用了内存模式检查存储路径配置切到本地持久化模式多线程写入报错连接跨线程未同步检查线程模型写操作加锁5.6 几条踩坑换来的经验第一别在写入路径上做重活。我一开始在add里同步调用嵌入模型模型推理慢的时候整个写入卡住。后来改成批量写入 异步嵌入吞吐上来了。记忆写入不是实时性要求极高的操作完全可以攒一批再处理。第二向量库和关系库的 ID 一定要共用。我见过有人两边各生成一套 ID靠一个映射表关联结果映射表一丢全乱套。共用 UUID 是最省心的做法。第三迁移脚本要定期演练。别等到真要迁移了才发现脚本跑不通。我每个月会做一次导出到临时目录再恢复的演练确保迁移链路始终可用。第四给记忆加个重要性字段。这是我后来加的检索时按重要性加权避免一些琐碎的闲聊记忆挤掉真正重要的内容。实现上就是在 SQLite 加一列检索排序时把score * importance_weight作为最终排序依据。第五定期做数据量监控。记忆是只增不减的时间长了 SQLite 和 Qdrant 都会膨胀。我加了个统计任务每周报一次记忆总数和存储占用超过阈值就提醒清理或归档。这套系统跑了大半年从最初的几百条记忆到现在几万条检索延迟依然稳定在几十毫秒。真正让我省心的是那层 MemoryProvider 抽象——中间我换过一次嵌入模型只改了配置和重建 collectionAgent 侧代码一行没动。如果你也在给 Agent 做记忆强烈建议先把这层抽象立起来后面所有的演进都会轻松很多。