Agent记忆系统实战：基于向量数据库的长期记忆存储与检索优化

1次阅读

没有评论

共计 2119 个字符，预计需要花费 6 分钟才能阅读完成。

在构建具备长期记忆能力的 Agent 系统时，开发团队通常会遇到三个典型问题：

上下文窗口爆炸：当对话历史超过 10 轮后，直接将所有历史记录拼接到 prompt 中会导致 API 调用成本激增
模糊检索失效：使用传统数据库的 LIKE 操作符无法有效匹配 ” 用户去年提过的那个图像处理需求 ” 这类语义查询
记忆碎片化：简单的 KV 存储难以建立记忆之间的语义关联，比如 ” 用户喜欢猫 ” 和 ” 用户养过布偶猫 ” 应当被自动关联

对比传统关系型数据库的局限性，向量数据库展现出三大核心能力：

高维数据处理：典型文本嵌入向量的 768~1024 维特征空间，比 B 树索引更适合表达语义
近似最近邻 (ANN) 搜索：在亿级数据集中实现毫秒级检索，比精确搜索快 1000 倍以上
动态 schema 支持：可随时添加新的记忆字段（如时间戳、置信度）而不需要重建索引

实测数据显示，对于 100 万条记忆记录：

方案	QPS	延迟(ms)	内存占用
PostgreSQL+pgvector	1200	8.2	12GB
FAISS(IVF2048)	8500	1.1	3.2GB
Milvus(HNSW)	6200	1.8	4.7GB

推荐使用多语言 sentence-transformers 模型，平衡性能与精度：

from sentence_transformers import SentenceTransformer

# 建议初始化时预加载模型
encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2', 
                             device='cuda' if torch.cuda.is_available() else 'cpu')

def embed_text(text: str) -> np.ndarray:
    """生成标准化后的嵌入向量"""
    vector = encoder.encode(text, normalize_embeddings=True)
    return vector.astype(np.float32)  # 多数向量数据库要求 float32

根据数据规模选择适当算法：

小规模数据(<1M)：HNSW 提供最佳召回率
配置示例：index = faiss.IndexHNSWFlat(dim, 32)
中大规模数据(1M-100M)：IVF+PQ 组合优化
典型设置：nlist=4096, m=32, nbits=8
超大规模(>100M)：分布式 Milvus 集群
需要配置 shard 和 query 节点比例

带相似度阈值和元数据过滤的检索示例：

def search_memories(query: str, top_k: int = 3, threshold: float = 0.65) -> List[dict]:
    """
    参数:
        threshold: 最小余弦相似度阈值
    """
    query_vec = embed_text(query)

    # FAISS 搜索
    distances, ids = index.search(np.array([query_vec]), top_k)

    # 结果过滤
    results = []
    for i in range(top_k):
        if distances[0][i] < threshold:
            continue

        memory_id = ids[0][i]
        raw_memory = redis_client.hgetall(f"memory:{memory_id}")
        results.append({'content': raw_memory['text'],
            'score': float(distances[0][i]),
            'last_accessed': raw_memory['timestamp']
        })

    return sorted(results, key=lambda x: x['score'], reverse=True)

查询加速：对高频记忆建立独立的小型 HNSW 索引
内存控制：使用 OPQ 量化将 768 维向量压缩到 64 字节
冷启动方案：
预生成热门问题的标准回答向量
采用层级索引（先粗筛后精搜）

维度对齐：确保所有写入向量的维度数与索引创建时一致
精度权衡：召回率每提升 5%，搜索延迟可能增加 2 - 3 倍
数据预热 ：定期运行index.reconstruct_n(0, index.ntotal) 避免内存碎片

当前实现仍存在两个关键挑战：

跨会话关联：需要引入图数据库建立记忆节点间的显式关系

动态权重：考虑实现基于访问频率和时效性的衰减函数：

def decay_weight(original_score: float, 
               last_access_time: int, 
               half_life: int = 30*24*3600) -> float:
    """按时间半衰期衰减记忆权重"""
    elapsed = time.time() - last_access_time
    return original_score * (0.5 ** (elapsed / half_life))

实际部署中，建议从每天 1 亿次检索的中等规模开始验证，逐步优化 ANN 参数。对于需要严格一致性的场景，可以结合 PostgreSQL 的 ACID 特性实现双写机制。

正文完