RAG 索引持久化：为什么 FAISS 要存成两个文件？

从0手敲RAG学习日记 · 索引篇

上一篇把 RAG 基础流水线里的 10 个函数串了一遍，今天继续深入一个更实际的问题：索引建好之后，要不要存到磁盘？

一、索引是什么

1. 知识库被切成很多小段（chunks）。
2. 每段用 M3E 变成一串数字（向量）。
3. 索引 = 把这些向量整理好，方便按「相似度」查找。

一句话：没有索引，每次提问都要把问题向量和所有块逐个比一遍，块一多就慢；有索引，FAISS 用专门结构做近邻搜索，很快取出 Top-K。

和数据库索引的相似点

维度	数据库索引	向量索引
目的	加快查找	加快查找
查什么	按字段精确查	按语义相似度查向量

二、为什么要保存索引

建索引的成本很高，完整流程是：读文档 → 切块 → M3E 编码（最慢）→ 写入 FAISS。

若每次 python main.py 都重做一遍，启动要等很久。持久化的目的就一句话：把算好的向量索引落到磁盘，下次启动直接读，跳过重新编码。

三、磁盘上存了什么？为什么是两个文件

FAISS 只存向量和下标，不存原文。检索时还要用下标找回 chunks[i] / sources[i]，所以必须另存元数据：

文件	内容
index_store/faiss.index	向量索引（二进制）
index_store/meta.json	chunks、sources，以及当时的 chunk_size / chunk_overlap

注意：meta 里记下切块参数，是为了防止「向量和下标对不上新切块」。你改了 .env 里的 chunk_size / chunk_overlap 后，旧索引会失效，load_index 返回 None，强制重建。

四、关键函数拆解

1. save_index —— 保存索引

落盘两样东西：faiss.index 是向量索引本身，meta.json 是 chunks + sources，因为 FAISS 不存原文。

# ============================================================
# 索引持久化（可选，加速下次启动）
# ============================================================
def save_index(
    index: faiss.IndexFlatIP,
    chunks: list[str],
    sources: list[str],
    index_dir: Path | None = None,
) -> Path:
    """
    落盘两样东西：
    - faiss.index : 向量索引本身
    - meta.json   : chunks + sources（因为 FAISS 不存原文）
    """
    index_dir = index_dir or INDEX_DIR
    index_dir.mkdir(parents=True, exist_ok=True)
    # 写 FAISS 二进制索引
    faiss.write_index(index, str(index_dir / "faiss.index"))
    # 写文本元数据，保证下标还能对回原文
    meta = {
        "chunks": chunks,
        "sources": sources,
        # 记下切块参数：参数变了就视为旧索引，强制重建
        "chunk_size": CHUNK_SIZE,
        "chunk_overlap": CHUNK_OVERLAP,
    }
    (index_dir / "meta.json").write_text(
        json.dumps(meta, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    return index_dir

功能：保存索引，faiss.write_index + 写 meta.json。

2. load_index —— 加载本地索引

任一文件缺失，或者切块参数和当前配置不一致，就返回 None，让上层重新 build。

def load_index(
    index_dir: Path | None = None,
) -> tuple[faiss.IndexFlatIP, list[str], list[str]] | None:
    """
    从磁盘读回索引。
    任一文件缺失则返回 None，调用方应改为重新 build。
    """
    index_dir = index_dir or INDEX_DIR
    index_path = index_dir / "faiss.index"
    meta_path = index_dir / "meta.json"
    if not index_path.exists() or not meta_path.exists():
        return None
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    # 切块参数与当前配置不一致 → 当作没有索引，让上层重建
    if meta.get("chunk_size") != CHUNK_SIZE or meta.get("chunk_overlap") != CHUNK_OVERLAP:
        return None
    index = faiss.read_index(str(index_path))
    return index, meta["chunks"], meta["sources"]

功能：加载本地索引，缺文件或参数不一致就返回 None。

3. build_rag_index —— 一键构建整条索引流水线

把加载模型、切块、创建索引合并成一个方法；构建完成后可选落盘。

def build_rag_index(
    data_dir: Path | None = None,
    model: SentenceTransformer | None = None,
    *,
    persist: bool = True,
    index_dir: Path | None = None,
) -> tuple[SentenceTransformer, faiss.IndexFlatIP, list[str], list[str]]:
    """
    把前面步骤串起来：
    加载 M3E
    → build_corpus（读文档 + 切块）
    → encode_texts（块 → 向量）
    → build_faiss_index（向量 → 可检索索引）
    → 可选 save_index
    返回四件套，后续 search() 全都要用：
    model, index, chunks, sources
    """
    # ① 模型：可外部传入（避免重复加载），否则读本地
    model = model or load_m3e_model()
    # ② 语料：所有块 + 来源文件名
    chunks, sources = build_corpus(data_dir)
    if not chunks:
        # ③a 空知识库：仍建一个同维度空索引，方便以后上传文档再 rebuild
        dim = model.get_sentence_embedding_dimension()
        index = faiss.IndexFlatIP(dim)
    else:
        # ③b 有内容：向量化 + 建索引
        embeddings = encode_texts(model, chunks)
        index = build_faiss_index(embeddings)
    # ④ 是否写到 index_store/
    if persist:
        save_index(index, chunks, sources, index_dir)
    return model, index, chunks, sources

功能：构建并可选落盘，persist=True 会在本地保存索引。

4. load_or_build_rag_index —— 先加载，没有再构建

多一步判断，不用每次都重复构建。服务启动时常用：磁盘已有索引就直接加载，没有就重新 build。

def load_or_build_rag_index(
    data_dir: Path | None = None,
    model: SentenceTransformer | None = None,
    index_dir: Path | None = None,
) -> tuple[SentenceTransformer, faiss.IndexFlatIP, list[str], list[str]]:
    """
    启动服务时常用：
    - 磁盘已有索引 → 直接加载（快）
    - 没有 → 重新 build_rag_index（慢一次）
    """
    model = model or load_m3e_model()
    loaded = load_index(index_dir)
    if loaded is not None:
        index, chunks, sources = loaded
        return model, index, chunks, sources
    return build_rag_index(data_dir, model, persist=True, index_dir=index_dir)

功能：多一步判断，优先读盘，失败再走创建。

5. _refresh_state —— 刷新内存里的 RAG 四件套

服务启动、上传 / 删除文档、手动重建索引时都会调用它。它负责两件事：一是重建或加载索引，二是把磁盘文件列表和 SQLite documents 表对齐。

def _refresh_state(
    *,
    rebuild: bool = False,
    model=None,
) -> None:
    """
    刷新内存里的 RAG 四件套，并同步 documents 表。
    什么时候调用？
    - 服务启动（lifespan）
    - 上传 / 删除文档之后
    - 手动「重建索引」
    """
    # ① 尽量复用已在内存里的 model
    model = model or state.get("model")
    # ② 构建或加载索引
    if rebuild or model is None:
        model, index, chunks, sources = build_rag_index(DATA_DIR, model, persist=True)
    else:
        model, index, chunks, sources = load_or_build_rag_index(DATA_DIR, model)
    # ③ 写回全局 state，后续 /search、/ask 都从这里取
    state.update(
        {
            "model": model,
            "index": index,
            "chunks": chunks,
            "sources": sources,
        }
    )
    # ④ 磁盘文件列表 ↔ SQLite documents 表对齐
    db.sync_documents_from_disk(list_documents(DATA_DIR))

上传 / 删除文档时会调用 _refresh_state(rebuild=True)，等于重新 build 并再次 save_index，保证磁盘索引和当前知识库一致。

五、完整流程串起来

load_or_build_rag_index()
  ├─ load_index 成功？ → 直接用（快）
  └─ 失败 / 切块参数变了？ → 再走 build_rag_index 并重新 save

启动优先读盘，日常刷新用 load_or_build_rag_index；上传 / 删除文档后用 _refresh_state(rebuild=True)。这就是 RAG 索引创建以及保存的完整流程。

---

写在最后

索引持久化是 RAG 服务化的关键一步。它让「启动慢」变成「启动快」，让「每次重启都重新编码」变成「一次编码、多次复用」。

下一篇会继续聊文档管理：上传、删除、重建索引怎么和 FastAPI 接口结合起来。如果这篇对你有帮助，欢迎点赞收藏，后续更新我们接着聊 RAG 服务化。