Agent长期记忆从入门到实战：如何构建高效稳定的记忆存储系统

1次阅读

没有评论

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

在构建智能 Agent 时，长期记忆系统是其核心能力之一。它使 Agent 能够记住过去的交互、学习到的知识和用户偏好，从而实现更连贯、个性化的服务。然而，开发者在实现这一功能时常常面临以下挑战：

记忆碎片化：Agent 在不同时间点获取的信息可能分散存储，难以形成连贯的知识体系。
检索效率低：随着记忆量的增长，快速找到相关信息变得困难。
数据一致性问题：在多线程或分布式环境下，如何保证记忆的准确更新和同步。
冷启动问题：新 Agent 缺乏初始记忆，导致早期交互质量不佳。

优点：
– 天然适合存储和检索 Embedding 向量
– 支持近似最近邻搜索，检索效率高
– 可扩展性强，适合大规模记忆存储

缺点：
– 对结构化数据支持有限
– 需要额外的元数据管理

优点：
– 擅长表示实体间复杂关系
– 支持图遍历查询，适合推理场景
– 数据模型直观

缺点：
– 向量搜索性能不如专用向量数据库
– 大规模图查询可能变慢

优点：
– ACID 特性保证数据一致性
– 成熟的查询语言和工具生态
– 适合结构化数据

缺点：
– 不适合高维向量操作
– 扩展性受限

选型建议：对于以语义搜索为主的 Agent，向量数据库是最佳选择；若需要复杂推理，可考虑图数据库；关系型数据库适合辅助存储结构化元数据。

from typing import List, Dict
import numpy as np

class MemoryItem:
    """单个记忆项的数据结构"""
    def __init__(self, content: str, embedding: np.ndarray, metadata: Dict):
        self.content = content  # 原始内容
        self.embedding = embedding  # 向量表示
        self.metadata = metadata  # 时间戳、来源等元数据
        self.access_count = 0  # 访问计数
        self.last_accessed = 0  # 最后访问时间戳

class MemorySystem:
    """记忆系统核心类"""
    def __init__(self, vector_db):
        self.vector_db = vector_db  # 向量数据库客户端
        self.memories = []  # 本地记忆缓存

    def add_memory(self, content: str, embedding_model):
        """添加新记忆"""
        embedding = embedding_model.encode(content)
        metadata = {'timestamp': time.time(), 'source': 'user_input'}
        memory = MemoryItem(content, embedding, metadata)

        # 存储到向量数据库
        self.vector_db.upsert([memory.embedding], [memory.metadata])
        self.memories.append(memory)
        return memory

将查询转换为 Embedding 向量
在向量数据库执行近似最近邻搜索
按相关性分数排序结果
返回 Top- K 相关记忆

def retrieve(self, query: str, embedding_model, top_k=5) -> List[MemoryItem]:
    """检索相关记忆"""
    query_embedding = embedding_model.encode(query)

    # 向量数据库搜索
    results = self.vector_db.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )

    # 更新访问记录
    for mem in self.memories:
        if mem.metadata['id'] in [r['id'] for r in results]:
            mem.access_count += 1
            mem.last_accessed = time.time()

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