向量检索是 RAG 系统中承上启下的关键环节：它接收用户问题的向量表示，从已建立的知识库向量索引中快速找到语义最相关的文档片段，为后续的生成提供证据支撑。本节以最轻量的方式展示一个可实际运行的向量检索实现，帮助快速理解其核心机制。
12.3.1 向量检索的本质
给定一个查询向量 \( q \) 和一个包含 \( N \) 个文档片段的向量集合 \( D = \{d_1, d_2, \dots, d_N\} \)，向量检索的目标是找出与 \( q \) 最相似的 \( k \) 个文档向量。相似度通常用余弦相似度或欧氏距离衡量，其中最常用的是余弦相似度：
\[
\text{similarity}(q, d) = \frac{q \cdot d}{\|q\| \cdot \|d\|}
\]
实际工程中，为了效率会使用近似最近邻（ANN）算法，在准确率和速度之间取得平衡。
12.3.2 技术选型：为什么选择 FAISS
对于学习和中小规模应用，Meta 开源的 FAISS 是非常合适的起点：
- 轻量高效：纯 C++ 底层，Python 封装，能高效处理百万级向量。
- 开机即用：支持多种索引类型，可从简单暴力搜索开始，逐步过渡到倒排索引等高级结构。
- 社区成熟：文档丰富，案例众多，排查问题方便。
安装 FAISS CPU 版本最简单：
pip install faiss-cpu如果数据量和实时性要求更高，后续可以平滑迁移到 Milvus、Qdrant 等分布式向量数据库，但核心检索思想完全一致。
12.3.3 逐步实现向量检索
假设我们已经通过嵌入模型（如 OpenAI text-embedding-ada-002 或本地部署的 BGE 模型）将知识库中的每个文档片段转换成了向量，并保存为两个文件：
chunks.npy：形状为(N, dim)的 numpy 数组，每行是一个文档片段的向量。documents.json：包含原始文本和元数据的列表，与向量一一对应。
以下是从零构建检索器的完整代码。
步骤 1：加载向量和文档元数据
import numpy as np
import json
import faiss
# 加载向量矩阵和对应的文档信息
chunks = np.load("chunks.npy").astype('float32') # FAISS 要求 float32
with open("documents.json", "r", encoding="utf-8") as f:
documents = json.load(f)
assert chunks.shape[0] == len(documents), "向量数量与文档数量不匹配"
dim = chunks.shape[1]
print(f"已加载 {len(documents)} 个文档片段，向量维度 {dim}")步骤 2：构建 FAISS 索引
对于数据量不大（例如十万级以下）的场景，可以使用暴力搜索索引 IndexFlatIP（内积索引）。配合对向量进行 L2 归一化，就能实现余弦相似度检索，因为归一化后内积等于余弦相似度。
# 对向量进行 L2 归一化（使相似度计算等价于余弦相似度）
faiss.normalize_L2(chunks)
# 创建内积索引
index = faiss.IndexFlatIP(dim) # IP: Inner Product
index.add(chunks) # 将所有向量加入索引
print(f"索引构建完成，包含 {index.ntotal} 个向量")如果数据量较大，可以将 IndexFlatIP 替换为 faiss.IndexIVFFlat（需要训练）或 faiss.IndexHNSWFlat，以牺牲少量精度换取查询速度的数量级提升。这里先用暴力搜索保证 100% 精确。
步骤 3：编写检索函数
给定一段文本查询，先用同样的嵌入模型生成查询向量，然后从索引中检索最相似的 top-k 个结果，并返回对应的原始文档片段。
def retrieve(query_text: str, embedding_model, index, documents, top_k: int = 5):
"""
参数:
query_text: 用户自然语言问题
embedding_model: 能够将文本转为向量的嵌入模型（例如 openai.Embedding.create）
index: FAISS 索引对象
documents: 文档列表，每个元素包含 content、source 等字段
top_k: 返回的最相似片段数量
返回:
一个列表，每个元素是 (相似度分数, 文档内容, 元数据)
"""
# 1. 将查询文本转为向量
query_vec = embedding_model.embed(query_text) # 具体调用方式取决于模型接口
query_vec = np.array([query_vec]).astype('float32')
faiss.normalize_L2(query_vec) # 同样需要归一化
# 2. 在索引中检索
scores, indices = index.search(query_vec, top_k)
# 3. 整理检索结果
results = []
for score, idx in zip(scores[0], indices[0]):
if idx == -1: # FAISS 用 -1 表示未能填充的位（如结果不足 k 个）
continue
doc = documents[idx]
results.append({
"score": float(score),
"content": doc["content"],
"source": doc.get("source", ""),
"chunk_id": idx
})
return results步骤 4：简单测试
# 假设已有一个嵌入模型实例 embed_model
results = retrieve("怎样修改绑定手机号？", embed_model, index, documents, top_k=3)
for i, res in enumerate(results):
print(f"结果 {i+1} [相似度: {res['score']:.4f}]")
print(f"来源: {res['source']}")
print(res['content'][:200] + "...")
print("-" * 50)12.3.4 实用注意事项
- 向量必须归一化：如果希望使用内积代替余弦相似度，一定要在入库和查询时都做
faiss.normalize_L2，否则相似度值会受向量模长干扰，检索结果不准确。
- 索引类型选择：
- 小于 10 万条向量：
IndexFlatIP完全够用，检索速度毫秒级。 - 10 万到几百万：优先考虑
IndexIVFFlat（需进行小量训练）或IndexHNSWFlat（无需训练，构建稍慢但查询极快）。 - 更大规模：考虑分布式向量数据库（如 Milvus）。
- 向量维度的统一：查询向量由嵌入模型生成，其维度必须与索引中向量的维度完全一致。如果更换了嵌入模型导致维度变化，需要重建整个索引。
- 保存和加载索引：避免每次启动都重建索引，可以使用
faiss.write_index(index, "faiss_index.bin")持久化，启动时用faiss.read_index加载，显著加快就绪时间。
- 批量处理与并发：检索函数内部并未涉及异步，但
index.search支持批量查询（传入多行矩阵），在需要高吞吐时可一次查询多条问题。
通过以上实现，我们得到了一个可独立工作的基础向量检索模块。在实际 RAG 系统中，它就是生成答案前那个“翻资料”的步骤，为后续的大模型提供了扎实的事实基础。后续章节将在此基础上讨论如何与生成模型对接，以及如何评估并提升检索质量。