3.3 搜索算法优化技巧（上）

3.3 搜索算法优化技巧（上）

本节导读：深入掌握FAISS搜索算法的高级优化技巧，包括索引组合策略、预处理优化、并行处理和缓存机制，全面提升搜索系统的性能和可扩展性。

学习目标

• 掌握索引组合和层叠策略的高级方法
• 学习数据预处理和特征优化技巧
• 了解并行搜索和分布式优化方法
• 掌握缓存机制和结果优化策略
• 能够处理复杂场景下的搜索优化挑战

索引组合策略

层叠索引架构

class CascadedIndexSystem:
    """层叠索引系统"""
    
    def __init__(self, data, levels_config):
        """
        初始化层叠索引系统
        
        Args:
            data: 向量数据
            levels_config: 各层级配置
        """
        import faiss
        import numpy as np
        
        self.data = data
        self.levels = []
        self.level_configs = levels_config
        
        # 构建层级索引
        for i, config in enumerate(levels_config):
            level_index = self.create_level_index(data, config, i)
            self.levels.append({
                'index': level_index,
                'config': config,
                'name': f"Level_{i+1}"
            })
    
    def create_level_index(self, data, config, level_idx):
        """创建层级索引"""
        import faiss
        
        if config['type'] == 'coarse':
            # 粗粒度层：使用快速但精度较低的索引
            nlist = config.get('nlist', 100)
            quantizer = faiss.IndexFlatL2(data.shape[1])
            index = faiss.IndexIVFFlat(quantizer, data.shape[1], nlist)
            index.nprobe = min(5, nlist)  # 较少的探针
            
        elif config['type'] == 'medium':
            # 中等精度层：平衡精度和速度
            nlist = config.get('nlist', 200)
            quantizer = faiss.IndexFlatL2(data.shape[1])
            index = faiss.IndexIVFFlat(quantizer, data.shape[1], nlist)
            index.nprobe = min(20, nlist)
            
        elif config['type'] == 'fine':
            # 精细层：高精度索引
            index = faiss.IndexFlatL2(data.shape[1])
        
        elif config['type'] == 'pq':
            # PQ压缩层
            nlist = config.get('nlist', 100)
            m = config.get('m', 8)
            bits = config.get('bits', 8)
            quantizer = faiss.IndexFlatL2(data.shape[1])
            index = faiss.IndexIVFPQ(quantizer, data.shape[1], nlist, m, bits)
            index.nprobe = min(10, nlist)
        
        # 训练和添加数据
        if hasattr(index, 'train'):
            index.train(data)
        index.add(data)
        
        return index
    
    def search_cascaded(self, query_vector, k=10):
        """
        层叠搜索
        
        Args:
            query_vector: 查询向量
            k: 返回结果数量
            
        Returns:
            final_indices: 最终索引
            final_distances: 最终距离
        """
        import numpy as np
        
        # 第一层：快速粗搜索
        level1_result = self.levels[0]['index'].search(
            query_vector.reshape(1, -1), k * 2
        )
        coarse_indices = level1_result[1][0]
        coarse_distances = level1_result[0][0]
        
        # 第二层：在中等精度层中搜索
        if len(self.levels) > 1:
            level2_result = self.levels[1]['index'].search(
                query_vector.reshape(1, -1), k
            )
            medium_indices = level2_result[1][0]
            medium_distances = level2_result[0][0]
        
        # 第三层：在精细层中精确搜索候选结果
        if len(self.levels) > 2:
            # 获取候选向量
            candidate_vectors = self.data[coarse_indices[:k//2]]
            
            # 在精细层中搜索
            fine_result = self.levels[2]['index'].search(
                query_vector.reshape(1, -1), k
            )
            final_indices = fine_result[1][0]
            final_distances = fine_result[0][0]
        else:
            final_indices = medium_indices
            final_distances = medium_distances
        
        return final_indices, final_distances
    
    def search_parallel(self, query_vector, k=10):
        """
        并行层叠搜索
        
        Args:
            query_vector: 查询向量
            k: 返回结果数量
            
        Returns:
            combined_result: 合并后的结果
        """
        import numpy as np
        
        # 并行搜索各层级
        results = []
        for level in self.levels:
            result = level['index'].search(
                query_vector.reshape(1, -1), k
            )
            results.append(result)
        
        # 合并结果
        # 使用投票机制选择最佳结果
        votes = np.zeros(len(self.data))
        for distances, indices in results:
            for i, idx in enumerate(indices[0]):
                # 根据距离倒数进行投票
                votes[idx] += 1.0 / (distances[0][i] + 1e-6)
        
        # 选择投票最高的k个结果
        top_indices = np.argsort(votes)[-k:][::-1]
        top_distances = np.array([votes[i] for i in top_indices])
        
        return top_indices, top_distances
​

混合索引策略

class HybridIndexSystem:
    """混合索引系统"""
    
    def __init__(self, data):
        """
        初始化混合索引系统
        
        Args:
            data: 向量数据
        """
        import faiss
        import numpy as np
        
        self.data = data
        self.indexes = {}
        self.setup_hybrid_indexes()
    
    def setup_hybrid_indexes(self):
        """设置混合索引"""
        import faiss
        
        # 1. 精确索引（用于高精度需求）
        self.indexes['exact'] = faiss.IndexFlatL2(self.data.shape[1])
        
        # 2. IVF索引（用于快速搜索）
        nlist = min(100, int(np.sqrt(len(self.data))))
        quantizer = faiss.IndexFlatL2(self.data.shape[1])
        self.indexes['ivf'] = faiss.IndexIVFFlat(quantizer, self.data.shape[1], nlist)
        self.indexes['ivf'].nprobe = min(20, nlist)
        
        # 3. PQ索引（用于内存敏感场景）
        m = 8
        bits = 8
        quantizer = faiss.IndexFlatL2(self.data.shape[1])
        self.indexes['pq'] = faiss.IndexIVFPQ(quantizer, self.data.shape[1], nlist, m, bits)
        
        # 4. HNSW索引（用于高维数据）
        self.indexes['hnsw'] = faiss.IndexHNSWFlat(self.data.shape[1], 32)
        
        # 训练所有需要训练的索引
        for name, index in self.indexes.items():
            if hasattr(index, 'train'):
                index.train(self.data)
            index.add(self.data)
    
    def search_optimized(self, query_vector, k=10, use_case='general'):
        """
        优化搜索
        
        Args:
            query_vector: 查询向量
            k: 返回结果数量
            use_case: 使用场景 ('general', 'speed', 'accuracy', 'memory')
            
        Returns:
            indices: 索引结果
            distances: 距离结果
        """
        import numpy as np
        
        # 根据使用场景选择合适的索引
        if use_case == 'speed':
            # 速度优先：使用IVF索引
            index = self.indexes['ivf']
        elif use_case == 'accuracy':
            # 精度优先：使用精确索引
            index = self.indexes['exact']
        elif use_case == 'memory':
            # 内存优先：使用PQ索引
            index = self.indexes['pq']
        else:
            # 通用场景：使用HNSW索引
            index = self.indexes['hnsw']
        
        return index.search(query_vector.reshape(1, -1), k)
    
    def search_fallback(self, query_vector, k=10):
        """
        降级搜索策略
        
        Args:
            query_vector: 查询向量
            k: 返回结果数量
            
        Returns:
            final_result: 最终结果
        """
        import numpy as np
        
        # 尝试HNSW索引（最快）
        try:
            result = self.indexes['hnsw'].search(query_vector.reshape(1, -1), k)
            return result
        except Exception:
            pass
        
        # 尝试IVF索引
        try:
            result = self.indexes['ivf'].search(query_vector.reshape(1, -1), k)
            return result
        except Exception:
            pass
        
        # 最后使用精确索引
        return self.indexes['exact'].search(query_vector.reshape(1, -1), k)
​

数据预处理优化

特征标准化

def standardize_features(data, method='l2'):
    """
    特征标准化
    
    Args:
        data: 原始特征数据
        method: 标准化方法 ('l2', 'z-score', 'min-max')
        
    Returns:
        standardized_data: 标准化后的数据
        normalization_params: 标准化参数
    """
    import numpy as np
    
    if method == 'l2':
        # L2标准化（余弦相似度）
        norms = np.linalg.norm(data, axis=1, keepdims=True)
        norms[norms == 0] = 1  # 避免除零
        standardized_data = data / norms
        
    elif method == 'z-score':
        # Z-score标准化
        mean = np.mean(data, axis=0)
        std = np.std(data, axis=0)
        std[std == 0] = 1  # 避免除零
        standardized_data = (data - mean) / std
        
    elif method == 'min-max':
        # Min-Max标准化
        min_val = np.min(data, axis=0)
        max_val = np.max(data, axis=0)
        range_val = max_val - min_val
        range_val[range_val == 0] = 1  # 避免除零
        standardized_data = (data - min_val) / range_val
    else:
        raise ValueError(f"Unknown normalization method: {method}")
    
    normalization_params = {
        'method': method,
        'mean': mean if method == 'z-score' else None,
        'std': std if method == 'z-score' else None,
        'min_val': min_val if method == 'min-max' else None,
        'max_val': max_val if method == 'min-max' else None
    }
    
    return standardized_data, normalization_params

def apply_normalization(data, normalization_params):
    """
    应用标准化参数
    
    Args:
        data: 原始数据
        normalization_params: 标准化参数
        
    Returns:
        normalized_data: 标准化后的数据
    """
    import numpy as np
    
    method = normalization_params['method']
    
    if method == 'l2':
        norms = np.linalg.norm(data, axis=1, keepdims=True)
        norms[norms == 0] = 1
        return data / norms
    
    elif method == 'z-score':
        mean = normalization_params['mean']
        std = normalization_params['std']
        std[std == 0] = 1
        return (data - mean) / std
    
    elif method == 'min-max':
        min_val = normalization_params['min_val']
        max_val = normalization_params['max_val']
        range_val = max_val - min_val
        range_val[range_val == 0] = 1
        return (data - min_val) / range_val
    
    else:
        raise ValueError(f"Unknown normalization method: {method}")
​

维度优化

def optimize_dimension(data, target_dim=None, method='pca'):
    """
    维度优化
    
    Args:
        data: 原始数据
        target_dim: 目标维度，如果为None则自动选择
        method: 降维方法 ('pca', 'tsne', 'autoencoder', 'feature-selection')
        
    Returns:
        reduced_data: 降维后的数据
        dimension_info: 维度信息
    """
    import numpy as np
    from sklearn.decomposition import PCA
    from sklearn.manifold import TSNE
    
    if method == 'pca':
        # PCA降维
        if target_dim is None:
            # 自动选择维度（保留95%方差）
            pca = PCA(n_components=0.95)
            reduced_data = pca.fit_transform(data)
            explained_variance = pca.explained_variance_ratio_
            target_dim = reduced_data.shape[1]
        else:
            pca = PCA(n_components=target_dim)
            reduced_data = pca.fit_transform(data)
            explained_variance = pca.explained_variance_ratio_
        
        dimension_info = {
            'method': 'pca',
            'original_dim': data.shape[1],
            'reduced_dim': target_dim,
            'explained_variance': explained_variance,
            'total_variance': np.sum(explained_variance)
        }
        
    elif method == 'tsne':
        # t-SNE降维（主要用于可视化）
        if target_dim is None:
            target_dim = 2
        
        tsne = TSNE(n_components=target_dim, random_state=42)
        reduced_data = tsne.fit_transform(data)
        
        dimension_info = {
            'method': 'tsne',
            'original_dim': data.shape[1],
            'reduced_dim': target_dim,
            'perplexity': tsne.perplexity
        }
        
    elif method == 'autoencoder':
        # 自编码器降维（需要深度学习库）
        from tensorflow.keras.layers import Input, Dense
        from tensorflow.keras.models import Model
        
        input_dim = data.shape[1]
        encoding_dim = target_dim if target_dim else min(32, data.shape[1] // 4)
        
        # 构建自编码器
        input_layer = Input(shape=(input_dim,))
        encoded = Dense(encoding_dim, activation='relu')(input_layer)
        decoded = Dense(input_dim, activation='sigmoid')(encoded)
        
        autoencoder = Model(input_layer, decoded)
        encoder = Model(input_layer, encoded)
        
        # 编译和训练
        autoencoder.compile(optimizer='adam', loss='mse')
        autoencoder.fit(data, data, epochs=50, batch_size=256, verbose=0)
        
        # 编码数据
        reduced_data = encoder.predict(data)
        
        dimension_info = {
            'method': 'autoencoder',
            'original_dim': input_dim,
            'reduced_dim': encoding_dim,
            'epochs': 50,
            'batch_size': 256
        }
        
    elif method == 'feature-selection':
        # 特征选择
        from sklearn.feature_selection import SelectKBest, f_classif
        
        if target_dim is None:
            target_dim = min(data.shape[1] // 2, 100)
        
        selector = SelectKBest(score_func=f_classif, k=target_dim)
        reduced_data = selector.fit_transform(data, np.random.randint(0, 2, len(data)))
        
        dimension_info = {
            'method': 'feature-selection',
            'original_dim': data.shape[1],
            'reduced_dim': target_dim,
            'selected_features': selector.get_support()
        }
    
    else:
        raise ValueError(f"Unknown dimension reduction method: {method}")
    
    return reduced_data, dimension_info
​

数据分片策略

class DataShardingSystem:
    """数据分片系统"""
    
    def __init__(self, data, shard_size=100000):
        """
        初始化数据分片系统
        
        Args:
            data: 原始数据
            shard_size: 每个分片的大小
        """
        import numpy as np
        
        self.original_data = data
        self.shard_size = shard_size
        self.shards = self.create_shards(data)
        self.shard_indexes = {}
        self.create_shard_indexes()
    
    def create_shards(self, data):
        """创建数据分片"""
        import numpy as np
        
        n_shards = (len(data) + self.shard_size - 1) // self.shard_size
        shards = []
        
        for i in range(n_shards):
            start_idx = i * self.shard_size
            end_idx = min((i + 1) * self.shard_size, len(data))
            shard_data = data[start_idx:end_idx]
            shards.append({
                'data': shard_data,
                'start_idx': start_idx,
                'end_idx': end_idx,
                'size': len(shard_data),
                'shard_id': i
            })
        
        return shards
    
    def create_shard_indexes(self):
        """为每个分片创建索引"""
        import faiss
        
        for shard in self.shards:
            # 使用较小的索引参数
            quantizer = faiss.IndexFlatL2(shard['data'].shape[1])
            nlist = min(50, int(np.sqrt(len(shard['data']))))
            index = faiss.IndexIVFFlat(quantizer, shard['data'].shape[1], nlist)
            index.nprobe = min(10, nlist)
            
            index.train(shard['data'])
            index.add(shard['data'])
            
            self.shard_indexes[shard['shard_id']] = index
    
    def search_distributed(self, query_vector, k=10):
        """
        分布式搜索
        
        Args:
            query_vector: 查询向量
            k: 返回结果数量
            
        Returns:
            combined_results: 合并后的结果
        """
        import numpy as np
        
        all_results = []
        
        # 在每个分片中搜索
        for shard_id, index in self.shard_indexes.items():
            result = index.search(query_vector.reshape(1, -1), k)
            distances, indices = result
            
            # 将索引转换为全局索引
            global_indices = indices[0] + self.shards[shard_id]['start_idx']
            
            all_results.append({
                'distances': distances[0],
                'indices': global_indices,
                'shard_id': shard_id
            })
        
        # 合并所有结果
        if all_results:
            # 合并所有距离和索引
            all_distances = []
            all_indices = []
            
            for result in all_results:
                all_distances.extend(result['distances'])
                all_indices.extend(result['indices'])
            
            # 排序并选择最佳的k个结果
            sorted_pairs = sorted(zip(all_distances, all_indices), key=lambda x: x[0])
            
            if len(sorted_pairs) > k:
                final_distances = [pair[0] for pair in sorted_pairs[:k]]
                final_indices = [pair[1] for pair in sorted_pairs[:k]]
            else:
                final_distances = [pair[0] for pair in sorted_pairs]
                final_indices = [pair[1] for pair in sorted_pairs]
            
            return np.array(final_indices), np.array(final_distances)
        else:
            return np.array([]), np.array([])
    
    def search_parallel(self, query_vector, k=10):
        """
        并行搜索
        
        Args:
            query_vector: 查询向量
            k: 返回结果数量
            
        Returns:
            final_results: 最终结果
        """
        import numpy as np
        from concurrent.futures import ThreadPoolExecutor
        
        def search_shard(shard_id):
            index = self.shard_indexes[shard_id]
            result = index.search(query_vector.reshape(1, -1), k)
            distances, indices = result
            global_indices = indices[0] + self.shards[shard_id]['start_idx']
            return distances[0], global_indices
        
        # 并行搜索所有分片
        with ThreadPoolExecutor() as executor:
            futures = []
            for shard_id in self.shard_indexes.keys():
                futures.append(executor.submit(search_shard, shard_id))
            
            all_results = []
            for future in futures:
                distances, indices = future.result()
                all_results.append((distances, indices))
        
        # 合并结果
        all_distances = []
        all_indices = []
        
        for distances, indices in all_results:
            all_distances.extend(distances)
            all_indices.extend(indices)
        
        # 排序并选择最佳结果
        sorted_pairs = sorted(zip(all_distances, all_indices), key=lambda x: x[0])
        
        if len(sorted_pairs) > k:
            final_distances = [pair[0] for pair in sorted_pairs[:k]]
            final_indices = [pair[1] for pair in sorted_pairs[:k]]
        else:
            final_distances = [pair[0] for pair in sorted_pairs]
            final_indices = [pair[1] for pair in sorted_pairs]
        
        return np.array(final_indices), np.array(final_distances)
​

本节小结

通过本节学习，我们掌握了：

1. 索引组合策略：学习了层叠索引和混合索引的高级方法
2. 数据预处理优化：掌握了特征标准化、维度优化和数据分片策略
3. 并行搜索优化：了解了多线程和多进程搜索的实现方法

这些优化技巧将帮助您构建高性能、可扩展的搜索系统。

延伸阅读

• 高性能向量搜索系统设计
• 大规模分布式搜索算法
• 现代搜索系统的架构优化

关键词：索引组合, 层叠索引, 数据分片, 并行搜索, 特征优化
难度：进阶
预计阅读：30分钟