3.4 搜索性能评估与优化

3.4 搜索性能评估与优化

本节导读：掌握FAISS搜索性能的评估方法、优化策略和调优技巧，学习如何构建完整的性能监控体系。

学习目标

• 掌握FAISS性能评估的完整指标体系
• 学习性能基准测试方法和最佳实践
• 理解不同场景下的性能优化策略
• 掌握实时性能监控和调优技术

性能评估指标体系

核心性能指标

精度指标

import numpy as np

def calculate_recall_precision(indices, ground_truth, k=10):
    """计算召回率和精确率"""
    m = len(indices)
    total_correct = 0
    
    for i in range(m):
        predicted_set = set(indices[i])
        true_set = set(ground_truth[i])
        intersection = len(predicted_set & true_set)
        total_correct += intersection
    
    recall = total_correct / (m * k)
    precision = total_correct / (m * k)
    
    return recall, precision

def calculate_f1_score(recall, precision):
    """计算F1分数"""
    if recall + precision == 0:
        return 0.0
    return 2 * (recall * precision) / (recall + precision)
​

性能指标

import time

def calculate_performance_metrics(search_time, n_queries, k=10):
    """计算性能指标"""
    qps = n_queries / search_time
    avg_latency = search_time / n_queries * 1000
    
    return {
        'qps': qps,
        'avg_latency_ms': avg_latency,
        'search_time_seconds': search_time,
        'n_queries': n_queries,
        'k': k
    }
​

内存指标

import psutil
import faiss

def get_memory_usage(index=None):
    """获取内存使用情况"""
    process = psutil.Process()
    process_memory = process.memory_info().rss / 1024 / 1024  # MB
    
    system_memory = psutil.virtual_memory()
    
    index_memory = 0
    if index is not None:
        try:
            if hasattr(index, 'memory_usage'):
                index_memory = index.memory_usage() / 1024 / 1024
        except Exception:
            index_memory = 0
    
    return {
        'process_memory_mb': process_memory,
        'system_memory_percent': system_memory.percent,
        'index_memory_mb': index_memory,
        'total_memory_mb': process_memory + index_memory
    }
​

完整性能评估框架

import time
import numpy as np
import faiss
from datetime import datetime
import json
import os

class ComprehensivePerformanceEvaluator:
    """完整的性能评估系统"""
    
    def __init__(self, log_dir='performance_logs'):
        self.log_dir = log_dir
        os.makedirs(log_dir, exist_ok=True)
        self.evaluation_results = []
        
    def evaluate_index_performance(self, index, vectors, queries, k=10, 
                                ground_truth=None, test_name='default'):
        """评估索引性能"""
        print(f"开始性能评估: {test_name}")
        
        # 性能测试
        start_time = time.time()
        distances, indices = index.search(queries.astype('float32'), k)
        search_time = time.time() - start_time
        
        # 精度评估
        recall = precision = f1_score = None
        if ground_truth is not None:
            recall, precision = calculate_recall_precision(indices, ground_truth, k)
            f1_score = calculate_f1_score(recall, precision)
        
        # 性能指标计算
        performance_metrics = calculate_performance_metrics(search_time, len(queries), k)
        
        # 内存使用评估
        memory_metrics = get_memory_usage(index)
        
        # 生成评估结果
        evaluation_result = {
            'test_name': test_name,
            'timestamp': datetime.now().isoformat(),
            'index_type': type(index).__name__,
            'data_stats': {
                'n_vectors': len(vectors),
                'dimension': vectors.shape[1],
                'n_queries': len(queries),
                'k': k
            },
            'performance': performance_metrics,
            'memory': memory_metrics,
            'accuracy': {
                'recall': recall,
                'precision': precision,
                'f1_score': f1_score
            } if recall is not None else None
        }
        
        # 存储结果
        self.evaluation_results.append(evaluation_result)
        self._save_evaluation_result(evaluation_result)
        
        # 打印结果摘要
        self._print_evaluation_summary(evaluation_result)
        
        return evaluation_result
    
    def _print_evaluation_summary(self, result):
        """打印评估结果摘要"""
        print("\n" + "="*50)
        print(f"评估结果: {result['test_name']}")
        print("="*50)
        print(f"索引类型: {result['index_type']}")
        print(f"数据规模: {result['data_stats']['n_vectors']} 向量")
        print()
        
        # 性能指标
        perf = result['performance']
        print(f"性能指标:")
        print(f"  QPS: {perf['qps']:.1f}")
        print(f"  平均延迟: {perf['avg_latency_ms']:.2f} ms")
        print()
        
        # 精度指标
        if result['accuracy']:
            acc = result['accuracy']
            print(f"精度指标:")
            print(f"  召回率: {acc['recall']:.4f}")
            print(f"  精确率: {acc['precision']:.4f}")
            print()
        
        # 内存指标
        mem = result['memory']
        print(f"内存使用:")
        print(f"  进程内存: {mem['process_memory_mb']:.1f} MB")
        print(f"  索引内存: {mem['index_memory_mb']:.1f} MB")
        print("="*50)
​

多索引对比评估

class IndexComparisonEvaluator:
    """索引对比评估器"""
    
    def __init__(self, vectors, queries, ground_truth=None, k=10):
        self.vectors = vectors
        self.queries = queries
        self.ground_truth = ground_truth
        self.k = k
        self.evaluator = ComprehensivePerformanceEvaluator()
        
    def compare_indexes(self, indexes_configs, test_names=None):
        """对比多个索引的性能"""
        if test_names is None:
            test_names = [f"index_{i}" for i in range(len(indexes_configs))]
        
        comparison_results = []
        
        for i, (config, name) in enumerate(zip(indexes_configs, test_names)):
            print(f"\n测试索引 {i+1}/{len(indexes_configs)}: {name}")
            
            # 创建索引
            index = self._create_index_from_config(config)
            
            # 评估性能
            result = self.evaluator.evaluate_index_performance(
                index, self.vectors, self.queries, self.k, 
                self.ground_truth, name
            )
            
            comparison_results.append(result)
        
        # 生成对比报告
        return self._generate_comparison_report(comparison_results)
    
    def _create_index_from_config(self, config):
        """根据配置创建索引"""
        import faiss
        
        index_type = config['type']
        
        if index_type == 'flat':
            index = faiss.IndexFlatL2(config['dimension'])
        elif index_type == 'ivf':
            nlist = config.get('nlist', 100)
            quantizer = faiss.IndexFlatL2(config['dimension'])
            index = faiss.IndexIVFFlat(quantizer, config['dimension'], nlist)
            index.nprobe = config.get('nprobe', min(20, nlist))
        elif index_type == 'pq':
            nlist = config.get('nlist', 100)
            m = config.get('m', 8)
            bits = config.get('bits', 8)
            quantizer = faiss.IndexFlatL2(config['dimension'])
            index = faiss.IndexIVFPQ(quantizer, config['dimension'], nlist, m, bits)
            index.nprobe = config.get('nprobe', min(10, nlist))
        else:
            raise ValueError(f"不支持的索引类型: {index_type}")
        
        # 训练和添加数据
        if hasattr(index, 'train'):
            index.train(self.vectors)
        index.add(self.vectors)
        
        return index
    
    def _generate_comparison_report(self, results):
        """生成对比报告"""
        print("\n" + "="*70)
        print("索引性能对比报告")
        print("="*70)
        
        # 表格头部
        print(f"{'索引名称':<15} {'类型':<15} {'QPS':>10} {'延迟(ms)':>10}")
        print("-" * 60)
        
        # 排序结果（按QPS降序）
        sorted_results = sorted(results, key=lambda x: x['performance']['qps'], reverse=True)
        
        for result in sorted_results:
            name = result['test_name']
            index_type = result['index_type']
            qps = result['performance']['qps']
            latency = result['performance']['avg_latency_ms']
            
            print(f"{name:<15} {index_type:<15} {qps:>10.1f} {latency:>10.2f}")
        
        # 找出最佳索引
        best_qps = max(results, key=lambda x: x['performance']['qps'])
        print(f"\n最佳索引 (QPS): {best_qps['test_name']} ({best_qps['performance']['qps']:.1f} QPS)")
        print("="*70)
        return sorted_results
​

性能监控与实时调优

实时性能监控

import threading
import queue
from datetime import datetime, timedelta

class RealTimePerformanceMonitor:
    """实时性能监控器"""
    
    def __init__(self, monitor_interval=60, max_history=1000):
        self.monitor_interval = monitor_interval
        self.max_history = max_history
        self.monitoring = False
        
        # 性能数据队列
        self.performance_queue = queue.Queue()
        
        # 历史数据
        self.history = []
        
        # 统计信息
        self.stats = {
            'total_queries': 0,
            'total_search_time': 0,
            'avg_qps': 0,
            'max_qps': 0,
            'current_qps': 0,
            'alert_triggered': False
        }
        
    def start_monitoring(self):
        """开始监控"""
        if self.monitoring:
            return
            
        self.monitoring = True
        self.monitor_thread = threading.Thread(target=self._monitor_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
        
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring = False
        if hasattr(self, 'monitor_thread'):
            self.monitor_thread.join()
            
    def record_performance(self, search_time, n_queries, index_type="default"):
        """记录性能数据"""
        current_time = time.time()
        qps = n_queries / search_time if search_time > 0 else 0
        
        performance_data = {
            'timestamp': current_time,
            'search_time': search_time,
            'n_queries': n_queries,
            'qps': qps,
            'index_type': index_type,
            'avg_latency_ms': (search_time / n_queries * 1000) if n_queries > 0 else 0
        }
        
        # 更新统计信息
        self._update_stats(performance_data)
        
        # 添加到历史
        self.history.append(performance_data)
        if len(self.history) > self.max_history:
            self.history.pop(0)
            
        # 检查异常
        self._check_anomalies(performance_data)
        
    def _update_stats(self, data):
        """更新统计信息"""
        self.stats['total_queries'] += data['n_queries']
        self.stats['total_search_time'] += data['search_time']
        
        self.stats['current_qps'] = data['qps']
        self.stats['max_qps'] = max(self.stats['max_qps'], data['qps'])
        self.stats['min_qps'] = min(self.stats['min_qps'], data['qps'])
        
        if self.stats['total_search_time'] > 0:
            total_qps = self.stats['total_queries'] / self.stats['total_search_time']
            self.stats['avg_qps'] = total_qps
            
    def _check_anomalies(self, data):
        """检查性能异常"""
        # QPS异常检测
        qps_threshold = 100
        if data['qps'] < qps_threshold and not self.stats['alert_triggered']:
            print(f"⚠️ 性能警报: QPS过低 {data['qps']:.1f} (阈值: {qps_threshold})")
            self.stats['alert_triggered'] = True
            
        # 延迟异常检测
        latency_threshold = 100
        if data['avg_latency_ms'] > latency_threshold:
            print(f"⚠️ 延迟警报: 平均延迟过高 {data['avg_latency_ms']:.2f}ms")
​

自适应调优系统

class AdaptivePerformanceOptimizer:
    """自适应性能优化器"""
    
    def __init__(self, index_configs):
        self.index_configs = index_configs
        self.current_config_index = 0
        self.performance_monitor = RealTimePerformanceMonitor()
        self.performance_history = []
        
    def start_optimization(self, vectors, queries, k=10):
        """开始优化过程"""
        print("开始自适应性能优化...")
        
        # 启动监控
        self.performance_monitor.start_monitoring()
        
        # 初始化索引
        self._setup_current_index(vectors, k)
        
        # 执行优化循环
        self._optimization_loop(vectors, queries, k)
        
    def _setup_current_index(self, vectors, k):
        """设置当前索引"""
        import faiss
        
        config = self.index_configs[self.current_config_index]
        
        if config['type'] == 'ivf':
            nlist = config.get('nlist', 100)
            quantizer = faiss.IndexFlatL2(vectors.shape[1])
            self.current_index = faiss.IndexIVFFlat(quantizer, vectors.shape[1], nlist)
            self.current_index.nprobe = config.get('nprobe', min(20, nlist))
        elif config['type'] == 'pq':
            nlist = config.get('nlist', 100)
            m = config.get('m', 8)
            bits = config.get('bits', 8)
            quantizer = faiss.IndexFlatL2(vectors.shape[1])
            self.current_index = faiss.IndexIVFPQ(quantizer, vectors.shape[1], nlist, m, bits)
            self.current_index.nprobe = config.get('nprobe', min(10, nlist))
        
        # 训练和添加数据
        if hasattr(self.current_index, 'train'):
            self.current_index.train(vectors)
        self.current_index.add(vectors)
        
    def _optimization_loop(self, vectors, queries, k):
        """优化循环"""
        iteration = 0
        max_iterations = len(self.index_configs) * 2
        
        while iteration < max_iterations:
            iteration += 1
            print(f"\n优化迭代 {iteration}/{max_iterations}")
            
            # 测试当前配置
            start_time = time.time()
            distances, indices = self.current_index.search(queries.astype('float32'), k)
            search_time = time.time() - start_time
            
            n_queries = len(queries)
            qps = n_queries / search_time
            
            # 记录性能
            self.performance_monitor.record_performance(search_time, n_queries, 
                                                     self.index_configs[self.current_config_index]['type'])
            
            # 评估当前配置
            current_performance = {
                'iteration': iteration,
                'config_index': self.current_config_index,
                'config': self.index_configs[self.current_config_index],
                'qps': qps,
                'search_time': search_time,
                'timestamp': time.time()
            }
            
            self.performance_history.append(current_performance)
            
            # 决策是否切换配置
            if self._should_switch_config(current_performance):
                self._switch_to_next_config()
            else:
                print(f"保持当前配置: {self.index_configs[self.current_config_index]['type']} ({qps:.1f} QPS)")
                break
                
            # 短暂休息
            time.sleep(1)
            
        # 生成优化报告
        self._generate_optimization_report()
        
    def _should_switch_config(self, current_performance):
        """判断是否应该切换配置"""
        if len(self.performance_history) == 1:
            return False
            
        previous_performances = [p for p in self.performance_history if p['config_index'] != self.current_config_index]
        
        if not previous_performances:
            return False
            
        # 如果当前QPS低于之前平均QPS的80%，切换配置
        previous_avg_qps = np.mean([p['qps'] for p in previous_performances])
        return current_performance['qps'] < previous_avg_qps * 0.8
        
    def _switch_to_next_config(self):
        """切换到下一个配置"""
        self.current_config_index = (self.current_config_index + 1) % len(self.index_configs)
        next_config = self.index_configs[self.current_config_index]
        print(f"切换到配置: {next_config['type']}")
        
    def _generate_optimization_report(self):
        """生成优化报告"""
        print("\n" + "="*60)
        print("自适应性能优化报告")
        print("="*60)
        
        # 找出最佳配置
        best_performance = max(self.performance_history, key=lambda x: x['qps'])
        
        print(f"最佳配置: {best_performance['config']['type']}")
        print(f"最佳QPS: {best_performance['qps']:.1f}")
        print(f"搜索时间: {best_performance['search_time']:.2f}s")
        
        print("="*60)
​

最佳实践与避坑

实践1：性能测试环境标准化

def standardize_performance_test(vectors, queries, k=10, n_runs=3):
    """标准化性能测试"""
    import faiss
    import time
    
    # 创建索引
    index = faiss.IndexFlatL2(vectors.shape[1])
    index.add(vectors)
    
    # 多次运行取平均
    all_results = []
    
    for run in range(n_runs):
        start_time = time.time()
        distances, indices = index.search(queries.astype('float32'), k)
        search_time = time.time() - start_time
        
        result = {
            'search_time': search_time,
            'qps': len(queries) / search_time,
            'avg_latency_ms': search_time / len(queries) * 1000
        }
        
        all_results.append(result)
        
        # 冷却时间
        time.sleep(1)
    
    # 计算平均值
    average_results = {
        'avg_search_time': np.mean([r['search_time'] for r in all_results]),
        'avg_qps': np.mean([r['qps'] for r in all_results]),
        'avg_latency_ms': np.mean([r['avg_latency_ms'] for r in all_results]),
        'std_qps': np.std([r['qps'] for r in all_results])
    }
    
    return average_results
​

坑点1：缓存效应导致的性能偏差

问题描述：首次搜索通常较慢，后续搜索因缓存而更快

解决方案：

def exclude_cache_effect(results, warmup_runs=3):
    """排除缓存效应的影响"""
    if len(results) <= warmup_runs:
        return results
    
    # 排除前几次运行结果
    filtered_results = results[warmup_runs:]
    
    # 计算稳定性能
    stable_qps = [r['qps'] for r in filtered_results]
    avg_qps = np.mean(stable_qps)
    std_qps = np.std(stable_qps)
    
    print(f"稳定性能: QPS = {avg_qps:.1f} ± {std_qps:.1f}")
    print(f"变异系数: {std_qps/avg_qps:.3f}")
    
    return filtered_results
​

本节小结

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

1. 完整的性能评估指标体系：精度、性能、内存指标的综合评估
2. 多索引对比评估方法：系统化的索引性能对比和选择策略
3. 实时性能监控：持续的性能监控和异常检测机制
4. 自适应性能优化：基于实时数据的自动配置调优
5. 最佳实践与避坑：标准化测试和常见问题的解决方案

这些技术将帮助您构建智能化、自适应的高性能搜索系统。

关键词：性能评估, 指标体系, 监控系统, 自适应优化, 实时调优
难度：高级
预计阅读：40分钟