3.4 向量索引与检索优化（上）

F[全部向量] --> G[K-Means 聚类]
G --> H[簇1: 向量子集]
G --> I[簇2: 向量子集]
G --> J[簇3: 向量子集]
G --> K[簇N: 向量子集]
​

</div>

#### IVF-Flat：基础倒排索引

```python
import faiss
import numpy as np
import time

dimension = 768
num_vectors = 200000
np.random.seed(42)
vectors = np.random.rand(num_vectors, dimension).astype('float32')
faiss.normalize_L2(vectors)

# ==================== IVF-Flat 索引 ====================
nlist = 1000  # 聚类中心数

print("构建 IVF-Flat 索引...")

# 使用 Flat 量化器（不压缩，精确搜索）
quantizer = faiss.IndexFlatIP(dimension)
index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT)

# 训练聚类中心
train_size = min(num_vectors, 100000)
t0 = time.time()
index.train(vectors[:train_size])
print(f"训练耗时: {time.time()-t0:.2f} 秒")

# 添加向量
t0 = time.time()
index.add(vectors)
print(f"添加耗时: {time.time()-t0:.2f} 秒")

# 查询
query = np.random.rand(100, dimension).astype('float32')
faiss.normalize_L2(query)

# nprobe: 查询时探测的簇数
# nprobe=1: 最快但召回最低
# nprobe=nlist: 等价于暴力搜索，最慢但100%召回
for nprobe in [1, 10, 50, 100, 500, 1000]:
    index.nprobe = nprobe
    t0 = time.time()
    distances, indices = index.search(query, k=10)
    qtime = (time.time() - t0) / len(query) * 1000
    print(f"  nprobe={nprobe:>4d}: {qtime:.2f} ms/query")

# IVF-Flat 的 nprobe 调优建议
print("\n=== nprobe 调优建议 ===")
print("• nprobe = nlist × 1% 到 10% 是常见的有效范围")
print("• 追求极致速度: nprobe=1~10，召回率约 60-80%")
print("• 平衡模式: nprobe=50~100，召回率约 90-95%")
print("• 追求高召回: nprobe=200+，召回率 > 98%")
​

IVF-PQ：压缩倒排索引

IVF-PQ 在 IVF-Flat 的基础上加入 PQ 量化，大幅降低内存占用：

# IVF-PQ: 聚类 + 乘积量化
nlist = 1000   # 聚类中心数
m = 64         # PQ 子空间数
nbits = 8      # 每个子空间编码位数

quantizer = faiss.IndexFlatIP(dimension)
index = faiss.IndexIVFPQ(quantizer, dimension, nlist, m, nbits, faiss.METRIC_INNER_PRODUCT)

index.train(vectors[:train_size])
index.add(vectors)

index.nprobe = 50
distances, indices = index.search(query, k=10)

# 内存对比
flat_size = num_vectors * dimension * 4 / 1024 / 1024
pq_size = num_vectors * m * 1 / 1024 / 1024  # nbits=8 每个编码 1 字节
print(f"IVF-Flat 内存: {flat_size:.0f} MB")
print(f"IVF-PQ  内存: {pq_size:.0f} MB")
print(f"压缩比: {flat_size/pq_size:.1f}x")
​

索引选择决策框架

根据数据规模、内存预算和查询延迟要求，我给出一个实用的选择框架：

def recommend_index(data_size, dimension, memory_budget_mb, latency_ms, recall_target):
    """
    RAG 向量索引推荐工具
    
    参数:
        data_size: 向量数量
        dimension: 向量维度
        memory_budget_mb: 内存预算（MB）
        recall_target: 目标召回率 (0.0-1.0)
        latency_ms: 可接受的单次查询延迟（毫秒）
    
    返回:
        推荐的索引类型和参数配置
    """
    # 计算原始数据大小
    raw_size_mb = data_size * dimension * 4 / 1024 / 1024
    
    # 内存充足（> 原始数据 2 倍）→ 优先 HNSW
    if memory_budget_mb > raw_size_mb * 2:
        if recall_target >= 0.99:
            return {
                "index_type": "HNSW",
                "params": {"M": 32, "efConstruction": 400, "efSearch": 300},
                "reason": "内存充裕 + 高召回需求 → HNSW 高配"
            }
        elif recall_target >= 0.95:
            return {
                "index_type": "HNSW", 
                "params": {"M": 32, "efConstruction": 200, "efSearch": 100},
                "reason": "内存充裕 + 中高召回 → HNSW 标配"
            }
        else:
            return {
                "index_type": "HNSW",
                "params": {"M": 16, "efConstruction": 100, "efSearch": 50},
                "reason": "内存充裕 + 低延迟需求 → HNSW 低配"
            }
    
    # 内存紧张 → 量化索引
    elif memory_budget_mb > raw_size_mb * 0.1:
        if recall_target >= 0.95:
            return {
                "index_type": "IVF-SQ",
                "params": {"nlist": int(data_size ** 0.5), "nprobe": 100},
                "reason": "内存受限 + 高召回 → IVF + 标量量化"
            }
        else:
            return {
                "index_type": "IVF-PQ",
                "params": {"nlist": int(data_size ** 0.5), "m": 48, "nbits": 8, "nprobe": 50},
                "reason": "内存受限 → IVF + 乘积量化"
            }
    
    # 极致压缩
    else:
        return {
            "index_type": "IVF-PQ",
            "params": {"nlist": int(data_size ** 0.5), "m": 64, "nbits": 8, "nprobe": 30},
            "reason": "极端内存限制 → IVF-PQ 高压缩比配置"
        }

# 使用示例
result = recommend_index(
    data_size=1000000,      # 100万向量
    dimension=768,          # 768维
    memory_budget_mb=2000,  # 2GB 内存
    latency_ms=10,          # 10ms 延迟
    recall_target=0.95
)
print(f"推荐索引: {result['index_type']}")
print(f"参数配置: {result['params']}")
print(f"推荐理由: {result['reason']}")
​

FAISS 索引构建最佳实践

1. 数据预处理

在构建索引前，务必对向量做归一化处理。这不仅是余弦相似度的要求，也是保证各索引算法一致性的前提：

import faiss
import numpy as np

def prepare_vectors(raw_vectors, normalize=True):
    """向量预处理"""
    vectors = raw_vectors.astype('float32')
    
    if normalize:
        # L2 归一化：使内积等价于余弦相似度
        faiss.normalize_L2(vectors)
    
    # 检查异常值
    norms = np.linalg.norm(vectors, axis=1)
    zero_mask = norms < 1e-8
    if zero_mask.any():
        print(f"警告: 发现 {zero_mask.sum()} 个零向量，建议过滤")
        vectors = vectors[~zero_mask]
    
    # 检查 NaN/Inf
    if not np.all(np.isfinite(vectors)):
        print("警告: 发现 NaN 或 Inf 值，建议检查嵌入模型输出")
        vectors = vectors[np.all(np.isfinite(vectors), axis=1)]
    
    return vectors
​

2. 索引构建与持久化

import os

class VectorIndexManager:
    """向量索引管理器"""
    
    def __init__(self, dimension, index_dir="./index_data"):
        self.dimension = dimension
        self.index_dir = index_dir
        os.makedirs(index_dir, exist_ok=True)
    
    def build_hnsw(self, vectors, M=32, ef_construction=200):
        """构建 HNSW 索引"""
        index = faiss.IndexHNSWFlat(self.dimension, M, faiss.METRIC_INNER_PRODUCT)
        index.hnsw.efConstruction = ef_construction
        index.add(vectors)
        return index
    
    def save_index(self, index, name="default"):
        """保存索引到磁盘"""
        path = os.path.join(self.index_dir, f"{name}.index")
        faiss.write_index(index, path)
        print(f"索引已保存到: {path}")
        return path
    
    def load_index(self, name="default"):
        """从磁盘加载索引"""
        path = os.path.join(self.index_dir, f"{name}.index")
        if not os.path.exists(path):
            raise FileNotFoundError(f"索引文件不存在: {path}")
        index = faiss.read_index(path)
        print(f"索引已从 {path} 加载，共 {index.ntotal} 个向量")
        return index

# 使用示例
manager = VectorIndexManager(dimension=768)

# 构建并保存
vectors = np.random.rand(100000, 768).astype('float32')
faiss.normalize_L2(vectors)

index = manager.build_hnsw(vectors, M=32, ef_construction=200)
manager.save_index(index, "rag_main")

# 加载并查询
loaded_index = manager.load_index("rag_main")
loaded_index.hnsw.efSearch = 100

query = np.random.rand(1, 768).astype('float32')
faiss.normalize_L2(query)
distances, indices = loaded_index.search(query, k=5)
​

3. 增量更新策略

class IncrementalIndexManager:
    """支持增量更新的索引管理器"""
    
    def __init__(self, dimension):
        self.dimension = dimension
        self.index = None
        self.id_map = []  # 外部 ID 到内部索引的映射
    
    def initialize(self, initial_vectors, initial_ids, M=32):
        """初始化索引"""
        self.index = faiss.IndexHNSWFlat(self.dimension, M, faiss.METRIC_INNER_PRODUCT)
        self.index.hnsw.efConstruction = 200
        self.id_map = list(initial_ids)
        self.index.add(initial_vectors)
    
    def add_vectors(self, new_vectors, new_ids):
        """增量添加向量（HNSW 原生支持）"""
        self.index.add(new_vectors)
        self.id_map.extend(new_ids)
        print(f"已添加 {len(new_ids)} 个向量，当前总数: {self.index.ntotal}")
    
    def search_with_ids(self, query, k=10):
        """查询并返回外部 ID"""
        self.index.hnsw.efSearch = 100
        distances, internal_indices = self.index.search(query, k)
        
        # 将内部索引映射回外部 ID
        external_ids = []
        for idx_list in internal_indices:
            external_ids.append([self.id_map[idx] if idx >= 0 else None for idx in idx_list])
        
        return distances, external_ids
    
    def remove_vectors(self, ids_to_remove):
        """移除向量（FAISS 不支持直接删除，需要重建）"""
        # 标记要保留的向量
        remove_set = set(ids_to_remove)
        keep_indices = [i for i, eid in enumerate(self.id_map) if eid not in remove_set]
        
        if not keep_indices:
            raise ValueError("不能移除所有向量")
        
        # 提取保留的向量
        # 注意：FAISS 没有直接的向量提取接口，建议在外部维护向量副本
        print(f"警告: FAISS 不支持向量删除，需要重建索引")
        print(f"保留 {len(keep_indices)}/{len(self.id_map)} 个向量")
​

常见问题 FAQ

Q1：HNSW 和 IVF 哪个更适合 RAG 知识库？

A：大多数 RAG 场景（百万级以下、内存充足）我推荐 HNSW。它的召回率更高、查询延迟更稳定、不需要调 nprobe。IVF 在千万级超大规模数据下有优势，因为内存占用更可控。实际建议：先用 HNSW 默认参数（M=32, efConstruction=200），只在遇到内存瓶颈时再考虑 IVF-PQ。

Q2：FAISS 索引构建太慢怎么办？

A：构建慢通常有两个原因。第一是 efConstruction 设得太高——降到 100-200 不会有明显质量损失。第二是数据量太大——可以考虑先用少量数据训练聚类中心（IVF），或者分批构建 HNSW（HNSW 支持增量添加）。另外确保使用了 numpy 的 float32 类型，避免隐式类型转换。

Q3：向量索引的召回率怎么评估？

A：标准做法是"以暴力搜索为基准"计算 Recall@k。用同一个查询集，分别跑目标索引和暴力索引，看 top-k 结果的重合比例。一般 HNSW 在合理参数下 Recall@10 能达到 0.98+，IVF 在 nprobe 足够大时也能达到类似水平。建议每次参数调整后都跑一次召回率评估。

Q4：生产环境如何选择距离度量？

A：RAG 场景几乎都用余弦相似度（Cosine Similarity），因为文本语义相似度的衡量不依赖向量长度。在 FAISS 中实现余弦相似度的方式是：先 L2 归一化所有向量，然后使用内积（Inner Product）搜索。这样既准确又高效。

最佳实践与避坑

最佳实践：

• 构建索引前务必 L2 归一化，并检查零向量和 NaN
• HNSW 的 M=32 是一个很好的默认值，适合大多数场景
• efConstruction 对构建质量影响大，建议 ≥ 200
• efSearch 是运行时参数，可以根据实时流量动态调整
• 索引文件要定期保存备份，重建成本可能很高

常见坑点：

• 用 L2 距离构建的索引和用余弦相似度构建的索引，结果排序含义不同，不要混用
• IVF 的 nlist 设太小（如 10）会导致每个簇太大，查询效率反而下降
• PQ 的 m 参数必须能整除向量维度（768 维可被 48 整除，但 768/50 不行）
• HNSW 增量添加大量向量后，建议重建索引以优化图结构
• 忘记设置 efSearch 会导致使用默认值（通常很小），召回率很差

本节小结

向量索引是 RAG 系统性能的关键瓶颈和优化重点。本节我们系统学习了四大类索引结构：基于树的 KD-Tree/Annoy、基于量化的 PQ/SQ、基于图的 HNSW、基于聚类的 IVF 系列。

核心要点回顾：

1. 小规模数据（< 10 万）可以直接暴力搜索，不需要复杂索引
2. 百万级数据推荐 HNSW（M=32, efConstruction=200），综合性能最优
3. 内存紧张时考虑 IVF-PQ，可以用 10-20 倍压缩比换取可接受的精度损失
4. FAISS 是构建向量索引最成熟的工具库，几乎覆盖所有主流算法

下一节（3.4 下）我们将深入混合索引策略和查询优化进阶技术，学习如何将多种索引组合使用，以及如何在生产环境中持续优化检索性能。

延伸阅读

• FAISS 官方文档 v1.8 版本（文字描述，不带链接）
• HNSW 原始论文：Efficient and robust approximate nearest neighbor search using Hierarchical Navigable Small World graphs
• 相关章节：本教程 3.3 节向量数据库对比、3.5 节数据库性能调优

关键词：RAG知识库实战, 向量索引, HNSW, FAISS, IVF, 乘积量化, 检索优化, 索引构建
难度：进阶
预计阅读：35分钟