Agent长期记忆的工程实践：基于向量数据库的实现与优化

1次阅读

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

在构建具备长期记忆能力的 Agent 系统时，传统关系型数据库面临几个关键挑战：

模糊匹配效率低 ：Agent 需要根据语义相似度检索记忆，而 SQL 的 LIKE 或全文索引难以理解上下文
上下文关联困难 ：关系型数据库无法自然表达记忆之间的语义关联（如 ” 咖啡 ” 和 ” 提神 ” 的关系）
高维数据处理不足 ：记忆的向量表示通常是 768+ 维度的，传统索引结构（如 B +Tree）性能急剧下降

主流向量数据库对比（以 Agent 场景为评估重点）：

产品	优势	局限性	适用场景
Pinecone	全托管服务，自动优化索引	成本较高，定制化能力有限	快速原型开发
Milvus	支持多种索引算法，开源可定制	运维复杂度高	需要深度调优的生产环境
Weaviate	内置 NLP 模块，支持混合查询	社区版功能受限	语义搜索优先的场景

选型建议 ：
– 早期验证阶段推荐 Pinecone
– 大规模生产环境建议 Milvus+ 自建集群
– 需要结合结构化查询时选择 Weaviate

推荐使用预训练模型生成 embedding，关键考虑因素：

模型选择 ：
通用场景：text-embedding-ada-002（OpenAI）
中文优先：paraphrase-multilingual-MiniLM-L12-v2（HuggingFace）

# 使用 OpenAI 生成 embedding 的示例
import openai

def get_embedding(text: str) -> list[float]:
    response = openai.Embedding.create(
        input=text,
        model="text-embedding-ada-002"
    )
    return response['data'][0]['embedding']

HNSW（Hierarchical Navigable Small World）选择依据 ：

比 IVF 更高的召回率（98%+）
支持增量更新，适合 Agent 的持续学习
查询复杂度 O(log n)，适合高频检索

配置建议：

# Milvus 索引配置示例
index_params = {
    "metric_type": "L2",
    "index_type": "HNSW",
    "params": {
        "M": 16,     # 影响连接数
        "efConstruction": 200  # 影响构建质量
    }
}

查询重写 ：将用户输入扩展为同义词（如 ” 记性不好 ”→[“ 记忆 ”,” 遗忘 ”]）
混合查询 ：结合 metadata 过滤（如时间范围）缩小搜索空间
缓存策略 ：对高频查询的 embedding 做 LRU 缓存

import pymilvus
from tenacity import retry, stop_after_attempt

class MemoryManager:
    def __init__(self):
        self.collection = pymilvus.Collection("agent_memories")

    @retry(stop=stop_after_attempt(3))
    def store_memory(self, text: str, metadata: dict):
        try:
            embedding = get_embedding(text)
            data = [[embedding],          # vector field
                [metadata["timestamp"]],  # scalar fields
                [metadata["importance"]]
            ]
            self.collection.insert(data)
        except Exception as e:
            logging.error(f"存储失败: {str(e)}")
            raise

def recall_memories(query: str, top_k=5, threshold=0.7):
    query_embedding = get_embedding(query)
    search_params = {
        "metric_type": "L2",
        "params": {"ef": 32}  # 搜索时遍历的节点数
    }

    results = collection.search([query_embedding],
        "vector_field",
        search_params,
        limit=top_k,
        output_fields=["text", "importance"]
    )

    # 应用相似度阈值过滤
    return [(hit.entity.text, hit.score)
        for hit in results[0]
        if 1 - hit.score > threshold  # 转换为余弦相似度
    ]