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 模式，写操作会锁住整个数据库文件，读也得等写完成。而WAL（Write-Ahead Logging）模式把写操作先追加到-wal文件，读操作可以继续读主库，读写不互相阻塞。对记忆系统这种"一边写新记忆、一边查旧记忆"的场景，WAL 几乎是必选项。

开启方式很简单，连接后执行一句：

PRAGMA journal_mode=WAL;

但光开 WAL 还不够，我还会顺手配几个参数：

PRAGMA synchronous=NORMAL; -- 兼顾安全与性能，WAL 下 NORMAL 已经足够稳 PRAGMA busy_timeout=5000; -- 遇到锁时等 5 秒再报错，而不是立刻失败 PRAGMA foreign_keys=ON; -- 如果用了外键约束，记得打开

synchronous=NORMAL是个权衡点。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 写入流程：两边如何保持一致

写入是整套系统里最容易出问题的地方，因为要同时操作两个存储。我的流程是：

1. 生成 UUID 作为记忆 ID；
2. 先写 SQLite（元数据），此时向量还没写；
3. 再写 Qdrant（向量 + payload）；
4. 如果第 3 步失败，回滚第 2 步的写入。

为什么先写 SQLite？因为 SQLite 有事务，回滚成本低；Qdrant 的写入相对"重"，放后面。如果反过来先写向量，向量写成功但元数据失败，就会产生"孤儿向量"——检索能召回，但取不到原文，很尴尬。

def add(self, text, session_id, tags=None): 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_name=self.collection, points=[PointStruct( id=memory_id, vector=vector, 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_name=self.collection, vectors_config=VectorParams( size=self.vector_dim, # 必须与嵌入模型输出一致 distance=Distance.COSINE # 文本语义检索常用余弦距离 ) )

距离度量我选的是COSINE（余弦距离）。文本嵌入向量通常关心方向而非长度，余弦距离对向量模长不敏感，更适合语义相似度。如果你用的是已经归一化的向量，点积和余弦等价，但显式写 COSINE 更不容易出错。

SQLite 初始化就是执行前面那几段建表语句，注意连接后先开 WAL：

def init_meta_store(self): self.sqlite_conn = sqlite3.connect(self.db_path, check_same_thread=False) self.sqlite_conn.execute("PRAGMA journal_mode=WAL;") self.sqlite_conn.execute("PRAGMA synchronous=NORMAL;") self.sqlite_conn.execute("PRAGMA busy_timeout=5000;") self.sqlite_conn.executescript(SCHEMA_SQL)

check_same_thread=False这个参数要留意。SQLite 默认不允许跨线程使用同一个连接，但 Agent 场景经常是多线程的，所以关掉这个检查。关掉之后要自己保证线程安全，我的做法是给写操作加一把锁，读操作走 WAL 的并发读能力。

4.3 检索流程：向量召回 + 结构化过滤

检索是这套系统最能体现价值的地方。完整流程是：

1. 把查询文本向量化；
2. 用向量在 Qdrant 里召回 top_k * 3 条候选（多召回一些，给过滤留余量）；
3. 拿候选 ID 去 SQLite 查元数据，应用时间范围、标签等过滤；
4. 按向量相似度排序，返回 top_k 条。

def search(self, query, top_k=5, session_id=None, time_range=None): query_vector = self.embed(query) # 构造 Qdrant 过滤条件 qdrant_filter = None if session_id: qdrant_filter = Filter( must=[FieldCondition(key="session_id", match=MatchValue(value=session_id))] ) # 多召回，给后续过滤留空间 hits = self.qdrant.search( collection_name=self.collection, query_vector=query_vector, limit=top_k * 3, query_filter=qdrant_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 = f"SELECT 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(key=lambda x: x["score"], reverse=True) 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_ok=True) # 1. 导出 Qdrant 快照 snapshot_path = self.qdrant.create_snapshot(collection_name=self.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, indent=2)

导入则是反过来：读配置清单，恢复 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_name=self.collection, location=os.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_name=self.collection, limit=1000, offset=offset ) 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_name=self.collection, points_selector=list(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_thread=False之后有没有自己做线程同步，多线程同时写同一个连接会出问题。

我的做法是给所有写操作套一把threading.Lock，读操作不加锁（WAL 支持并发读）。这样既保证了写安全，又不牺牲读性能。

5.3 检索结果不相关

向量检索召回不准，八成是嵌入模型的问题，不是 Qdrant 的问题。先确认模型是否适合你的语言和领域——通用模型在专业领域（比如医疗、法律）表现会打折。其次检查文本预处理，如果写入时做了清洗但查询时没做，两边不一致会导致语义偏移。

还有一个隐蔽的坑：距离度量选错。如果向量没归一化却用了点积，或者该用余弦却用了欧氏，排序会乱。文本场景我统一用 COSINE，基本不会错。

5.4 迁移后检索不到旧记忆

迁移后如果新记忆能查、旧记忆查不到，多半是 Qdrant 快照没恢复成功，或者恢复到了错误的 collection 名。检查manifest.json里的 collection 名和当前配置是否一致。另一个可能是 SQLite 恢复了但 Qdrant 没恢复，导致向量孤儿被对账任务清掉了——所以迁移后先别急着跑对账，确认两边数据都在再说。

5.5 常见问题速查表

5.6 几条踩坑换来的经验

第一，别在写入路径上做重活。我一开始在add里同步调用嵌入模型，模型推理慢的时候整个写入卡住。后来改成批量写入 + 异步嵌入，吞吐上来了。记忆写入不是实时性要求极高的操作，完全可以攒一批再处理。

第二，向量库和关系库的 ID 一定要共用。我见过有人两边各生成一套 ID，靠一个映射表关联，结果映射表一丢全乱套。共用 UUID 是最省心的做法。

第三，迁移脚本要定期演练。别等到真要迁移了才发现脚本跑不通。我每个月会做一次"导出到临时目录再恢复"的演练，确保迁移链路始终可用。

第四，给记忆加个"重要性"字段。这是我后来加的，检索时按重要性加权，避免一些琐碎的闲聊记忆挤掉真正重要的内容。实现上就是在 SQLite 加一列，检索排序时把score * importance_weight作为最终排序依据。

第五，定期做数据量监控。记忆是只增不减的，时间长了 SQLite 和 Qdrant 都会膨胀。我加了个统计任务，每周报一次记忆总数和存储占用，超过阈值就提醒清理或归档。

这套系统跑了大半年，从最初的几百条记忆到现在几万条，检索延迟依然稳定在几十毫秒。真正让我省心的是那层 MemoryProvider 抽象——中间我换过一次嵌入模型，只改了配置和重建 collection，Agent 侧代码一行没动。如果你也在给 Agent 做记忆，强烈建议先把这层抽象立起来，后面所有的演进都会轻松很多。