第 22 章 · 扩展：复杂文档解析管线（PDF/PPT/OCR/转录）
本章目标：
- 用 PyMuPDF 对 PDF 做结构化抽取（文本、表格、版面块），为 RAG 提供高质量语料
- 用 python-pptx 解析 PPT，保留「页 → 标题 → 正文 → 备注」的层级结构
- 集成 OCR（Tesseract / PaddleOCR）处理扫描件与图片中的文字
- 用 trafilatura 提取网页正文，用 Whisper 完成音视频字幕转录
- 把以上能力封装成一个统一的「文档解析 Agent 工具」，接入 LangGraph 工作流
在实战一的 RAG 知识库 Agent 中，我们的数据入口只有 Markdown 文本。但真实业务中，知识往往散落在 PDF 报告、PPT 课件、扫描合同、会议录屏里。文档解析管线是 RAG 系统的第一公里——解析质量直接决定检索上限。本章把 JD 中「复杂文档解析」的加分项能力系统化，并封装为 Agent 可调用的工具。
22.1 PDF 结构化抽取：PyMuPDF
PDF 有两类：原生 PDF（文字可选中，直接抽取）和扫描 PDF（本质是图片，需要 OCR）。先处理前者。
bash
pip install pymupdf22.1.1 基础文本抽取与版面块
python
# agent_prod/parsers/pdf_parser.py
import fitz # PyMuPDF
def extract_pdf_pages(path: str) -> list[dict]:
"""按页抽取文本块，保留页码与坐标信息"""
doc = fitz.open(path)
pages = []
for i, page in enumerate(doc):
# blocks 返回 (x0, y0, x1, y1, text, block_no, block_type)
# block_type: 0=文本, 1=图片
blocks = [
{
"text": b[4].strip(),
"bbox": (b[0], b[1], b[2], b[3]),
"type": "text" if b[6] == 0 else "image",
}
for b in page.get_text("blocks")
if b[4].strip()
]
pages.append({"page": i + 1, "blocks": blocks})
doc.close()
return pages
def is_scanned_pdf(path: str, threshold: int = 50) -> bool:
"""判断是否为扫描件：全文字符数过少则视为扫描件"""
doc = fitz.open(path)
total_chars = sum(len(page.get_text()) for page in doc)
doc.close()
return total_chars < threshold * len(fitz.open(path))💡 为什么要保留坐标（bbox）？ 后续做「标题识别」时，字号大、居中的块通常是标题；做表格还原时，坐标能判断列对齐关系。丢弃版面信息等于把 PDF 退化成纯文本流。
22.1.2 标题层级识别与结构化输出
python
def extract_with_headings(path: str) -> list[dict]:
"""利用字号推断标题层级，输出结构化章节树"""
doc = fitz.open(path)
sections, current = [], None
# 第一遍：统计全文字号分布，出现最多的字号是正文
sizes = [s["size"] for page in doc for s in page.get_text("dict")["blocks"]
if s["type"] == 0 for line in s["lines"] for s in line["spans"]]
body_size = max(set(sizes), key=sizes.count)
for pno, page in enumerate(doc, 1):
for block in page.get_text("dict")["blocks"]:
if block["type"] != 0:
continue
for line in block["lines"]:
text = "".join(s["text"] for s in line["spans"]).strip()
if not text:
continue
size = max(s["size"] for s in line["spans"])
if size > body_size * 1.3:
current = {"level": 1, "title": text, "page": pno, "content": []}
sections.append(current)
elif size > body_size * 1.1 and current:
current["content"].append(f"### {text}")
elif current:
current["content"].append(text)
doc.close()
for s in sections:
s["content"] = "\n".join(s["content"])
return sections22.1.3 表格抽取
python
def extract_tables(path: str) -> list[dict]:
"""抽取 PDF 中的表格，返回每张表的页码与二维数组"""
doc = fitz.open(path)
tables = []
for pno, page in enumerate(doc, 1):
finder = page.find_tables() # PyMuPDF 内置表格识别
for t in finder.tables:
tables.append({
"page": pno,
"rows": t.extract(), # 二维列表
"row_count": t.row_count,
"col_count": t.col_count,
})
doc.close()
return tables
# 转成 Markdown 表格，便于直接进入 RAG 语料
def table_to_markdown(table: dict) -> str:
rows = table["rows"]
header = "| " + " | ".join(str(c or "") for c in rows[0]) + " |"
sep = "| " + " | ".join("---" for _ in rows[0]) + " |"
body = "\n".join("| " + " | ".join(str(c or "") for c in r) + " |" for r in rows[1:])
return f"{header}\n{sep}\n{body}"22.2 PPT 结构化解析：python-pptx
PPT 的价值在于「页级语义单元」：每页有标题、正文要点、演讲者备注——天然适合做检索的分块（chunk）。
bash
pip install python-pptxpython
# agent_prod/parsers/pptx_parser.py
from pptx import Presentation
from pptx.enum.shapes import MSO_SHAPE_TYPE
def parse_pptx(path: str) -> list[dict]:
"""解析 PPT，输出页级结构：标题/正文/表格/备注"""
prs = Presentation(path)
slides = []
for idx, slide in enumerate(prs.slides, 1):
item = {"slide": idx, "title": "", "bullets": [], "tables": [], "notes": ""}
for shape in slide.shapes:
if shape.has_text_frame:
text = shape.text_frame.text.strip()
if not text:
continue
# 占位符类型决定角色：标题占位符优先
if shape == slide.shapes.title:
item["title"] = text
else:
item["bullets"].extend(
p.text.strip() for p in shape.text_frame.paragraphs if p.text.strip()
)
elif shape.has_table:
rows = [[cell.text.strip() for cell in row.cells] for row in shape.table.rows]
item["tables"].append(rows)
elif shape.shape_type == MSO_SHAPE_TYPE.PICTURE:
item.setdefault("images", []).append(shape.image.blob[:0]) # 记录存在图片
if slide.has_notes_slide:
item["notes"] = slide.notes_slide.notes_text_frame.text.strip()
slides.append(item)
return slides
def slide_to_markdown(slide: dict) -> str:
"""页级转 Markdown：标题 + 要点 + 备注，一个 slide 一个 chunk"""
parts = [f"# {slide['title']}"] if slide["title"] else []
parts += [f"- {b}" for b in slide["bullets"]]
for table in slide["tables"]:
parts.append(table_to_markdown({"rows": table}))
if slide["notes"]:
parts.append(f"> 备注：{slide['notes']}")
return "\n".join(parts)💡 分块策略：PPT 天然按页分块，每页转成一段 Markdown 后直接送入 embedding，检索命中时能定位到「第 N 页」，方便回溯原文。
22.3 OCR：处理扫描件与图片文字
22.3.1 Tesseract：轻量通用方案
bash
# macOS: brew install tesseract tesseract-lang（含中文包）
# Ubuntu: apt install tesseract-ocr tesseract-ocr-chi-sim
pip install pytesseractpython
# agent_prod/parsers/ocr.py
import fitz
import pytesseract
from PIL import Image
import io
def ocr_scanned_pdf(path: str, lang: str = "chi_sim+eng") -> list[dict]:
"""扫描 PDF → 逐页渲染为图片 → OCR"""
doc = fitz.open(path)
results = []
for pno, page in enumerate(doc, 1):
# 300 DPI 渲染：dpi 太低会丢笔画，太高浪费算力
pix = page.get_pixmap(dpi=300)
img = Image.open(io.BytesIO(pix.tobytes("png")))
text = pytesseract.image_to_string(img, lang=lang)
results.append({"page": pno, "text": text.strip()})
doc.close()
return results
def ocr_image(image_bytes: bytes, lang: str = "chi_sim+eng") -> str:
"""单张图片 OCR（Agent 工具常用入口）"""
img = Image.open(io.BytesIO(image_bytes))
return pytesseract.image_to_string(img, lang=lang).strip()22.3.2 PaddleOCR：中文场景的更高精度选择
Tesseract 对中文排版（竖排、多栏、艺术字）效果一般。生产中中文 OCR 首选 PaddleOCR：
python
# pip install paddleocr
from paddleocr import PaddleOCR
ocr_engine = PaddleOCR(use_angle_cls=True, lang="ch") # 初始化一次，全局复用
def paddle_ocr(image_path: str) -> str:
"""PaddleOCR：返回按阅读顺序拼接的文本"""
result = ocr_engine.ocr(image_path, cls=True)
lines = []
for page in result:
for line in (page or []):
lines.append(line[1][0]) # (bbox, (text, confidence))
return "\n".join(lines)⚠️ 工程要点：OCR 引擎初始化成本高（模型加载 1-3 秒），必须做成进程级单例，不要在每次请求里重复创建。在 FastAPI 中放到
lifespan里初始化。
22.4 网页正文提取与字幕转录
22.4.1 trafilatura：网页正文提取
爬下来的 HTML 充满导航栏、广告、页脚。正则清洗不可维护，trafilatura 用可读性算法自动识别正文：
bash
pip install trafilaturapython
# agent_prod/parsers/web_parser.py
import trafilatura
def extract_web_content(url: str) -> dict | None:
"""提取网页正文 + 元数据，失败返回 None"""
downloaded = trafilatura.fetch_url(url)
if not downloaded:
return None
text = trafilatura.extract(
downloaded,
include_comments=False, # 排除评论区
include_tables=True, # 保留表格
favor_recall=True, # 倾向召回更多内容
)
metadata = trafilatura.extract_metadata(downloaded)
return {
"text": text,
"title": metadata.title if metadata else "",
"url": url,
}22.4.2 Whisper：音视频转录为字幕文本
会议录音、课程视频是重要的知识载体。OpenAI Whisper（本地开源版）可离线转录：
bash
pip install openai-whisper
# ffmpeg 是 whisper 的依赖：brew install ffmpeg / apt install ffmpegpython
# agent_prod/parsers/transcribe.py
import whisper
# 模型分级：tiny < base < small < medium < large-v3
# 中文推荐 small 起步；首次调用会自动下载模型
_model = None
def get_model(size: str = "small"):
global _model
if _model is None:
_model = whisper.load_model(size)
return _model
def transcribe_to_srt(audio_path: str, size: str = "small") -> dict:
"""转录并输出带时间戳的分段，可转 SRT 字幕"""
model = get_model(size)
result = model.transcribe(audio_path, language="zh", verbose=False)
segments = [
{"start": seg["start"], "end": seg["end"], "text": seg["text"].strip()}
for seg in result["segments"]
]
return {"text": result["text"], "segments": segments}
def segments_to_srt(segments: list[dict]) -> str:
def fmt(ts: float) -> str:
h, m = int(ts // 3600), int(ts % 3600 // 60)
s, ms = int(ts % 60), int(ts % 1 * 1000)
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
lines = []
for i, seg in enumerate(segments, 1):
lines += [str(i), f"{fmt(seg['start'])} --> {fmt(seg['end'])}", seg["text"], ""]
return "\n".join(lines)💡 成本权衡：本地 Whisper 免费但吃 GPU/CPU；如果已有 OpenAI API Key，
client.audio.transcriptions.create(model="whisper-1")更省事。在 agent-prod ch15 的多网关成本框架下，把「转录」也纳入模型路由决策：短音频走 API，长音频走本地。
22.5 封装为 Agent 工具：统一解析入口
把所有解析能力收敛到一个工具函数，让 LangGraph 的 Agent 按文件类型自动路由：
python
# agent_prod/parsers/router.py
from pathlib import Path
from agent_prod.parsers.pdf_parser import extract_with_headings, is_scanned_pdf
from agent_prod.parsers.ocr import ocr_scanned_pdf
from agent_prod.parsers.pptx_parser import parse_pptx, slide_to_markdown
from agent_prod.parsers.web_parser import extract_web_content
from agent_prod.parsers.transcribe import transcribe_to_srt
def parse_document(source: str) -> dict:
"""统一文档解析入口：按类型路由到对应解析器
source: 本地文件路径或 URL
返回: {"kind": 类型, "chunks": [markdown 片段列表]}
"""
if source.startswith(("http://", "https://")):
if any(source.endswith(ext) for ext in (".mp3", ".mp4", ".wav", ".m4a")):
result = transcribe_to_srt(source)
return {"kind": "transcript", "chunks": [result["text"]]}
web = extract_web_content(source)
return {"kind": "web", "chunks": [web["text"]] if web else []}
path = Path(source)
suffix = path.suffix.lower()
if suffix == ".pdf":
if is_scanned_pdf(source):
pages = ocr_scanned_pdf(source)
return {"kind": "pdf-ocr", "chunks": [p["text"] for p in pages if p["text"]]}
sections = extract_with_headings(source)
return {"kind": "pdf", "chunks": [f"# {s['title']}\n{s['content']}" for s in sections]}
if suffix == ".pptx":
slides = parse_pptx(source)
return {"kind": "pptx", "chunks": [slide_to_markdown(s) for s in slides]}
if suffix in (".md", ".txt"):
return {"kind": "text", "chunks": [path.read_text(encoding="utf-8")]}
raise ValueError(f"不支持的文件类型: {suffix}")接入 LangGraph（复用实战一的 RAG 图，只替换数据入口）：
python
# agent_prod/graphs/ingest_graph.py
from langgraph.graph import StateGraph, END
from typing import TypedDict
from agent_prod.parsers.router import parse_document
# 假设实战一已有 embed_and_store / chunk_text 两个能力
from agent_prod.rag import chunk_text, embed_and_store
class IngestState(TypedDict):
source: str
kind: str
chunks: list[str]
stored: int
def parse_node(state: IngestState) -> dict:
result = parse_document(state["source"])
return {"kind": result["kind"], "chunks": result["chunks"]}
def chunk_node(state: IngestState) -> dict:
# 长段落二次切分，控制 chunk 在 512 token 以内
fine_chunks = [c for chunk in state["chunks"] for c in chunk_text(chunk, max_tokens=512)]
return {"chunks": fine_chunks}
def store_node(state: IngestState) -> dict:
return {"stored": embed_and_store(state["chunks"], metadata={"kind": state["kind"]})}
def build_ingest_graph() -> StateGraph:
g = StateGraph(IngestState)
g.add_node("parse", parse_node)
g.add_node("chunk", chunk_node)
g.add_node("store", store_node)
g.set_entry_point("parse")
g.add_edge("parse", "chunk")
g.add_edge("chunk", "store")
g.add_edge("store", END)
return g.compile()
# 使用：一条命令把一份 PDF 课件灌进知识库
# ingest = build_ingest_graph()
# ingest.invoke({"source": "reports/q3-review.pdf"})22.6 生产化要点
- 异步化：OCR 和 Whisper 都是 CPU/GPU 密集操作，在 FastAPI 中必须丢进
run_in_executor或 Celery 任务队列（见 ch02），避免阻塞事件循环。 - 幂等入库：对文件内容算 SHA-256 作为文档 ID，重复上传直接跳过，防止向量库重复灌入（呼应 ch03 的幂等设计）。
- 解析质量监控：给每个解析任务记录
char_count、ocr_confidence等指标上报到 Langfuse/CloudWatch（见 ch14/ch18），空解析率突增说明上游格式变了。 - 失败降级：PaddleOCR 失败时降级到 Tesseract；Whisper 本地 OOM 时降级到 API 转录——降级链路要有告警。
本章小结
- PDF：PyMuPDF 抽取文本/表格/版面，用字号分布推断标题层级；扫描件走 OCR 分支
- PPT：python-pptx 按「页 → 标题/要点/备注」结构化，天然适合页级分块
- OCR：Tesseract 通用轻量，中文场景首选 PaddleOCR；引擎必须进程级单例
- 网页：trafilatura 自动剥离噪音提取正文；转录：Whisper 本地或 API 按成本路由
- 统一入口：
parse_document按类型路由，输出 Markdown chunks 直接对接 RAG 图 - 生产四件套：异步化、幂等入库、质量监控、失败降级
🛠️ 动手实践
- 解析质量评估集：找 3 份不同类型的 PDF（原生/扫描/带表格），编写 pytest 用例断言解析输出的最小字符数与表格行数，把「文档解析」纳入 CI（呼应 pytest 课程 ch19）。
- 元数据检索过滤：扩展
store_node，把kind、文件名、页码写入向量库 metadata，并在检索时支持filter={"kind": "pptx"}。 - 批量灌库脚本：写一个 CLI 脚本遍历 S3 桶（boto3，复用 ch16 的凭证配置），把所有 PDF/PPTX 增量灌入知识库，用文件哈希实现幂等。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 判断一份 PDF 是否为扫描件，最可靠的启发式方法是什么？
2. 为什么 PPT 特别适合按页（slide 级）做 RAG 分块？
3. 在 FastAPI 服务中使用 PaddleOCR，正确的初始化方式是什么？
4. 文档解析管线的「幂等入库」通常如何实现？