📚 课程目录第3章暴力搜索与线性扫描

3. 暴力搜索与线性扫描：KNN暴力搜索实现

说到向量检索，很多人一上来就聊IVF、HNSW这些花哨的算法。但说实话，我做了这么多年向量数据库，最常用的反而是最朴素的暴力搜索。为什么？因为很多场景下，它才是那个最靠谱的选择。

暴力搜索，也叫线性扫描，说白了就是挨个比。你把查询向量和库里的每一个向量都算一遍距离，然后挑出最近的K个。听起来很笨是吧？但有时候，笨办法反而是最好的办法。

3.1 KNN暴力搜索的实现原理

KNN（K-Nearest Neighbors）暴力搜索的核心逻辑就三步：

遍历：把查询向量和数据集中的每个向量逐一计算距离
排序：维护一个大小为K的最近邻列表
输出：返回距离最近的K个向量的索引和距离值

这里有个细节——距离度量。最常用的是欧氏距离（L2）和余弦相似度。我个人习惯在文本场景用余弦，在图像特征上用L2。当然这不是的，得看你的向量是怎么训练出来的。

核心公式（欧氏距离）：

d(p, q) = √(∑(p_i - q_i)²)

说白了就是两个向量对应位置相减，平方，求和，再开根号。

3.2 时间复杂度分析

咱们来算笔账。假设数据集有N个向量，每个向量的维度是D。那么一次暴力搜索的时间复杂度是：

操作	时间复杂度	说明
距离计算	O(N × D)	每个向量都要算D维的差值
Top-K维护	O(N × log K)	用堆排序维护最近邻列表
总复杂度	O(N × (D + log K))	通常D远大于log K，所以近似O(N × D)

举个例子。100万个128维的向量，暴力搜索一次大概要算1亿次浮点运算。在普通CPU上，这大概需要几十到几百毫秒。嗯，听起来还行？但如果是1亿个向量呢？那就得几秒甚至几十秒了。

注意：我曾经在一个项目里踩过坑——数据集只有50万向量，但维度是768（BERT输出的）。暴力搜索一次要1.2秒，用户直接骂娘。后来发现是没做SIMD优化，换了个支持AVX的库，直接降到200毫秒。

3.3 何时该用暴力搜索？

你可能会问：既然有那么多索引算法，为什么还要用暴力搜索？我总结了几个场景：

数据量小（N < 10万）：这时候暴力搜索的性能完全可以接受，没必要引入索引的额外开销
要求100%召回率：近似算法（如IVF、HNSW）会牺牲精度，暴力搜索是唯一保证精确结果的方法
频繁插入/删除：索引重建很贵，暴力搜索不需要维护索引结构
作为基准测试：我每次做新算法评估，都会先用暴力搜索跑一遍，拿它的结果当ground truth

我的经验：在工业级系统中，我通常把暴力搜索作为"兜底方案"。当近似索引的置信度不够高时，回退到暴力搜索做二次验证。这种混合策略在很多场景下效果很好。

3.4 Python代码实战

好了，理论说完了，咱们直接上代码。我会实现一个完整的KNN暴力搜索类，包含欧氏距离和余弦相似度两种度量方式。

import numpy as np
from heapq import heappush, heappop
from typing import List, Tuple

class BruteForceKNN:
    """暴力搜索KNN实现"""
    
    def __init__(self, metric: str = 'euclidean'):
        """
        Args:
            metric: 距离度量方式，'euclidean' 或 'cosine'
        """
        self.metric = metric
        self.data = None
        self._is_fitted = False
        
    def fit(self, X: np.ndarray):
        """存储数据集"""
        self.data = X.astype(np.float32)
        self._is_fitted = True
        print(f"数据集已加载，形状: {self.data.shape}")
        
    def _compute_distance(self, query: np.ndarray, point: np.ndarray) -> float:
        """计算单个距离"""
        if self.metric == 'euclidean':
            diff = query - point
            return np.sqrt(np.dot(diff, diff))
        elif self.metric == 'cosine':
            dot = np.dot(query, point)
            norm_q = np.linalg.norm(query)
            norm_p = np.linalg.norm(point)
            # 余弦相似度转距离：1 - cos
            return 1.0 - dot / (norm_q * norm_p + 1e-10)
        else:
            raise ValueError(f"不支持的度量方式: {self.metric}")
    
    def query(self, query: np.ndarray, k: int = 5) -> Tuple[List[int], List[float]]:
        """
        查询K个最近邻
        
        Args:
            query: 查询向量
            k: 返回的最近邻数量
            
        Returns:
            indices: 最近邻的索引列表
            distances: 对应的距离列表
        """
        if not self._is_fitted:
            raise RuntimeError("请先调用fit()加载数据")
        
        query = query.astype(np.float32).flatten()
        n = self.data.shape[0]
        
        # 用最大堆维护K个最近邻（存负距离实现最小堆效果）
        heap = []
        
        for i in range(n):
            dist = self._compute_distance(query, self.data[i])
            
            if len(heap) < k:
                heappush(heap, (-dist, i))
            else:
                # 如果当前距离比堆中最大的还小，替换
                if dist < -heap[0][0]:
                    heappop(heap)
                    heappush(heap, (-dist, i))
        
        # 提取结果并排序
        results = sorted([(-d, idx) for d, idx in heap])
        indices = [idx for _, idx in results]
        distances = [d for d, _ in results]
        
        return indices, distances
    
    def batch_query(self, queries: np.ndarray, k: int = 5) -> Tuple[List[List[int]], List[List[float]]]:
        """批量查询"""
        all_indices = []
        all_distances = []
        
        for q in queries:
            idx, dist = self.query(q, k)
            all_indices.append(idx)
            all_distances.append(dist)
            
        return all_indices, all_distances

# ========== 使用示例 ==========
if __name__ == "__main__":
    # 生成测试数据：1000个128维的随机向量
    np.random.seed(42)
    n_samples = 1000
    n_dim = 128
    X = np.random.randn(n_samples, n_dim).astype(np.float32)
    
    # 初始化暴力搜索
    knn = BruteForceKNN(metric='euclidean')
    knn.fit(X)
    
    # 单条查询
    query = np.random.randn(n_dim)
    indices, distances = knn.query(query, k=5)
    
    print("查询结果：")
    for i, (idx, dist) in enumerate(zip(indices, distances)):
        print(f"  第{i+1}近邻: 索引={idx}, 距离={dist:.4f}")
    
    # 批量查询
    queries = np.random.randn(10, n_dim)
    all_idx, all_dist = knn.batch_query(queries, k=3)
    print(f"\n批量查询完成，共处理 {len(queries)} 条查询")

代码要点：

用堆（heap）维护Top-K，比全排序快一个数量级
float32精度足够，用float64反而浪费内存和计算
余弦距离用1-cos表示，保证距离越小越相似

3.5 性能优化技巧

暴力搜索虽然简单，但优化空间其实不小。我分享几个实战中常用的技巧：

向量化计算：用numpy的广播机制替代for循环，能快10-50倍
SIMD指令：现代CPU支持AVX/SSE，可以一次算多个维度
剪枝策略：如果当前部分距离已经超过已知最差候选，提前终止计算
内存布局：按行存储（row-major）比按列存储快，因为CPU缓存命中率高

避坑指南：我曾经在一个项目里用暴力搜索处理100万条记录，结果发现每次查询都要重新加载数据到内存。后来改成内存映射（mmap），查询速度直接提升了3倍。记住：I/O往往是瓶颈，不是计算。

3.6 知识体系总览

下面这张图总结了本章的核心知识点，帮你快速回顾：

暴力搜索虽然简单，但它是理解所有高级索引算法的基础。你想想看，IVF不就是把暴力搜索做了个分桶吗？HNSW不就是用图结构加速了搜索过程吗？所以别小看这个"笨办法"，它背后的思想贯穿了整个向量检索领域。