从 PDF 到 FAISS 向量索引：构建本地 RAG 数据预处理流水线

【学习记录】从 PDF 到 FAISS 向量索引：构建本地 RAG 数据预处理流水线

在本地 RAG（检索增强生成）系统中，数据预处理的第一步往往是将 PDF 文档转换成可检索的向量索引。然而，PDF 文件既有可直接提取文字的电子文档，也有扫描图片型的文档。本文基于 Python 实现了一套完整的流水线：自动判断文档类型，必要时调用 OCR，然后进行文本分块、向量化（使用中文嵌入模型）并构建 FAISS 索引。代码完全开源，可直接用于本地知识库的预处理。

📌 目录

1️⃣ 代码功能概述

本代码实现了一个端到端的数据预处理流水线：

最终产出一个可被直接加载的向量索引，供下游的语义搜索、问答系统使用。

2️⃣ 环境配置与依赖

Python 库依赖

pip install pdfplumber pytesseract pdf2image faiss-cpu llama-index transformers huggingface_hub

系统依赖（必须安装）

对于 Windows 或 macOS，请参考官方文档安装对应版本。

3️⃣ 输入与输出

输入

• PDF_PATH：待处理的 PDF 文件路径（例如 ./data/y0664.pdf）。
• 配置参数：
  • CHUNK_SIZE = 512：每个文本块的最大字符数。
  • CHUNK_OVERLAP = 50：相邻块之间的重叠字符数。
  • EMBED_DIM = 512：嵌入向量维度（与模型 bge-small-zh 一致）。
  • INDEX_DIR = "./storage/faiss_index"：索引保存目录。

输出

• 文本节点列表（nodes）：每个节点包含一段文本和元数据，长度约 512 字符。
• FAISS 索引文件：vector_store.faiss
• LlamaIndex 元数据：保存在 INDEX_DIR 目录下（包括文档存储、索引结构等）。

生成的文件可用于后续的向量检索，例如：

from llama_index.core import StorageContext, load_index_from_storage

storage_context = StorageContext.from_defaults(persist_dir=INDEX_DIR)
index = load_index_from_storage(storage_context)
query_engine = index.as_query_engine()
response = query_engine.query("文档中提到了哪些技术规范？")
print(response)

4️⃣ 核心逻辑解析

4.1 PDF 文本提取与 OCR 回退

def extract_pdf(pdf_path):
text = ""
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\\n"
return text

使用 pdfplumber 尝试提取文字层。如果总字符数不足 5000（可根据文档长度调整阈值），则认为该 PDF 是扫描件或图片型，触发 OCR。

def ocr_pdf(pdf_path):
images = convert_from_path(pdf_path, dpi=300)
full_text = ""
for i, img in enumerate(images):
text = pytesseract.image_to_string(img, lang="chi_sim+eng")
full_text += text + "\\n"
return full_text

• pdf2image.convert_from_path 将每页转为 PIL 图片。
• pytesseract.image_to_string 识别文字，语言包指定简体中文和英文。

4.2 文本分块

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)
nodes = splitter.get_nodes_from_documents(documents)

• SentenceSplitter 会尽量在句子边界切分，避免切断语义。
• 重叠（chunk_overlap）可以防止关键信息正好落在两个块的边缘而被遗漏。

4.3 向量化与 FAISS 索引构建

Settings.embed_model = HuggingFaceEmbedding(model_name=EMBED_MODEL, device="cpu")

faiss_index = faiss.IndexFlatL2(EMBED_DIM)
vector_store = FaissVectorStore(faiss_index=faiss_index)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(nodes, storage_context=storage_context)

• HuggingFaceEmbedding 加载中文轻量级模型 BAAI/bge-small-zh-v1.5，输出 512 维向量。
• faiss.IndexFlatL2 使用欧氏距离进行精确检索，适合百万级以下的向量。
• 将节点和向量存储到 FaissVectorStore，并构建 LlamaIndex 的 VectorStoreIndex。

4.4 索引持久化

storage_context.persist(persist_dir=INDEX_DIR)
faiss.write_index(faiss_index, os.path.join(INDEX_DIR, "vector_store.faiss"))

• persist 保存 LlamaIndex 的元数据（节点文本、文档存储等）。
• faiss.write_index 保存原生 FAISS 索引，方便直接加载到内存。

5️⃣ 完整代码

import os
import pdfplumber
import pytesseract
from pdf2image import convert_from_path

from llama_index.core import (
Document,
VectorStoreIndex,
StorageContext,
Settings
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.vector_stores.faiss import FaissVectorStore
import faiss

# ==================================================
# 配置
# ==================================================

PDF_PATH = "./data/y0664.pdf"
INDEX_DIR = "./storage/faiss_index"
EMBED_MODEL = "BAAI/bge-small-zh-v1.5"
EMBED_DIM = 512
CHUNK_SIZE = 512
CHUNK_OVERLAP = 50

os.makedirs(INDEX_DIR, exist_ok=True)

# ==================================================
# OCR
# ==================================================

def ocr_pdf(pdf_path):
print("开始OCR…")
images = convert_from_path(pdf_path, dpi=300)
full_text = ""
for i, img in enumerate(images):
text = pytesseract.image_to_string(img, lang="chi_sim+eng")
print(f"第{i+1}页 OCR字符数:", len(text))
full_text += text + "\\n"
return full_text

# ==================================================
# PDF文本提取
# ==================================================

def extract_pdf(pdf_path):
text = ""
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\\n"
return text

# ==================================================
# 主程序
# ==================================================

if __name__ == "__main__":
print("加载Embedding模型…")
Settings.embed_model = HuggingFaceEmbedding(model_name=EMBED_MODEL, device="cpu")

print("读取PDF…")
raw_text = extract_pdf(PDF_PATH)
print("PDF字符数:", len(raw_text))

if len(raw_text) < 5000:
print("文本过少，自动OCR…")
raw_text = ocr_pdf(PDF_PATH)

print("最终字符数:", len(raw_text))

documents = [Document(text=raw_text, metadata={"source": "YY/T0664-2020"})]

splitter = SentenceSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)
nodes = splitter.get_nodes_from_documents(documents)
print("节点数:", len(nodes))

# 创建FAISS索引
faiss_index = faiss.IndexFlatL2(EMBED_DIM)
vector_store = FaissVectorStore(faiss_index=faiss_index)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

print("构建索引…")
index = VectorStoreIndex(nodes, storage_context=storage_context)

print("保存索引…")
storage_context.persist(persist_dir=INDEX_DIR)
faiss.write_index(faiss_index, os.path.join(INDEX_DIR, "vector_store.faiss"))

print("\\n文件检查")
for f in os.listdir(INDEX_DIR):
fp = os.path.join(INDEX_DIR, f)
print(f, os.path.getsize(fp))

print("\\n完成")
print("索引保存路径:", os.path.abspath(INDEX_DIR))

6️⃣ 运行与测试

启动预处理

python build_index.py

预期输出

加载Embedding模型…
读取PDF…
PDF字符数: 3200
文本过少，自动OCR…
开始OCR…
第1页 OCR字符数: 1250
第2页 OCR字符数: 1180
…
最终字符数: 12500
节点数: 25
构建索引…
保存索引…
文件检查
docstore.json 1280
index_store.json 560
vector_store.faiss 512000
…
完成
索引保存路径: /home/user/project/storage/faiss_index

7️⃣ 注意事项与优化建议

8️⃣ 总结

本文提供了一个生产可用的 PDF → 向量索引流水线，具备以下亮点：

• ✅ 自动区分文本型 PDF 和扫描件，回退到 OCR。
• ✅ 支持中文嵌入（bge-small-zh），兼顾效果与速度。
• ✅ 基于 LlamaIndex 和 FAISS，代码简洁且易于扩展。
• ✅ 索引持久化，可重复加载使用。

你可以将此流水线集成到本地 RAG 系统中，作为知识库的预处理环节。下一步可以：

• 增加文件格式支持（如 DOCX、HTML）。
• 将 FAISS 索引迁移到更强大的向量数据库（如 Milvus、Qdrant）。
• 添加文档元数据（作者、时间）以支持过滤检索。

希望本文对你的项目有所帮助。欢迎在评论区分享你的实践经验！