AI Agent记忆系统设计：如何构建有效的短期和长期记忆机制

@dataclass
class MemoryItem:
    content: str          # 记忆内容
    timestamp: float      # Unix时间戳
    importance: float     # 重要度评分(0-1)
    embedding: Optional[np.ndarray] = None  # 向量嵌入
    access_count: int = 0                   # 访问次数
    last_access: float = None              # 最后访问时间
    tags: List[str] = None                 # 标签分类
    source: str = ""                       # 记忆来源
    memory_type: str = "observation"       # 记忆类型

# 向量检索流程示例
class VectorMemory:
    def __init__(self):
        self.vector_store = WeaviateClient()
        self.text_encoder = SentenceTransformer()
    
    def store_memory(self, content: str, metadata: dict):
        embedding = self.text_encoder.encode(content)
        self.vector_store.add_object({
            'content': content,
            'embedding': embedding,
            'metadata': metadata
        })
    
    def retrieve_similar(self, query: str, top_k: int=5):
        query_embedding = self.text_encoder.encode(query)
        return self.vector_store.search(query_embedding, top_k)

class ExperienceReplay:
    def __init__(self, capacity: int = 10000):
        self.memory = deque(maxlen=capacity)
        self.batch_size = 32
    
    def add_experience(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))
    
    def sample_batch(self):
        # 优先采样高回报经验
        indices = self._prioritized_sampling()
        batch = [self.memory[i] for i in indices]
        return self._process_batch(batch)
    
    def _prioritized_sampling(self):
        # 基于TD误差的优先级采样
        priorities = [abs(reward) for _, _, reward, _, _ in self.memory]
        probabilities = priorities / np.sum(priorities)
        return np.random.choice(len(self.memory), self.batch_size, p=probabilities)

class MemoryConsolidator:
    def __init__(self):
        self.cluster_threshold = 0.7  # 语义相似度阈值
    
    def consolidate_cluster(self, memories: List[MemoryItem]) -> MemoryItem:
        """将相似记忆压缩为摘要"""
        if len(memories) < 3:
            return memories[0]  # 小集群不压缩
        
        # 语义聚类
        clusters = self._semantic_clustering(memories)
        
        # 生成摘要
        summary = self._generate_summary(clusters)
        
        # 继承重要度
        max_importance = max(m.importance for m in memories)
        
        return MemoryItem(
            content=summary,
            importance=max_importance * 1.1,  # 压缩记忆提升重要度
            memory_type="consolidated"
        )
    
    def _semantic_clustering(self, memories):
        # 使用DBSCAN进行语义聚类
        embeddings = [m.embedding for m in memories]
        clusterer = DBSCAN(eps=0.3, min_samples=3)
        return clusterer.fit_predict(embeddings)

class MemoryOrchestrator:
    def __init__(self):
        self.stm = ShortTermMemory()      # 短期记忆
        self.episodic = EpisodicMemory()  # 情节记忆
        self.semantic = SemanticMemory()  # 语义记忆
        self.ltm = LongTermMemory()       # 长期记忆
        
    def store(self, memory_type: str, data: dict, metadata: dict):
        """统一存储接口"""
        if memory_type == "short_term":
            return self.stm.store(data, metadata)
        elif memory_type == "episodic":
            return self.episodic.store(data, metadata)
        # ... 其他记忆类型
        
    def retrieve(self, query: str, memory_types: List[str], top_k: int=10):
        """混合检索策略"""
        results = []
        
        # 并行检索不同记忆类型
        for mem_type in memory_types:
            if mem_type == "short_term":
                stm_results = self.stm.retrieve(query, top_k)
                results.extend(stm_results)
            # ... 其他记忆类型
        
        # 综合评分排序
        return self._rank_results(results, query)
    
    def _rank_results(self, results, query):
        """三重权重综合评分"""
        ranked = []
        for result in results:
            # 语义相关性
            relevance = self._semantic_similarity(result.content, query)
            
            # 衰减后重要度
            decayed_importance = result.importance * (0.99 ** days_passed)
            
            # 时效性得分
            recency_score = np.exp(-hours_passed / 24)
            
            # 综合评分
            total_score = (
                0.3 * relevance +           # 语义相关性权重30%
                0.35 * decayed_importance +  # 重要度权重35%
                0.35 * recency_score        # 时效性权重35%
            )
            
            ranked.append((total_score, result))
        
        return sorted(ranked, reverse=True)[:top_k]

# 性能优化参数配置
optimization_config = {
    "max_capacity": 50000,           # 单机最大记忆容量
    "vector_index_type": "HNSW",     # 向量索引类型
    "batch_size": 1000,              # 批处理大小
    "cache_ttl": 3600,               # 缓存过期时间
    "compression_threshold": 30      # 压缩阈值（天）
}