Agent长期记忆实战：基于向量数据库的实现原理与避坑指南

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

背景痛点

传统 Agent 系统在处理多轮对话或长期任务时，常因缺乏记忆能力导致重复学习或上下文丢失。例如用户昨天说『我喜欢吃辣』，今天询问餐厅推荐时，Agent 却需要重新确认口味偏好。这种『金鱼脑』行为严重影响用户体验。

技术选型对比

核心实现

1. Embedding 生成

from sentence_transformers import SentenceTransformer

# 建议初始化时指定模型维度
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2', device='cpu')

def generate_embedding(text: str) -> np.ndarray:
    """生成 384 维的文本向量"""
    return model.encode(text, normalize_embeddings=True)

2. 向量存储与检索

import faiss
import numpy as np

# 创建索引时需显式定义维度
d = 384  # 必须与 embedding 维度一致
index = faiss.IndexFlatIP(d)  # 内积相似度

# 添加向量时需类型检查
def add_memory(vector: np.ndarray, metadata: dict):
    assert vector.shape == (d,), f"维度必须为{d}"
    index.add(np.expand_dims(vector, axis=0))
    # 同时存储关联元数据...

3. 记忆更新策略

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key):
        if key not in self.cache:
            return None
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

性能优化

索引类型对比（测试数据）

分布式部署建议

• 使用 Milvus 的读写分离架构
• 对热数据采用内存缓存
• 批量写入时开启auto_flush=False

避坑指南

1. 维度不一致问题
2. 解决方案：在 add_memory() 方法中添加严格校验

3. 错误示例：faiss.AssertionError: expected 384 dimensions, got 512

4. 冷启动优化

5. 预加载高频问答对
6. 使用 faiss.read_index() 加载已有索引

延伸思考

当记忆条目达到百万级时，如何平衡：
– 检索精度（召回率）
– 存储成本（内存 /SSD）
– 实时性要求（延迟）

欢迎在评论区分享你的实践经验！