RAG 系列 5：使用 FAISS 实现本地向量检索

langchain qwen rag embedding faiss vector-search
rag
发布日期:   2026-02-12
文章字数:   3.1k
阅读时长:   13 分
阅读次数:  

上一篇比较了 FAISS、Chroma 和 Milvus。这一篇先使用最轻量的 FAISS，不启动数据库服务，也不使用 LangChain 的 VectorStore。

我们会让本地 Qwen3-Embedding-0.6B 生成 1024 维向量，使用 IndexFlatIP 完成 Top-K 检索，再把索引与元数据写入磁盘。最后用手工二维向量对比 L2 距离与余弦相似度，避免把两种分数方向混淆。

1. 整体流程

FAISS 原生向量检索流程

这张图需要分成两条路径理解：

  • 建立索引时，文档先生成归一化的 float32 向量，再通过 index.add() 写入 IndexFlatIP。
  • 执行查询时，查询文本单独生成查询向量，再通过 index.search() 与索引中的文档向量比较。查询向量不会写入索引。

IndexFlatIP 会逐个比较索引中的全部向量，因此这里完成的是精确检索，不是近似最近邻检索。搜索结果包含分数和位置编号，程序还要使用位置编号回到业务列表中读取文本与元数据。

FAISS 原生索引保存向量及其位置编号。文本、分类和业务 ID 仍由程序管理，因此本篇会保存两个文件：

life_notes.faiss  -> 向量索引
life_notes.json   -> 原始文本和元数据

两个文件必须保持相同的记录顺序。FAISS 返回位置 0 时，程序就读取 JSON 列表中的第 0 条记录。只保存 .faiss 文件并不能恢复原文，只保存 .json 文件也无法直接进行向量检索。

2. 为什么单独创建 Conda 环境

本机是 Apple Silicon，安装的是 CPU 版 FAISS。FAISS 官方安装说明推荐使用 Conda 包，GPU 包面向 CUDA 环境，不适用于 M1 Max 的 Metal GPU。

本项目把 FAISS 放到独立的 .conda_faiss，不与 Chroma 和 PyMilvus 混装。这样可以避开 macOS 上多个 OpenMP 运行库同时加载时出现的错误：

OMP: Error #15: Initializing libomp.dylib,
but found libomp.dylib already initialized.

不要用 KMP_DUPLICATE_LIB_OK=TRUE 掩盖这个错误。环境隔离更容易保证结果稳定。

environment.yml 内容如下：

name: llm_learning_faiss
channels:
  - pytorch
  - conda-forge
dependencies:
  - python=3.12
  - faiss-cpu=1.13.2
  - pip
  - pip:
      - numpy==2.2.6
      - sentence-transformers==5.1.2
      - transformers==4.57.6

进入项目目录并创建环境：

cd source/_posts/llm_learning

conda env create \
  --prefix .conda_faiss \
  --file rag/p05_faiss_native/environment.yml

conda activate "$(pwd)/.conda_faiss"
python -m pip check

本机检查结果为：

No broken requirements found.

3. 创建 FAISS 索引并检索

代码文件为 01_create_and_search_index.py：

# 这个文件使用 Qwen3 Embedding 生成向量，并通过 FAISS 完成 Top-K 相似度检索。
# FAISS 只保存向量和位置编号，原始文本与元数据由 Python 列表单独管理。

from pathlib import Path

import faiss
import torch
from sentence_transformers import SentenceTransformer


PROJECT_ROOT = Path(__file__).resolve().parents[2]
MODEL_PATH = PROJECT_ROOT / "embedding_model"

documents = [
    {"id": "doc-1", "text": "西湖边有步道和树荫，很适合周末散步。", "category": "travel"},
    {"id": "doc-2", "text": "苹果富含膳食纤维，是常见的健康水果。", "category": "food"},
    {"id": "doc-3", "text": "汽车需要定期更换机油并检查轮胎。", "category": "car"},
    {"id": "doc-4", "text": "杭州植物园环境安静，适合慢慢游览。", "category": "travel"},
]
query = "周末想找一个适合散步的地方"

device = "mps" if torch.backends.mps.is_available() else "cpu"
model = SentenceTransformer(str(MODEL_PATH), device=device)

# 文档向量和查询向量都做归一化，内积就等于余弦相似度。
document_vectors = model.encode(
    [item["text"] for item in documents],
    normalize_embeddings=True,
).astype("float32")
query_vector = model.encode(
    [query],
    prompt_name="query",
    normalize_embeddings=True,
).astype("float32")

dimension = document_vectors.shape[1]
index = faiss.IndexFlatIP(dimension)
index.add(document_vectors)

# k 不能超过索引中的向量数，否则空缺位置会返回 -1。
k = min(3, index.ntotal)

# 返回的 positions 是向量在插入列表中的位置，scores 越大越相似。
scores, positions = index.search(query_vector, k=k)

print(f"运行设备：{device}")
print(f"向量维度：{dimension}")
print(f"索引向量数：{index.ntotal}")
print(f"查询：{query}")

for rank, (position, score) in enumerate(zip(positions[0], scores[0]), start=1):
    item = documents[int(position)]
    print(
        f"{rank}. score={score:.4f} id={item['id']} "
        f"category={item['category']} text={item['text']}"
    )

3.1 为什么必须转换成 float32

SentenceTransformer 返回的是 NumPy 数组。FAISS 的常用浮点索引要求输入为连续的 float32 数据，因此代码中明确执行：

.astype("float32")

如果传入维度错误或不兼容的数据类型，add()、search() 可能直接失败。

3.2 为什么使用 IndexFlatIP

文档向量和查询向量都设置了：

normalize_embeddings=True

归一化后每个向量的范数接近 1，此时内积等于余弦相似度。IndexFlatIP 返回的分数越大，语义越接近。

IndexFlatIP 是精确索引，不需要训练。它会直接比较所有向量，适合教学、小数据和结果基线验证。

3.3 位置编号不是业务 ID

当前示例按下面的顺序写入四条向量：

位置 0 -> doc-1
位置 1 -> doc-2
位置 2 -> doc-3
位置 3 -> doc-4

positions 返回的是 FAISS 中的位置，不是 doc-1 这类业务 ID，所以代码必须通过 documents[int(position)] 取回完整记录。更复杂的项目可以使用显式整数 ID 映射，但业务文本和元数据仍然需要由应用保存。

运行代码：

python rag/p05_faiss_native/01_create_and_search_index.py

本机真实输出为：

运行设备：mps
向量维度：1024
索引向量数：4
查询：周末想找一个适合散步的地方
1. score=0.6498 id=doc-1 category=travel text=西湖边有步道和树荫，很适合周末散步。
2. score=0.4660 id=doc-4 category=travel text=杭州植物园环境安静，适合慢慢游览。
3. score=0.2183 id=doc-2 category=food text=苹果富含膳食纤维，是常见的健康水果。

第一名与查询都提到了周末散步，第二名虽然没有出现“散步”，但“环境安静、慢慢游览”的语义仍然接近。

4. 保存并重新加载索引

内存中的 index 会随着 Python 进程退出而消失。FAISS 提供 write_index() 和 read_index() 保存、恢复索引。

FAISS 索引与元数据的持久化边界

保存阶段必须分别写入索引文件和业务数据文件；恢复阶段也要同时读取两者。加载后至少应检查记录数和向量维度，再执行查询。记录数相同只能发现文件缺失或数量变化，不能发现 JSON 被重新排序，因此顺序一致仍然是应用必须维护的约束。

代码文件为 02_save_and_load_index.py：

# 这个文件演示如何保存和重新加载 FAISS 索引。
# FAISS 索引写入 .faiss 文件，文本和元数据另外写入 JSON 文件。

import json
from pathlib import Path

import faiss
import torch
from sentence_transformers import SentenceTransformer


PROJECT_ROOT = Path(__file__).resolve().parents[2]
MODEL_PATH = PROJECT_ROOT / "embedding_model"
DATA_DIR = PROJECT_ROOT / "data" / "p05_faiss_native"
INDEX_FILE = DATA_DIR / "life_notes.faiss"
METADATA_FILE = DATA_DIR / "life_notes.json"

documents = [
    {"id": "doc-1", "text": "西湖边有步道和树荫，很适合周末散步。", "category": "travel"},
    {"id": "doc-2", "text": "苹果富含膳食纤维，是常见的健康水果。", "category": "food"},
    {"id": "doc-3", "text": "汽车需要定期更换机油并检查轮胎。", "category": "car"},
    {"id": "doc-4", "text": "杭州植物园环境安静，适合慢慢游览。", "category": "travel"},
]

DATA_DIR.mkdir(parents=True, exist_ok=True)
device = "mps" if torch.backends.mps.is_available() else "cpu"
model = SentenceTransformer(str(MODEL_PATH), device=device)

document_vectors = model.encode(
    [item["text"] for item in documents],
    normalize_embeddings=True,
).astype("float32")

index = faiss.IndexFlatIP(document_vectors.shape[1])
index.add(document_vectors)

# 向量索引和业务数据分别保存，但二者的顺序必须保持一致。
faiss.write_index(index, str(INDEX_FILE))
METADATA_FILE.write_text(
    json.dumps(documents, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

# 模拟程序重启后重新读取索引与元数据。
loaded_index = faiss.read_index(str(INDEX_FILE))
loaded_documents = json.loads(METADATA_FILE.read_text(encoding="utf-8"))

# 索引位置与 JSON 列表按顺序对应，记录数不一致时应立即停止检索。
if loaded_index.ntotal != len(loaded_documents):
    raise ValueError("FAISS 索引与 JSON 元数据的记录数不一致")
if loaded_index.ntotal == 0:
    raise ValueError("FAISS 索引中没有可检索的向量")

query = "我想去安静的地方走一走"
query_vector = model.encode(
    [query],
    prompt_name="query",
    normalize_embeddings=True,
).astype("float32")

# 查询向量的维度必须与保存索引时的维度一致。
if loaded_index.d != query_vector.shape[1]:
    raise ValueError("查询向量维度与 FAISS 索引维度不一致")

k = min(2, loaded_index.ntotal)
scores, positions = loaded_index.search(query_vector, k=k)

print(f"索引文件：{INDEX_FILE.relative_to(PROJECT_ROOT)}")
print(f"元数据文件：{METADATA_FILE.relative_to(PROJECT_ROOT)}")
print(f"重新加载后的向量数：{loaded_index.ntotal}")
print(f"查询：{query}")

for rank, (position, score) in enumerate(zip(positions[0], scores[0]), start=1):
    item = loaded_documents[int(position)]
    print(f"{rank}. score={score:.4f} id={item['id']} text={item['text']}")

运行代码：

python rag/p05_faiss_native/02_save_and_load_index.py

真实输出为：

索引文件：data/p05_faiss_native/life_notes.faiss
元数据文件：data/p05_faiss_native/life_notes.json
重新加载后的向量数：4
查询：我想去安静的地方走一走
1. score=0.5688 id=doc-4 text=杭州植物园环境安静，适合慢慢游览。
2. score=0.4935 id=doc-1 text=西湖边有步道和树荫，很适合周末散步。

这里先读取了 .faiss 和 .json，检查两边记录数及查询维度后再执行搜索，说明索引和业务数据已经能够从磁盘配套恢复。

5. 对比 L2 距离和余弦相似度

为了看清两种度量的差异，下面使用两个手工二维向量：

查询向量：[1.0, 0.0]
向量 A：[3.0, 0.0]，方向完全相同，但距离较远
向量 B：[0.8, 0.6]，方向相近，而且距离较近

代码文件为 03_compare_l2_and_cosine.py：

# 这个文件使用手工二维向量，对比 L2 距离和余弦相似度的排序规则。
# 手工向量只用于讲解距离度量，不是 Embedding 模型的真实输出。

import faiss
import numpy as np


labels = ["同方向但更远", "方向相近且距离近"]
vectors = np.array(
    [
        [3.0, 0.0],
        [0.8, 0.6],
    ],
    dtype="float32",
)
query = np.array([[1.0, 0.0]], dtype="float32")

# IndexFlatL2 返回平方 L2 距离，数值越小越接近。
l2_index = faiss.IndexFlatL2(2)
l2_index.add(vectors)
l2_distances, l2_positions = l2_index.search(query, k=2)

# 先归一化，再使用内积；此时分数就是余弦相似度，数值越大越接近。
cosine_vectors = vectors.copy()
cosine_query = query.copy()
faiss.normalize_L2(cosine_vectors)
faiss.normalize_L2(cosine_query)

cosine_index = faiss.IndexFlatIP(2)
cosine_index.add(cosine_vectors)
cosine_scores, cosine_positions = cosine_index.search(cosine_query, k=2)

print("L2 距离排序（越小越接近）：")
for position, distance in zip(l2_positions[0], l2_distances[0]):
    print(f"{labels[int(position)]}: {distance:.4f}")

print("\n余弦相似度排序（越大越接近）：")
for position, score in zip(cosine_positions[0], cosine_scores[0]):
    print(f"{labels[int(position)]}: {score:.4f}")

真实输出为：

L2 距离排序（越小越接近）：
方向相近且距离近: 0.4000
同方向但更远: 4.0000

余弦相似度排序（越大越接近）：
同方向但更远: 1.0000
方向相近且距离近: 0.8000

L2 更关注空间中的直线距离，余弦更关注方向。选择索引时必须先确定模型和业务使用的度量，不能只看一个分数大小。

6. 常见问题

6.1 ModuleNotFoundError: No module named ‘faiss’

先确认激活的是 .conda_faiss：

conda activate "$(pwd)/.conda_faiss"
which python
python -c "import faiss; print(faiss.__version__)"

6.2 向量维度不一致

创建索引时的维度必须与写入向量一致：

dimension = document_vectors.shape[1]
index = faiss.IndexFlatIP(dimension)

本文实测维度为 1024。

6.3 分数方向理解反了

IndexFlatIP：越大越接近
IndexFlatL2：越小越接近

6.4 重新加载后文本对应错了

FAISS 返回的是位置编号。如果保存索引后又单独调整 JSON 的顺序，向量和文本就会错位。本文增加的记录数检查只能发现数量变化，不能识别顺序变化。生产代码通常需要更严格的 ID 映射、版本号或校验机制。

6.5 Top-K 大于索引中的向量数

当 k 大于 index.ntotal 时，FAISS 会使用 -1 填充没有结果的位置。如果直接把 -1 当作 Python 列表下标，会错误地读取最后一条记录。本文使用下面的写法限制查询数量：

k = min(3, index.ntotal)

7. 小结

本篇没有数据库服务，只有 Python 进程、本地 Qwen3 Embedding 和 FAISS 索引。真实运行结果证明：

检查项 结果
FAISS 版本 1.13.2
运行平台 Apple Silicon，CPU 版 FAISS
Embedding 设备 MPS
向量维度 1024
索引类型 IndexFlatIP
相似度规则 归一化向量的内积，越大越接近
索引向量数 4
持久化 .faiss 索引 + .json 元数据
重新加载 成功，Top-2 检索正常

FAISS 足够轻量，但文本、元数据和一致性需要应用自己管理。下一篇使用 Chroma，把 ID、向量、文本和元数据放进同一个持久化 Collection。

文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/4166cc7c.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
langchain qwen rag embedding faiss vector-search
评论
  目录