Agent 的长期记忆如何设计？向量库和 KV 存储如何配合？

长期记忆的设计，最忌讳的就是“一个向量库梭哈”。实际落地时，我常用的组合是 “向量库做索引，KV 存储做正本”，用分离的方式兼顾语义检索的灵活性和高速读取的稳定性。

---

🎯 核心设计理念：索引与存储分离

你可以把向量库想象成图书馆的 “主题索引卡片”，只记录“哪本书在哪个架位”；而 KV 存储就是 “书架本身”，存的是书的完整内容。

这样分工的好处很直接：检索时轻量化，取数据时零解析损耗。

💻 实现示例：MemoryManager

下面这段代码演示了如何用 Chroma（向量库） 配合 Redis（KV） 构建一个完整的记忆存取流。

import chromadb
import redis
import json
import uuid
from datetime import datetime
from sentence_transformers import SentenceTransformer

class LongTermMemory:
    def __init__(self, redis_host='localhost', redis_port=6379):
        # 向量库：只存 embedding + 元数据 + 指向 KV 的 key
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection("memories")
        # KV 存储：存完整记忆体
        self.kv = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
        # 嵌入模型
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

    def store(self, content: str, metadata: dict = None):
        """存储一条长期记忆，返回记忆 ID"""
        mem_id = str(uuid.uuid4())
        # 生成向量
        embedding = self.encoder.encode(content).tolist()
        # 构造 KV 中的完整对象
        mem_object = {
            "content": content,
            "metadata": metadata or {},
            "created_at": datetime.now().isoformat()
        }
        # ① 先存入 KV（正本）
        self.kv.set(mem_id, json.dumps(mem_object))
        # ② 再存入向量库（索引），注意只存元数据和 ID，不存大段文本
        self.collection.add(
            embeddings=[embedding],
            documents=[content],          # 可选，用于某些向量库自带的文本展示
            metadatas=[{"mem_id": mem_id}],
            ids=[mem_id]
        )
        return mem_id

    def retrieve(self, query: str, top_k: int = 3):
        """语义检索相关记忆，返回完整对象列表"""
        # ① 向量检索，拿到最相似的 ID
        q_emb = self.encoder.encode(query).tolist()
        results = self.collection.query(query_embeddings=[q_emb], n_results=top_k)
        # 从 metadatas 中提取 mem_id 列表
        mem_ids = []
        for meta in results.get('metadatas', [[]])[0]:
            if meta and 'mem_id' in meta:
                mem_ids.append(meta['mem_id'])

        # ② 用 pipeline 批量从 KV 拉取完整记忆
        pipe = self.kv.pipeline()
        for mid in mem_ids:
            pipe.get(mid)
        full_objects = pipe.execute()

        # 解析 JSON
        memories = []
        for obj in full_objects:
            if obj:
                memories.append(json.loads(obj))
        return memories

调用示例：

memory = LongTermMemory()
memory.store("用户偏好 Python 代码生成，风格偏简洁。")
memory.store("上次用户要求生成一个图像分类器，使用了 PyTorch。")

results = memory.retrieve("需要写一段机器学习代码")
# 会返回上面两条相关记忆，完整的文本都在里面

🧩 为什么这样配合？三个实打实的好处

查看内嵌表格

另外，当记忆体特别大（比如整篇对话记录、多模态描述）时，向量库只放摘要的 embedding，KV 放原始数据，这种架构天然支持 “轻索引+重正本”，避免向量库被撑爆。

---

⚠️ 实际落地必踩的两个坑

1. 一致性问题 向量库里删了索引，但 KV 里的正本忘记删，变成“幽灵记忆”；或者反过来。 解法：统一用 store() 和 delete() 方法管理，内部双写并用事务（或至少先写 KV 再写索引，保证能读回来）。

2. 过期与遗忘策略 向量库通常不支持 TTL（生存时间），但 KV 可以。如果某类记忆应该 30 天后失效，你需要在检索后多一步判断：从 KV 取出后，检查 created_at 或通过 Redis 自带的 TTL 做逻辑删除。 建议：在 retrieve() 里加入时间衰减过滤，不直接依赖底层物理删除。

---

🔄 从“向量+KV”到“多模态检索”的演进

如果业务发展到不仅靠语义，还需要按时间范围、用户 ID 等精确条件过滤，那么我们会在向量检索前加一层 关系数据库（如 PostgreSQL），用 SQL 做粗筛，再拿种子 ID 进向量库做相似度排序。

这时候就不是“向量库+KV”两兄弟了，而是 “SQL 做定位，向量做排序，KV 做正本” 的三层结构——长期记忆的设计，终究是要跟着业务复杂度一起长的。

---