基于向量数据库的Agent记忆系统设计与实现：从短期记忆到长期记忆的演进

1次阅读

没有评论

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

在构建智能 Agent 系统时，记忆管理是一个核心挑战。传统的记忆管理方法通常采用简单的内存缓存或关系型数据库存储，但这些方法存在明显的局限性：

内存占用高 ：随着对话历史的增长，内存缓存会迅速膨胀，导致资源紧张。
检索效率低 ：关系型数据库虽然可以持久化存储，但在语义检索场景下表现不佳，尤其是在处理复杂上下文关联时。
上下文关联弱 ：传统方法难以捕捉对话中的语义关联，导致记忆检索的准确性和相关性不足。

这些痛点促使我们探索更高效的记忆管理方案，尤其是基于向量数据库的混合架构。

在设计 Agent 记忆系统时，我们对比了几种常见的存储方案：

内存缓存 ：如 Redis，适用于短期记忆，速度快但容量有限。
关系型数据库 ：如 PostgreSQL，支持结构化查询，但不擅长语义检索。
向量数据库 ：如 FAISS 或 Pinecone，专为高维向量设计，支持高效的语义检索。

向量数据库因其在语义检索和上下文关联方面的优势，成为长期记忆存储的理想选择。

我们的架构采用分层设计，结合短期记忆和长期记忆：

短期记忆 ：使用内存缓存（如 Redis）存储最近的对话上下文，确保低延迟访问。
长期记忆 ：使用向量数据库存储历史对话的语义向量，支持高效的语义检索。

记忆编码是将对话上下文转化为向量表示的关键步骤。我们使用 Sentence-BERT 等预训练模型生成句子的语义向量：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

def encode_memory(text):
    return model.encode(text)

为了加速记忆检索，我们采用近似最近邻（ANN）算法，如 FAISS 的 IVF 索引或 HNSW 图算法：

import faiss

# 创建 FAISS 索引
dimension = 384  # Sentence-BERT 向量维度
index = faiss.IndexFlatIP(dimension)

# 添加向量到索引
vectors = [encode_memory("Hello, world!")]
index.add(vectors)

# 检索相似记忆
query_vector = encode_memory("Hi there!")
D, I = index.search(query_vector.reshape(1, -1), k=3)

我们使用 Sentence-BERT 模型将文本转换为向量，并支持批量编码：

class MemoryEncoder:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)

    def encode(self, texts):
        return self.model.encode(texts)

基于 FAISS 的向量数据库交互模块，支持动态添加和检索记忆：

class VectorMemoryStore:
    def __init__(self, dimension=384):
        self.index = faiss.IndexFlatIP(dimension)
        self.memories = []

    def add_memory(self, text, vector):
        self.memories.append(text)
        self.index.add(vector.reshape(1, -1))

    def retrieve(self, query_vector, k=3):
        D, I = self.index.search(query_vector.reshape(1, -1), k)
        return [self.memories[i] for i in I[0]]

结合短期记忆和长期记忆的检索策略，优先从短期记忆获取最近上下文：

class MemoryManager:
    def __init__(self, encoder, vector_store, max_short_term=10):
        self.encoder = encoder
        self.vector_store = vector_store
        self.short_term_memories = []
        self.max_short_term = max_short_term

    def add_memory(self, text):
        vector = self.encoder.encode(text)
        self.short_term_memories.append((text, vector))
        if len(self.short_term_memories) > self.max_short_term:
            old_text, old_vector = self.short_term_memories.pop(0)
            self.vector_store.add_memory(old_text, old_vector)

    def retrieve(self, query_text, k=3):
        query_vector = self.encoder.encode(query_text)
        short_term_results = [text for text, _ in self.short_term_memories[-k:]]
        long_term_results = self.vector_store.retrieve(query_vector, k)
        return short_term_results + long_term_results

我们测试了不同规模下的性能表现：