1. 这份教程怎么读 #

IndexFlatL2 是 Faiss 向量检索库里最简单的一种索引。它简单到什么程度呢——你只要理解"把查询向量和库里每一个向量都比一遍，挑出最近的几个"这句话，就已经理解了它的全部算法。

但简单不等于不重要。恰恰相反，它是整个向量检索领域的地基：所有花哨的近似算法，最终都要拿 IndexFlatL2 的结果当标准答案来打分。而且你可能已经在用它了——LangChain 的 FAISS 向量库，底层默认建的就是 IndexFlatL2（这一点我们在第 12 章会亲手验证）。

真正会让新手栽跟头的，从来不是算法，而是那些"不报错但结果是错的"的细节。这份教程把重点放在这些地方。

1.1 你会学到什么 #

按顺序读下来，你会依次掌握：

本文所有代码块都是完整、可以直接复制运行的独立脚本，所有输出都是在真机上跑出来的真实结果，没有一行是手写编造的。

1.2 先纠正三个流传很广的错误说法 #

网上（包括这份文档的旧版本）有几个说法已经过时了，先说清楚，免得你被误导：

说法一："Faiss 不支持 float64，必须手动转成 float32。"

不对。现在的 faiss Python 包装层会在 add() 和 search() 里自动调用 np.ascontiguousarray(x, dtype="float32")，你传 float64、int32、甚至内存不连续的切片，它都照单全收，不报错。真正的问题是精度会被悄悄截断——这才是值得提醒的地方。第 7 章有实测。

说法二："距离是平方值这件事无关紧要，反正排序一样。"

排序确实一样，但只要你的代码里出现了"距离小于某个阈值才算匹配"这种判断，含义就完全变了。$5$ 的平方是 $25$，你拿阈值 $3$ 去卡 $25$，会把本该匹配的结果全漏掉。第 13 章坑 8 演示了这个。

说法三："IndexFlatL2 不能删除向量。"

能删，remove_ids() 是支持的。但删完之后所有向量的行号会重新排列，你在外面记的"行号到文档"的对照表会全部错位。第 8 章和第 13 章坑 7 都讲了这个。

1.3 本文的运行环境 #

下面所有输出都来自这套环境。你的数字（尤其是性能测试那一章）会和这里有出入，但结论应该是一致的。

项目 版本
Python 3.13.14
faiss-cpu 1.15.0
numpy 2.5.1
langchain-community 0.4.2
CPU 线程数 12

2. 前置知识：向量、距离、最近邻 #

这一章解释三个词。如果你已经知道"用 384 维向量表示一段文本、然后找余弦最近的那几条"是在干什么，可以直接跳到第 3 章。

2.1 向量就是一串数字 #

在向量检索里，向量（vector）就是一个固定长度的数字列表，仅此而已。比如 [0.12, -0.85, 0.33, 0.07] 就是一个 4 维向量，"4 维"的意思就是它有 4 个数字。

这串数字从哪来？来自嵌入模型（embedding model）。它的作用是把人类看得懂的东西——一句话、一张图、一段音频——转换成一串数字，并且保证一件事：意思相近的东西，转换出来的数字也相近。

比如把三句话喂给嵌入模型：

前两句意思接近，它们的向量也会靠得很近；第三句离得很远。这样一来，"找语义相似的句子"这个模糊的问题，就变成了"找数字上最接近的向量"这个可以精确计算的问题。

常见嵌入模型的维度，让你对 $d$ 这个数字有个概念：

模型 维度 $d$
all-MiniLM-L6-v2 384
BGE-base-zh 768
OpenAI text-embedding-3-small 1536
OpenAI text-embedding-3-large 3072

关键点：一个索引里所有向量的维度必须完全一致。 这是 faiss 最常见的报错来源。同一个索引里不能混用不同模型产出的向量，因为它们的维度和语义空间都不一样。

2.2 欧氏距离：中学那个公式推广到 n 维 #

"两个向量有多接近"需要一个具体的数字来衡量，最直观的衡量方式就是欧氏距离，也叫 L2 距离。

从中学的平面几何说起。平面上两个点 $(x_1, y_1)$ 和 $(x_2, y_2)$，距离是勾股定理：

$$ \text{dist} = \sqrt{(x_1 - x_2)^2 + (y_1 - y_2)^2} $$

三维空间里多加一项：

$$ \text{dist} = \sqrt{(x_1 - x_2)^2 + (y_1 - y_2)^2 + (z_1 - z_2)^2} $$

规律很清楚了——每个维度上的差取平方，全部加起来，再开方。推广到 $d$ 维（从这里开始，$d$ 一律表示维度，距离改用 $|\cdot|_2$ 这个记号），设两个向量 $\mathbf{q} = (q_1, q_2, \ldots, q_d)$ 和 $\mathbf{x} = (x_1, x_2, \ldots, x_d)$：

$$ |\mathbf{q} - \mathbf{x}|2 = \sqrt{\sum{i=1}^{d} (q_i - x_i)^2} $$

这就是 IndexFlatL2 名字里 "L2" 的含义。距离越小，两个向量越接近。

举个能心算的例子：$\mathbf{q} = (0,0,0,0)$，$\mathbf{x} = (3,4,0,0)$。

$$ |\mathbf{q} - \mathbf{x}|_2 = \sqrt{(0-3)^2 + (0-4)^2 + 0 + 0} = \sqrt{9 + 16} = \sqrt{25} = 5 $$

勾三股四弦五。记住这个例子，第 13 章会用它演示一个很常见的坑。

2.3 为什么 faiss 返回的是平方距离 #

这是新手第一个必须知道的细节：faiss 返回的不是上面那个距离，而是它的平方，也就是少了最外面那层开方：

$$ |\mathbf{q} - \mathbf{x}|2^2 = \sum{i=1}^{d} (q_i - x_i)^2 $$

为什么要省掉开方？两个原因：

第一，开方很慢而且没必要。 开方是个相对昂贵的运算。一次查询要算 $N$ 个距离，$N$ 可能是几十万，那就是几十万次开方。而排序只关心大小关系——因为平方根函数在非负数上是单调递增的，也就是说，对任意 $a \geq 0$、$b \geq 0$：

$$ a \geq b \quad \Longleftrightarrow \quad \sqrt{a} \geq \sqrt{b} $$

开不开方，排出来的名次一模一样。既然名次一样，那几十万次开方就纯属浪费。

第二，要真实距离你自己开方就行。 你只需要对最终返回的 $k$ 个结果开方，$k$ 通常就是 10 或者 100，成本可以忽略：

真实距离 = np.sqrt(faiss返回的距离)

什么时候必须开方？ 只要你要把这个数字拿去和一个有物理意义的阈值比较，或者要展示给用户看，就得开方。纯粹排个名次、取前 k 个，不用管。

2.4 什么是"最近邻搜索" #

最近邻搜索（Nearest Neighbor Search，简称 NN 搜索）说的是这么一件事：

给你一个查询向量 $\mathbf{q}$，和一堆候选向量 ${\mathbf{x}_1, \mathbf{x}_2, \ldots, \mathbf{x}_N}$，找出其中离 $\mathbf{q}$ 最近的那 $k$ 个。

因为要找 $k$ 个而不是 1 个，完整叫法是 k 近邻搜索（k-NN）。

搜索方法分两大类，这个分类是理解整个 Faiss 的关键：

精确搜索：老老实实把 $N$ 个距离全算一遍，保证找到的就是真正最近的那 $k$ 个。IndexFlatL2 属于这一类。代价是慢，耗时和 $N$ 成正比。

近似搜索（Approximate Nearest Neighbor，ANN）：用各种技巧只检查一小部分候选，快得多，但有可能漏掉真正的最近邻。IndexIVFFlat、IndexHNSW 都属于这一类。

衡量近似搜索好坏的指标叫召回率（recall）：

$$ \text{recall@}k = \frac{|\,\text{近似搜索返回的 } k \text{ 个} \;\cap\; \text{真正的 } k \text{ 个最近邻}\,|}{k} $$

分母那个"真正的 $k$ 个最近邻"，就得靠 IndexFlatL2 跑出来当标准答案。这就是为什么即使你最后用的是近似索引，也一定绕不开 IndexFlatL2。

2.5 这里的"索引"不是数据库里那个索引 #

看到"索引"（index）这个词，学过数据库的人容易想歪。在 Faiss 里，索引指的是"一个装着向量、并且能对它做搜索的对象"，它同时扮演了"存储容器"和"搜索引擎"两个角色。

你可以这样理解：

不同的索引类型，区别在于"倒进去之后怎么组织这些向量"以及"搜索时用什么策略"。IndexFlatL2 的策略最朴素：原样堆着，搜索时全部扫一遍。

3. 安装与验证 #

Faiss 是 C++ 写的，Python 这边是一层薄薄的包装。好消息是官方提供了预编译的 wheel 包，装起来就一条命令，不需要你有编译环境。

3.1 安装 #

Faiss 在 PyPI 上的包名是 faiss-cpu（注意不叫 faiss）：

pip install faiss-cpu

有几点值得说明：

3.2 验证装好了没有 #

跑一下这段，能打印出最后那行就说明一切正常：

from importlib.metadata import version

import numpy as np

import faiss

# 装的包叫 faiss-cpu，但导入时写 faiss，这一点经常让人困惑
print("faiss-cpu 版本:", version("faiss-cpu"))
print("numpy 版本:", version("numpy"))

# 建一个 4 维的索引，能建出来就说明装好了
index = faiss.IndexFlatL2(4)
# 塞两个向量进去
index.add(np.array([[1, 0, 0, 0], [0, 1, 0, 0]], dtype="float32"))
# 查一下离 [1,0,0,0] 最近的 1 个
D, I = index.search(np.array([[1, 0, 0, 0]], dtype="float32"), 1)

print("索引里有几个向量:", index.ntotal)
print("最近邻的行号:", I[0][0], "（应该是 0）")
print("最近邻的距离:", D[0][0], "（查自己，应该是 0）")
print("一切正常，可以开始了")
faiss-cpu 版本: 1.15.0
numpy 版本: 2.5.1
索引里有几个向量: 2
最近邻的行号: 0 （应该是 0）
最近邻的距离: 0.0 （查自己，应该是 0）
一切正常，可以开始了

注意最后一行：用 [1,0,0,0] 去查，找到的最近邻就是它自己，距离是 0.0。这个"查自己距离为零"的性质后面会反复用到。

4. IndexFlatL2 到底是什么 #

上一章你已经跑通了它。这一章补上原理：它内部是怎么组织数据的、搜索时到底在算什么、以及这种做法的代价在哪。理解了这些，第 11 章的性能数据和第 14 章的选型建议才不会是死记硬背。

4.1 把名字拆开看 #

IndexFlatL2 这个名字由三部分拼成，每一部分都在告诉你它的一个特性：

部分 含义
Index 这是一个 Faiss 索引对象，负责装向量 + 提供搜索
Flat "平铺"存储，向量原封不动挨个排好，不压缩、不量化、不分组
L2 用 L2 距离（欧氏距离）衡量远近

Flat 是这里最关键的词。 它意味着索引里存的就是你给它的原始向量，一个比特都没改。好处是精确——既然数据没被动过手脚，算出来的距离自然是准的，而且随时能把原向量完整取回来（第 8 章的 reconstruct）。代价是内存，$N$ 个 $d$ 维向量老老实实占 $N \times d \times 4$ 字节，一分不少。

与之相对，IndexPQ 里的 PQ（乘积量化）会把向量压缩成很短的编码，内存能省几十倍，但存进去的向量取不回来了，算出来的距离也只是个近似值。

4.2 工作原理只有三步 #

第一步，存。 add() 把向量按顺序追加到一块连续内存里。第 1 个进来的编号是 0，第 2 个是 1，依此类推。这个编号就是后面搜索返回的"行号"。没有任何预处理，所以 add 非常快。

第二步，算。 收到查询向量 $\mathbf{q}$ 后，和库里每一个 $\mathbf{x}_j$ 都算一次平方距离：

$$ d_j^2 = \sum_{i=1}^{d} (q_i - x_{j,i})^2, \qquad j = 1, 2, \ldots, N $$

实际实现上，faiss 不会傻乎乎地写双重循环，而是用这个恒等式把它变成矩阵乘法：

$$ |\mathbf{q} - \mathbf{x}|^2 = |\mathbf{q}|^2 - 2\,\mathbf{q} \cdot \mathbf{x} + |\mathbf{x}|^2 $$

其中 $|\mathbf{x}|^2$ 对每个库向量是固定的，可以提前算好存着；$\mathbf{q} \cdot \mathbf{x}$ 这一项对一批查询来说就是一次大矩阵乘法，可以交给高度优化的 BLAS 库去做，还能多线程并行。这就是为什么 faiss 能比你自己写的 numpy 快好几倍（第 11 章有实测对比）。

第三步，挑。 从 $N$ 个距离里选出最小的 $k$ 个。faiss 用的是堆（heap），不需要把 $N$ 个数全排序，复杂度是 $O(N \log k)$ 而不是 $O(N \log N)$。

4.3 代价：时间和内存 #

时间复杂度 是 $O(N \cdot d)$ ——每个向量要算 $d$ 个维度的差。注意这里 $d$ 也在里面，所以 1536 维的向量比 384 维的慢大约 4 倍。

内存占用 是一个精确到字节的公式：

$$ \text{内存} = N \times d \times 4 \ \text{字节} $$

那个 4 是因为 faiss 内部统一用 float32 存储，一个 float32 占 4 个字节。这个公式在第 11 章有实测验证，一个字节都不差。

心里有个数：

向量数 $N$ 维度 $d$ 内存
1 万 128 4.9 MB
10 万 384 146.5 MB
100 万 1536 5859.4 MB（约 5.7 GB）

最后那一行说明了为什么大规模场景不能用 IndexFlatL2——光是存向量就要 5.7 GB 内存，还没算查询开销。

5. 第一个完整例子 #

理论说完了，来跑代码。下面这段是一个从头到尾的完整流程，六个步骤对应实际使用中你一定会写的六段代码。

代码里用了 np.random.default_rng(2024) 固定随机种子，这样你在自己机器上跑出来的数字会和文档里一模一样，方便对照。

import numpy as np

import faiss

# 固定随机种子，保证你跑出来的数字和文档里一模一样
rng = np.random.default_rng(2024)

# ---------- 第一步：决定向量维度 ----------
# d 是每个向量有多少个数字。真实场景里这个数由你的嵌入模型决定，
# 比如 all-MiniLM-L6-v2 是 384 维，OpenAI 的 text-embedding-3-small 是 1536 维
d = 8

# ---------- 第二步：建索引 ----------
# IndexFlatL2 只需要告诉它维度，不需要任何其他参数
index = faiss.IndexFlatL2(d)
# is_trained 表示"能不能直接用"。有些索引要先喂数据学习分布，IndexFlatL2 不用，所以一上来就是 True
print("需要训练吗:", not index.is_trained)
# 刚建好，里面一个向量都没有
print("当前向量数:", index.ntotal)

# ---------- 第三步：准备数据 ----------
# 造 6 个 8 维向量当作"数据库"。必须是二维数组：行是向量，列是维度
data = rng.random((6, d)).astype("float32")
print("数据形状:", data.shape, "| 数据类型:", data.dtype)

# ---------- 第四步：把向量加进索引 ----------
index.add(data)
print("加完之后的向量数:", index.ntotal)

# ---------- 第五步：查询 ----------
# 造 2 个查询向量。查询也必须是二维数组，哪怕只查一条
query = rng.random((2, d)).astype("float32")
# k 表示每个查询要返回几个最近的邻居
k = 3
# search 返回两个数组：距离 D 和行号 I
D, I = index.search(query, k)

print()
print("D 的形状:", D.shape, "（2 个查询 x 3 个邻居）")
print("I 的形状:", I.shape)
print()

# ---------- 第六步：读结果 ----------
# 逐个查询打印它的近邻
for qi in range(len(query)):
    print(f"第 {qi} 个查询向量的 {k} 个最近邻：")
    # zip 把行号和距离配对，enumerate 给出名次
    for rank, (row, dist) in enumerate(zip(I[qi], D[qi]), start=1):
        # dist 是平方距离，开方才是真正的欧氏距离
        print(f"  第 {rank} 名: data 的第 {row} 行, 平方距离={dist:.4f}, 实际距离={np.sqrt(dist):.4f}")
需要训练吗: False
当前向量数: 0
数据形状: (6, 8) | 数据类型: float32
加完之后的向量数: 6

D 的形状: (2, 3) （2 个查询 x 3 个邻居）
I 的形状: (2, 3)

第 0 个查询向量的 3 个最近邻：
  第 1 名: data 的第 3 行, 平方距离=0.2676, 实际距离=0.5173
  第 2 名: data 的第 1 行, 平方距离=0.3403, 实际距离=0.5833
  第 3 名: data 的第 5 行, 平方距离=0.4513, 实际距离=0.6718
第 1 个查询向量的 3 个最近邻：
  第 1 名: data 的第 5 行, 平方距离=0.8517, 实际距离=0.9229
  第 2 名: data 的第 1 行, 平方距离=0.8682, 实际距离=0.9318
  第 3 名: data 的第 2 行, 平方距离=0.9045, 实际距离=0.9511

有几处值得停下来看一眼：

is_trained 一上来就是 True。 很多 Faiss 索引（比如 IndexIVFFlat、IndexPQ）必须先喂一批样本数据 train() 一下，学习数据的分布规律，才能开始用。IndexFlatL2 什么都不用学，建好就能用。这是它最省心的地方。

建索引只需要一个参数 $d$。 没有任何需要调的超参数。对比一下，HNSW 有 M、efConstruction、efSearch 三个参数要调，调不好性能差好几倍。

查询向量必须是二维的。 就算只查一条，也得写成 (1, d) 的形状，不能是 (d,)。这是最高频的报错，第 7 章会详细说。

每个查询的 3 个结果是按距离从近到远排好序的，第 1 名就是最近的。这一点由 faiss 保证，不用你自己排。

6. 读懂返回的 D 和 I #

search() 返回两个 numpy 数组，习惯上叫 D 和 I。它们的形状都是 (查询条数, k)。

下面这段代码把关于这两个数组你需要知道的一切都验证了一遍。

import numpy as np

import faiss

# 固定随机种子，你跑出来的数字会和文档里完全一样
rng = np.random.default_rng(7)
# 用 4 维，方便一眼看完整个向量
d = 4

# 造一个小数据库，只有 5 个向量，方便手工核对
data = rng.random((5, d)).astype("float32")
# 建索引，参数只有维度
index = faiss.IndexFlatL2(d)
# 把 5 个向量一次性装进去
index.add(data)

# 一个查询向量
query = rng.random((1, d)).astype("float32")

print("=" * 60)
print("一、D 到底是平方距离还是欧氏距离？")
print("=" * 60)
# k 取 5，也就是把全部向量都返回，方便和手算结果比
D, I = index.search(query, 5)
print("faiss 返回的 D[0]:", np.round(D[0], 6))

# 手动算平方距离：每个维度的差取平方，再沿维度方向求和
manual_sq = ((query[0] - data) ** 2).sum(axis=1)
print("手算的平方距离  :", np.round(np.sort(manual_sq), 6))
# 开方才是课本上的欧氏距离
print("手算的欧氏距离  :", np.round(np.sort(np.sqrt(manual_sq)), 6))
# allclose 用来比较浮点数组，避免因为末位误差判成不相等
print("D 等于平方距离吗:", np.allclose(np.sort(D[0]), np.sort(manual_sq)))
print("要拿真实距离，自己开方即可: np.sqrt(D)")

print()
print("=" * 60)
print("二、I 是什么？")
print("=" * 60)
print("I[0] =", I[0])
print("它是 data 的行号，用它可以直接把原向量取回来：")
# 用最近邻的行号去 data 里取出那一行
nearest_row = I[0][0]
print(f"  最近的是第 {nearest_row} 行:", np.round(data[nearest_row], 4))
print("  查询向量本身       :", np.round(query[0], 4))

print()
print("=" * 60)
print("三、结果一定是按距离从近到远排好序的")
print("=" * 60)
print("D[0] =", np.round(D[0], 4))
# np.diff 算相邻元素的差，全部大于等于 0 就说明是升序
print("是升序吗:", bool(np.all(np.diff(D[0]) >= 0)))

print()
print("=" * 60)
print("四、k 比库里的向量数还大会怎样？")
print("=" * 60)
# 库里只有 5 个，却要 8 个
D2, I2 = index.search(query, 8)
print("k=8 但库里只有 5 个：")
print("  I =", I2[0])
print("  D =", D2[0])
print("  不够的位置：行号填 -1，距离填 float32 的最大值 3.4028235e+38")
print("  所以拿到结果后要过滤掉 -1，不能直接拿去当下标用")
# 正确的过滤方式：只保留行号不等于 -1 的
valid = I2[0] != -1
print("  过滤后的有效行号:", I2[0][valid])

print()
print("=" * 60)
print("五、查询自己，距离是 0")
print("=" * 60)
# 把 data 的第 2 行当查询向量，最近邻必然是它自己
# 注意写成 data[2:3] 而不是 data[2]，切片才能保持二维形状
D3, I3 = index.search(data[2:3], 2)
print("拿 data[2] 当查询：")
print("  最近的行号:", I3[0][0], "（就是它自己）")
print("  距离:", D3[0][0])
print("  第二近的行号:", I3[0][1], "距离:", round(float(D3[0][1]), 4))
print("  这个特性很有用：做去重时，第 2 名才是真正的最近邻居")
============================================================
一、D 到底是平方距离还是欧氏距离？
============================================================
faiss 返回的 D[0]: [0.354022 0.583916 0.770571 1.489013 1.901844]
手算的平方距离  : [0.354022 0.583916 0.770571 1.489013 1.901844]
手算的欧氏距离  : [0.594997 0.764144 0.877822 1.220251 1.379074]
D 等于平方距离吗: True
要拿真实距离，自己开方即可: np.sqrt(D)

============================================================
二、I 是什么？
============================================================
I[0] = [3 2 0 1 4]
它是 data 的行号，用它可以直接把原向量取回来：
  最近的是第 3 行: [0.2549 0.4451 0.5045 0.5535]
  查询向量本身       : [0.2153 0.1602 0.6125 0.0439]

============================================================
三、结果一定是按距离从近到远排好序的
============================================================
D[0] = [0.354  0.5839 0.7706 1.489  1.9018]
是升序吗: True

============================================================
四、k 比库里的向量数还大会怎样？
============================================================
k=8 但库里只有 5 个：
  I = [ 3  2  0  1  4 -1 -1 -1]
  D = [3.5402149e-01 5.8391619e-01 7.7057058e-01 1.4890131e+00 1.9018438e+00
 3.4028235e+38 3.4028235e+38 3.4028235e+38]
  不够的位置：行号填 -1，距离填 float32 的最大值 3.4028235e+38
  所以拿到结果后要过滤掉 -1，不能直接拿去当下标用
  过滤后的有效行号: [3 2 0 1 4]

============================================================
五、查询自己，距离是 0
============================================================
拿 data[2] 当查询：
  最近的行号: 2 （就是它自己）
  距离: 0.0
  第二近的行号: 3 距离: 0.4108
  这个特性很有用：做去重时，第 2 名才是真正的最近邻居

6.1 D 是平方距离，实测确认 #

输出第一段把三组数字并排放在一起：faiss 返回的 [0.354022, 0.583916, ...] 和手算的平方距离完全一致，和手算的欧氏距离 [0.594997, 0.764144, ...] 对不上。这就实锤了——D 是平方距离。

6.2 I 是行号，不是别的 #

I[0] = [3 2 0 1 4] 的意思是：最近的是 data 的第 3 行，第二近的是第 2 行，依此类推。

这个行号是 add() 进去时的顺序编号，从 0 开始。faiss 只存向量，不存你的原始数据（文本、图片路径之类的），所以你必须自己维护一份"行号 → 原始数据"的对照表：

文档列表 = ["第一段文本", "第二段文本", ...]
# 向量的第 i 行对应 文档列表[i]
命中的文本 = [文档列表[row] for row in I[0]]

第 12 章会看到，LangChain 的 FAISS 向量库帮你做的主要就是这件事。

6.3 结果自带排序 #

D[0] 是升序的（是升序吗: True），第 1 列永远是最近的。不用自己排。

唯一的例外是 range_search()，它返回的结果不排序，需要自己处理，第 8 章会提到。

6.4 k 超过总数时会出现 -1 #

这是很容易出事的一个点。库里只有 5 个向量，你要 8 个，faiss 不会报错，而是：

危险在于 Python 的负数下标是合法的：文档列表[-1] 会返回列表的最后一个元素，而不是报错。于是你会拿到一条毫不相干的数据，还完全察觉不到。

所以拿到结果之后，第一件事是过滤：

valid = I[0] != -1
有效行号 = I[0][valid]

或者从源头避免，让 k 不超过总数：

k = min(想要的数量, index.ntotal)

6.5 查自己，距离是 0 #

拿 data[2] 当查询向量，返回的第 1 名就是行号 2 本身，距离 0.0。

这个性质在去重场景特别有用：如果你想找"每个向量最像的另一个向量"，就要 k=2 然后取第 2 名，因为第 1 名永远是它自己。

7. 喂给 faiss 的数据要长什么样 #

这一章讲数据格式。看起来琐碎，但新手绝大多数报错都出在这里。

先说结论，标准答案是：形状为 (n, d) 的 float32 二维 numpy 数组。

下面逐条看，如果不符合这个标准会发生什么。

import numpy as np

import faiss

# 固定随机种子，保证输出可复现
rng = np.random.default_rng(11)

print("=" * 62)
print("一、必须是二维数组，一维会报错")
print("=" * 62)
# 建一个 4 维索引
index = faiss.IndexFlatL2(4)
# 一维数组：只有 4 个数字，faiss 不知道这是"1 个 4 维向量"还是"4 个 1 维向量"
one_dim = rng.random(4).astype("float32")
print("一维数组的形状:", one_dim.shape)
try:
    # 直接加一维数组，会在解包形状时失败
    index.add(one_dim)
except Exception as e:
    print("  报错:", type(e).__name__, "-", str(e).splitlines()[0])
# 正确做法：用 reshape(1, -1) 把它变成"1 行 4 列"，-1 表示这一维自动算
fixed = one_dim.reshape(1, -1)
print("reshape(1, -1) 之后:", fixed.shape)
# 改成二维之后就能正常添加了
index.add(fixed)
print("  加进去了，ntotal =", index.ntotal)

print()
print("=" * 62)
print("二、float64 会怎样？（很多老教程说「不支持」，其实不对）")
print("=" * 62)
# numpy 默认造出来就是 float64
d64 = np.array([[1.123456789012345, 2.5, 3.0, 4.0]])
print("数组 dtype:", d64.dtype)
# 新建一个索引来做这个实验
ix = faiss.IndexFlatL2(4)
# 直接加，不报错
ix.add(d64)
print("  加进去了，ntotal =", ix.ntotal)
# 但取回来的时候，dtype 变成了 float32
back = ix.reconstruct(0)
print("  取回来的 dtype:", back.dtype)
print("  原始值  :", repr(d64[0][0]))
print("  取回的值:", repr(float(back[0])))
print("  差了    :", abs(d64[0][0] - float(back[0])))
print("  结论：faiss 的 Python 包装层会自动转成 float32，不报错但会丢精度")
print("  建议还是自己写 .astype('float32')，一来省一次内存拷贝，二来意图清楚")

print()
print("=" * 62)
print("三、整数数组也会被自动转换")
print("=" * 62)
# 再开一个索引做整数实验
ix2 = faiss.IndexFlatL2(4)
# 故意用整数类型
ints = np.array([[1, 2, 3, 4]], dtype="int32")
print("传进去的是:", ints.dtype, ints[0])
# 整数一样能加进去
ix2.add(ints)
print("  取回来的是:", ix2.reconstruct(0).dtype, ix2.reconstruct(0))

print()
print("=" * 62)
print("四、内存不连续的数组也能用")
print("=" * 62)
# 先造一个 8 列的大数组
big = rng.random((5, 8)).astype("float32")
# 按列切片，取出来的数组在内存里是跳着存的
sliced = big[:, :4]
print("切片后 C 连续吗:", sliced.flags["C_CONTIGUOUS"])
# 拿这个不连续的数组去建索引
ix3 = faiss.IndexFlatL2(4)
# 照样能加，包装层会帮你整理内存
ix3.add(sliced)
print("  照样能加，ntotal =", ix3.ntotal)
print("  因为包装层内部调了 np.ascontiguousarray()，会悄悄复制一份")

print()
print("=" * 62)
print("五、维度对不上：报一个没有任何提示的 AssertionError")
print("=" * 62)
# 建一个 4 维索引
ix4 = faiss.IndexFlatL2(4)
try:
    # 索引是 4 维，数据却是 8 维
    ix4.add(rng.random((3, 8)).astype("float32"))
except Exception as e:
    # 把报错信息取出来看看到底说了什么
    msg = str(e).strip()
    print("  报错类型:", type(e).__name__)
    print("  报错信息:", repr(msg), "<- 空的，什么都不说")
    print("  遇到光秃秃的 AssertionError，第一个要怀疑的就是维度对不上")
print("  自己先检查一下最保险:")
# 再造一份 8 维数据，演示怎么提前发现问题
data8 = rng.random((3, 8)).astype("float32")
# 加之前先比对维度，出问题时能给出人能看懂的提示
if data8.shape[1] != ix4.d:
    print(f"    索引维度 {ix4.d}，数据维度 {data8.shape[1]}，对不上")

print()
print("=" * 62)
print("六、推荐的写法：加一个统一的预处理函数")
print("=" * 62)


def prepare(vectors, dim):
    """把任意输入整理成 faiss 要的格式，并且提前把问题暴露出来。"""
    # np.asarray 能接受列表、元组、numpy 数组，统一转成 numpy
    arr = np.asarray(vectors, dtype="float32")
    # 一维就补成一行
    if arr.ndim == 1:
        # -1 让 numpy 自己算列数
        arr = arr.reshape(1, -1)
    # 维度不符时给一个说人话的报错
    if arr.ndim != 2 or arr.shape[1] != dim:
        raise ValueError(f"需要形状 (n, {dim}) 的数组，实际拿到 {arr.shape}")
    # 确保内存连续，避免包装层再复制一次
    return np.ascontiguousarray(arr)


# 用这个函数试试各种输入
ix5 = faiss.IndexFlatL2(4)
# 直接喂 Python 列表也行
ix5.add(prepare([[1, 2, 3, 4], [5, 6, 7, 8]], 4))
print("  用列表喂进去，ntotal =", ix5.ntotal)
# 喂一维的也行
ix5.add(prepare([9, 9, 9, 9], 4))
print("  用一维列表喂进去，ntotal =", ix5.ntotal)
# 维度错了会得到清楚的提示
try:
    # 只有 2 列，但索引要 4 列
    ix5.add(prepare([[1, 2]], 4))
except ValueError as e:
    print("  维度错了会说清楚:", e)
==============================================================
一、必须是二维数组，一维会报错
==============================================================
一维数组的形状: (4,)
  报错: ValueError - not enough values to unpack (expected 2, got 1)
reshape(1, -1) 之后: (1, 4)
  加进去了，ntotal = 1

==============================================================
二、float64 会怎样？（很多老教程说「不支持」，其实不对）
==============================================================
数组 dtype: float64
  加进去了，ntotal = 1
  取回来的 dtype: float32
  原始值  : np.float64(1.123456789012345)
  取回的值: 1.1234568357467651
  差了    : 4.673442011160489e-08
  结论：faiss 的 Python 包装层会自动转成 float32，不报错但会丢精度
  建议还是自己写 .astype('float32')，一来省一次内存拷贝，二来意图清楚

==============================================================
三、整数数组也会被自动转换
==============================================================
传进去的是: int32 [1 2 3 4]
  取回来的是: float32 [1. 2. 3. 4.]

==============================================================
四、内存不连续的数组也能用
==============================================================
切片后 C 连续吗: False
  照样能加，ntotal = 5
  因为包装层内部调了 np.ascontiguousarray()，会悄悄复制一份

==============================================================
五、维度对不上：报一个没有任何提示的 AssertionError
==============================================================
  报错类型: AssertionError
  报错信息: '' <- 空的，什么都不说
  遇到光秃秃的 AssertionError，第一个要怀疑的就是维度对不上
  自己先检查一下最保险:
    索引维度 4，数据维度 8，对不上

==============================================================
六、推荐的写法：加一个统一的预处理函数
==============================================================
  用列表喂进去，ntotal = 2
  用一维列表喂进去，ntotal = 3
  维度错了会说清楚: 需要形状 (n, 4) 的数组，实际拿到 (1, 2)

7.1 必须是二维数组 #

一维数组会报 ValueError: not enough values to unpack (expected 2, got 1)。这个报错信息其实挺清楚的：faiss 想从数组形状里解包出"行数"和"列数"两个值，但一维数组只有一个值。

修复方式是 reshape(1, -1)，-1 表示"这一维你自己算"。(4,) 变成 (1, 4)，意思是"1 个 4 维向量"。

这个坑最常出现在只查询一条的时候：

# 错的
index.search(向量, 5)
# 对的
index.search(向量.reshape(1, -1), 5)

7.2 dtype：自动转换的真相 #

这一段推翻了开头说的那个流传很广的错误说法。实测结果是：

原因在 faiss 的 Python 包装层里，add() 的实现第一行就是：

x = np.ascontiguousarray(x, dtype="float32")

这一句同时干了三件事：转 dtype、拉平内存布局、必要时复制一份。所以你传什么它都能接。

那还需要自己写 .astype("float32") 吗？需要，有两个理由：

  1. 省一次内存拷贝。 如果你传的已经是 float32 的连续数组，ascontiguousarray 会直接返回原数组，零拷贝。否则它要新分配一块内存复制过去。数据量大的时候这个开销不小——100 万个 1536 维向量，多复制一次就是 5.7 GB 的内存峰值。

  2. 意图清楚。 float32 只有约 7 位有效数字，这个精度损失是真实存在的。显式写出来，等于告诉读代码的人"我知道这里会降精度，是有意为之"。

7.3 维度对不上：一个哑巴报错 #

这是最坑的一个：维度不匹配时 faiss 抛出的 AssertionError 报错信息是空字符串，什么线索都不给。

看到光秃秃的 AssertionError，第一反应就该是查维度：

print(index.d, data.shape)

7.4 推荐写法：写一个 prepare 函数 #

与其每次都小心翼翼，不如封装一个预处理函数，一次性把所有问题挡在门外。输出最后一段演示了效果：Python 列表、一维列表都能直接喂，维度错了会得到 需要形状 (n, 4) 的数组，实际拿到 (1, 2) 这种说人话的提示，而不是空白的 AssertionError。

第 15 章的模板里有这个函数的完整版，还加上了 NaN 检查。

8. 常用操作 #

前面只用了 add 和 search。这一章补齐日常会用到的其他操作，其中删除那一节请务必看完，那里有一个特别容易出事的行为。

import numpy as np

import faiss

# 固定随机种子，保证输出可复现
rng = np.random.default_rng(3)
# 用 4 维，向量短一点方便打印
d = 4

print("=" * 62)
print("一、add 可以分多次调用，向量会往后接着排")
print("=" * 62)
# 建一个空索引
index = faiss.IndexFlatL2(d)
# 第一批 3 个向量
first = rng.random((3, d)).astype("float32")
# 第二批 2 个向量
second = rng.random((2, d)).astype("float32")
# 第一批
index.add(first)
print("加了 3 个之后 ntotal =", index.ntotal)
# 第二批，行号从 3 开始接着排
index.add(second)
print("再加 2 个之后 ntotal =", index.ntotal)
print("第 3 行是第二批的第 0 个吗:", np.array_equal(index.reconstruct(3), second[0]))

print()
print("=" * 62)
print("二、reconstruct：按行号把原向量取回来")
print("=" * 62)
# 单个取
print("第 0 行:", np.round(index.reconstruct(0), 4))
# 批量取，从第 0 行开始取 5 个
allv = index.reconstruct_n(0, 5)
print("全部取出来的形状:", allv.shape)
print("和原始数据一致吗:", np.array_equal(allv, np.vstack([first, second])))
print("说明 IndexFlatL2 原样保存了向量，没有做任何压缩")

print()
print("=" * 62)
print("三、remove_ids：能删，但行号会往前挪（重要的坑）")
print("=" * 62)
# 单独建一个索引做删除实验
ix = faiss.IndexFlatL2(d)
# 放 5 个向量进去
v = rng.random((5, d)).astype("float32")
ix.add(v)
print("删除前 ntotal =", ix.ntotal)
# 记下第 4 行长什么样，待会儿看它挪到哪去了
row4_before = ix.reconstruct(4).copy()
# 要删的 id 必须是 int64 数组
removed = ix.remove_ids(np.array([1, 3], dtype="int64"))
print(f"删掉了 {removed} 个，剩余 ntotal = {ix.ntotal}")
print()
print("原来的第 4 行，现在跑到哪了？")
# 把剩下的每一行都和之前记下的比一比
for i in range(ix.ntotal):
    # 逐行比对内容是否完全相同
    same = np.array_equal(ix.reconstruct(i), row4_before)
    print(f"  现在的第 {i} 行 == 原来的第 4 行 ? {same}")
print()
print("结论：删除后所有行号都会重排，你之前记下来的行号全部作废。")
print("如果你的业务要靠 id 关联数据，别用行号，改用下面的 IndexIDMap。")

print()
print("=" * 62)
print("四、IndexIDMap：用自己的 ID，不受行号变动影响")
print("=" * 62)
# IndexIDMap 是个包装层，把真正干活的索引包在里面
base = faiss.IndexFlatL2(d)
# 用 IndexIDMap 把它包起来
idmap = faiss.IndexIDMap(base)
# 准备 3 个向量
vecs = rng.random((3, d)).astype("float32")
# 自己指定 id，必须是 int64
my_ids = np.array([1001, 1002, 1003], dtype="int64")
# 注意这里用 add_with_ids 而不是 add
idmap.add_with_ids(vecs, my_ids)
# 拿第 1 个向量去查，看返回什么
D, I = idmap.search(vecs[1:2], 3)
print("查询结果的 id:", I[0])
print("返回的是我们给的 1002 这类 id，不是 0/1/2")
# 删掉一个 id
idmap.remove_ids(np.array([1002], dtype="int64"))
# 删完再查一次，看剩下的 id 有没有变
D2, I2 = idmap.search(vecs[1:2], 3)
print("删掉 1002 之后再查:", I2[0], "（-1 表示凑不够 k 个）")
print("剩下的 id 没有变，还是 1001 和 1003")

print()
print("=" * 62)
print("五、range_search：不要前 k 个，要「距离小于某个值」的全部")
print("=" * 62)
# 再建一个索引，放多一点数据
ix2 = faiss.IndexFlatL2(d)
# 20 个向量当候选池
pool = rng.random((20, d)).astype("float32")
ix2.add(pool)
# 一个查询向量
q = rng.random((1, d)).astype("float32")
# 半径也是按平方距离算的。想按真实距离 r 来筛，这里要传 r*r
radius = 0.2
# 返回三个值，和 search 的两个值不一样
lims, D3, I3 = ix2.range_search(q, radius)
# lims 记录每个查询的结果在 D3/I3 里的起止位置
start, end = lims[0], lims[1]
print(f"平方距离小于 {radius} 的共有 {end - start} 个:")
# 按 lims 给的范围把这个查询的结果切出来
for i in range(start, end):
    print(f"  行号 {I3[i]:2d}  平方距离 {D3[i]:.4f}")
print()
print("注意 range_search 的结果没有排序，需要自己排:")
# argsort 给出从小到大的下标顺序
order = np.argsort(D3[start:end])
print("  排序后的行号:", I3[start:end][order])

print()
print("=" * 62)
print("六、reset：清空索引重新来")
print("=" * 62)
print("清空前 ntotal =", ix2.ntotal)
# reset 把向量全部清掉，但索引对象本身还在
ix2.reset()
print("清空后 ntotal =", ix2.ntotal)
print("维度还在:", ix2.d, "，可以直接接着 add")
==============================================================
一、add 可以分多次调用，向量会往后接着排
==============================================================
加了 3 个之后 ntotal = 3
再加 2 个之后 ntotal = 5
第 3 行是第二批的第 0 个吗: True

==============================================================
二、reconstruct：按行号把原向量取回来
==============================================================
第 0 行: [0.0856 0.2368 0.8013 0.5822]
全部取出来的形状: (5, 4)
和原始数据一致吗: True
说明 IndexFlatL2 原样保存了向量，没有做任何压缩

==============================================================
三、remove_ids：能删，但行号会往前挪（重要的坑）
==============================================================
删除前 ntotal = 5
删掉了 2 个，剩余 ntotal = 3

原来的第 4 行，现在跑到哪了？
  现在的第 0 行 == 原来的第 4 行 ? False
  现在的第 1 行 == 原来的第 4 行 ? False
  现在的第 2 行 == 原来的第 4 行 ? True

结论：删除后所有行号都会重排，你之前记下来的行号全部作废。
如果你的业务要靠 id 关联数据，别用行号，改用下面的 IndexIDMap。

==============================================================
四、IndexIDMap：用自己的 ID，不受行号变动影响
==============================================================
查询结果的 id: [1002 1003 1001]
返回的是我们给的 1002 这类 id，不是 0/1/2
删掉 1002 之后再查: [1003 1001   -1] （-1 表示凑不够 k 个）
剩下的 id 没有变，还是 1001 和 1003

==============================================================
五、range_search：不要前 k 个，要「距离小于某个值」的全部
==============================================================
平方距离小于 0.2 的共有 4 个:
  行号  1  平方距离 0.1540
  行号  9  平方距离 0.0749
  行号 11  平方距离 0.0823
  行号 18  平方距离 0.0826

注意 range_search 的结果没有排序，需要自己排:
  排序后的行号: [ 9 11 18  1]

==============================================================
六、reset：清空索引重新来
==============================================================
清空前 ntotal = 20
清空后 ntotal = 0
维度还在: 4 ，可以直接接着 add

8.1 add 可以分批调用 #

不需要一次性把所有数据准备好。add() 可以调用任意多次，新向量接在后面排号。先加 3 个（行号 0、1、2），再加 2 个（行号 3、4）。

这对增量场景很友好：新文档来了直接 add 就行，不用重建索引。这也是 IndexFlatL2 相比 IVF 类索引的一个实际优势——IVF 加太多新数据后聚类中心会失准，需要重新训练。

8.2 reconstruct：把向量原样取回来 #

reconstruct(i) 取第 $i$ 行，reconstruct_n(起始, 个数) 批量取。

输出里 和原始数据一致吗: True 这一行很有意义：它证明了 Flat 是真的一个比特都没改。对比一下，如果换成 IndexPQ，取回来的只是压缩后近似还原的向量，跟原始数据对不上。

这个能力的实际用途：你可以只保存 faiss 索引文件，需要原向量时再取出来，不用另外存一份。

8.3 remove_ids：能删，但行号会重排 #

这是本章最重要的一段，请仔细看输出。

删掉行号 1 和 3 之后，原来的第 4 行跑到了第 2 行。也就是说，faiss 把剩下的向量往前挪，重新排了号。

为什么会这样？因为 Flat 索引就是一块连续内存，删除中间的元素后，faiss 的做法是把最后面的元素搬过来填坑，这样最省事。代价就是编号全乱。

这会造成什么后果？ 假设你在外面维护了一个对照表：

对照表 = {0: "文档A", 1: "文档B", 2: "文档C", 3: "文档D", 4: "文档E"}

删除之后，索引返回行号 2，你去查对照表得到"文档C"，但实际上那个位置现在装的是"文档E"的向量。搜索结果和实际内容对不上了，而且没有任何报错。

两个解决办法：

  1. 用 IndexIDMap（下一节），让 faiss 帮你维护稳定的 id。
  2. 不真删，改用软删除：维护一个"已删除"的集合，搜索时多要一些结果再过滤掉。数据量不大时这样最简单，也不用担心行号变动。

8.4 IndexIDMap：用自己的 ID #

IndexIDMap 是一个包装层，把真正干活的索引包在里面，额外维护一张"内部行号 → 你的 id"的映射表。

用法上有两处不同：

好处在输出里很明显：搜索返回的是 [1002 1003 1001] 这样你自己定义的 id，删掉 1002 之后，剩下的 1001 和 1003 编号完全不变。你的对照表永远有效。

代价是每个向量多占 8 字节存 id，以及多一层间接寻址。对绝大多数场景，这点开销换来的稳定性绝对值得。

8.5 range_search：按距离阈值搜 #

search() 回答的是"最近的 $k$ 个是谁"。有时候你想问的是另一个问题："所有距离小于某个值的都给我，有几个算几个。"这时用 range_search()。

它的返回值有三个，形式和 search 不一样：

lims, D, I = index.range_search(查询, 半径)

因为每个查询命中的数量不一样，没法排成整齐的矩阵，所以 faiss 把所有结果拼成一个长数组 D/I，再用 lims 记录每个查询的结果在里面的起止位置。第 $i$ 个查询的结果是 I[lims[i]:lims[i+1]]。

两个必须注意的点：

  1. 半径也是平方距离。 想筛"真实距离小于 $r$"，要传 $r^2$。比如想要距离 0.5 以内，参数应该写 0.25。
  2. 结果没有排序。 看输出，行号是 1, 9, 11, 18，距离是 0.1540, 0.0749, 0.0823, 0.0826，明显乱序。要排序得自己用 np.argsort()。

8.6 reset：清空重来 #

reset() 把所有向量清空，但索引对象还在，维度设置也还在，可以直接接着 add。适合需要周期性全量重建的场景。

9. 保存与加载 #

索引建好之后总不能每次重启程序都重算一遍。Faiss 提供了两种持久化方式：存成文件，或者序列化成字节串。

import os
import tempfile

import numpy as np

import faiss

# 固定随机种子，保证输出可复现
rng = np.random.default_rng(5)
d = 16
n = 100

# ---------- 建一个索引并填数据 ----------
index = faiss.IndexFlatL2(d)
# 造 100 个 16 维向量
data = rng.random((n, d)).astype("float32")
# 装进索引
index.add(data)
print("原索引:", index.ntotal, "个", d, "维向量")

# ---------- 存到磁盘 ----------
# 用临时目录，避免在你的项目里留下垃圾文件；实际用的时候换成自己的路径就行
tmpdir = tempfile.mkdtemp()
# 拼出完整的文件路径，用 os.path.join 保证跨平台
path = os.path.join(tmpdir, "demo.index")
# write_index 把整个索引（包括所有向量）序列化到一个文件
faiss.write_index(index, path)

# 看看生成的文件有多大
size = os.path.getsize(path)
print()
print("文件大小:", size, "字节")
# 纯向量数据应该占 n * d * 4 字节
raw = n * d * 4
print("纯向量数据:", raw, "字节")
print("文件头开销:", size - raw, "字节（存了维度、索引类型这些元信息）")

# ---------- 从磁盘读回来 ----------
loaded = faiss.read_index(path)
print()
print("读回来的索引:", loaded.ntotal, "个", loaded.d, "维向量")
print("类型还是 IndexFlatL2 吗:", type(loaded).__name__)

# ---------- 验证读回来的和原来完全一样 ----------
query = rng.random((3, d)).astype("float32")
# 用原索引搜一次
D1, I1 = index.search(query, 5)
# 用读回来的索引搜一次
D2, I2 = loaded.search(query, 5)
print()
print("原索引和读回来的索引，搜索结果一样吗:")
print("  行号完全相同:", np.array_equal(I1, I2))
print("  距离完全相同:", np.array_equal(D1, D2))

# ---------- 存到内存字节串（不落盘） ----------
print()
print("=" * 62)
print("不想落盘？可以序列化成字节串")
print("=" * 62)
# serialize_index 返回一个 numpy 的 uint8 数组
buf = faiss.serialize_index(index)
print("序列化后的类型:", type(buf).__name__, "| dtype:", buf.dtype, "| 长度:", len(buf))
# 反序列化还原
restored = faiss.deserialize_index(buf)
print("还原后的 ntotal:", restored.ntotal)
# 再搜一次，确认和原来完全一致
D3, I3 = restored.search(query, 5)
print("搜索结果和原来一致吗:", np.array_equal(I1, I3))
print()
print("这个字节串可以塞进 Redis、数据库或者用 pickle 存，比存文件灵活")

# ---------- 清理 ----------
os.remove(path)
# 目录空了才能删掉
os.rmdir(tmpdir)
print()
print("临时文件已清理")
原索引: 100 个 16 维向量

文件大小: 6445 字节
纯向量数据: 6400 字节
文件头开销: 45 字节（存了维度、索引类型这些元信息）

读回来的索引: 100 个 16 维向量
类型还是 IndexFlatL2 吗: IndexFlatL2

原索引和读回来的索引，搜索结果一样吗:
  行号完全相同: True
  距离完全相同: True

==============================================================
不想落盘？可以序列化成字节串
==============================================================
序列化后的类型: ndarray | dtype: uint8 | 长度: 6445
还原后的 ntotal: 100
搜索结果和原来一致吗: True

这个字节串可以塞进 Redis、数据库或者用 pickle 存，比存文件灵活

临时文件已清理

9.1 存文件：write_index 和 read_index #

两个函数，用法直白：

faiss.write_index(index, "我的索引.index")
index = faiss.read_index("我的索引.index")

文件大小值得看一眼。 100 个 16 维向量，纯数据是 $100 \times 16 \times 4 = 6400$ 字节，实际文件 6445 字节，文件头只占 45 字节。这印证了 Flat 索引确实就是把原始向量原样堆着，几乎没有额外开销。你可以用这个估算索引文件会有多大，基本就等于 $N \times d \times 4$。

读回来的对象是完全等价的。 输出里"行号完全相同"和"距离完全相同"都是 True，说明序列化是无损的，连浮点数的最后一位都一样。类型也自动还原成了 IndexFlatL2，read_index 会从文件头里读出原本是什么索引类型。

9.2 存字节串：serialize_index 和 deserialize_index #

如果不想落盘——比如要把索引存进 Redis、放进数据库的 BLOB 字段，或者通过网络传给另一个服务——用这一对：

buf = faiss.serialize_index(index)      # 得到 numpy 的 uint8 数组
index = faiss.deserialize_index(buf)    # 还原

字节串长度和文件大小完全一样（都是 6445），说明这两种方式底层是同一套序列化逻辑，只是一个落盘一个留在内存。

9.3 别忘了原始数据 #

这是个容易忽略的地方：faiss 只存向量，不存你的原始文本、图片路径、元数据。

索引文件读回来之后，你能拿到的只有"第 3 行最近"这样的信息，至于第 3 行对应哪篇文档，faiss 一无所知。所以你必须自己把那份"行号 → 原始数据"的对照表也存下来（用 json、pickle 都行），并且保证两者同步更新。

第 12 章会看到 LangChain 就是这么做的：save_local() 会生成两个文件，index.faiss 是向量，index.pkl 是文本和元数据。

10. 想用余弦相似度怎么办 #

做文本检索的话，你听到的大概率是"余弦相似度"而不是"欧氏距离"。这一章讲清楚两者的关系，以及在 faiss 里正确的做法。

10.1 两者的区别：长度算不算数 #

余弦相似度衡量的是两个向量的夹角，公式是点积除以两个模长的乘积：

$$ \cos(\mathbf{a}, \mathbf{b}) = \frac{\mathbf{a} \cdot \mathbf{b}}{|\mathbf{a}| \, |\mathbf{b}|} $$

取值范围是 $[-1, 1]$，越大越相似（注意方向和距离相反）。

关键区别在于：余弦只看方向，完全不管长度。把一个向量整体放大 10 倍，它和别人的夹角一点没变，余弦相似度也就一点没变。但欧氏距离会变得很大——放大 10 倍后，这个点在空间里跑到很远的地方去了。

在文本检索里，向量的模长往往和文本长度之类的因素相关，并不代表语义。所以大多数时候我们只关心方向，也就是用余弦。

下面的代码故意把第 0 个向量放大 10 倍，看看两种度量会给出多么不同的答案。

import numpy as np

import faiss

# 固定随机种子，保证输出可复现
rng = np.random.default_rng(9)
d = 6
n = 8

# 造一批向量。故意让它们长度差别很大，这样才能看出归一化的作用
data = rng.random((n, d)).astype("float32")
# 把第 0 行放大 10 倍：方向没变，但长度变成原来的 10 倍
data[0] *= 10
# 一个查询向量，保持原样不放大
query = rng.random((1, d)).astype("float32")

print("=" * 64)
print("一、L2 距离会被向量长度影响，余弦相似度不会")
print("=" * 64)
# 每行的模长（也就是向量的"长度"）
norms = np.linalg.norm(data, axis=1)
print("各向量的模长:", np.round(norms, 3))
print("第 0 行被放大了 10 倍，模长明显比别人大")

# 不归一化，直接用 L2 搜
ix_raw = faiss.IndexFlatL2(d)
# 把没归一化的原始向量装进去
ix_raw.add(data)
# k 取 n，把全部结果都要出来看排名
D_raw, I_raw = ix_raw.search(query, n)
print()
print("直接用 L2 搜索的排名:", I_raw[0])
# 手算余弦相似度：点积除以两个模长的乘积
cos = (data @ query[0]) / (norms * np.linalg.norm(query[0]))
# argsort 默认升序，加 [::-1] 变降序，余弦是越大越相似
print("按余弦相似度的排名:", np.argsort(cos)[::-1])
print("两个排名不一样，因为 L2 把「长度差异」也算成了「不相似」")

print()
print("=" * 64)
print("二、正确做法：先 L2 归一化，再用 IndexFlatIP（内积）")
print("=" * 64)
# 复制一份，因为 normalize_L2 是原地修改，会改掉原数组
data_n = data.copy()
# 查询向量同样要复制一份再归一化
query_n = query.copy()
# faiss 自带的归一化函数，把每一行缩放成模长为 1
faiss.normalize_L2(data_n)
# 查询向量也必须归一化，否则算出来的不是余弦
faiss.normalize_L2(query_n)
print("归一化后各行的模长:", np.round(np.linalg.norm(data_n, axis=1), 6))

# 归一化之后，内积就等于余弦相似度
ix_ip = faiss.IndexFlatIP(d)
# 装的是归一化后的向量
ix_ip.add(data_n)
# IP 索引返回的是相似度，降序排列
D_ip, I_ip = ix_ip.search(query_n, n)
print()
print("IndexFlatIP 返回的分数:", np.round(D_ip[0], 4))
print("手算的余弦相似度      :", np.round(np.sort(cos)[::-1], 4))
print("完全一致吗:", np.allclose(np.sort(D_ip[0])[::-1], np.sort(cos)[::-1]))
print("排名:", I_ip[0], "<- 和上面手算的余弦排名一致")
print()
print("注意：IndexFlatIP 返回的是相似度，越大越好，所以它是降序排列的；")
print("而 IndexFlatL2 返回的是距离，越小越好，是升序排列的。")

print()
print("=" * 64)
print("三、归一化之后，用 L2 搜也能得到同样的排名")
print("=" * 64)
# 同样的归一化数据，这次用 L2 索引
ix_l2n = faiss.IndexFlatL2(d)
# 装归一化后的向量
ix_l2n.add(data_n)
# 用归一化后的查询向量去搜
D_l2n, I_l2n = ix_l2n.search(query_n, n)
print("归一化后 L2 的排名:", I_l2n[0])
print("归一化后 IP 的排名:", I_ip[0])
print("排名相同吗:", np.array_equal(I_l2n[0], I_ip[0]))
print()
print("原因是一个简单的恒等式。设 a、b 都已归一化（模长为 1）：")
print("  ||a-b||^2 = ||a||^2 - 2*a·b + ||b||^2 = 1 - 2*cos + 1 = 2 - 2*cos")
print("平方距离越小 <=> 余弦越大，两者是严格的单调关系，所以排名必然一样。")
print()
# 用数字验证这个恒等式
print("用数字验证 2 - 2*cos 是不是等于平方距离:")
# 只看前 3 名就够说明问题了
for rank in range(3):
    # 第 rank 名对应的行号
    row = I_ip[0][rank]
    # IP 索引返回的就是余弦值
    cos_val = D_ip[0][rank]
    # 手算这两个归一化向量之间的平方距离
    l2_val = ((query_n[0] - data_n[row]) ** 2).sum()
    print(f"  行 {row}: 余弦={cos_val:.6f}  2-2*余弦={2 - 2 * cos_val:.6f}  实际平方距离={l2_val:.6f}")

print()
print("=" * 64)
print("四、该选哪个？")
print("=" * 64)
print("要余弦相似度 -> 归一化 + IndexFlatIP，分数直接就是余弦值，好解释")
print("要欧氏距离   -> 不归一化 + IndexFlatL2，长度差异也算进相似度里")
print("两者排名虽然可以一致，但返回的数值含义完全不同，别混用")
================================================================
一、L2 距离会被向量长度影响，余弦相似度不会
================================================================
各向量的模长: [17.773  1.419  1.694  1.81   1.812  1.184  1.242  1.485]
第 0 行被放大了 10 倍，模长明显比别人大

直接用 L2 搜索的排名: [6 3 1 4 5 2 7 0]
按余弦相似度的排名: [3 0 4 6 1 2 5 7]
两个排名不一样，因为 L2 把「长度差异」也算成了「不相似」

================================================================
二、正确做法：先 L2 归一化，再用 IndexFlatIP（内积）
================================================================
归一化后各行的模长: [1. 1. 1. 1. 1. 1. 1. 1.]

IndexFlatIP 返回的分数: [0.9127 0.8883 0.8607 0.8468 0.8059 0.6988 0.6422 0.6261]
手算的余弦相似度      : [0.9127 0.8883 0.8607 0.8468 0.8059 0.6988 0.6422 0.6261]
完全一致吗: True
排名: [3 0 4 6 1 2 5 7] <- 和上面手算的余弦排名一致

注意：IndexFlatIP 返回的是相似度，越大越好，所以它是降序排列的；
而 IndexFlatL2 返回的是距离，越小越好，是升序排列的。

================================================================
三、归一化之后，用 L2 搜也能得到同样的排名
================================================================
归一化后 L2 的排名: [3 0 4 6 1 2 5 7]
归一化后 IP 的排名: [3 0 4 6 1 2 5 7]
排名相同吗: True

原因是一个简单的恒等式。设 a、b 都已归一化（模长为 1）：
  ||a-b||^2 = ||a||^2 - 2*a·b + ||b||^2 = 1 - 2*cos + 1 = 2 - 2*cos
平方距离越小 <=> 余弦越大，两者是严格的单调关系，所以排名必然一样。

用数字验证 2 - 2*cos 是不是等于平方距离:
  行 3: 余弦=0.912684  2-2*余弦=0.174632  实际平方距离=0.174632
  行 0: 余弦=0.888338  2-2*余弦=0.223325  实际平方距离=0.223325
  行 4: 余弦=0.860708  2-2*余弦=0.278584  实际平方距离=0.278584

================================================================
四、该选哪个？
================================================================
要余弦相似度 -> 归一化 + IndexFlatIP，分数直接就是余弦值，好解释
要欧氏距离   -> 不归一化 + IndexFlatL2，长度差异也算进相似度里
两者排名虽然可以一致，但返回的数值含义完全不同，别混用

10.2 长度会毁掉排名：一个直观的例子 #

看输出第一段。第 0 行的模长是 17.773，其他都在 1.2 到 1.8 之间——因为我们故意把它放大了 10 倍。

结果是：

同一个向量，一个说它最不像，一个说它第二像。差别就来自那个放大 10 倍——方向完全没变，纯粹是长度把 L2 距离拉大了。

这在真实场景里是会出事的。如果你的嵌入模型输出的向量模长不统一（很多模型确实如此），直接用 L2 搜索，结果会被模长带偏。

10.3 正确做法：归一化 + IndexFlatIP #

标准流程是两步：

第一步，把所有向量归一化，也就是缩放到模长为 1：

$$ \hat{\mathbf{a}} = \frac{\mathbf{a}}{|\mathbf{a}|} $$

faiss 提供了 faiss.normalize_L2(数组) 来做这件事。注意它是原地修改，会直接改掉你传进去的数组，想保留原数据得先 .copy()。这是第 13 章坑 9。

第二步，用 IndexFlatIP 而不是 IndexFlatL2。 IP 是 Inner Product（内积）的缩写。

为什么归一化之后内积就等于余弦？回头看余弦公式，分母是 $|\mathbf{a}| |\mathbf{b}|$。归一化之后这两个模长都是 1，分母变成 1，于是：

$$ \cos(\hat{\mathbf{a}}, \hat{\mathbf{b}}) = \frac{\hat{\mathbf{a}} \cdot \hat{\mathbf{b}}}{1 \times 1} = \hat{\mathbf{a}} \cdot \hat{\mathbf{b}} $$

余弦相似度就是内积本身。输出验证了这一点：IndexFlatIP 返回的 [0.9127, 0.8883, ...] 和手算余弦一模一样。

一个方向上的区别要记牢：

索引 返回什么 排序方向
IndexFlatL2 距离 升序，越小越相似
IndexFlatIP 相似度 降序，越大越相似

这两个方向是反的，写业务逻辑（比如"分数低于阈值就丢弃"）时最容易搞混。

10.4 归一化之后，L2 和 IP 排名相同 #

输出第三段验证了一件有点反直觉的事：归一化之后，用 IndexFlatL2 搜和用 IndexFlatIP 搜，排名完全一样（都是 [3 0 4 6 1 2 5 7]）。

原因是一个两行就能推完的恒等式。设 $\hat{\mathbf{a}}$、$\hat{\mathbf{b}}$ 都已归一化，展开平方距离：

$$ |\hat{\mathbf{a}} - \hat{\mathbf{b}}|^2 = |\hat{\mathbf{a}}|^2 - 2\,\hat{\mathbf{a}} \cdot \hat{\mathbf{b}} + |\hat{\mathbf{b}}|^2 $$

因为模长都是 1，$|\hat{\mathbf{a}}|^2 = |\hat{\mathbf{b}}|^2 = 1$，代入得到：

$$ \boxed{\ |\hat{\mathbf{a}} - \hat{\mathbf{b}}|^2 = 2 - 2\cos(\hat{\mathbf{a}}, \hat{\mathbf{b}})\ } $$

平方距离是余弦的严格单调递减函数：余弦越大，距离越小。既然是严格单调，排名必然一致。

输出最后用具体数字验证了这个公式：余弦 $0.912684$，$2 - 2 \times 0.912684 = 0.174632$，实际平方距离也是 $0.174632$，小数点后六位都对上了。

这个公式有个实用价值： 如果你手上只有归一化后的 L2 平方距离 $s$，想换算回余弦值，直接用

$$ \cos = 1 - \frac{s}{2} $$

第 12 章会用这个公式把 LangChain 返回的距离换算成余弦。

10.5 到底选哪个 #

虽然归一化后两者排名一致，但返回的数值含义完全不同，别混着用。

11. 性能实测：什么时候该换索引 #

这一章全是在真机上跑出来的数字。目的是回答一个很实际的问题：我的数据量还能不能用 IndexFlatL2？

由于性能测试受 CPU 型号、内存带宽、后台负载影响很大，你跑出来的绝对数字一定和这里不同，但趋势和结论是通用的。

import time

import numpy as np

import faiss

# 固定随机种子，保证每次测的是同一批数据
rng = np.random.default_rng(1)


def timeit(fn, repeat=3):
    """跑几遍取最快的一次，减少系统抖动带来的干扰。"""
    # 先设一个不可能达到的初始值
    best = float("inf")
    # 重复若干次
    for _ in range(repeat):
        # perf_counter 是高精度计时器，适合测短耗时
        t0 = time.perf_counter()
        # 执行被测函数
        fn()
        # 只保留最快的那次，慢的那些多半是被系统打断了
        best = min(best, time.perf_counter() - t0)
    return best * 1000  # 换成毫秒


print("=" * 68)
print("一、查询耗时随数据量增长")
print("=" * 68)
d = 128
nq = 100
k = 10
# 查询向量固定，只改数据库大小
xq = rng.random((nq, d)).astype("float32")
# 记录上一轮的耗时，用来算增长倍数
prev_ms = None
prev_n = None
# 表头不用中文对齐：中文在等宽字体里占两格，f-string 的对齐参数数不准
print(" 库里向量数 | 100条总耗时 | 平均单次查询 | 数据翻倍时耗时涨了几倍")
# 每一档都是上一档的 2 倍
for n in (50_000, 100_000, 200_000, 400_000, 800_000):
    # 造这一档大小的数据库
    xb = rng.random((n, d)).astype("float32")
    # 每一档都建一个全新的索引
    index = faiss.IndexFlatL2(d)
    # 装进去
    index.add(xb)
    # add 之后 faiss 内部已经复制了一份，原数组可以扔掉省内存
    del xb
    # 先跑一次热身，让 CPU 缓存和线程池就位
    index.search(xq[:1], k)
    # 正式计时
    ms = timeit(lambda: index.search(xq, k))
    # 每次数据量都是上一行的 2 倍，看耗时涨了几倍
    grow = "—" if prev_ms is None else f"{ms / prev_ms:.2f}x"
    print(f"{n:11,d} | {ms:8.1f} ms | {ms / nq:9.2f} ms | {grow:>10s}")
    # 存下这一轮的结果，供下一轮对比
    prev_ms, prev_n = ms, n
print()
print("数据翻倍，耗时至少翻倍——这就是 O(N)，没有任何捷径。")
print("前几行涨得比 2 倍还多，是因为几 MB 的向量本来能整个装进 CPU 缓存，")
print("跑得格外快；数据一旦超过缓存，就得老老实实从内存搬，成本才回到常态。")
print()
print("选型时最该盯的是「平均单次查询」那一列：")
print("十几万向量还在毫秒级，到八十万就要十毫秒左右，这时候就该换索引了。")

print()
print("=" * 68)
print("二、faiss 比自己写 numpy 快多少")
print("=" * 68)
# 用 10 万条数据做对比
n = 100_000
xb = rng.random((n, d)).astype("float32")
# 建索引
index = faiss.IndexFlatL2(d)
index.add(xb)
# 热身一次
index.search(xq[:1], k)


def by_faiss():
    """用 faiss 搜。"""
    return index.search(xq, k)


def by_numpy():
    """用展开式 ||a-b||^2 = ||a||^2 - 2ab + ||b||^2 算，避免造巨大的中间数组。"""
    # 每个库向量的模长平方，只跟库有关
    xb_sq = (xb ** 2).sum(axis=1)
    # 每个查询向量的模长平方
    xq_sq = (xq ** 2).sum(axis=1)
    # 广播相加得到完整的距离矩阵，形状是 (查询数, 库大小)
    d2 = xq_sq[:, None] - 2 * (xq @ xb.T) + xb_sq[None, :]
    # argpartition 只把前 k 小的挑到前面，比全排序快
    idx = np.argpartition(d2, k, axis=1)[:, :k]
    # 前 k 个内部还没排好序，再排一次
    rows = np.arange(len(xq))[:, None]
    # 按距离给这 k 个重新排序
    order = np.argsort(d2[rows, idx], axis=1)
    return idx[rows, order]


# 分别计时
t_faiss = timeit(by_faiss)
t_numpy = timeit(by_numpy)
# 再各跑一次拿结果，用来比对是否一致
_, I_faiss = by_faiss()
I_numpy = by_numpy()
print(f"faiss  : {t_faiss:8.1f} ms")
print(f"numpy  : {t_numpy:8.1f} ms")
print(f"faiss 快 {t_numpy / t_faiss:.1f} 倍")
print("两者结果一致率:", f"{(I_faiss == I_numpy).mean():.1%}")
print("一致率 100% 说明 faiss 确实是精确搜索，没有近似")

print()
print("=" * 68)
print("三、批量查询远快于逐条查询")
print("=" * 68)
# 100 条一次性查完
t_batch = timeit(lambda: index.search(xq, k))


def one_by_one():
    """把同样 100 条查询拆成 100 次调用。"""
    # 每次只查一条，循环 100 次
    for i in range(len(xq)):
        # 用切片保持二维形状
        index.search(xq[i:i + 1], k)


# 逐条查询很慢，跑一遍就够了
t_loop = timeit(one_by_one, repeat=1)
print(f"100 条一次性查完 : {t_batch:8.1f} ms")
print(f"100 条循环逐条查 : {t_loop:8.1f} ms")
print(f"批量快 {t_loop / t_batch:.1f} 倍")
print("原因：批量查询能把距离计算变成一次大矩阵乘法，还能多线程并行；")
print("逐条查询每次都要付出调用开销，多线程也用不起来")

print()
print("=" * 68)
print("四、多线程的效果")
print("=" * 68)
# 问 faiss 本机最多能用几个线程
max_threads = faiss.omp_get_max_threads()
print("本机可用线程数:", max_threads)
# 用来记录单线程的基准耗时
base_ms = None
# 依次试 1、2、4 和满线程
for nt in (1, 2, 4, max_threads):
    # 设置 faiss 内部使用的线程数
    faiss.omp_set_num_threads(nt)
    # 换完线程数要重新热身
    index.search(xq[:1], k)
    # 计时
    ms = timeit(lambda: index.search(xq, k))
    # 第一次跑的就是单线程，拿它当基准
    if base_ms is None:
        base_ms = ms
    print(f"  {nt:2d} 线程: {ms:8.1f} ms   相对单线程加速 {base_ms / ms:4.1f}x")
# 用完记得恢复默认，免得影响后面的代码
faiss.omp_set_num_threads(max_threads)

print()
print("=" * 68)
print("五、内存占用：就是 N × d × 4 字节，一分不多")
print("=" * 68)
# 列几组常见规模，感受一下量级
for n_v, d_v in ((10_000, 128), (100_000, 384), (1_000_000, 1536)):
    # float32 一个数占 4 字节
    mb = n_v * d_v * 4 / 1024 / 1024
    print(f"  {n_v:9,d} 个 {d_v:5d} 维向量 -> {mb:9.1f} MB")

# 实测验证一下公式对不对
n_v, d_v = 20_000, 64
# 建一个这个规模的索引
ix = faiss.IndexFlatL2(d_v)
# 填满
ix.add(rng.random((n_v, d_v)).astype("float32"))
# index.codes 是 faiss 内部真正存向量的那块字节数组
actual = faiss.vector_to_array(ix.codes).nbytes
print()
print(f"实测 {n_v} 个 {d_v} 维向量:")
print(f"  公式算出来: {n_v * d_v * 4} 字节")
print(f"  实际占用  : {actual} 字节")
print(f"  一致吗: {actual == n_v * d_v * 4}")
====================================================================
一、查询耗时随数据量增长
====================================================================
 库里向量数 | 100条总耗时 | 平均单次查询 | 数据翻倍时耗时涨了几倍
     50,000 |     14.2 ms |      0.14 ms |          —
    100,000 |     41.0 ms |      0.41 ms |      2.89x
    200,000 |    173.4 ms |      1.73 ms |      4.23x
    400,000 |    410.4 ms |      4.10 ms |      2.37x
    800,000 |    994.7 ms |      9.95 ms |      2.42x

数据翻倍，耗时至少翻倍——这就是 O(N)，没有任何捷径。
前几行涨得比 2 倍还多，是因为几 MB 的向量本来能整个装进 CPU 缓存，
跑得格外快；数据一旦超过缓存，就得老老实实从内存搬，成本才回到常态。

选型时最该盯的是「平均单次查询」那一列：
十几万向量还在毫秒级，到八十万就要十毫秒左右，这时候就该换索引了。

====================================================================
二、faiss 比自己写 numpy 快多少
====================================================================
faiss  :     46.6 ms
numpy  :     91.8 ms
faiss 快 2.0 倍
两者结果一致率: 100.0%
一致率 100% 说明 faiss 确实是精确搜索，没有近似

====================================================================
三、批量查询远快于逐条查询
====================================================================
100 条一次性查完 :     40.6 ms
100 条循环逐条查 :    326.8 ms
批量快 8.1 倍
原因：批量查询能把距离计算变成一次大矩阵乘法，还能多线程并行；
逐条查询每次都要付出调用开销，多线程也用不起来

====================================================================
四、多线程的效果
====================================================================
本机可用线程数: 12
   1 线程:    284.8 ms   相对单线程加速  1.0x
   2 线程:    140.3 ms   相对单线程加速  2.0x
   4 线程:     73.7 ms   相对单线程加速  3.9x
  12 线程:     38.9 ms   相对单线程加速  7.3x

====================================================================
五、内存占用：就是 N × d × 4 字节，一分不多
====================================================================
     10,000 个   128 维向量 ->       4.9 MB
    100,000 个   384 维向量 ->     146.5 MB
  1,000,000 个  1536 维向量 ->    5859.4 MB

实测 20000 个 64 维向量:
  公式算出来: 5120000 字节
  实际占用  : 5120000 字节
  一致吗: True

11.1 耗时随数据量线性增长 #

第一张表是 IndexFlatL2 最本质的性质：数据翻倍，耗时至少翻倍。这就是 $O(N)$，没有任何优化空间，因为算法本身就要求把每个向量都摸一遍。

表的最后两行（40 万→80 万）增长倍数是 2.37 和 2.42，已经很接近理论上的 2.0 了。前面几行涨得更猛，不是测量误差，而是 CPU 缓存在起作用：5 万个 128 维向量只占 24 MB，相当一部分能留在 CPU 缓存里，访问速度比主内存快一个数量级，所以小数据量时显得格外便宜。数据量一旦远超缓存容量，就只能老老实实从内存搬，单位成本才稳定下来。

选型时真正该看的是"平均单次查询"那一列。 这是用户等待的时间。什么算慢，取决于你的场景：

单次查询耗时 体感
< 1 ms 完全无感，随便用
1 ~ 10 ms 可以接受，但已经是数据库查询级别的开销了
> 10 ms 用户能感觉到卡，高并发下会堆积

11.2 faiss 比手写 numpy 快好几倍 #

这一段是给"我自己用 numpy 算不就行了"这个想法的回答。

对比的 numpy 版本已经是优化过的写法了——用了 $|a-b|^2 = |a|^2 - 2ab + |b|^2$ 展开式避免生成巨大的中间数组，用了 argpartition 而不是全排序。即便如此，faiss 还是明显更快，我反复跑了几次，倍数在 2 到 3 倍之间浮动。

差距来自三个地方：faiss 用了 SIMD 指令做向量化、用 OpenMP 做多线程、用堆结构做 top-k 而不是部分排序。这些都是 C++ 层面的优化，numpy 层面做不到。

"两者结果一致率 100%"这一行也很重要，它证明了 IndexFlatL2 确实是精确搜索——和暴力计算的结果一个不差。近似索引就做不到这一点。

11.3 一定要批量查询 #

批量比逐条快 6 到 8 倍，这是最容易拿到的性能提升。

原因有两个：一是批量查询可以把距离计算变成一次大矩阵乘法，BLAS 库对此有极致优化；二是多线程需要足够的工作量才划得来，一次只查一条根本喂不饱 12 个线程。

实践建议： 如果你的服务是一条一条来请求的，考虑攒一个小批次（比如攒够 16 条或者等 10 毫秒）再一起查。这是很常见的优化手段。

11.4 多线程默认就开着 #

从 1 线程到 12 线程，加速 6 到 7 倍。注意不是 12 倍——多线程有调度开销，而且这个任务主要瓶颈在内存带宽，线程再多也搬不快。

你不需要做任何配置，faiss 默认就会用满所有核心。 只有在一种情况下需要手动干预：你自己已经在用多进程或多线程了，faiss 再开一层会导致线程数爆炸、互相抢 CPU。这时候把它限制成单线程：

faiss.omp_set_num_threads(1)

11.5 内存占用可以精确预测 #

最后一段实测验证了公式：2 万个 64 维向量，公式算出 5120000 字节，实际占用 5120000 字节，一个字节都不差。

所以你可以放心地用 $N \times d \times 4$ 来做容量规划。注意这只是索引本身，你的原始文本、元数据要另算。

11.6 到底什么时候该换索引 #

综合上面的数据，给一个可操作的判断标准：

向量规模 建议
< 10 万 放心用 IndexFlatL2，单次查询毫秒级，还省去调参的麻烦
10 万 ~ 100 万 看要求。能接受几十毫秒延迟、并发不高，可以继续用；否则换 HNSW
> 100 万 换索引。首选 IndexHNSWFlat（不用训练、召回率高），内存吃紧就上 IndexIVFPQ

还有两个额外的信号提示你该换了：

12. 在 LangChain 里使用 #

如果你是通过 LangChain 接触到 faiss 的，这一章把两边连起来。核心结论：LangChain 的 FAISS 向量库，底层默认建的就是 IndexFlatL2，前面学的所有东西在这里全部适用。

下面的例子用 DeterministicFakeEmbedding 这个假嵌入模型，好处是不需要任何 API key 就能跑起来。代价是它生成的向量是随机数、没有语义，所以搜索结果和字面意思对不上——但我们这一章要看的是机制，不是语义质量。

import numpy as np
from langchain_core.documents import Document
from langchain_core.embeddings import DeterministicFakeEmbedding

from langchain_community.vectorstores import FAISS

# DeterministicFakeEmbedding 是 LangChain 自带的假嵌入模型：
# 同样的文字永远得到同样的向量，但向量本身是随机数，没有语义。
# 用它是为了让这个例子不需要 API key 就能跑起来，专心看 faiss 这一层怎么工作。
# 真实项目里请换成 OpenAIEmbeddings、HuggingFaceEmbeddings 之类的真模型。
embeddings = DeterministicFakeEmbedding(size=32)

# 五段用来入库的示例文字
texts = [
    "猫是一种常见的宠物",
    "狗是人类忠诚的朋友",
    "苹果是一种水果",
    "香蕉富含钾元素",
    "Python 是一门编程语言",
]

print("=" * 64)
print("一、LangChain 的 FAISS 向量库，底层默认就是 IndexFlatL2")
print("=" * 64)
# from_texts 会先把每段文字送去嵌入，再建索引
store = FAISS.from_texts(texts, embeddings)
# .index 属性就是原生的 faiss 索引对象，可以直接拿出来看
print("底层索引类型:", type(store.index).__name__)
print("向量维度 d  :", store.index.d)
print("向量条数    :", store.index.ntotal)
print("需要训练吗  :", not store.index.is_trained)
print()
print("也就是说，你用 LangChain 建向量库，其实就是在用 IndexFlatL2。")
print("前面学的所有东西在这里全部适用。")

print()
print("=" * 64)
print("二、相似度搜索，以及分数是什么")
print("=" * 64)
# 查询词
query = "宠物"
# similarity_search 只返回文档，不给分数
docs = store.similarity_search(query, k=2)
print(f"查询「{query}」，返回 {len(docs)} 条:")
# 逐条打印命中的文本
for i, doc in enumerate(docs, 1):
    print(f"  {i}. {doc.page_content}")

print()
# 带分数的版本，能看到底层的距离值
pairs = store.similarity_search_with_score(query, k=3)
print("带分数的版本:")
# 每一项是 (文档, 分数) 的二元组
for doc, score in pairs:
    print(f"  分数 {score:.4f}  {doc.page_content}")
print()
print("这个 score 就是 faiss 返回的平方 L2 距离，越小越相似。")
print("注意别被「相似度」这个词误导，它是距离，不是相似度，是越小越好。")
print()
print("（用的是假嵌入模型，所以结果和语义无关，看机制就行）")

print()
print("=" * 64)
print("三、自己算一遍，验证 score 确实是平方 L2 距离")
print("=" * 64)
# 手动把查询文字变成向量
qv = np.array([embeddings.embed_query(query)], dtype="float32")
# 直接调底层 faiss 索引搜索
D, I = store.index.search(qv, 3)
print("直接调 faiss 拿到的距离:", np.round(D[0], 4))
# score 是 numpy 的 float32，套一层 float() 打印出来才干净
print("LangChain 给的分数     :", [round(float(s), 4) for _, s in pairs])
print("一样吗:", np.allclose(D[0], [float(s) for _, s in pairs]))

print()
print("=" * 64)
print("四、想用余弦相似度，加一个参数就行")
print("=" * 64)
# normalize_L2=True 会在存入和查询前都做归一化
store_cos = FAISS.from_texts(texts, embeddings, normalize_L2=True)
print("归一化后的索引类型:", type(store_cos.index).__name__, "（还是 L2，但向量都被归一化了）")
# 归一化之后再搜一次
pairs_cos = store_cos.similarity_search_with_score(query, k=3)
# 把每条结果的平方距离换算成余弦值
for doc, score in pairs_cos:
    # 归一化后平方距离范围是 [0, 4]，用 1 - d/2 换算回余弦值
    print(f"  平方距离 {score:.4f} -> 余弦相似度 {1 - score / 2:.4f}  {doc.page_content}")
print()
print("如果想让分数直接就是余弦值，改用 MAX_INNER_PRODUCT 策略：")
# DistanceStrategy 是 LangChain 用来指定距离度量的枚举
from langchain_community.vectorstores.utils import DistanceStrategy

# 指定内积策略后，底层索引会换成 IndexFlatIP
store_ip = FAISS.from_texts(
    # 同样的文本
    texts,
    # 同样的嵌入模型
    embeddings,
    # 归一化仍然要开，否则内积不等于余弦
    normalize_L2=True,
    # 关键就是这个参数
    distance_strategy=DistanceStrategy.MAX_INNER_PRODUCT,
)
print("  这时底层索引变成:", type(store_ip.index).__name__)
# 这次拿到的分数直接就是余弦值，不用换算
for doc, score in store_ip.similarity_search_with_score(query, k=3):
    print(f"  余弦相似度 {score:.4f}  {doc.page_content}")

print()
print("=" * 64)
print("五、增量添加与保存")
print("=" * 64)
print("添加前:", store.index.ntotal, "条")
# add_texts 底层就是 faiss 的 index.add
store.add_texts(["老虎是猫科动物", "橘子和柚子是近亲"])
print("添加后:", store.index.ntotal, "条")

# 带元数据的写法：用 Document 对象
store.add_documents([Document(page_content="鲸鱼是哺乳动物", metadata={"来源": "百科"})])
print("再加一条带元数据的:", store.index.ntotal, "条")
# 用一模一样的原文去查：同样的文字得到同样的向量，距离必然是 0，一定排第一
found = store.similarity_search("鲸鱼是哺乳动物", k=1)
print("查到的内容:", found[0].page_content)
print("查到的元数据:", found[0].metadata)
print("元数据存在 index.pkl 里，faiss 那边只管向量")

print()
# save_local 会存两个文件：index.faiss 是索引，index.pkl 是文字和元数据
import tempfile

# 存到临时目录，跑完就删掉
tmpdir = tempfile.mkdtemp()
# 把整个向量库保存下来
store.save_local(tmpdir)
import os

print("保存到本地，生成的文件:", sorted(os.listdir(tmpdir)))
print("  index.faiss 是 faiss 索引本身（就是 write_index 存的那个）")
print("  index.pkl   是原文和元数据，faiss 只存向量不存文字，所以要单独存")

# 读回来。allow_dangerous_deserialization 是因为 pkl 用 pickle，
# 加载别人给的文件有安全风险，所以 LangChain 要求你显式确认
loaded = FAISS.load_local(tmpdir, embeddings, allow_dangerous_deserialization=True)
print("读回来的条数:", loaded.index.ntotal)
# 同一个查询在两边应该得到同样的结果
print("搜索结果一致吗:", loaded.similarity_search("宠物", k=1)[0].page_content
      == store.similarity_search("宠物", k=1)[0].page_content)

# 清理临时文件
for f in os.listdir(tmpdir):
    # 先把目录里的文件逐个删掉
    os.remove(os.path.join(tmpdir, f))
# 空目录才能删除
os.rmdir(tmpdir)
print("临时文件已清理")
================================================================
一、LangChain 的 FAISS 向量库，底层默认就是 IndexFlatL2
================================================================
底层索引类型: IndexFlatL2
向量维度 d  : 32
向量条数    : 5
需要训练吗  : False

也就是说，你用 LangChain 建向量库，其实就是在用 IndexFlatL2。
前面学的所有东西在这里全部适用。

================================================================
二、相似度搜索，以及分数是什么
================================================================
查询「宠物」，返回 2 条:
  1. 猫是一种常见的宠物
  2. 狗是人类忠诚的朋友

带分数的版本:
  分数 49.7319  猫是一种常见的宠物
  分数 55.1507  狗是人类忠诚的朋友
  分数 57.7116  Python 是一门编程语言

这个 score 就是 faiss 返回的平方 L2 距离，越小越相似。
注意别被「相似度」这个词误导，它是距离，不是相似度，是越小越好。

（用的是假嵌入模型，所以结果和语义无关，看机制就行）

================================================================
三、自己算一遍，验证 score 确实是平方 L2 距离
================================================================
直接调 faiss 拿到的距离: [49.7319 55.1507 57.7116]
LangChain 给的分数     : [49.7319, 55.1507, 57.7116]
一样吗: True

================================================================
四、想用余弦相似度，加一个参数就行
================================================================
归一化后的索引类型: IndexFlatL2 （还是 L2，但向量都被归一化了）
  平方距离 1.5089 -> 余弦相似度 0.2455  香蕉富含钾元素
  平方距离 1.5124 -> 余弦相似度 0.2438  苹果是一种水果
  平方距离 1.5323 -> 余弦相似度 0.2338  狗是人类忠诚的朋友

如果想让分数直接就是余弦值，改用 MAX_INNER_PRODUCT 策略：
  这时底层索引变成: IndexFlatIP
  余弦相似度 0.2455  香蕉富含钾元素
  余弦相似度 0.2438  苹果是一种水果
  余弦相似度 0.2338  狗是人类忠诚的朋友

================================================================
五、增量添加与保存
================================================================
添加前: 5 条
添加后: 7 条
再加一条带元数据的: 8 条
查到的内容: 鲸鱼是哺乳动物
查到的元数据: {'来源': '百科'}
元数据存在 index.pkl 里，faiss 那边只管向量

保存到本地，生成的文件: ['index.faiss', 'index.pkl']
  index.faiss 是 faiss 索引本身（就是 write_index 存的那个）
  index.pkl   是原文和元数据，faiss 只存向量不存文字，所以要单独存
读回来的条数: 8
搜索结果一致吗: True
临时文件已清理

12.1 底层就是 IndexFlatL2 #

store.index 这个属性直接暴露了底层的原生 faiss 索引对象。输出显示它的类型就是 IndexFlatL2，维度 32（和嵌入模型的输出维度一致），is_trained 是 True。

这意味着：

12.2 similarity_search 返回的"分数"是距离 #

这是 LangChain 用户最容易搞混的地方。similarity_search_with_score() 名字里带 similarity，直觉上应该是"越大越相似"，但它返回的其实是 faiss 的平方 L2 距离，越小越相似。

输出第三段做了交叉验证：手动把查询文本嵌入成向量，直接调 store.index.search()，拿到的距离和 LangChain 给的分数完全一致。

所以如果你写了这样的过滤逻辑：

# 这是错的！score 是距离，不是相似度
好结果 = [d for d, score in 结果 if score > 0.8]

方向就反了，你留下的恰恰是最不相关的那些。正确的写法是 score < 阈值。

12.3 想要余弦相似度：两种改法 #

改法一，加 normalize_L2=True。 这会在存入和查询时都做归一化，底层索引仍然是 IndexFlatL2，但返回的距离已经是归一化空间里的了。用第 10.4 节的公式可以换算回余弦：

$$ \cos = 1 - \frac{\text{平方距离}}{2} $$

输出里用这个公式算出的 0.2455、0.2438、0.2338，和改法二直接得到的余弦值一模一样，公式得到了验证。

改法二，改用 MAX_INNER_PRODUCT 策略。 这时底层索引会变成 IndexFlatIP，返回的分数直接就是余弦值，不用换算，方向也变成"越大越好"：

from langchain_community.vectorstores.utils import DistanceStrategy

store = FAISS.from_texts(
    文本列表,
    嵌入模型,
    normalize_L2=True,
    distance_strategy=DistanceStrategy.MAX_INNER_PRODUCT,
)

推荐用改法二，因为分数含义清晰，定阈值的时候不容易搞反方向。

12.4 保存时会生成两个文件 #

save_local() 生成 index.faiss 和 index.pkl 两个文件，这正好印证了第 9.3 节说的那件事：

加载时要显式传 allow_dangerous_deserialization=True，因为 .pkl 用的是 Python 的 pickle 格式，加载来路不明的 pickle 文件可以执行任意代码，是真实的安全风险。只加载你自己生成的文件。

13. 踩坑清单 #

这一章把最容易出问题的九个点集中起来，每个都有可复现的代码和真实报错。分成两类：会报错的（相对好办，至少你知道出事了）和不报错但结果是错的（危险得多）。

import numpy as np

import faiss

# 固定随机种子，保证输出可复现
rng = np.random.default_rng(13)


def demo(title, fn):
    """跑一段会出问题的代码，把真实报错原样打出来。"""
    print()
    print("-" * 62)
    print(title)
    print("-" * 62)
    try:
        # 执行那段有问题的代码
        fn()
    except Exception as e:
        # 有些报错信息是空的，要特殊处理，不然会看得一头雾水
        msg = str(e).strip()
        # 类型名永远有，先打类型
        print(f"  报错类型: {type(e).__name__}")
        print(f"  报错信息: {msg if msg else '（空的，什么都不说）'}")


print("=" * 62)
print("会报错的坑")
print("=" * 62)


def err_dim():
    """维度不匹配。"""
    # 索引建成 4 维，却往里加 8 维的数据
    ix = faiss.IndexFlatL2(4)
    # 这一行会抛 AssertionError
    ix.add(rng.random((3, 8)).astype("float32"))


# 跑给你看
demo("坑 1：维度对不上", err_dim)
print("  原因：index 的 d 和数据的列数不一致")
print("  这个 AssertionError 一个字都不提示，看到它先去查维度")
print("  排查：print(index.d, data.shape)")


def err_1d():
    """忘了把一维数组变成二维。"""
    # 忘了 reshape，传了一维数组
    ix = faiss.IndexFlatL2(4)
    # 形状是 (4,) 而不是 (1, 4)
    ix.add(np.array([1, 2, 3, 4], dtype="float32"))


# 跑给你看
demo("坑 2：传了一维数组", err_1d)
print("  原因：faiss 要 (n, d) 的二维数组，一维它没法拆成行和列")
print("  修：data.reshape(1, -1)")


def err_k0():
    """k 传了 0。"""
    # 先正常建索引、加数据
    ix = faiss.IndexFlatL2(4)
    # 放 3 个向量
    ix.add(rng.random((3, 4)).astype("float32"))
    # k 必须是正整数
    ix.search(rng.random((1, 4)).astype("float32"), 0)


# 跑给你看
demo("坑 3：k 传了 0", err_k0)
print("  又是一个没有提示的 AssertionError，k 必须 >= 1")


def err_ids():
    """id 数组用了 int32 而不是 int64。"""
    # IndexIDMap 才支持自定义 id
    ix = faiss.IndexIDMap(faiss.IndexFlatL2(4))
    # id 数组必须是 int64，这里给了 int32
    ix.add_with_ids(rng.random((2, 4)).astype("float32"),
                    # 故意用 int32 看看会不会出事
                    np.array([1, 2], dtype="int32"))
    print("  居然没报错，说明 int32 也被自动转成 int64 了，ntotal =", ix.ntotal)


# 跑给你看
demo("坑 4：id 用了 int32（其实没事）", err_ids)


print()
print()
print("=" * 62)
print("不报错、但结果是错的坑（更危险）")
print("=" * 62)

print()
print("-" * 62)
print("坑 5：空索引搜索，静默返回 -1")
print("-" * 62)
# 建好索引但一个向量都不加
empty = faiss.IndexFlatL2(4)
# 直接搜，不会报错
D, I = empty.search(rng.random((1, 4)).astype("float32"), 3)
print("  ntotal =", empty.ntotal, "却照样能搜")
print("  I =", I[0])
print("  D =", D[0])
print("  如果你不检查就拿 I 当下标去取数据，会取到列表的最后一个元素（-1）")
print("  养成习惯：搜完先过滤 I != -1")

print()
print("-" * 62)
print("坑 6：向量里混进 NaN，那条数据会人间蒸发")
print("-" * 62)
# 造 4 个正常向量
v = rng.random((4, 4)).astype("float32")
# 第 2 行的第 0 个数字弄成 NaN，模拟数据清洗没做干净的情况
v[2][0] = np.nan
# 建索引
ix = faiss.IndexFlatL2(4)
# 含 NaN 的数据照样能加进去，不报错
ix.add(v)
# k 取 4，也就是把库里全部向量都要出来
D, I = ix.search(rng.random((1, 4)).astype("float32"), 4)
print("  库里有 4 个向量，第 2 行含 NaN")
print("  搜 k=4 的结果 I =", I[0])
print("  第 2 行没有出现在结果里，反而多了个 -1")
print("  原因：任何数和 NaN 比大小都返回 False，这条向量永远赢不了排序")
print("  它不报错，只是悄悄消失，非常难查")
print("  自检方法:")
# np.isnan 逐个元素判断，any() 只要有一个是 NaN 就返回 True
print("    有 NaN 吗:", bool(np.isnan(v).any()))
# inf 同样会捣乱，一起查掉
print("    有 inf 吗:", bool(np.isinf(v).any()))
# any(axis=1) 按行判断，np.where 给出出问题的行号
print("    哪几行有问题:", np.where(np.isnan(v).any(axis=1))[0])

print()
print("-" * 62)
print("坑 7：删除之后行号全变了")
print("-" * 62)
# 建索引放 5 个向量
ix = faiss.IndexFlatL2(4)
# 填数据
ix.add(rng.random((5, 4)).astype("float32"))
# 假设业务代码里记了一份"行号 -> 文档"的对照表
mapping = {0: "文档A", 1: "文档B", 2: "文档C", 3: "文档D", 4: "文档E"}
# 记下第 4 行的内容，一会儿找它去哪了
target = ix.reconstruct(4).copy()
# 删掉行号 1
ix.remove_ids(np.array([1], dtype="int64"))
print("  删掉行号 1 之后，ntotal =", ix.ntotal)
# 找找原来的第 4 行现在在哪
now = [i for i in range(ix.ntotal) if np.array_equal(ix.reconstruct(i), target)]
print(f"  原来的第 4 行（{mapping[4]}）现在变成了第 {now[0]} 行")
print(f"  但对照表还认为第 {now[0]} 行是 {mapping[now[0]]}，数据全错位了")
print("  解决：用 IndexIDMap 配自己的稳定 id，别依赖行号")

print()
print("-" * 62)
print("坑 8：把平方距离当成了欧氏距离")
print("-" * 62)
# 原点
a = np.array([[0, 0, 0, 0]], dtype="float32")
# 距离原点 5 个单位的点（勾三股四弦五）
b = np.array([[3, 4, 0, 0]], dtype="float32")
# 把 b 放进索引
ix = faiss.IndexFlatL2(4)
ix.add(b)
# 用 a 去查
D, I = ix.search(a, 1)
print("  两个点是 (0,0,0,0) 和 (3,4,0,0)，勾三股四弦五，真实距离应该是 5")
print("  faiss 返回:", D[0][0], "<- 这是 5 的平方")
print("  开方之后:", np.sqrt(D[0][0]))
print("  如果你把阈值定成「距离小于 3 就算相似」，直接拿 D 去比会漏掉一大片")

print()
print("-" * 62)
print("坑 9：normalize_L2 是原地修改，会改掉你的原数组")
print("-" * 62)
# 一个模长为 5 的向量
orig = np.array([[3.0, 4.0, 0.0, 0.0]], dtype="float32")
print("  归一化前:", orig[0])
# 注意这里没有接收返回值，函数直接改了 orig 本身
faiss.normalize_L2(orig)
print("  归一化后:", orig[0], "<- 原来那个变量已经被改了")
print("  想保留原数据，先 .copy() 一份再归一化")
==============================================================
会报错的坑
==============================================================

--------------------------------------------------------------
坑 1：维度对不上
--------------------------------------------------------------
  报错类型: AssertionError
  报错信息: （空的，什么都不说）
  原因：index 的 d 和数据的列数不一致
  这个 AssertionError 一个字都不提示，看到它先去查维度
  排查：print(index.d, data.shape)

--------------------------------------------------------------
坑 2：传了一维数组
--------------------------------------------------------------
  报错类型: ValueError
  报错信息: not enough values to unpack (expected 2, got 1)
  原因：faiss 要 (n, d) 的二维数组，一维它没法拆成行和列
  修：data.reshape(1, -1)

--------------------------------------------------------------
坑 3：k 传了 0
--------------------------------------------------------------
  报错类型: AssertionError
  报错信息: （空的，什么都不说）
  又是一个没有提示的 AssertionError，k 必须 >= 1

--------------------------------------------------------------
坑 4：id 用了 int32（其实没事）
--------------------------------------------------------------
  居然没报错，说明 int32 也被自动转成 int64 了，ntotal = 2


==============================================================
不报错、但结果是错的坑（更危险）
==============================================================

--------------------------------------------------------------
坑 5：空索引搜索，静默返回 -1
--------------------------------------------------------------
  ntotal = 0 却照样能搜
  I = [-1 -1 -1]
  D = [3.4028235e+38 3.4028235e+38 3.4028235e+38]
  如果你不检查就拿 I 当下标去取数据，会取到列表的最后一个元素（-1）
  养成习惯：搜完先过滤 I != -1

--------------------------------------------------------------
坑 6：向量里混进 NaN，那条数据会人间蒸发
--------------------------------------------------------------
  库里有 4 个向量，第 2 行含 NaN
  搜 k=4 的结果 I = [ 0  3  1 -1]
  第 2 行没有出现在结果里，反而多了个 -1
  原因：任何数和 NaN 比大小都返回 False，这条向量永远赢不了排序
  它不报错，只是悄悄消失，非常难查
  自检方法:
    有 NaN 吗: True
    有 inf 吗: False
    哪几行有问题: [2]

--------------------------------------------------------------
坑 7：删除之后行号全变了
--------------------------------------------------------------
  删掉行号 1 之后，ntotal = 4
  原来的第 4 行（文档E）现在变成了第 3 行
  但对照表还认为第 3 行是 文档D，数据全错位了
  解决：用 IndexIDMap 配自己的稳定 id，别依赖行号

--------------------------------------------------------------
坑 8：把平方距离当成了欧氏距离
--------------------------------------------------------------
  两个点是 (0,0,0,0) 和 (3,4,0,0)，勾三股四弦五，真实距离应该是 5
  faiss 返回: 25.0 <- 这是 5 的平方
  开方之后: 5.0
  如果你把阈值定成「距离小于 3 就算相似」，直接拿 D 去比会漏掉一大片

--------------------------------------------------------------
坑 9：normalize_L2 是原地修改，会改掉你的原数组
--------------------------------------------------------------
  归一化前: [3. 4. 0. 0.]
  归一化后: [0.6 0.8 0.  0. ] <- 原来那个变量已经被改了
  想保留原数据，先 .copy() 一份再归一化

13.1 会报错的坑 #

坑 1：维度对不上 → 空白的 AssertionError

最常见，也最难受，因为报错信息是空字符串。看到光秃秃的 AssertionError，第一件事就是 print(index.d, data.shape) 对一下维度。

坑 2：传了一维数组 → ValueError: not enough values to unpack

忘了 reshape(1, -1)。这个报错至少还给了线索。

坑 3：k 传了 0 → 又一个空白的 AssertionError

常见于 k = min(想要的, index.ntotal) 但索引恰好是空的时候。

坑 4：id 用了 int32 → 其实没事

文档常说 id 必须是 int64，实测 int32 也会被自动转换，不报错。但显式写 dtype="int64" 更保险，别指望这个自动转换在所有版本里都存在。

13.2 不报错但结果是错的坑（更危险） #

坑 5：空索引搜索，静默返回 -1

ntotal = 0 的索引照样能搜，返回一堆 -1。要命的是 Python 里负数下标合法，文档列表[-1] 会返回最后一个元素而不是报错。于是你拿到一条完全不相干的数据，还毫无察觉。

养成习惯：搜完先过滤 I != -1。

坑 6：向量里混进 NaN，那条数据会人间蒸发

这是最难查的一个。库里 4 个向量，第 2 行含 NaN，搜 k=4 的结果是 [0 3 1 -1]——第 2 行没了，多了个 -1。

原因是浮点数的特性：任何数和 NaN 比较（包括 <、>、==）都返回 False。所以含 NaN 的向量在排序里永远赢不了任何一场比较，永远进不了结果。

它不报错，只是让一部分数据静悄悄地检索不到。数据清洗没做干净的时候特别容易中招。

入库前自检：

assert np.isfinite(向量数组).all(), "有 NaN 或 inf"

坑 7：删除之后行号全变了

第 8.3 节详细讲过。删掉行号 1 之后，原来的第 4 行变成了第 3 行，而你的对照表还认为第 3 行是"文档D"，数据全错位，没有任何报错。

解决：用 IndexIDMap，或者软删除。

坑 8：把平方距离当成了欧氏距离

用勾三股四弦五验证：两点真实距离是 5，faiss 返回 25。

如果你定了个规则"距离小于 3 就算相似"，直接拿 faiss 的返回值去比，等于实际要求距离小于 $\sqrt{3} \approx 1.73$，会漏掉一大片本该匹配的结果。而且这个错误不会报警，只会让召回率莫名其妙地低。

坑 9：normalize_L2 是原地修改

faiss.normalize_L2(orig) 会直接改掉 orig 本身。[3, 4, 0, 0] 归一化后变成 [0.6, 0.8, 0, 0]，原变量已经不是原来那个了。

如果后面还要用原始向量（比如要展示模长、或者要用另一种方式再算一遍），一定先 .copy()。

14. 和其他索引怎么选 #

Faiss 提供了几十种索引，入门阶段只需要认识这几个。

14.1 对比表 #

下面五个索引覆盖了绝大多数场景。看表时抓住两个维度：精不精确（会不会漏掉真正的最近邻）和要不要训练（能不能建完就用）。IndexFlatL2 在这两项上都是最省心的那一档，代价全部体现在最后两列。

索引类型 精确? 需要训练? 内存 查询速度 什么时候用
IndexFlatL2 精确 不用 高（$N d \times 4$ 字节） 慢，$O(N)$ 10 万条以内；做基准答案
IndexFlatIP 精确 不用 同上 同上 同上，但要余弦相似度时
IndexHNSWFlat 近似 不用 更高（还要存图结构） 很快 百万级，最省心的近似方案
IndexIVFFlat 近似 要 略高于 Flat 快 百万级，内存够
IndexIVFPQ 近似 要 低（压缩几十倍） 快 千万级或内存吃紧

14.2 怎么读这张表 #

"需要训练"是个重要区分。 IVF 和 PQ 系列必须先用一批有代表性的样本调用 train()，让算法学习数据的分布（IVF 要聚类出中心点，PQ 要学码本）。这带来两个麻烦：一是要额外准备训练数据，二是数据分布随时间漂移后，索引质量会下降，需要定期重建。IndexFlatL2 和 HNSW 都不用训练，省心得多。

HNSW 的内存比 Flat 还高。 这一点很多人没想到。HNSW 除了存原始向量，还要存一张多层的邻接图，通常要多占 50% 到 100%。它换来的是速度——百万级数据上单次查询能做到零点几毫秒。

PQ 是唯一真正省内存的。 它把向量压缩成很短的编码，能省几十倍内存。代价是向量取不回来了，距离也只是近似值，召回率会下降。

14.3 升级路线 #

从 IndexFlatL2 出发，按这个顺序考虑：

  1. 数据涨到几十万，觉得慢了 → 换 IndexHNSWFlat。不用训练，代码改动最小，召回率通常能到 95% 以上。
  2. 内存不够了 → 换 IndexIVFPQ。要训练，要调参，但内存能降一个数量级。
  3. 不知道换完效果如何 → 用 IndexFlatL2 跑出标准答案，算一下召回率再决定。这正是它最大的价值。

第 3 条值得展开一句：评估近似索引的召回率，方法就是拿同一批查询，分别用近似索引和 IndexFlatL2 搜一遍，看重合了多少。 所以即使生产环境用的是 HNSW，开发阶段你还是会用到 IndexFlatL2。

15. 速查表与标准模板 #

前面十四章的内容，浓缩成三张表和一份可以直接抄走的代码。写代码时忘了某个方法怎么调、或者拿不准某个返回值的含义，翻这一章就够了。

15.1 API 速查 #

日常会用到的全部方法都在这里。记住两条贯穿始终的约定：传进去的向量永远是 (n, d) 的 float32 二维数组，返回的距离永远是平方值。

操作 代码
建索引 index = faiss.IndexFlatL2(d)
添加向量 index.add(数组) — 形状 (n, d)，float32
搜索 D, I = index.search(查询, k) — 查询也要二维
按半径搜 lims, D, I = index.range_search(查询, 半径平方)
有几个向量 index.ntotal
维度 index.d
取回原向量 index.reconstruct(行号) / index.reconstruct_n(起点, 个数)
删除 index.remove_ids(np.array([...], dtype="int64")) — 注意行号会变
清空 index.reset()
存文件 faiss.write_index(index, "路径")
读文件 index = faiss.read_index("路径")
存字节串 buf = faiss.serialize_index(index)
读字节串 index = faiss.deserialize_index(buf)
归一化 faiss.normalize_L2(数组) — 原地修改
设线程数 faiss.omp_set_num_threads(n)

15.2 关键数字 #

这些是最容易记混、又最容易导致线上问题的几个具体数值和结论，建议单独记一下。

事项 结论
search 返回的 D 平方距离，要真实距离自己 np.sqrt()
内存占用 $N \times d \times 4$ 字节，精确
索引文件大小 约等于内存占用，文件头只有几十字节
凑不够 k 个时 行号填 -1，距离填 3.4028235e+38
查询自己 距离是 0.0，排第一名
归一化后 平方距离 $= 2 - 2\cos$，即 $\cos = 1 - \frac{距离}{2}$
建议上限 10 万条以内放心用，超过 100 万换索引

15.3 标准模板 #

把前面所有的防坑经验封装成两个函数，可以直接抄进你的项目。

import os
import tempfile

import numpy as np

import faiss


def build_index(vectors, dim):
    """建一个 IndexFlatL2 并把向量装进去。

    vectors: 任意能转成 numpy 的东西，形状要能变成 (n, dim)
    dim:     向量维度
    """
    # 统一转成 float32 的连续数组，避免踩 dtype 和内存布局的坑
    arr = np.ascontiguousarray(np.asarray(vectors, dtype="float32"))
    # 一维就补成一行，方便调用方偷懒
    if arr.ndim == 1:
        # -1 让 numpy 自己算列数
        arr = arr.reshape(1, -1)
    # 维度不对就早点报错，别等 faiss 抛那个没提示的 AssertionError
    if arr.ndim != 2 or arr.shape[1] != dim:
        raise ValueError(f"需要 (n, {dim}) 的数组，实际是 {arr.shape}")
    # 数据脏了要提前发现，NaN 会让向量在搜索里静默消失
    if not np.isfinite(arr).all():
        # 找出具体哪几行有问题，方便定位
        bad = np.where(~np.isfinite(arr).all(axis=1))[0]
        raise ValueError(f"第 {bad.tolist()} 行含 NaN 或 inf")

    # IndexFlatL2 不需要训练，建完直接 add
    index = faiss.IndexFlatL2(dim)
    # 把整理好的向量装进去
    index.add(arr)
    return index


def search(index, queries, k, return_euclidean=True):
    """查询，并把结果整理成好用的形式。

    返回一个列表，每个查询对应一个 [(行号, 距离), ...]
    """
    # 查询向量同样要规整
    q = np.ascontiguousarray(np.asarray(queries, dtype="float32"))
    # 只查一条时允许传一维
    if q.ndim == 1:
        # 补成 (1, d)
        q = q.reshape(1, -1)
    # k 不能超过库里的总数，超了只会得到一堆 -1
    k = min(k, index.ntotal)
    # 空索引直接返回空结果，省得后面处理 -1
    if k <= 0:
        return [[] for _ in range(len(q))]

    # 正式搜索
    D, I = index.search(q, k)
    # faiss 返回平方距离，按需开方成真正的欧氏距离
    if return_euclidean:
        # 开方只对最终的 k 个结果做，成本可以忽略
        D = np.sqrt(D)

    results = []
    # 一个查询一个查询地整理
    for row_ids, dists in zip(I, D):
        # 过滤掉 -1：凑不够 k 个时会用它填充
        valid = row_ids != -1
        # 打包成 (行号, 距离) 的列表，顺便转成 Python 原生类型
        results.append(list(zip(row_ids[valid].tolist(), dists[valid].tolist())))
    return results


# ==================== 用起来 ====================
if __name__ == "__main__":
    # 固定随机种子，保证输出可复现
    rng = np.random.default_rng(2025)
    d = 16

    # 1. 准备数据
    data = rng.random((100, d)).astype("float32")

    # 2. 建索引
    index = build_index(data, d)
    print("索引建好了:", index.ntotal, "个", index.d, "维向量")
    print("占用内存:", index.ntotal * index.d * 4 / 1024, "KB")

    # 3. 查询
    query = rng.random((2, d)).astype("float32")
    # k=3，每个查询要 3 个最近邻
    results = search(index, query, k=3)
    # 逐个查询打印结果
    for qi, hits in enumerate(results):
        print(f"\n第 {qi} 个查询:")
        # enumerate 从 1 开始，正好当名次
        for rank, (row, dist) in enumerate(hits, 1):
            print(f"  第 {rank} 名: 行号 {row:3d}, 欧氏距离 {dist:.4f}")

    # 4. 增量添加
    index.add(rng.random((10, d)).astype("float32"))
    print("\n又加了 10 个，现在有", index.ntotal, "个")

    # 5. 存盘和读回
    tmpdir = tempfile.mkdtemp()
    # 拼出索引文件路径
    path = os.path.join(tmpdir, "my.index")
    # 写盘
    faiss.write_index(index, path)
    # 读回来
    reloaded = faiss.read_index(path)
    # 用同样的查询验证结果一致
    print("存盘再读回:", reloaded.ntotal, "个，搜索结果一致吗:",
          # 注意 results 是加那 10 个之前算的，所以这里应当一致
          search(reloaded, query, 3) == results)

    # 6. 演示防错：脏数据会被拦下来
    dirty = data.copy()
    # 人为制造一个 NaN
    dirty[5][0] = np.nan
    try:
        # build_index 会在入库前把它拦住
        build_index(dirty, d)
    except ValueError as e:
        print("\n脏数据被拦住了:", e)

    # 清理临时文件
    os.remove(path)
    # 空目录才能删掉
    os.rmdir(tmpdir)
索引建好了: 100 个 16 维向量
占用内存: 6.25 KB

第 0 个查询:
  第 1 名: 行号  97, 欧氏距离 0.9068
  第 2 名: 行号  52, 欧氏距离 1.1233
  第 3 名: 行号  80, 欧氏距离 1.2667

第 1 个查询:
  第 1 名: 行号  46, 欧氏距离 1.2338
  第 2 名: 行号  23, 欧氏距离 1.2659
  第 3 名: 行号  21, 欧氏距离 1.3135

又加了 10 个，现在有 110 个
存盘再读回: 110 个，搜索结果一致吗: True

脏数据被拦住了: 第 [5] 行含 NaN 或 inf

模板里做了这几件防护，每一件都对应前面讲过的一个坑：

防护 防的是哪个坑
np.asarray(..., dtype="float32") 第 7.2 节，避免隐式转换和额外拷贝
reshape(1, -1) 第 7.1 节，一维数组
显式检查维度并抛 ValueError 第 7.3 节，空白的 AssertionError
np.isfinite().all() 检查 坑 6，NaN 导致数据静默消失
k = min(k, index.ntotal) 第 6.4 节，k 超出总数
过滤 row_ids != -1 坑 5，-1 当下标
np.sqrt(D) 坑 8，平方距离当成欧氏距离

15.4 最后总结 #

IndexFlatL2 的算法只有一句话：把查询向量和库里每一个向量都算一遍距离，挑最近的 $k$ 个。 没有近似、没有训练、没有参数要调，结果永远是精确的。

它的价值在两个地方：

一是够用。 10 万条向量以内，单次查询毫秒级，完全不需要更复杂的东西。绝大多数个人项目和中小型应用，从头到尾用 IndexFlatL2 就够了。过早引入 HNSW 或 IVF，换来的往往是调参的麻烦和难以察觉的召回率损失。

二是当标准答案。 所有近似索引都要靠它来验证效果。这个角色是不可替代的。

真正需要你留心的，从来不是算法本身，而是那些细节：D 是平方距离、凑不够时填 -1、删除后行号会变、NaN 会让数据静默消失。这些都写在第 13 章了，用之前过一遍。

← 上一节 HNSW 下一节 mermaid →