4.3 记忆检索与更新机制

C --> G[向量索引]
C --> H[倒排索引]
C --> I[图索引]
G --> J[相似度搜索]
H --> J
I --> J
​

</div>

### 记忆更新模式

1. **增量更新**：只更新变化的部分，保持整体结构稳定
2. **批量更新**：定期批量处理所有记忆条目
3. **实时更新**：每次访问时立即更新相关信息
4. **异步更新**：后台异步处理更新任务

## 分步实战

### 步骤 1：向量搜索引擎设计与实现

```python
class MemorySearchEngine:
    """记忆搜索引擎"""
    
    def __init__(self, embedding_dim: int = 768, index_type: str = "hnsw"):
        self.embedding_dim = embedding_dim
        self.index_type = index_type
        self.vector_index = None
        self.text_index = {}
        self.metadata_index = {}
        self.memory_store = {}
        
        # 初始化索引
        self._init_index()
    
    def _init_index(self):
        """初始化搜索索引"""
        if self.index_type == "hnsw":
            # 使用FAISS的HNSW索引
            self.vector_index = faiss.IndexHNSWFlat(self.embedding_dim, 32)
        elif self.index_type == "ivf":
            # 使用FAISS的IVF索引
            nlist = 100  # 聚类中心数量
            quantizer = faiss.IndexFlatL2(self.embedding_dim)
            self.vector_index = faiss.IndexIVFFlat(quantizer, self.embedding_dim, nlist)
            # 需要先训练索引
            self._train_index()
    
    def _train_index(self):
        """训练向量索引"""
        # 生成训练数据（实际应用中应该使用真实数据）
        training_data = np.random.rand(1000, self.embedding_dim).astype('float32')
        self.vector_index.train(training_data)
    
    def add_memory(self, memory_id: str, content: str, embedding: np.ndarray,
                   metadata: Dict[str, Any] = None):
        """添加记忆到搜索引擎"""
        # 存储原始数据
        self.memory_store[memory_id] = {
            'content': content,
            'embedding': embedding,
            'metadata': metadata or {},
            'timestamp': datetime.now(),
            'access_count': 0,
            'last_access': datetime.now(),
            'version': 1
        }
        
        # 添加到向量索引
        if self.vector_index is not None:
            self.vector_index.add(embedding.reshape(1, -1))
        
        # 构建文本索引
        self._update_text_index(memory_id, content)
        
        # 构建元数据索引
        self._update_metadata_index(memory_id, metadata or {})
    
    def search_similar(self, query: str, query_embedding: np.ndarray = None,
                      top_k: int = 10, threshold: float = 0.7) -> List[Dict[str, Any]]:
        """搜索相似记忆"""
        if query_embedding is None:
            query_embedding = self._generate_embedding(query)
        
        results = []
        
        # 向量相似度搜索
        if self.vector_index is not None:
            vector_results = self._vector_search(query_embedding, top_k, threshold)
            results.extend(vector_results)
        
        # 文本关键词搜索
        text_results = self._text_search(query, top_k)
        results.extend(text_results)
        
        # 元数据过滤搜索
        metadata_results = self._metadata_search(query, top_k)
        results.extend(metadata_results)
        
        # 融合和排序结果
        final_results = self._rank_and_deduplicate(results, top_k)
        
        return final_results
    
    def _vector_search(self, query_embedding: np.ndarray, top_k: int, 
                      threshold: float) -> List[Dict[str, Any]]:
        """向量相似度搜索"""
        if self.vector_index is None:
            return []
        
        # 执行搜索
        distances, indices = self.vector_index.search(
            query_embedding.reshape(1, -1), top_k
        )
        
        results = []
        for i, (distance, idx) in enumerate(zip(distances[0], indices[0])):
            if idx < 0 or distance > (1 - threshold):  # 距离转换为相似度
                continue
            
            # 获取记忆ID（简化实现）
            memory_id = self._get_memory_id_from_index(idx)
            if memory_id and memory_id in self.memory_store:
                memory = self.memory_store[memory_id]
                similarity_score = 1 - distance
                
                results.append({
                    'memory_id': memory_id,
                    'type': 'vector',
                    'similarity_score': similarity_score,
                    'memory': memory,
                    'search_weight': 0.8  # 向量搜索权重
                })
        
        return results
    
    def _text_search(self, query: str, top_k: int) -> List[Dict[str, Any]]:
        """文本关键词搜索"""
        query_words = set(query.lower().split())
        results = []
        
        for memory_id, memory in self.memory_store.items():
            content_words = set(memory['content'].lower().split())
            
            # 计算关键词匹配度
            intersection = query_words.intersection(content_words)
            jaccard_similarity = len(intersection) / len(query_words.union(content_words))
            
            if jaccard_similarity > 0.1:  # 相似度阈值
                results.append({
                    'memory_id': memory_id,
                    'type': 'text',
                    'similarity_score': jaccard_similarity,
                    'memory': memory,
                    'search_weight': 0.5  # 文本搜索权重
                })
        
        # 按相似度排序
        results.sort(key=lambda x: x['similarity_score'], reverse=True)
        return results[:top_k]
    
    def _rank_and_deduplicate(self, results: List[Dict[str, Any]], top_k: int) -> List[Dict[str, Any]]:
        """结果融合和去重"""
        # 按memory_id去重
        unique_results = {}
        
        for result in results:
            memory_id = result['memory_id']
            if memory_id not in unique_results:
                unique_results[memory_id] = result
            else:
                # 如果已存在，选择分数更高的
                existing_score = unique_results[memory_id]['similarity_score'] * unique_results[memory_id]['search_weight']
                new_score = result['similarity_score'] * result['search_weight']
                if new_score > existing_score:
                    unique_results[memory_id] = result
        
        # 计算最终分数并排序
        final_results = []
        for result in unique_results.values():
            final_score = result['similarity_score'] * result['search_weight']
            result['final_score'] = final_score
            final_results.append(result)
        
        # 按最终分数排序
        final_results.sort(key=lambda x: x['final_score'], reverse=True)
        return final_results[:top_k]
    
    def _generate_embedding(self, text: str) -> np.ndarray:
        """生成文本embedding"""
        # 使用TF-IDF生成特征向量（简化实现）
        vectorizer = TfidfVectorizer(max_features=self.embedding_dim)
        features = vectorizer.fit_transform([text])
        return features.toarray()[0]
    
    def update_memory(self, memory_id: str, new_content: str = None,
                     new_metadata: Dict[str, Any] = None):
        """更新记忆"""
        if memory_id not in self.memory_store:
            return False
        
        memory = self.memory_store[memory_id]
        updated = False
        
        # 更新内容
        if new_content and new_content != memory['content']:
            memory['content'] = new_content
            memory['version'] += 1
            self._update_text_index(memory_id, new_content)
            updated = True
        
        # 更新元数据
        if new_metadata:
            # 移除旧的元数据索引
            self._remove_metadata_index(memory_id, memory['metadata'])
            
            # 更新元数据
            memory['metadata'].update(new_metadata)
            memory['version'] += 1
            
            # 添加新的元数据索引
            self._update_metadata_index(memory_id, memory['metadata'])
            updated = True
        
        if updated:
            memory['last_access'] = datetime.now()
            # 重新生成embedding并更新向量索引
            new_embedding = self._generate_embedding(memory['content'])
            memory['embedding'] = new_embedding
            
            # 重新训练向量索引（简化实现）
            if self.index_type == "hnsw":
                # 重新构建索引
                all_embeddings = np.array([m['embedding'] for m in self.memory_store.values()])
                self.vector_index = faiss.IndexHNSWFlat(self.embedding_dim, 32)
                self.vector_index.add(all_embeddings)
        
        return updated
​

步骤 2：记忆版本管理系统

class MemoryVersionManager:
    """记忆版本管理器"""
    
    def __init__(self, max_versions: int = 10):
        self.max_versions = max_versions
        self.memory_versions = {}  # memory_id -> [versions]
        self.version_metadata = {}  # (memory_id, version) -> metadata
    
    def add_version(self, memory_id: str, content: str, 
                   metadata: Dict[str, Any] = None):
        """添加记忆版本"""
        if memory_id not in self.memory_versions:
            self.memory_versions[memory_id] = []
        
        # 创建新版本
        version = len(self.memory_versions[memory_id]) + 1
        version_data = {
            'version': version,
            'content': content,
            'timestamp': datetime.now(),
            'metadata': metadata or {}
        }
        
        # 添加版本
        self.memory_versions[memory_id].append(version_data)
        
        # 存储版本元数据
        self.version_metadata[(memory_id, version)] = {
            'size': len(content),
            'hash': self._calculate_hash(content),
            'metadata': metadata or {}
        }
        
        # 保持版本数量限制
        if len(self.memory_versions[memory_id]) > self.max_versions:
            # 移除最旧的版本
            old_version = self.memory_versions[memory_id].pop(0)
            old_key = (memory_id, old_version['version'])
            if old_key in self.version_metadata:
                del self.version_metadata[old_key]
    
    def get_version(self, memory_id: str, version: int = None) -> Optional[Dict[str, Any]]:
        """获取指定版本的记忆"""
        if memory_id not in self.memory_versions:
            return None
        
        versions = self.memory_versions[memory_id]
        
        if version is None:
            # 返回最新版本
            return versions[-1] if versions else None
        else:
            # 返回指定版本
            for v in versions:
                if v['version'] == version:
                    return v
            return None
    
    def get_version_history(self, memory_id: str) -> List[Dict[str, Any]]:
        """获取记忆版本历史"""
        if memory_id not in self.memory_versions:
            return []
        
        return self.memory_versions[memory_id].copy()
    
    def compare_versions(self, memory_id: str, version1: int, version2: int) -> Dict[str, Any]:
        """比较两个版本的差异"""
        v1_data = self.get_version(memory_id, version1)
        v2_data = self.get_version(memory_id, version2)
        
        if not v1_data or not v2_data:
            return {'error': '版本不存在'}
        
        # 简单的文本差异分析
        diff_result = {
            'version1': version1,
            'version2': version2,
            'timestamp1': v1_data['timestamp'],
            'timestamp2': v2_data['timestamp'],
            'size_diff': len(v2_data['content']) - len(v1_data['content']),
            'content_changed': v1_data['content'] != v2_data['content'],
            'metadata_changed': v1_data['metadata'] != v2_data['metadata']
        }
        
        return diff_result
    
    def rollback_to_version(self, memory_id: str, version: int) -> bool:
        """回滚到指定版本"""
        if memory_id not in self.memory_versions:
            return False
        
        # 查找指定版本
        target_version = None
        for v in self.memory_versions[memory_id]:
            if v['version'] == version:
                target_version = v
                break
        
        if not target_version:
            return False
        
        # 创建新版本作为回滚记录
        rollback_metadata = {
            'action': 'rollback',
            'rolled_from_version': version,
            'rolled_to_content': target_version['content']
        }
        
        # 添加回滚版本
        self.add_version(
            memory_id,
            target_version['content'],
            rollback_metadata
        )
        
        return True
    
    def _calculate_hash(self, content: str) -> str:
        """计算内容哈希值"""
        import hashlib
        return hashlib.md5(content.encode()).hexdigest()
​

常见问题 FAQ

Q1：记忆检索如何平衡速度和准确性？

A：记忆检索需要在速度和准确性之间找到平衡，主要通过以下策略：

1. 多级索引：使用向量索引+文本索引+元数据索引的多级检索
2. 权重分配：不同检索类型分配不同的权重，如向量搜索权重0.8，文本搜索0.5
3. 缓存机制：热门结果缓存，减少重复计算
4. 分页处理：支持分页检索，避免大数据量下的性能问题
5. 并行处理：不同索引类型并行检索，最后合并结果

Q2：记忆版本管理如何控制存储空间？

A：记忆版本管理通过以下方式控制存储空间：

1. 版本数量限制：每个记忆保留最多N个历史版本
2. 增量存储：只存储变化的增量部分，而非完整内容
3. 自动清理：定期清理过期的旧版本
4. 压缩策略：对旧版本进行压缩存储
5. 重要性过滤：保留重要版本，清理次要版本

Q3：记忆更新如何避免频繁写入？

A：避免频繁写入的记忆更新策略：

1. 批量处理：将多个更新请求批量处理，减少IO次数
2. 异步更新：使用异步更新队列，避免阻塞主线程
3. 更新阈值：只有达到一定重要程度的更新才会执行
4. 智能合并：合并多个小的更新为一次大的更新
5. 延迟写入：非关键信息延迟写入，集中处理

本节小结

本节深入探讨了记忆检索与更新机制，从向量搜索引擎到版本管理系统，全面介绍了记忆系统的核心技术实现。通过学习本节内容，读者应该能够：

1. 掌握记忆检索的核心算法和实现技术
2. 理解相似度计算和向量搜索的优化方法
3. 学习记忆版本管理和更新策略

下一节我们将探讨记忆系统的实现与优化，完成第4章的全部内容。

关键词：Agent智能体开发实战, 记忆系统, 检索算法, 向量搜索, 版本管理
难度：高级
预计阅读：30 分钟