RAG 服务中 PDF 解析方案全景
PDF 解析是 RAG 质量的地基 —— 解析质量直接决定召回质量,是整个 RAG 链路中最容易被忽视但影响最大的环节
为什么 PDF 解析很难?
PDF 的本质是"打印格式",不是"语义格式"
┌─────────────────────────────────────────────────────────┐
│ PDF 内部存储的是: │
│ "在坐标(x=120, y=340)处,用12号字体画'销售额'" │
│ "在坐标(x=200, y=340)处,用12号字体画'100万'" │
│ │
│ 没有"这是表格"、"这是标题"、"这是正文"的概念 │
│ 一切都是坐标 + 字符 + 样式 │
└─────────────────────────────────────────────────────────┘
真实 PDF 的挑战:
✅ 纯文字 PDF → 简单,直接提取
⚠️ 双栏/多栏排版 → 文字顺序乱
⚠️ 表格 → 难以还原结构
⚠️ 图文混排 → 图片中有关键信息
⚠️ 扫描件(图片PDF) → 根本没有文字,全是图片
⚠️ 数学公式 → 特殊符号丢失
⚠️ 页眉页脚 → 噪音,需过滤
⚠️ 水印 → 噪音
⚠️ 加密 PDF → 需解密
⚠️ 中文竖排 → 方向识别错误
解析方案全景图
┌─────────────────────────────────────────────────────────────┐
│ PDF 解析方案分层 │
├─────────────────────────────────────────────────────────────┤
│ Layer 4:商业/云服务 API │
│ Azure Document Intelligence / AWS Textract / 阿里云 │
│ 优点:开箱即用,效果好;缺点:费用高,数据上云 │
├─────────────────────────────────────────────────────────────┤
│ Layer 3:AI 驱动的开源框架 │
│ MinerU / Docling / Unstructured / Marker │
│ 优点:效果优秀,可本地部署;缺点:资源消耗大 │
├─────────────────────────────────────────────────────────────┤
│ Layer 2:传统解析库 │
│ PyMuPDF / pdfplumber / pdfminer / PyPDF2 │
│ 优点:轻量快速;缺点:复杂版面效果差 │
├─────────────────────────────────────────────────────────────┤
│ Layer 1:OCR 引擎 │
│ PaddleOCR / Tesseract / EasyOCR │
│ 适用:扫描件/图片型 PDF │
└─────────────────────────────────────────────────────────────┘
主流方案详细对比
一、传统解析库
PyMuPDF(fitz)⭐ 最常用基础库
import fitz # pip install pymupdf
def parse_pdf_pymupdf(pdf_path):
doc = fitz.open(pdf_path)
full_text = []
for page_num, page in enumerate(doc):
# 基础文本提取
text = page.get_text("text") # 纯文本
# text = page.get_text("blocks") # 按块提取(带坐标)
# text = page.get_text("dict") # 详细字典格式
full_text.append({
"page": page_num + 1,
"content": text
})
# 提取图片
image_list = page.get_images()
for img_index, img in enumerate(image_list):
xref = img[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
# 可进一步做 OCR
doc.close()
return full_text
# 特点:
# ✅ 速度极快(C 底层)
# ✅ 支持坐标信息
# ✅ 支持图片提取
# ❌ 复杂排版(双栏、表格)效果差
# ❌ 扫描件无法处理
pdfplumber ⭐ 表格提取最佳传统方案
import pdfplumber # pip install pdfplumber
def parse_with_tables(pdf_path):
results = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
page_data = {}
# 提取文本
page_data["text"] = page.extract_text()
# 提取表格(pdfplumber 强项)
tables = page.extract_tables()
page_data["tables"] = []
for table in tables:
# table 是二维列表
# [['姓名', '年龄', '部门'],
# ['张三', '28', '技术部'],
# ['李四', '32', '产品部']]
if table:
# 转换为 Markdown 表格(对 LLM 友好)
md_table = table_to_markdown(table)
page_data["tables"].append(md_table)
results.append(page_data)
return results
def table_to_markdown(table):
"""将表格转为 Markdown 格式"""
if not table or not table[0]:
return ""
lines = []
# 表头
header = "| " + " | ".join(str(c or "") for c in table[0]) + " |"
separator = "| " + " | ".join(["---"] * len(table[0])) + " |"
lines.append(header)
lines.append(separator)
# 数据行
for row in table[1:]:
line = "| " + " | ".join(str(c or "") for c in row) + " |"
lines.append(line)
return "\n".join(lines)
# 特点:
# ✅ 表格提取是传统库中最好的
# ✅ 可以获取字符坐标信息
# ❌ 速度比 PyMuPDF 慢
# ❌ 扫描件无法处理
二、AI 驱动开源框架(RAG 推荐)
🏆 MinerU —— 当前最强开源 PDF 解析
项目:https://github.com/opendatalab/MinerU
开发:上海人工智能实验室(OpenDataLab)
Stars:⭐ 25k+
特点:专为 RAG/LLM 数据准备设计
核心能力:
✅ 版面分析(Layout Analysis)识别标题/正文/表格/图片区域
✅ 公式识别(LaTeX 输出)
✅ 表格识别(HTML/Markdown 输出)
✅ 阅读顺序重排(解决双栏问题)
✅ 图片区域 OCR
✅ 输出 Markdown(LLM 最友好格式)
✅ 支持 GPU 加速
# pip install magic-pdf[full]
# 或 pip install mineru
from magic_pdf.data.data_reader_writer import FileBasedDataWriter
from magic_pdf.data.dataset import PymuDocDataset
from magic_pdf.model.doc_analyze_by_custom_model import doc_analyze
from magic_pdf.config.enums import SupportedPdfParseMethod
def parse_with_mineru(pdf_path: str, output_dir: str):
import os
# 读取 PDF
with open(pdf_path, "rb") as f:
pdf_bytes = f.read()
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
writer = FileBasedDataWriter(output_dir)
# 数据集封装
ds = PymuDocDataset(pdf_bytes)
# 自动判断解析模式(文字版/扫描版)
if ds.classify() == SupportedPdfParseMethod.OCR:
# 扫描件:走 OCR 模式
infer_result = ds.apply(doc_analyze, ocr=True)
pipe_result = infer_result.pipe_ocr_mode(writer)
else:
# 数字版:走文字提取模式
infer_result = ds.apply(doc_analyze, ocr=False)
pipe_result = infer_result.pipe_txt_mode(writer)
# 获取 Markdown 内容
md_content = pipe_result.get_markdown(output_dir)
return md_content
# 输出示例:
# # 第一章 产品概述
#
# ## 1.1 核心功能
#
# 本产品具备以下核心功能:
#
# | 功能 | 描述 | 状态 |
# |------|------|------|
# | 搜索 | 全文检索 | ✅ |
# | 推荐 | 个性化推荐 | ✅ |
#
# $$E = mc^2$$ ← 公式完整保留
MinerU 处理流程:
PDF 输入
↓
版面检测(YOLOv8/LayoutLMv3)
├── 识别:标题/段落/表格/图片/公式/页眉页脚
↓
内容提取
├── 文字区域 → 直接提取文字
├── 表格区域 → 表格结构识别 → Markdown/HTML
├── 公式区域 → 公式识别 → LaTeX
└── 图片区域 → OCR 或图片描述
↓
阅读顺序重排
└── 解决双栏/多栏排版顺序问题
↓
Markdown 输出
└── 结构化、对 LLM 友好
Docling —— IBM 开源,企业级
# pip install docling
from docling.document_converter import DocumentConverter
def parse_with_docling(pdf_path: str):
converter = DocumentConverter()
# 解析文档(支持 PDF/Word/Excel/PPT/HTML)
result = converter.convert(pdf_path)
# 导出 Markdown
markdown = result.document.export_to_markdown()
# 导出结构化 JSON
doc_dict = result.document.export_to_dict()
# 直接获取适合 RAG 的 Chunks
chunks = []
for element in result.document.iterate_items():
chunks.append({
"text": element.text,
"type": element.label, # title/text/table/figure
"page": element.prov[0].page_no if element.prov else None
})
return markdown, chunks
# Docling 特点:
# ✅ IBM 出品,企业级可靠性
# ✅ 多格式支持(不只是 PDF)
# ✅ 与 LlamaIndex / LangChain 深度集成
# ✅ 表格、公式识别优秀
# ✅ 内置 RAG 友好的 Chunking
Unstructured —— LangChain 生态最常用
# pip install unstructured[pdf]
from unstructured.partition.pdf import partition_pdf
from unstructured.staging.base import elements_to_json
def parse_with_unstructured(pdf_path: str):
# 基础解析
elements = partition_pdf(
filename=pdf_path,
strategy="hi_res", # auto/fast/hi_res/ocr_only
infer_table_structure=True, # 识别表格结构
extract_images_in_pdf=True, # 提取图片
extract_image_block_types=["Image", "Table"],
)
# 按类型分类处理
result = []
for element in elements:
elem_type = type(element).__name__
# Title / NarrativeText / Table / Image /
# ListItem / Header / Footer
if elem_type == "Table":
# 表格转 HTML
result.append({
"type": "table",
"content": element.metadata.text_as_html,
"page": element.metadata.page_number
})
elif elem_type in ("Header", "Footer"):
pass # 过滤页眉页脚
else:
result.append({
"type": elem_type,
"content": str(element),
"page": element.metadata.page_number
})
return result
# Strategy 选择:
# "fast" → 纯文字提取,最快,不识别版面
# "auto" → 自动选择(推荐)
# "hi_res" → 高精度版面分析,慢但准
# "ocr_only" → 全页 OCR(扫描件专用)
Marker —— 速度与质量平衡
# pip install marker-pdf
from marker.convert import convert_single_pdf
from marker.models import load_all_models
def parse_with_marker(pdf_path: str):
# 加载模型(首次加载较慢)
models = load_all_models()
# 解析(直接输出 Markdown)
markdown, images, metadata = convert_single_pdf(
pdf_path,
models,
max_pages=None, # None = 全部页
langs=["Chinese", "English"], # 指定语言提高准确率
batch_multiplier=2 # GPU 批处理倍数
)
return markdown, metadata
# Marker 特点:
# ✅ 速度比 MinerU 快(轻量模型)
# ✅ 输出干净的 Markdown
# ✅ 去除页眉页脚
# ✅ 公式转 LaTeX
# ⚠️ 复杂表格效果不如 MinerU
三、OCR 引擎(扫描件专用)
PaddleOCR —— 中文最强 OCR
# pip install paddlepaddle paddleocr
from paddleocr import PaddleOCR
import fitz
import numpy as np
from PIL import Image
import io
def parse_scanned_pdf(pdf_path: str):
"""处理扫描件 PDF"""
ocr = PaddleOCR(
use_angle_cls=True, # 文字方向检测
lang='ch', # 中英文混合
use_gpu=True, # GPU 加速
layout=True, # 版面分析
table=True, # 表格识别
)
doc = fitz.open(pdf_path)
all_text = []
for page_num in range(len(doc)):
page = doc[page_num]
# PDF 页面渲染为图片(300 DPI 精度更高)
mat = fitz.Matrix(300/72, 300/72)
pix = page.get_pixmap(matrix=mat)
img_bytes = pix.tobytes("png")
# 转为 numpy array
img = np.array(Image.open(io.BytesIO(img_bytes)))
# OCR 识别
result = ocr.ocr(img, cls=True)
# 提取文字(按 Y 坐标排序保证阅读顺序)
page_text = []
if result and result[0]:
lines = sorted(
result[0],
key=lambda x: x[0][0][1] # 按 Y 坐标排序
)
for line in lines:
bbox, (text, confidence) = line
if confidence > 0.8: # 置信度过滤
page_text.append(text)
all_text.append({
"page": page_num + 1,
"content": "\n".join(page_text)
})
return all_text
# PaddleOCR 特点:
# ✅ 中文识别最强(百度出品)
# ✅ 版面分析 + 表格识别
# ✅ 支持 80+ 语言
# ✅ GPU 加速,速度快
四、商业/云服务 API
┌──────────────────────────────────────────────────────────────┐
│ Azure Document Intelligence(前 Form Recognizer) │
│ ├── 效果:★★★★★ 业界最强 │
│ ├── 价格:$1.5 / 1000页 │
│ ├── 优点:表格、手写、复杂版面全搞定 │
│ └── 适合:对质量要求极高,预算充足 │
├──────────────────────────────────────────────────────────────┤
│ AWS Textract │
│ ├── 效果:★★★★☆ │
│ ├── 价格:$1.5 / 1000页 │
│ ├── 优点:AWS 生态集成好 │
│ └── 适合:已在 AWS 上的项目 │
├──────────────────────────────────────────────────────────────┤
│ 阿里云文档智能 │
│ ├── 效果:★★★★☆ │
│ ├── 价格:¥0.1 / 页(按量) │
│ ├── 优点:中文支持最好,国内访问快 │
│ └── 适合:国内项目,文件不出境要求低 │
├──────────────────────────────────────────────────────────────┤
│ 百度智能云文档解析 │
│ ├── 效果:★★★★☆ │
│ └── 优点:中文、表格效果好 │
└──────────────────────────────────────────────────────────────┘
# Azure Document Intelligence 示例
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.core.credentials import AzureKeyCredential
def parse_with_azure(pdf_path: str):
client = DocumentIntelligenceClient(
endpoint="https://xxx.cognitiveservices.azure.com/",
credential=AzureKeyCredential("your-key")
)
with open(pdf_path, "rb") as f:
poller = client.begin_analyze_document(
"prebuilt-layout", # 版面分析模型
body=f,
output_content_format="markdown" # 直接输出 Markdown!
)
result = poller.result()
# 直接获取 Markdown(Azure 原生支持)
markdown = result.content
# 表格信息
for table in result.tables:
print(f"表格:{table.row_count}行 x {table.column_count}列")
return markdown
核心问题处理
1. 双栏/多栏排版处理
问题:双栏 PDF 直接提取,文字顺序错乱
原始布局: 错误提取结果:
┌────┬────┐ "第一列第一行 第二列第一行
│左1 │右1 │ → 第一列第二行 第二列第二行..."
│左2 │右2 │
└────┴────┘
正确结果应为:
"第一列第一行 第一列第二行...(左栏完整)
第二列第一行 第二列第二行...(右栏完整)"
import fitz
def parse_two_column_pdf(pdf_path):
doc = fitz.open(pdf_path)
result = []
for page in doc:
width = page.rect.width
# 左右两栏分别提取
left_clip = fitz.Rect(0, 0, width/2, page.rect.height)
right_clip = fitz.Rect(width/2, 0, width, page.rect.height)
left_text = page.get_text("text", clip=left_clip)
right_text = page.get_text("text", clip=right_clip)
# 先左后右拼接
full_text = left_text.strip() + "\n" + right_text.strip()
result.append(full_text)
return result
# 更好的方案:用 MinerU/Docling(自动处理多栏)
2. 表格转 Markdown(RAG 友好)
def table_to_rag_friendly(table_data):
"""
将表格数据转换为 RAG 友好的格式
策略:Markdown 表格 + 自然语言描述
原因:LLM 对 Markdown 表格理解最好
"""
if not table_data:
return ""
lines = []
headers = table_data[0]
# Markdown 表格
lines.append("| " + " | ".join(str(h or "") for h in headers) + " |")
lines.append("| " + " | ".join(["---"] * len(headers)) + " |")
for row in table_data[1:]:
lines.append("| " + " | ".join(str(c or "") for c in row) + " |")
markdown_table = "\n".join(lines)
# 同时生成自然语言描述(增强检索)
description = f"表格包含 {len(table_data)-1} 行数据,"
description += f"列名为:{', '.join(str(h) for h in headers if h)}。"
return f"{description}\n\n{markdown_table}"
3. 图片中的文字(图片型 PDF)
def is_scanned_pdf(pdf_path: str) -> bool:
"""判断是否为扫描件(图片型)PDF"""
doc = fitz.open(pdf_path)
total_pages = len(doc)
text_pages = 0
for page in doc:
text = page.get_text().strip()
if len(text) > 50: # 有足够文字
text_pages += 1
doc.close()
# 超过 50% 页面无文字 → 判断为扫描件
return text_pages / total_pages < 0.5
def smart_parse(pdf_path: str):
"""智能选择解析策略"""
if is_scanned_pdf(pdf_path):
print("检测到扫描件,使用 OCR 模式")
return parse_scanned_pdf(pdf_path) # PaddleOCR
else:
print("检测到数字 PDF,使用文字提取模式")
return parse_with_mineru(pdf_path) # MinerU
RAG 专用解析最佳实践
完整 RAG PDF 解析流水线
import fitz
import hashlib
from pathlib import Path
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class ParsedChunk:
content: str
chunk_type: str # title/text/table/image_caption
page_num: int
doc_title: str
chunk_index: int
metadata: dict
class RagPdfParser:
"""RAG 专用 PDF 解析器"""
def __init__(self, use_gpu: bool = True):
self.use_gpu = use_gpu
def parse(self, pdf_path: str) -> List[ParsedChunk]:
pdf_path = Path(pdf_path)
# Step 1: 判断 PDF 类型
if self._is_scanned(pdf_path):
raw_elements = self._ocr_parse(pdf_path)
else:
raw_elements = self._digital_parse(pdf_path)
# Step 2: 清洗噪音
cleaned = self._clean_elements(raw_elements)
# Step 3: 语义切片
chunks = self._semantic_chunk(cleaned, pdf_path.stem)
return chunks
def _is_scanned(self, pdf_path) -> bool:
doc = fitz.open(str(pdf_path))
text_ratio = sum(
1 for page in doc if len(page.get_text().strip()) > 100
) / len(doc)
doc.close()
return text_ratio < 0.5
def _digital_parse(self, pdf_path):
"""数字 PDF 解析(MinerU)"""
from magic_pdf.pipe.UNIPipe import UNIPipe
# ... MinerU 解析逻辑
pass
def _ocr_parse(self, pdf_path):
"""扫描件解析(PaddleOCR)"""
from paddleocr import PaddleOCR
# ... PaddleOCR 解析逻辑
pass
def _clean_elements(self, elements):
"""清洗噪音数据"""
cleaned = []
for elem in elements:
content = elem.get("content", "").strip()
# 过滤规则
if len(content) < 10: continue # 太短
if self._is_header_footer(content): continue # 页眉页脚
if self._is_noise(content): continue # 纯数字/符号
cleaned.append(elem)
return cleaned
def _is_header_footer(self, text: str) -> bool:
"""检测页眉页脚"""
import re
# 纯页码
if re.match(r'^\d+$', text.strip()):
return True
# 常见页眉模式
if re.match(r'^第\s*\d+\s*页', text.strip()):
return True
return False
def _is_noise(self, text: str) -> bool:
"""检测噪音文本"""
# 超过 50% 是特殊字符
special_ratio = sum(1 for c in text if not c.isalnum()) / len(text)
return special_ratio > 0.7
def _semantic_chunk(
self, elements: list, doc_title: str
) -> List[ParsedChunk]:
"""
语义切片:按标题层级切分,保持上下文完整性
"""
chunks = []
current_section = []
current_title = ""
chunk_index = 0
for elem in elements:
elem_type = elem.get("type", "text")
content = elem.get("content", "")
page = elem.get("page", 1)
if elem_type == "title":
# 遇到新标题,保存当前片段
if current_section:
chunks.append(ParsedChunk(
content="\n".join(current_section),
chunk_type="text",
page_num=page,
doc_title=doc_title,
chunk_index=chunk_index,
metadata={"section": current_title}
))
chunk_index += 1
current_title = content
current_section = [f"# {content}"]
elif elem_type == "table":
# 表格单独成 chunk
if current_section:
chunks.append(ParsedChunk(
content="\n".join(current_section),
chunk_type="text",
page_num=page,
doc_title=doc_title,
chunk_index=chunk_index,
metadata={"section": current_title}
))
chunk_index += 1
current_section = []
chunks.append(ParsedChunk(
content=content,
chunk_type="table",
page_num=page,
doc_title=doc_title,
chunk_index=chunk_index,
metadata={"section": current_title}
))
chunk_index += 1
else:
current_section.append(content)
# 单个 chunk 不超过 1000 字
if len("\n".join(current_section)) > 1000:
chunks.append(ParsedChunk(
content="\n".join(current_section),
chunk_type="text",
page_num=page,
doc_title=doc_title,
chunk_index=chunk_index,
metadata={"section": current_title}
))
chunk_index += 1
current_section = []
return chunks
各方案横向对比
┌────────────────┬────────┬────────┬────────┬────────┬────────┐
│ 方案 │ 速度 │ 效果 │ 表格 │ 扫描件 │ 部署 │
├────────────────┼────────┼────────┼────────┼────────┼────────┤
│ PyMuPDF │ ⭐⭐⭐⭐⭐ │ ⭐⭐⭐ │ ⭐⭐ │ ❌ │ 极简 │
│ pdfplumber │ ⭐⭐⭐⭐ │ ⭐⭐⭐ │ ⭐⭐⭐⭐ │ ❌ │ 极简 │
│ MinerU │ ⭐⭐⭐ │ ⭐⭐⭐⭐⭐ │ ⭐⭐⭐⭐⭐ │ ✅ │ 需GPU │
│ Docling │ ⭐⭐⭐ │ ⭐⭐⭐⭐⭐ │ ⭐⭐⭐⭐⭐ │ ✅ │ 需GPU │
│ Marker │ ⭐⭐⭐⭐ │ ⭐⭐⭐⭐ │ ⭐⭐⭐ │ ✅ │ 需GPU │
│ Unstructured │ ⭐⭐⭐ │ ⭐⭐⭐⭐ │ ⭐⭐⭐⭐ │ ✅ │ 中等 │
│ PaddleOCR │ ⭐⭐⭐ │ ⭐⭐⭐⭐ │ ⭐⭐⭐ │ ✅✅ │ 需GPU │
│ Azure DI │ ⭐⭐⭐⭐ │ ⭐⭐⭐⭐⭐ │ ⭐⭐⭐⭐⭐ │ ✅✅ │ 云服务 │
└────────────────┴────────┴────────┴────────┴────────┴────────┘
选型决策
文件类型?
│
├── 扫描件(图片 PDF)
│ ├── 中文为主 ──────────→ PaddleOCR + MinerU(OCR模式)
│ └── 多语言 ──────────→ Azure DI / Docling
│
└── 数字 PDF(可选中文字)
│
├── 有复杂表格/公式?
│ ├── 是,且可用GPU ─→ 🏆 MinerU 或 Docling
│ ├── 是,无GPU ─→ 🏆 pdfplumber + PyMuPDF
│ └── 数据敏感不上云,效果要最好 → MinerU
│
├── 简单纯文字 PDF?
│ └── PyMuPDF(最快最轻量)
│
├── 已在 LangChain 生态?
│ └── Unstructured(无缝集成)
│
└── 预算充足,效果优先?
└── Azure Document Intelligence
总结
┌───────────────────────────────────────────────────────┐
│ RAG PDF 解析选型推荐 │
├──────────────────┬────────────────────────────────────┤
│ 场景 │ 推荐方案 │
├──────────────────┼────────────────────────────────────┤
│ 生产级RAG系统 │ 🏆 MinerU(最强开源) │
│ 企业文档处理 │ 🏆 Docling(IBM出品,稳定) │
│ 有扫描件 │ MinerU + PaddleOCR(中文) │
│ 轻量快速 │ PyMuPDF + pdfplumber │
│ LangChain集成 │ Unstructured │
│ 效果第一不计费 │ Azure Document Intelligence │
│ 国内项目 │ MinerU / 阿里云文档智能 │
└──────────────────┴────────────────────────────────────┘
一句话:RAG 项目中 PDF 解析首选 MinerU(国内团队出品、专为LLM设计、完全开源、输出Markdown),有扫描件则搭配 PaddleOCR,对质量要求极致且预算充足选 Azure Document Intelligence。