索引

3. 索引

3.1 索引类型全面解析

FAISS提供了多种索引类型，每种索引在精度、速度和内存使用之间有不同的权衡。

精确搜索索引

• IndexFlatL2：暴力搜索，使用L2距离
• IndexFlatIP：暴力搜索，使用内积相似度

近似搜索索引

• IVF索引（Inverted File Index）：基于倒排文件的索引
• HNSW索引：基于层级导航小世界图
• PQ索引（Product Quantization）：乘积量化索引
• 复合索引：结合多种技术的混合索引

3.2 索引选择指南与使用场景

IndexFlatL2/IP - 精确搜索索引

使用场景：

• 数据量较小（通常<10,000个向量）
• 需要100%准确率
• 开发调试阶段

特点：

• 结果精确
• 搜索速度慢（O(n)复杂度）
• 内存占用高

def flat_index_demo():
    dimension = 256
    n_vectors = 5000
    
    data = np.random.random((n_vectors, dimension)).astype('float32')
    
    # 创建L2距离索引
    index_l2 = faiss.IndexFlatL2(dimension)
    index_l2.add(data)
    
    # 创建内积相似度索引
    index_ip = faiss.IndexFlatIP(dimension) 
    index_ip.add(data)
    
    return index_l2, index_ip

index_l2, index_ip = flat_index_demo()
​

IVF索引 - 倒排文件索引

使用场景：

• 中等规模数据（10万-1000万向量）
• 查询速度优先，可接受少量精度损失
• 内存相对充足

特点：

• 通过聚类加速搜索
• 需要训练阶段
• 可通过nprobe参数平衡速度与精度

import faiss
import numpy as np

def ivf_index_demo():
    dimension = 128
    n_vectors = 50000
    n_clusters = 1024  # 聚类中心数量
    print(f"初始化参数 - 维度: {dimension}, 向量数量: {n_vectors}, 聚类中心数量: {n_clusters}")
    
    # 生成数据
    data = np.random.random((n_vectors, dimension)).astype('float32')
    print("数据生成完成，数据形状:", data.shape)
    
    # 创建IVF索引
    quantizer = faiss.IndexFlatL2(dimension)  # 量化器
    index = faiss.IndexIVFFlat(quantizer, dimension, n_clusters, faiss.METRIC_L2)
    print("IVF索引创建完成")
    print("-------------------------------------------------------------------------------------------")

    # 训练索引
    assert not index.is_trained
    print("开始训练索引...")
    index.train(data)
    assert index.is_trained
    print("索引训练完成")
    print("-------------------------------------------------------------------------------------------")
    
    # 添加数据
    index.add(data)
    print("数据添加到索引完成")
    
    # 设置搜索时检查的聚类数量（平衡速度与精度）
    index.nprobe = 16
    print(f"设置搜索时检查的聚类数量为: {index.nprobe}")
    
    # 搜索
    queries = np.random.random((10, dimension)).astype('float32')
    print("查询向量生成完成，形状:", queries.shape)
    distances, indices = index.search(queries, 5)
    print("-------------------------------------------------------------------------------------------")
    print("搜索完成")
    print("搜索结果 - 距离:", distances)
    print("搜索结果 - 索引:", indices)
    
    return index

ivf_index = ivf_index_demo()
​

HNSW索引 - 图结构索引

使用场景：

• 高维数据
• 需要高召回率和快速查询
• 实时搜索应用

特点：

• 基于图结构的近似算法
• 无需训练
• 构建时间较长，但查询速度快

HNSW索引属于图索引，它通过构建节点之间的连接关系来组织向量数据。
构建过程只是将新向量添加到图中并建立相应的连接，因此不需要训练阶段。
而在FAISS中，需要训练的索引通常是基于量化（Quantization）或聚类（Clustering）的索引，
例如 IndexIVFFlat、IndexIVFPQ、IndexFlatL2 等。这些索引在添加数据前需要先通过训练数据学习量化或聚类的参数。

import faiss
import numpy as np
def hnsw_index_demo():
    dimension = 128
    n_vectors = 1000
    
    data = np.random.random((n_vectors, dimension)).astype('float32')
    
    # 创建HNSW索引
    index = faiss.IndexHNSWFlat(dimension, 32)  # 32表示每个节点的连接数
    print("创建HNSW索引")
    print("-------------------------------------------------------------------------------------------")
    
    # 设置构建参数
    index.hnsw.efConstruction = 200  # 构建时考虑的邻居数量
    index.hnsw.efSearch = 50         # 搜索时考虑的邻居数量    
    # 添加数据
    index.add(data)
    print(f"已向索引中添加 {n_vectors} 条数据")
    print("-------------------------------------------------------------------------------------------")
    
    # 搜索
    queries = np.random.random((5, dimension)).astype('float32')
    distances, indices = index.search(queries, 3)
    print("搜索完成，搜索结果示例:")
    print("距离:", distances)
    print("索引:", indices)
    
    return index

hnsw_index = hnsw_index_demo()
​

PQ索引 - 乘积量化索引

使用场景：

• 超大规模数据（百万级以上）
• 内存受限环境
• 可接受一定精度损失

特点：

• 大幅压缩向量存储
• 搜索速度快
• 需要训练，有量化误差

def pq_index_demo():
    dimension = 128
    n_vectors = 100000
    
    data = np.random.random((n_vectors, dimension)).astype('float32')
    
    # 创建PQ索引参数
    m = 8  # 子量化器数量（必须能被dimension整除）
    n_bits = 8  # 每个子量化器的位数
    
    # 创建PQ索引
    index = faiss.IndexPQ(dimension, m, n_bits)
    
    # 训练并添加数据
    index.train(data)
    index.add(data)
    
    print(f"PQ索引大小：{index.ntotal} 个向量")
    
    return index

pq_index = pq_index_demo()
​

复合索引 - IVF+PQ

使用场景：

• 超大规模数据集
• 需要在速度和内存使用间取得最佳平衡

import faiss
import numpy as np
def ivfpq_index_demo():
    dimension = 128
    n_vectors = 1000000
    n_clusters = 1024
    
    data = np.random.random((n_vectors, dimension)).astype('float32')
    
    # 创建IVFPQ索引
    quantizer = faiss.IndexFlatL2(dimension)
    m = 16  # 字节数（压缩后）
    index = faiss.IndexIVFPQ(quantizer, dimension, n_clusters, m, 8)
    
    # 训练并添加数据
    index.train(data)
    index.add(data)
    
    index.nprobe = 32
    
    print(f"聚类中心数量: {n_clusters}")
    print(f"压缩后字节数: {m}")
    
    return index

ivfpq_index = ivfpq_index_demo()
​

3.3 索引原理解析

IVF索引原理

IVF（Inverted File Index）索引的核心思想是"分而治之"：

1. 聚类阶段：使用k-means算法将所有向量分配到不同的聚类中心
2. 倒排列表：为每个聚类中心建立包含该簇所有向量的倒排列表
3. 搜索阶段：对于查询向量，只在与它最接近的nprobe个簇中进行搜索

数学原理：

• 聚类中心数量：nlist
• 搜索簇数量：nprobe
• 搜索复杂度从O(N)降低到O(nprobe × N/nlist)

HNSW索引原理

HNSW（Hierarchical Navigable Small World）基于小世界网络理论：

1. 层级结构：构建多层图，上层稀疏，下层密集
2. 导航机制：从上层开始搜索，逐步向下层细化
3. 贪婪搜索：在每一层使用最佳优先搜索算法

优势：

• 搜索复杂度近似O(log N)
• 对高维数据友好
• 无需训练数据

PQ索引原理

乘积量化通过向量分解和标量化化实现压缩：

1. 向量分割：将D维向量分割为m个D/m维子向量
2. 子空间量化：对每个子空间独立进行k-means聚类
3. 编码存储：用聚类中心ID代替原始子向量

压缩效果：

• 原始存储：D × 4字节（float32）
• 压缩后：m × 1字节（256个聚类中心）
• 压缩比：4D/m

3.4 性能比较

def benchmark_indices():
    """不同索引的性能比较"""
    dimension = 128
    n_vectors = 100000
    n_queries = 1000
    
    # 生成测试数据
    data = np.random.random((n_vectors, dimension)).astype('float32')
    queries = np.random.random((n_queries, dimension)).astype('float32')
    
    indices = {
        'FlatL2': faiss.IndexFlatL2(dimension),
        'IVFFlat': faiss.IndexIVFFlat(faiss.IndexFlatL2(dimension), dimension, 1024),
        'HNSW': faiss.IndexHNSWFlat(dimension, 32),
        'IVFPQ': faiss.IndexIVFPQ(faiss.IndexFlatL2(dimension), dimension, 1024, 16, 8)
    }
    
    # 训练需要训练的索引
    for name, index in indices.items():
        if name != 'FlatL2' and name != 'HNSW':
            index.train(data)
        index.add(data)
        if name == 'IVFFlat' or name == 'IVFPQ':
            index.nprobe = 32
    
    # 测试搜索性能
    results = {}
    for name, index in indices.items():
        import time
        start = time.time()
        distances, indices_result = index.search(queries, 10)
        end = time.time()
        
        results[name] = {
            'time': end - start,
            'throughput': n_queries / (end - start)
        }
    
    return results

# 性能测试结果
performance_results = benchmark_indices()
for name, result in performance_results.items():
    print(f"{name}: {result['throughput']:.1f} queries/second")
​