1. 什么是向量相似度搜索？ #

1.1 生活中的例子 #

想象一下，你在一个巨大的图书馆里，想要找到与"人工智能"相关的所有书籍。传统的方法是：

• 遍历每一本书，检查是否包含"人工智能"关键词
• 这非常慢，特别是当图书馆有几百万本书时

向量相似度搜索提供了一种更高效的方法：

1. 将每本书的内容转换成向量（一组数字）
2. 将查询"人工智能"也转换成向量
3. 快速找到与查询向量最相似的书籍向量

1.2 为什么需要 FAISS？ #

当你有大量向量（比如百万、千万甚至十亿个）需要搜索时，普通的遍历方法太慢了。FAISS 提供了高效的索引和搜索算法，可以在毫秒级时间内找到最相似的向量。

1.3 FAISS 是什么？ #

FAISS（Facebook AI Similarity Search） 是 Facebook AI Research 开发的开源库，专门用于高效相似性搜索和密集向量聚类。

主要特点：

• 高性能：使用优化的 C++ 内核，搜索速度极快
• 支持大规模：可以处理百万到十亿级别的向量
• 多种算法：提供多种索引类型，适应不同场景
• 易于使用：提供完整的 Python 接口

2. 前置知识 #

在学习 FAISS 之前，我们需要了解一些基础概念。

2.1 什么是向量（Vector）？ #

向量是一组有序的数字，用来表示数据。在计算机中，文本、图像、音频等都可以转换成向量。

例如：

• 文本"人工智能"可能被转换成向量：[0.2, 0.8, 0.1, 0.5, ...]（通常有几百个数字）
• 图像可能被转换成向量：[0.1, 0.3, 0.9, 0.2, ...]（通常有几千个数字）

2.2 NumPy 基础 #

FAISS 使用 NumPy 数组来存储向量，所以我们需要了解一些 NumPy 基础知识。

# 导入 NumPy 库
import numpy as np

# 创建一个简单的向量（一维数组）
vector = np.array([1.0, 2.0, 3.0])
print(f"向量: {vector}")
print(f"向量形状: {vector.shape}")

# 创建多个向量（二维数组）
# 每行是一个向量
vectors = np.array([
    [1.0, 2.0, 3.0],
    [4.0, 5.0, 6.0],
    [7.0, 8.0, 9.0]
])
print(f"\n多个向量:")
print(vectors)
print(f"形状: {vectors.shape}")  # (3, 3) 表示3个向量，每个向量3维

# 生成随机向量
random_vectors = np.random.random((5, 10)).astype('float32')
print(f"\n随机向量形状: {random_vectors.shape}")  # 5个向量，每个10维
print(f"数据类型: {random_vectors.dtype}")  # float32

2.3 相似度计算 #

相似度用来衡量两个向量有多相似。常用的相似度计算方法有：

2.3.1 欧氏距离（L2 距离） #

欧氏距离越小，向量越相似。

# 导入 NumPy
import numpy as np

# 定义两个向量
vector_a = np.array([1.0, 2.0, 3.0])
vector_b = np.array([4.0, 5.0, 6.0])

# 计算欧氏距离
# 公式：√[(x1-y1)² + (x2-y2)² + ...]
distance = np.sqrt(np.sum((vector_a - vector_b) ** 2))
print(f"欧氏距离: {distance:.4f}")

# 使用 NumPy 的 linalg.norm 函数（更简单）
distance2 = np.linalg.norm(vector_a - vector_b)
print(f"欧氏距离（方法2）: {distance2:.4f}")

2.3.2 余弦相似度 #

余弦相似度越大，向量越相似（范围 -1 到 1）。

# 导入 NumPy
import numpy as np

# 定义两个向量
vector_a = np.array([1.0, 2.0, 3.0])
vector_b = np.array([4.0, 5.0, 6.0])

# 计算余弦相似度
# 公式：cos(θ) = (A·B) / (|A| × |B|)
dot_product = np.dot(vector_a, vector_b)  # 点积
norm_a = np.linalg.norm(vector_a)  # 向量A的长度
norm_b = np.linalg.norm(vector_b)  # 向量B的长度
cosine_similarity = dot_product / (norm_a * norm_b)

print(f"余弦相似度: {cosine_similarity:.4f}")

# 归一化后，内积就等于余弦相似度
vector_a_norm = vector_a / np.linalg.norm(vector_a)
vector_b_norm = vector_b / np.linalg.norm(vector_b)
cosine_similarity2 = np.dot(vector_a_norm, vector_b_norm)
print(f"余弦相似度（归一化后）: {cosine_similarity2:.4f}")

2.4 什么是索引（Index）？ #

索引是一种数据结构，用来快速查找数据。就像书的目录一样，索引帮助我们快速找到想要的内容，而不需要遍历所有数据。

在 FAISS 中，索引定义了如何存储向量和如何搜索向量。

3. 安装 FAISS #

3.1 安装 CPU 版本 #

FAISS 有 CPU 版本和 GPU 版本。对于大多数用户，CPU 版本就足够了。

# 使用 pip 安装 CPU 版本
pip install faiss-cpu

# 或者使用 conda 安装
conda install -c conda-forge faiss-cpu

3.2 验证安装 #

安装完成后，可以验证是否安装成功：

# 导入 faiss 库
import faiss

# 打印 FAISS 版本信息
print(f"FAISS 版本: {faiss.__version__}")

# 测试基本功能：创建一个简单的索引
d = 64  # 向量维度
index = faiss.IndexFlatL2(d)  # 创建 L2 距离索引
print(f"索引创建成功！向量维度: {d}")

4. 找到最相似的向量 #

让我们从一个最简单的例子开始，了解 FAISS 的基本用法。

# 导入必要的库
import numpy as np
import faiss

# 设置随机种子，确保结果可重复
np.random.seed(42)

# 准备数据
# d: 向量维度（每个向量有多少个数字）
d = 64
# nb: 数据库中的向量数量
nb = 1000
# nq: 查询向量的数量
nq = 5

# 生成随机向量作为数据库
# 形状：(1000, 64) 表示1000个向量，每个向量64维
xb = np.random.random((nb, d)).astype('float32')

# 生成随机向量作为查询
# 形状：(5, 64) 表示5个查询向量，每个向量64维
xq = np.random.random((nq, d)).astype('float32')

print(f"数据库向量形状: {xb.shape}")
print(f"查询向量形状: {xq.shape}")

# 创建索引
# IndexFlatL2: 使用 L2 距离（欧氏距离）的精确搜索索引
index = faiss.IndexFlatL2(d)

# 查看索引中的向量数量（初始为0）
print(f"\n初始索引向量数: {index.ntotal}")

# 将数据库向量添加到索引中
index.add(xb)

# 再次查看索引中的向量数量
print(f"添加后索引向量数: {index.ntotal}")

# 执行搜索
# k: 返回最相似的 k 个向量
k = 4
# search 方法返回两个结果：
# D: 距离矩阵（每个查询向量与最相似向量的距离）
# I: 索引矩阵（每个查询向量对应的最相似向量在数据库中的索引）
D, I = index.search(xq, k)

print(f"\n距离矩阵形状: {D.shape}")  # (5, 4) 表示5个查询，每个返回4个结果
print(f"索引矩阵形状: {I.shape}")  # (5, 4)

# 显示搜索结果
print("\n搜索结果:")
for i in range(nq):
    print(f"\n查询 {i+1}:")
    print(f"  查询向量: {xq[i][:5]}...")  # 只显示前5个数字
    print(f"  最相似的 {k} 个向量:")
    for j in range(k):
        idx = I[i][j]  # 数据库中的索引
        dist = D[i][j]  # 距离
        print(f"    {j+1}. 索引 {idx}, 距离 {dist:.4f}")

FAISS 的基本使用流程：

1. 准备数据：创建向量数组（必须是 float32 类型）
2. 创建索引：选择合适的索引类型
3. 添加向量：将数据库向量添加到索引中
4. 执行搜索：使用查询向量搜索最相似的向量

5. 索引类型详解 #

FAISS 提供了多种索引类型，适应不同的场景。让我们详细了解几种常用的索引。

5.1 IndexFlatL2：精确搜索（最简单） #

特点：

• 精度：100% 精确（找到真正最相似的向量）
• 速度：较慢（需要计算所有向量的距离）
• 内存：高（存储所有原始向量）
• 适用场景：小数据集（< 10万向量），需要精确结果

# 导入必要的库
import numpy as np
import faiss

# 准备数据
d = 64  # 向量维度
nb = 10000  # 数据库大小
nq = 10  # 查询数量

np.random.seed(42)
xb = np.random.random((nb, d)).astype('float32')
xq = np.random.random((nq, d)).astype('float32')

# 创建 IndexFlatL2 索引
# L2 表示使用欧氏距离
index = faiss.IndexFlatL2(d)

# 添加向量
index.add(xb)

# 搜索
k = 5
D, I = index.search(xq, k)

# 显示第一个查询的结果
print(f"查询 1 的前 {k} 个最相似向量:")
for i in range(k):
    print(f"  索引: {I[0][i]}, 距离: {D[0][i]:.4f}")

5.2 IndexFlatIP：内积搜索（用于余弦相似度） #

特点：

• 使用内积（点积）作为相似度度量
• 归一化后，内积等于余弦相似度
• 其他特点与 IndexFlatL2 相同

# 导入必要的库
import numpy as np
import faiss

# 准备数据
d = 64
nb = 1000
nq = 5

np.random.seed(42)
xb = np.random.random((nb, d)).astype('float32')
xq = np.random.random((nq, d)).astype('float32')

# 归一化向量（重要！）
# 归一化后，内积就等于余弦相似度
faiss.normalize_L2(xb)  # 归一化数据库向量
faiss.normalize_L2(xq)  # 归一化查询向量

# 创建 IndexFlatIP 索引
# IP 表示 Inner Product（内积）
index = faiss.IndexFlatIP(d)

# 添加向量
index.add(xb)

# 搜索
k = 3
D, I = index.search(xq, k)

# 显示结果
# 注意：D 中的值是相似度（越大越相似），不是距离
print("使用余弦相似度搜索:")
for i in range(nq):
    print(f"\n查询 {i+1}:")
    for j in range(k):
        idx = I[i][j]
        similarity = D[i][j]  # 这是相似度，不是距离
        print(f"  索引 {idx}, 相似度 {similarity:.4f}")

5.3 IndexIVFFlat：快速近似搜索 #

特点：

• 精度：高（接近精确结果）
• 速度：快（比 IndexFlatL2 快很多）
• 内存：高（存储原始向量）
• 适用场景：大数据集（> 10万向量），需要快速搜索

工作原理：

1. 先将向量分成多个聚类（cluster）
2. 搜索时只搜索部分聚类，而不是所有向量
3. 通过 nprobe 参数控制搜索的聚类数量

# 导入必要的库
import numpy as np
import faiss

# 准备数据
d = 64  # 向量维度
nb = 50000  # 数据库大小（较大）
nq = 10  # 查询数量

np.random.seed(42)
xb = np.random.random((nb, d)).astype('float32')
xq = np.random.random((nq, d)).astype('float32')

# 创建 IndexIVFFlat 索引
# nlist: 聚类中心的数量（通常设为 sqrt(nb) 到 nb/10 之间）
nlist = 100
# quantizer: 量化器，用于计算距离
quantizer = faiss.IndexFlatL2(d)
# 创建 IVF 索引
index = faiss.IndexIVFFlat(quantizer, d, nlist)

# 训练索引（重要！IVF 索引需要先训练）
# 训练数据应该足够多（至少 nlist 个向量）
print(f"索引是否已训练: {index.is_trained}")
index.train(xb)
print(f"训练后，索引是否已训练: {index.is_trained}")

# 添加向量
index.add(xb)
print(f"索引中的向量数: {index.ntotal}")

# 设置 nprobe：搜索时检查的聚类数量
# nprobe 越大，精度越高，但速度越慢
index.nprobe = 10

# 搜索
k = 5
D, I = index.search(xq, k)

# 显示结果
print("\n搜索结果（前3个查询）:")
for i in range(min(3, nq)):
    print(f"\n查询 {i+1}:")
    for j in range(k):
        print(f"  索引: {I[i][j]}, 距离: {D[i][j]:.4f}")

5.4 IndexHNSW：基于图的快速搜索 #

特点：

• 精度：高
• 速度：很快
• 内存：中等
• 适用场景：需要快速搜索，内存充足

工作原理：

• 使用图结构组织向量
• 每个向量是图中的一个节点
• 通过图的边快速找到相似向量

# 导入必要的库
import numpy as np
import faiss

# 准备数据
d = 64
nb = 10000
nq = 5

np.random.seed(42)
xb = np.random.random((nb, d)).astype('float32')
xq = np.random.random((nq, d)).astype('float32')

# 创建 IndexHNSW 索引
# M: 每个节点的连接数（通常设为 16-64）
# M 越大，精度越高，但内存占用越大
M = 16
index = faiss.IndexHNSWFlat(d, M)

# 添加向量（HNSW 不需要训练）
index.add(xb)
print(f"索引中的向量数: {index.ntotal}")

# 搜索
k = 5
D, I = index.search(xq, k)

# 显示结果
print("\n搜索结果:")
for i in range(nq):
    print(f"\n查询 {i+1}:")
    for j in range(k):
        print(f"  索引: {I[i][j]}, 距离: {D[i][j]:.4f}")

5.5 索引选择指南 #

索引类型	精度	速度	内存	适用场景
IndexFlatL2	100% 精确	慢	高	小数据集（< 10万），需要精确结果
IndexFlatIP	100% 精确	慢	高	小数据集，使用余弦相似度
IndexIVFFlat	高（> 95%）	快	高	大数据集（> 10万），平衡精度和速度
IndexHNSW	高（> 95%）	很快	中	需要快速搜索，内存充足

选择建议：

• 小数据集（< 1万）：使用 IndexFlatL2 或 IndexFlatIP
• 中等数据集（1万-100万）：使用 IndexIVFFlat 或 IndexHNSW
• 大数据集（> 100万）：使用 IndexIVFFlat，考虑使用量化版本

6. 实际应用示例 #

6.1 文本相似度搜索 #

下面是一个完整的文本相似度搜索示例：

# 导入必要的库
import numpy as np
import faiss

# 模拟文本向量（实际应用中，你需要使用文本嵌入模型）
# 这里我们使用随机向量来模拟
np.random.seed(42)

# 准备文档
documents = [
    "人工智能是计算机科学的一个分支",
    "机器学习是人工智能的核心技术",
    "深度学习是机器学习的一个子领域",
    "自然语言处理是人工智能的重要应用",
    "计算机视觉可以识别图像中的物体",
    "强化学习通过与环境交互来学习",
    "神经网络模拟人脑的工作方式",
    "Python 是数据科学中常用的编程语言"
]

# 模拟生成文档向量（实际中需要使用嵌入模型）
# 假设每个文档的向量维度是 128
d = 128
doc_vectors = np.random.random((len(documents), d)).astype('float32')

# 归一化向量（用于余弦相似度）
faiss.normalize_L2(doc_vectors)

# 创建索引（使用内积实现余弦相似度）
index = faiss.IndexFlatIP(d)
index.add(doc_vectors)

# 查询
query_text = "人工智能和机器学习"
# 模拟生成查询向量（实际中需要使用相同的嵌入模型）
query_vector = np.random.random((1, d)).astype('float32')
faiss.normalize_L2(query_vector)

# 搜索最相似的文档
k = 3
D, I = index.search(query_vector, k)

# 显示结果
print(f"查询: '{query_text}'")
print(f"\n最相似的 {k} 个文档:")
for i, (idx, similarity) in enumerate(zip(I[0], D[0])):
    print(f"{i+1}. {documents[idx]} (相似度: {similarity:.4f})")

6.2 简单的推荐系统 #

下面是一个简单的推荐系统示例：

# 导入必要的库
import numpy as np
import faiss

# 模拟推荐系统
# 假设我们有用户和物品的向量表示
np.random.seed(42)

# 用户数量
n_users = 1000
# 物品数量
n_items = 5000
# 向量维度
d = 64

# 生成用户向量（模拟用户偏好）
user_vectors = np.random.random((n_users, d)).astype('float32')
# 生成物品向量（模拟物品特征）
item_vectors = np.random.random((n_items, d)).astype('float32')

# 归一化向量
faiss.normalize_L2(user_vectors)
faiss.normalize_L2(item_vectors)

# 创建物品索引
item_index = faiss.IndexFlatIP(d)
item_index.add(item_vectors)

# 为用户推荐物品
def recommend_items(user_id, k=10):
    """为用户推荐 k 个物品"""
    # 获取用户向量
    user_vec = user_vectors[user_id].reshape(1, -1)

    # 搜索最相似的物品
    D, I = item_index.search(user_vec, k)

    # 返回推荐结果
    recommendations = []
    for i in range(k):
        recommendations.append({
            'item_id': int(I[0][i]),
            'score': float(D[0][i])
        })
    return recommendations

# 测试：为用户 0 推荐物品
user_id = 0
recommendations = recommend_items(user_id, k=5)

print(f"为用户 {user_id} 推荐的物品:")
for i, rec in enumerate(recommendations, 1):
    print(f"{i}. 物品 ID: {rec['item_id']}, 相似度: {rec['score']:.4f}")

7. 保存和加载索引 #

在实际应用中，我们通常需要保存索引以便后续使用。

7.1 保存索引到文件 #

# 导入必要的库
import numpy as np
import faiss
import os

# 准备数据
d = 64
nb = 1000
np.random.seed(42)
xb = np.random.random((nb, d)).astype('float32')

# 创建并训练索引
nlist = 100
quantizer = faiss.IndexFlatL2(d)
index = faiss.IndexIVFFlat(quantizer, d, nlist)
index.train(xb)
index.add(xb)

# 保存索引到文件
index_file = "my_index.faiss"
faiss.write_index(index, index_file)
print(f"索引已保存到: {index_file}")

# 从文件加载索引
loaded_index = faiss.read_index(index_file)
print(f"索引已加载，向量数: {loaded_index.ntotal}")

# 测试加载的索引
xq = np.random.random((5, d)).astype('float32')
D, I = loaded_index.search(xq, 3)
print(f"\n搜索测试成功，找到 {len(I[0])} 个结果")

# 清理：删除测试文件
if os.path.exists(index_file):
    os.remove(index_file)
    print(f"\n已删除测试文件: {index_file}")

8. 常见问题和注意事项 #

8.1 数据类型必须是 float32 #

FAISS 要求向量必须是 float32 类型，不能是 float64 或其他类型。

# 导入必要的库
import numpy as np
import faiss

# 错误：使用 float64
vectors_wrong = np.random.random((100, 64))  # 默认是 float64
# index.add(vectors_wrong)  # 可能会报错

# 正确：转换为 float32
vectors_correct = np.random.random((100, 64)).astype('float32')
index = faiss.IndexFlatL2(64)
index.add(vectors_correct)  # 正确
print("向量添加成功！")

8.2 向量维度必须匹配 #

索引创建时指定的维度必须与添加的向量维度一致。

# 导入必要的库
import numpy as np
import faiss

# 创建索引，指定维度为 64
d = 64
index = faiss.IndexFlatL2(d)

# 正确：向量维度匹配
vectors = np.random.random((100, 64)).astype('float32')
index.add(vectors)
print("向量添加成功！")

# 错误：向量维度不匹配
# vectors_wrong = np.random.random((100, 32)).astype('float32')
# index.add(vectors_wrong)  # 会报错

8.3 IVF 索引需要先训练 #

使用 IndexIVFFlat 等需要训练的索引时，必须先调用 train() 方法。

# 导入必要的库
import numpy as np
import faiss

# 准备数据
d = 64
nb = 1000
np.random.seed(42)
xb = np.random.random((nb, d)).astype('float32')

# 创建 IVF 索引
nlist = 100
quantizer = faiss.IndexFlatL2(d)
index = faiss.IndexIVFFlat(quantizer, d, nlist)

# 检查是否需要训练
print(f"索引是否需要训练: {not index.is_trained}")

# 必须先训练
index.train(xb)
print(f"训练后，索引是否已训练: {index.is_trained}")

# 然后才能添加向量
index.add(xb)
print(f"向量添加成功，索引中的向量数: {index.ntotal}")

8.4 处理空索引 #

搜索前应该检查索引中是否有向量。

# 导入必要的库
import numpy as np
import faiss

# 创建索引
index = faiss.IndexFlatL2(64)

# 检查索引是否为空
if index.ntotal == 0:
    print("警告：索引为空，无法搜索")
else:
    xq = np.random.random((1, 64)).astype('float32')
    D, I = index.search(xq, 5)
    print("搜索成功")

9. 性能优化建议 #

9.1 批量添加向量 #

批量添加向量比逐个添加更高效。

# 导入必要的库
import numpy as np
import faiss

# 创建索引
d = 64
index = faiss.IndexFlatL2(d)

# 准备大量向量
nb = 100000
xb = np.random.random((nb, d)).astype('float32')

# 方法1：一次性添加（推荐）
index.add(xb)
print(f"一次性添加完成，向量数: {index.ntotal}")

# 方法2：批量添加（如果内存有限）
index2 = faiss.IndexFlatL2(d)
batch_size = 10000
for i in range(0, nb, batch_size):
    batch = xb[i:i+batch_size]
    index2.add(batch)
print(f"批量添加完成，向量数: {index2.ntotal}")

9.2 批量搜索 #

批量搜索比逐个搜索更高效。

# 导入必要的库
import numpy as np
import faiss

# 创建索引并添加数据
d = 64
nb = 10000
nq = 100

np.random.seed(42)
index = faiss.IndexFlatL2(d)
xb = np.random.random((nb, d)).astype('float32')
index.add(xb)

# 准备查询向量
xq = np.random.random((nq, d)).astype('float32')

# 批量搜索（推荐）
k = 5
D, I = index.search(xq, k)
print(f"批量搜索完成，查询数: {nq}, 每个查询返回 {k} 个结果")

10. 总结 #

10.1 FAISS 的核心价值 #

1. 高效搜索：可以在毫秒级时间内搜索百万级向量
2. 多种算法：提供多种索引类型，适应不同场景
3. 易于使用：简单的 Python API，几行代码就能使用
4. 开源免费：完全开源，可以自由使用和修改

10.2 使用流程 #

1. 准备数据：将数据转换为向量（float32 类型）
2. 选择索引：根据数据量和需求选择合适的索引类型
3. 创建索引：创建索引对象
4. 训练索引：如果需要（如 IVF 索引）
5. 添加向量：将向量添加到索引中
6. 执行搜索：使用查询向量搜索最相似的向量