RAG离线入库全流程实战：从PDF解析到向量库的完整指南

说实话，RAG 项目我前后折腾了两年多，踩过的坑比写过的代码还多。最讽刺的是，真正让检索效果崩盘的，往往不是 prompt 写得不好，也不是 rerank 没调明白，而是最容易被忽略的“入库”环节——文档从 PDF 变成向量库里那一条条记录的过程。这一环要是拉胯，后面召回再精准、生成再流畅，也只能在垃圾数据上自嗨。

这两年我逐渐把整个离线入库流程收敛到一条相对固定的方案上，整套代码加起来 1200 行上下，从 PDF 解析到向量库可检索，全程离线可复现。这篇文章把这条流程掰开揉碎讲清楚，包括每一层的设计思路、选型理由、完整实操步骤，以及那些我交过学费才知道的坑。不管你是刚搭好 RAG demo 正被分块困扰，还是已经在做生产级知识库但召回率一直上不去，这篇文章应该都能给你一些直接能用的东西。

1. 为什么入库环节最容易翻车

1.1 检索效果的天花板在入库时就已经定死了

很多人理解 RAG 是一条“召回 → 排序 → 生成”的链路，觉得入口是 query 进来那一刻。但实际决定这条链路能走多远的，是更早一步的入库。你在入库时存进去什么，决定了检索时能查到什么。这不是绕口令，是物理事实。

举个例子。有一份 PDF 里写了一段流程说明，中间包含一个三级标题“异常处理机制”，标题后面跟着三个步骤。如果你在解析时把标题弄丢了，或者在分块时把标题留在了上一个切片里，后面的内容block就变成了无主信息。用户问“异常处理机制是什么流程”，query 里的关键词和那个无主 block 里的内容匹配度可能不够，它要么召回不到，要么召回结果排到了十名开外。等到了 rerank 阶段，模型再强，也只能从错误的候选集里捞结果。

换句话说，入库环节决定了检索的上限，生成阶段只是在逼近这个上限。这也是我后来把 70% 的精力都砸在入库上的根本原因。

1.2 我最常看到的三个翻车现场

项目里遇到过的翻车，基本可以归成三类。

第一类是 PDF 解析直接拉胯。文本型 PDF 抽出来是一堆乱码，或者明明有文字却抽不出内容。扫描件更不用说了，OCR 没调好就硬上，识别出来的中文缺胳膊少腿。这一关挂了，后面全是脏数据。

第二类是分块切碎了语义。最常见的是固定窗口切分，比如 512 个 token 一刀切。运气不好时，一刀下去把一个函数定义切成了两半，或者把“为什么不能用 XX”这种问答对拆得七零八落。召回率自然上不去。

第三类是入库后检索行为诡异。比如向量库里存了几万条数据，但搜索时永远只命中一小部分。我排查过几次，发现要么是 embedding 向量全一样（模型加载出了问题），要么是主键设计有误导致写入互相覆盖，要么是没给标量字段建索引，过滤条件直接把接口拖崩。

这些问题在 demo 阶段都不明显，因为测试数据量小、场景简单。一旦上了真实 PDF 库，立刻暴露。也是因为这些踩坑经历，我才下定决心把入库流程标准化，做成一条任何文档进来都能走完的流水线。

2. 离线入库的整体架构与方案选型

2.1 分层流程，先看全局

整个离线入库流程可以拆成六个环节，顺序如下：

1. 文档收集与预处理：把零散 PDF 汇总到统一目录，做去重和格式检查。
2. 内容解析：文本型 PDF 抽文字，扫描件走 OCR。
3. 清洗与结构化：去掉页眉页脚、页码、水印，处理表格。
4. 分块：把长文档切成适合检索和 embedding 的切片。
5. 向量化：用 embedding 模型把文本转成向量。
6. 写入向量库：带上 metadata 和主键，批量写入并建立索引。

这六个环节层层递进，每一层都是独立的 Python 模块。这样做的好处是，任何一层出了问题可以单独替换，不需要把整条链路推到重来。比如今天想换更强的 OCR，不需要动后面的分块逻辑。

2.2 工具选型，我用过的几种方案

我试过不少组合，这里直接给对比表，帮你少走弯路。

这套组合里，最核心的是“PyMuPDF 抽文本 + PaddleOCR 补漏 + BGE 向量化 + Milvus 存储”。它不是唯一解，但胜在稳定、离线、可控。

2.3 为什么我偏偏选这套组合

先聊 PDF 解析。PyMuPDF 的性能是几个库里最突出的，它不依赖外部渲染引擎，直接操作 PDF 内部结构，提取速度很快。它的问题在于对某些复杂版式（比如多栏 PDF）的处理不够智能，但这个问题可以用“先抽文本，再按块排序”来解决，后面实操部分会细说。

扫描件为什么必须配 OCR？因为扫描件本质是图片，里面压根没有文字层，PyMuPDF 抽出来的只能是空字符串。PaddleOCR 强在中文识别和版面分析，它能把文字块、表格、图片位置一次给全，后期做结构化清洗很方便。

Embedding 选 BGE 而不是 OpenAI，核心原因是离线。很多知识库场景对数据安全有硬性要求，文档不能出内网。BGE 模型可以本地部署，且中文语义理解能力不输闭源接口。这里说一个经验：用 BGE-M3 时可以同时得到 dense 向量和 sparse 向量，后者在做关键词精确匹配时效果意外的好，后面洗数据时我会用 sparse 向量做初筛。

向量库选 Milvus 主要是看中它的标量过滤能力。知识库场景经常需要过滤条件，比如“只看最近三个月的文档”“只看产品 A 的文档”，Milvus 可以直接在检索时带上过滤表达式，不用先把数据捞出来再过滤。

3. 保姆级实操：从 PDF 到可检索的完整实现

这部分的代码我按模块来讲，每个模块都给出关键代码和设计思路，整体串联起来就是你自己的 1200 行入库流水线。我在自己项目里的目录结构大概长这样：

rag_ingest/ ├── config.py # 全局配置 ├── pdf_parser.py # PDF 解析与 OCR ├── text_cleaner.py # 清洗与结构化 ├── chunker.py # 分块策略 ├── embedder.py # 向量化 ├── writer.py # 写入向量库 └── ingest.py # 入口调度

3.1 环境准备与依赖安装

建议直接用 conda 建独立环境，避免依赖冲突。我常用的安装命令如下，Python 版本选 3.10 或 3.11 都行。

conda create -n rag_ingest python=3.11 -y conda activate rag_ingest pip install pymupdf pdfplumber paddleocr paddlepaddle pip install milvus-pymongo # 或者用 pymilvus pip install sentence-transformers

如果机器上有 NVIDIA GPU，paddlepaddle 可以装 GPU 版本，OCR 速度会快很多。没有 GPU 也能跑，就是慢一些，后面会有优化建议。

3.2 PDF 解析，先判断是文本型还是扫描件

有人拿到 PDF 直接无脑 OCR，这是最大的浪费时间。一个文本型 PDF 本来几秒钟就能抽完，硬是用 OCR 跑了几分钟，识别率还不一定比直接抽取高。所以第一步是自动判断 PDF 类型。

import fitz # PyMuPDF def detect_pdf_type(pdf_path: str) -> str: doc = fitz.open(pdf_path) total_text_len = 0 for page in doc: text = page.get_text() total_text_len += len(text.strip()) doc.close() if total_text_len > 50: return "text" return "scanned"

逻辑很简单：如果整份 PDF 能抽取出来的文本长度超过 50 个字符，就认为它带文字层，不需要 OCR。50 这个阈值不是随便定的，因为有些 PDF 只有封面有文字，正文全是扫描图，这种要自动识别出来。

扫描件接着走 PaddleOCR。这里有一个很关键的细节：PaddleOCR 直接传 PDF 路径是跑不了的，必须先按页转成图片。直接用 fitz 渲染，代码如下：

def pdf_to_images(pdf_path: str, dpi: int = 200) -> list: doc = fitz.open(pdf_path) images = [] for page_idx in range(len(doc)): page = doc[page_idx] pix = page.get_pixmap(dpi=dpi) img_path = f"/tmp/pdf_ocr_{page_idx}.png" pix.save(img_path) images.append(img_path) doc.close() return images

dpi 参数我一般用 200，太小了小字号识别不出来，太大了渲染时间长且 OCR 内存暴涨。200 是我测试下来速度和质量比较均衡的值。

PaddleOCR 的具体调用相信你已经很熟了，不贴完整代码。提醒一句：PaddleOCR 返回的结果里每个文字块带位置坐标，这个坐标后面清洗时用处很大。

3.3 清洗与结构化，别让垃圾进库

从 PDF 抽出来的文本通常带着一堆噪音：页眉页码、页脚版权信息、水印、目录页、重复的空行。如果不清洗直接分块，这些噪音会被切进多个 chunk 里，污染检索结果。

这里我分享一套实际好用的清洗流程：

1. 按页处理，先把每页文本按行切分。
2. 用正则和位置信息去掉页眉、页脚、页码。PyMuPDF 提供了文本块的 bounding box 坐标，通常页眉和页脚都固定在页面顶部或底部 5% 区域内，直接按坐标过滤比用正则稳得多。
3. 检测并去除水印。水印文字通常是斜体、低透明度或重复多次，特征比较明显。我的做法是统计同一文本在同一页出现的次数，超过三次且内容较短，就按水印直接丢弃。
4. 压缩连续空行为单个空行，避免分块时产生空切片。
5. 最后把处理后的文本重新拼接，按页面顺序输出。

这里有一个容易忽略的问题——多栏 PDF。很多 PDF 是双栏排版，如果直接按页面纵向抽取，抽出来的文字顺序会乱。第一栏还没读完就跳去读第二栏，语义完全中断。我处理这个问题的思路是：拿到 PyMuPDF 的文本块坐标后，按 x 坐标先做列聚类，把文本块归到不同的栏里，再按 y 坐标排序每个栏内的文本块，最后按栏的顺序拼接。

这个过程在代码里大概长这样：

def sort_text_blocks(blocks, page_width: float): # blocks: 从 fitz 拿到的 text blocks, 每个 block 有 (x0, y0, x1, y1, text) # 简单判断是否是两栏或三栏 if page_width > 0: col_count = 2 if page_width > 900 else 1 cols = [[] for _ in range(col_count)] col_width = page_width / col_count for block in blocks: center_x = (block[0] + block[2]) / 2 col_idx = min(int(center_x // col_width), col_count - 1) cols[col_idx].append(block) result = [] for col in cols: col.sort(key=lambda b: (b[1], b[0])) # 按 y 排序, 同一行按 x for block in col: if block[4].strip(): result.append(block[4]) return "\n".join(result)

这个逻辑不算复杂，但在处理双栏文档时把所有文本块按“逻辑阅读顺序”重新排了一遍，效果立竿见影。

3.4 分块策略，既要控制长度也要保住语义

分块是入库流程里最容易改坏的部分，也是争议最大的部分。我尝试过 LangChain 的 RecursiveCharacterTextSplitter、按标题切分、语义切分，最后沉淀下来的是一套“固定窗口 + 语义边界”的混合策略。

思路分成三层：

第一层，优先按标题或大纲切分。如果文档本身有清晰的目录结构（比如 README、技术规范），先用标题级别切出大段。一个一级标题下面对应的是一个语义完整的章节，这一层切分质量最高。

第二层，对每个大段继续切。如果大段文本超长，就按段落边界切，而不是按字符硬切。段落是自然语义单元，一个段落内部通常讨论同一件事。

第三层，对极长的段落使用固定窗口 + overlap。窗口大小我一般设 400~500 token，overlap 设 100~150 token。overlap 是为了避免语义在窗口边界中断，确保关键信息至少在一个完整窗口内出现。

代码实现上，我封装了一个Chunker类，核心逻辑如下：

from typing import List class MixedChunker: def __init__(self, chunk_size=450, overlap=120): self.chunk_size = chunk_size self.overlap = overlap def split_by_headings(self, text: str) -> List[str]: # 简单按 Markdown 标题切割 import re lines = text.split("\n") sections = [] cur_section = [] heading_pattern = re.compile(r'^(#{1,4})\s+(.*)$') for line in lines: if heading_pattern.match(line) and cur_section: sections.append("\n".join(cur_section)) cur_section = [line] else: cur_section.append(line) if cur_section: sections.append("\n".join(cur_section)) return sections def split_long_text(self, text: str) -> List[str]: if len(text) <= self.chunk_size: return [text] chunks = [] start = 0 step = self.chunk_size - self.overlap while start < len(text): end = start + self.chunk_size chunk = text[start:end] # 尽量在段落边界截断 if end < len(text): last_newline = chunk.rfind("\n\n") if last_newline > len(chunk) * 0.6: end = start + last_newline chunk = text[start:end] chunks.append(chunk) start = end - self.overlap if start < 0: break return chunks def chunk(self, text: str) -> List[str]: sections = self.split_by_headings(text) result = [] for sec in sections: result.extend(self.split_long_text(sec)) return result

这套逻辑的关键是“先保边界，再控长度”。优先级从高到低依次是文档结构、段落、固定窗口。如果一上来就用固定窗口，会失去结构信息；但只要先做了标题和段落边界，固定窗口的副作用就会被压到最小。

3.5 向量化与批量写入 Milvus

分块完成后进入向量化和入库环节。向量化我用 sentence-transformers 加载 BGE 模型：

from sentence_transformers import SentenceTransformer model = SentenceTransformer("BAAI/bge-large-zh-v1.5") embeddings = model.encode(chunks, normalize_embeddings=True, batch_size=32) chunks_embeddings = embeddings.tolist()

这里有个细节：BGE 官方推荐在 encode 时给文本加上代表任务的前缀吗？严格来说，BGE 系列在 query 和 document 两端使用了不同的指令模式。具体来说，在生成 document embedding 时，原版 BGE 推荐给文本加上“为这个句子生成表示以用于检索相关文章：”这样的前缀，而在对应 query 端也要加同样的前缀。这在实验里能提点分数。不过如果你用的是带 instruction 的模型，需要注意。我在生产环境里为了部署简单，直接用了不加前缀的中间模型，效果也可以接受。关键是 query 和 doc 两端保持一致，别一边加一边不加。

写入 Milvus 时，最核心的是设计好主键和 metadata 字段。我用哈希或自增 ID，配合doc_id来标识数据来自哪个源文档，同时保存page_no、chunk_index、source_file、created_at等标量字段。建 collection 的代码如下：

from pymilvus import ( connections, FieldSchema, CollectionSchema, DataType, Collection, utility ) connections.connect(alias="default", host="localhost", port="19530") fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="chunk_index", dtype=DataType.INT32), FieldSchema(name="source_file", dtype=DataType.VARCHAR, max_length=512), FieldSchema(name="page_no", dtype=DataType.INT32), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=8192), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024), ] schema = CollectionSchema(fields) collection = Collection("knowledge_base", schema, consistency_level="Strong") collection.create_index( field_name="embedding", index_params={"index_type": "HNSW", "metric_type": "IP", "params": {"M": 16, "efConstruction": 200}} ) collection.create_index(field_name="doc_id", index_name="idx_doc_id")

Batch 写入时要注意别把整批数据一次性塞进去。我一般按 512 条一个 batch 插入，每插完一个 batch 就 flush 一次。数据量大了以后，一次性插入几万条反而容易触发 timeout 或者内存问题。写入代码建议自己包一层重试机制，catch 到异常就重试两三回。

还有一点必须提醒：dim要和 embedding 模型输出维度一致。BGE-large-zh-v1.5 的维度是 1024，BGE-M3 默认 dense 向量维度也是 1024。如果你中途换模型，旧 collection 的向量维度对不上，只能重建或另建 collection。

3.6 入库后的检索验证，别等线上才发现问题

写入完成后，马上做一轮检索冒烟测试。我的习惯是拿一份源文档里的典型问答对来测试，看能不能召回正确答案。直接在 Milvus 里查：

from pymilvus import Collection, utility collection = Collection("knowledge_base") collection.load() query_embedding = model.encode([query]) results = collection.search( data=query_embedding, anns_field="embedding", param={"metric_type": "IP", "params": {"ef": 128}}, limit=10, output_fields=["text", "source_file", "page_no", "doc_id"] ) for hits in results: for hit in hits: print(f"距离: {hit.distance:.4f}, 来源: {hit.entity.get('source_file')}, 页码: {hit.entity.get('page_no')}") print(hit.entity.get("text")[:200])

这个冒烟测试不只是看一眼结果对不对，还要关注排序。正确答案在不在 top 5？如果是 top 10 开外，说明入库环节大概率有问题，不用急着去调 holder 或者 prompt。

同时我还会做一个“空召回”测试：用一个明显不在知识库里的 query 去查，看它是否真的返回很少或没有结果。如果随随便便一个无关问题都能召回出一堆高相似度结果，那说明 embedding 模型和向量相似度计算可能出了问题。最典型的例子是向量全为零或者全局归一化出 bug，导致所有向量之间的 cosine 相似度都一样。所以每次批量入库完，我会随机抽几个 embedding 向量的范数值，如果发现大量向量范数接近 0，立刻停下排查。

4. 常见问题与排查技巧实录

4.1 解析阶段：PDF 抽出来是空的或有乱码

这是刚接触 PDF 解析时最容易撞的问题。先分情况排查：

• 如果Document.get_text()返回空字符串，大概率是扫描件，老老实实走 OCR。
• 如果返回内容包含大量乱码字母和符号，一般是 PDF 字体用了非标准编码（常见于某些国产软件生成的 PDF），或者字体子集有问题。这时候换个解析库，用 pdfplumber 再抽一次，经常能抽出来正常文本。我遇到过一个案例，PyMuPDF 抽出满屏GILD字样，换 pdfplumber 后文本完全正常。
• 如果文字能抽出来但顺序混乱，就是多栏问题，按 3.3 节的方法排一下序。

OCR 阶段最常见的错误是漏识别小字号的注释。我的解决办法是：第一次 OCR 用 dpi=200 跑一遍，如果页面文字量明显偏低，再用 dpi=300 重跑一次。代价是多花时间，但总比进库后再返工便宜。

4.2 分块阶段：为什么检索结果总是不带上下文

分块之后可以先人工观察几个 chunk 的前后文本。我一般写个脚本随机抽样 100 个 chunk，把每个 chunk 的首尾 50 字打印出来看。分块质量差的问题主要有两种：

一种是 chunk 边界把一句话切成两半。这种情况要看你是不是用了纯字符切分，没有在段落边界做截断。混合分块策略里我在split_long_text中做了“在最近换行处截断”的处理，专门解决这个问题。

另一种是 chunk 太短导致信息不足。比如表格里的每一个单元格被单独拆成一个 chunk，检索时命中了单元格，但模型拿到的上下文只是一格，完全不知道这格属于哪张表。这种情况要做“父子分块”，把细粒度 chunk 和它所在的粗粒度段落关联起来。具体说，在写入 metadata 里保存parent_id，检索时用细 chunk 找到对应粗 chunk，然后把粗 chunk 整体拼进上下文。实现起来不复杂，但对表格类文档非常有效。

4.3 向量化和入库阶段：维度、命中率、长尾问题

维度不一致是换模型后最容易踩的雷。历史数据用旧模型跑，向量维度是 768，新模型是 1024，导入时报错才醒悟。所以我在 config 里单独留了一个embedding_dim配置项，每次换模型先改配置再跑入库，同时旧 collection 要么删掉重建，要么换个新名字。

命中率低且集中在少数文档上，通常是 embedding 模型没适配领域。通用模型在垂直领域（比如法律、医疗、内网技术文档）效果会打折。对这种场景，我建议先用 pipeline 入库一部分数据，然后用真实 query 测一轮，人工标注 top 10 里是否包含正确答案。如果普遍不理想，再考虑领域适配（微调或换更专业的基础模型）。不要一上来就迷信模型越高配越好，先跑通数据链路再说。

Milvus 检索慢或者召回异常，先检查索引参数。HNSW 的M和efConstruction参数会影响建索引速度和精度，ef是搜索参数，调大ef能提升召回精度但会加大耗时。前面给的参数（M=16，efConstruction=200，ef=128）是我在中等规模数据量下的常用配置，数据量到千万级以后需要重新调。

4.4 增量更新与去重：入库不是一次性的事

知识库是活的，文档会新增、会过期。我最初做增量更新时吃了不少亏，这里直接给经验：

• 对源 PDF 计算 hash（如 MD5/SHA1），写入 metadata。每次新增文档时先查 hash，一样的就跳过，避免重复入库。
• 对已更新内容的文档，先删除旧doc_id下的所有 chunk，再重新入库。不要试图“原地更新”某些 chunk，逻辑太容易出 bug。
• 定期清理废弃数据：用 Milvus 的 delete 表达式按doc_id删除，线上知识库建议先跑在副本或测试环境，确认没问题再接正式环境。
• Milvus 里删除数据后，检索不会立刻生效到所有 segment，有时会有 lag。建议在删除后执行一次collection.compact()再用搜索验证，避免索引状态不一致。

这里强烈建议把所有入库中间产物（解析后的纯文本、分块结果、embedding 缓存）持久化保存下来。很多问题（比如后面发现某个文档解析错了）可以不用重新跑全流程，直接改对中间文件重新分块或重新 embedding 就行，速度能快很多。

4.5 一个真实案例：从召回率 20% 到 80% 的排查过程

最后分享一个真实案例，帮你理解排查思路。

之前帮一个团队调知识库，技术文档都是 PDF 扫描件。他们的 rag 系统召回率感人，20% 都不到。我接手后，第一步没有看任何模型代码，先把他们的入库流程跑了一遍，发现几个叠加问题：

• 扫描件 OCR 用的是通用英文模型，中文识别率惨不忍睹，大量错字导致 embedding 向量畸变。
• 分块逻辑是 2000 字符一刀切，直接把文档里的表格、列表全部切碎。
• 向量库里的 metadata 字段里没有页码，定位原文只能靠猜。
• 多栏 PDF 的阅读顺序错乱，跨栏拼接，导致 chunk 内部语义断裂。

改了四个地方：OCR 换成中文模型、分块改成“标题 + 段落边界 + 窗口 overlap”的混合策略、metadata 加上页码和 doc_id、解析阶段做了双栏排序。重新跑完一遍后，召回率直接拉到 80% 以上。

这个案例说明，很多时候 RAG 用不好，不是你 prompt 写得不对，而是入库这一步埋了太多雷。把入库流程标准化，做扎实，会有奇效。

5. 最后分享一个长期有效的习惯

这几年来，我最大的体会是，入库链路一定要做成“可复现、可审计、可观测”的流水线。可复现是任何一次入库都能重跑，结果一致；可审计是每一条数据都能知道它来自哪个文件的第几页；可观测是每个环节的输入输出都有日志，出问题能快速定位。

还有一个很实用的习惯：每份文档入库后，都生成一个小的“入库报告”，包括 PDF 类型识别结果、抽出的文本长度、OCR 置信度分布、分块数量、embedding 维度、写入成功条数。这个报告看起来不起眼，但它会在你跟人协作、线上问题回溯时，帮你省下大把时间。

如果你现在正准备搭 RAG，不妨把入库这条动线当成第一优先级来设计。先把 PDF 到向量库这段彻底搞稳了，再往前端 prompt 上发力。相信我，检索效果会给你惊喜的。