ARTICLE · 1154972
PDF解析三策略工程取舍
标题：PDF解析三策略工程取舍
digest：PDF解析三策略：文字vs Vision，附并发代码
引子：文档处理是多模态工程的高频需求
上一篇讲了 Vision API 的请求结构与多图处理。今天我们把视线拉到一个更常见的场景——PDF 与文档解析。
你会发现，客户系统里流转的合同、财报、发票、研报，绝大多数都以 PDF 形式存在。做 RAG 要索引它们、做 Agent 要读取它们、做数据抽取要结构化它们。PDF 处理表面上是"读文件"，实际上是多模态工程绕不开的第一道关卡。
坑在哪里？PDF 是个非常"混血"的格式：
有的 PDF 是原生数字文档，文字层完整（Word 导出、LaTeX 编译） 有的是扫描件，本质上就是一堆图片 有的是"半原生"——文字层是乱码或空的，图形却是清晰的（常见于第三方工具生成） 更多是图文混排——表格、图表、印章、手写签名混在一起
如果你只用一种策略处理所有 PDF，要么慢得离谱，要么精度差到不能用。这篇文章讲清楚三种主流策略、它们的适用场景，以及如何在工程上把它们组合成一个自动选路的解析管线。
三种策略：一张表看懂取舍
核心原则：能用文字提取就别用 Vision——不是精度问题，是速度和成本差了两个数量级。但是必须有一套"检测机制"知道什么时候文字提取不够用，需要降级到 Vision。这个检测机制才是策略C的价值所在。
代码实现：统一的 PDFParser 接口
我们先搭一个骨架。三种策略对外暴露统一接口 parse(pdf_path) -> ParseResult，内部由策略选择器决定走哪条路。所有代码都支持 mock 模式（无需真实 API key），方便本地跑通。
"""PDF 解析工程 - 三策略统一接口运行方式：python pdf_parser.py （默认 mock 模式）"""from __future__ import annotationsimport asyncioimport jsonimport randomimport timefrom dataclasses import dataclass, field, asdictfrom enum import Enumfrom pathlib import Pathfrom typing import Anyclass Strategy(str, Enum): TEXT = "text" # 方案A：纯文字提取 VISION = "vision" # 方案B：Vision 识别 HYBRID = "hybrid" # 方案C：混合，自动选路@dataclassclass PageResult: page_no: int strategy_used: Strategy text: str tables: list[dict] = field(default_factory=list) charts: list[dict] = field(default_factory=list) confidence: float = 1.0 cost_tokens: int = 0 latency_ms: int = 0@dataclassclass ParseResult: pdf_path: str pages: list[PageResult] = field(default_factory=list) total_latency_ms: int = 0 total_cost_tokens: int = 0 def to_dict(self) -> dict: return { "pdf_path": self.pdf_path, "total_latency_ms": self.total_latency_ms, "total_cost_tokens": self.total_cost_tokens, "pages": [asdict(p) for p in self.pages], }
文字提取路径（方案A）
真实实现会用 pdfplumber，mock 模式下我们模拟三种情况：正常提取、乱码、空白。
class TextExtractor: """方案A：直接从 PDF 文字层提取""" def __init__(self, mock: bool = True): self.mock = mock def extract_page(self, pdf_path: str, page_no: int) -> PageResult: t0 = time.time() if self.mock: # 模拟三种情况，用文件名决定行为 name = Path(pdf_path).stem if "scan" in name: text = "" # 扫描件文字层为空 conf = 0.0 elif "garbled" in name: text = "��� 乱码 ��" conf = 0.15 else: text = f"[第{page_no}页] 这是一段原生 PDF 提取到的正文内容。" conf = 0.95 else: import pdfplumber with pdfplumber.open(pdf_path) as pdf: page = pdf.pages[page_no - 1] text = page.extract_text() or "" conf = self._estimate_confidence(text) latency = int((time.time() - t0) * 1000) or 1 return PageResult( page_no=page_no, strategy_used=Strategy.TEXT, text=text, confidence=conf, cost_tokens=0, latency_ms=latency, ) @staticmethod def _estimate_confidence(text: str) -> float: """极简置信度：文本长度 + 非乱码字符比例""" if not text or len(text) < 20: return 0.0 # 检测替换符/乱码 garbled = sum(1 for c in text if c in "�" or ord(c) < 32) ratio = 1 - garbled / max(len(text), 1) return max(0.0, min(1.0, ratio * (min(len(text), 500) / 500)))
Vision 识别路径（方案B）
真实实现会调用 pdf2image 渲染出 PNG，再送到 Vision API（通义千问VL、DeepSeek-VL2 等）。mock 模式下直接生成"看起来合理"的结果。
class VisionExtractor: """方案B：pdf2image 渲染 + Vision 识别""" def __init__(self, mock: bool = True, dpi: int = 200): self.mock = mock self.dpi = dpi async def extract_page(self, pdf_path: str, page_no: int) -> PageResult: t0 = time.time() if self.mock: # 模拟 Vision 调用延迟 await asyncio.sleep(random.uniform(0.05, 0.15)) text = f"[第{page_no}页/Vision识别] 扫描件正文内容，包含图表和表格。" tables = [{ "title": "月度销售", "headers": ["月份", "销售额（万元）"], "rows": [["1月", "128"], ["2月", "156"], ["3月", "203"]], }] conf = 0.92 tokens = 2400 # 200 DPI 单页图约 2000-3000 tokens else: # 真实路径：pdf2image → base64 → Vision API from pdf2image import convert_from_path import base64, io images = convert_from_path( pdf_path, dpi=self.dpi, first_page=page_no, last_page=page_no, ) buf = io.BytesIO() images[0].save(buf, format="PNG") img_b64 = base64.b64encode(buf.getvalue()).decode() text, tables, tokens, conf = await self._call_vision(img_b64) latency = int((time.time() - t0) * 1000) or 1 return PageResult( page_no=page_no, strategy_used=Strategy.VISION, text=text, tables=tables, confidence=conf, cost_tokens=tokens, latency_ms=latency, ) async def _call_vision(self, img_b64: str): # 生产实现：调用 messages.create 传入 image + text prompt # 要求模型返回 JSON 格式：{"text": ..., "tables": [...]} raise NotImplementedError("真实 Vision 调用需 API key")
策略选择器（方案C）
这是整篇文章最关键的部分。决策树如下：
PDF 输入 │ ├─ 是扫描件？（文字层空 or 只有图像流）→ 直接走 Vision │ ├─ 有文字层？ │ │ │ ├─ 尝试提取 │ ├─ 置信度 ≥ 0.7 → 采用文字结果 │ └─ 置信度 < 0.7 → 降级 Vision │ └─ 页面包含复杂表格/图表？（可通过启发式检测） → 文字 + Vision 都跑一遍，Vision 结果补充表格字段
class PDFParser: """统一入口，自动选择策略""" def __init__(self, strategy: Strategy = Strategy.HYBRID, mock: bool = True): self.strategy = strategy self.text_extractor = TextExtractor(mock=mock) self.vision_extractor = VisionExtractor(mock=mock) self.confidence_threshold = 0.7 async def parse(self, pdf_path: str, num_pages: int = 3) -> ParseResult: result = ParseResult(pdf_path=pdf_path) t0 = time.time() for page_no in range(1, num_pages + 1): page = await self._parse_page(pdf_path, page_no) result.pages.append(page) result.total_cost_tokens += page.cost_tokens result.total_latency_ms = int((time.time() - t0) * 1000) return result async def _parse_page(self, pdf_path: str, page_no: int) -> PageResult: if self.strategy == Strategy.TEXT: return self.text_extractor.extract_page(pdf_path, page_no) if self.strategy == Strategy.VISION: return await self.vision_extractor.extract_page(pdf_path, page_no) # HYBRID：先文字，置信度不达标降级 Vision text_page = self.text_extractor.extract_page(pdf_path, page_no) if text_page.confidence >= self.confidence_threshold: return text_page # 降级 vision_page = await self.vision_extractor.extract_page(pdf_path, page_no) vision_page.text = f"[降级Vision, 文字置信度={text_page.confidence:.2f}] " + vision_page.text return vision_page
表格提取：为什么纯文字不够
pdfplumber 也能提表格（page.extract_tables()），但真实项目里你会发现三个问题：
- 跨页表格
：表头在上一页、数据行在下一页，解析后完全散开 - 合并单元格
：pdfplumber 得到的是矩阵结构，但语义上是"合并"，恢复成本高 - 无边框表格
：靠视觉对齐的表格，pdfplumber 只能抽出文本行，不认它是表格
Vision 路径的优势是让模型"理解"表格语义——不是解析像素坐标，而是识别这是一个"3列4行的销售数据表"。我们用结构化 JSON 输出来强制模型给出规范格式：
class TableExtractor: """表格提取：强制 JSON schema 输出""" JSON_SCHEMA = { "type": "object", "properties": { "tables": { "type": "array", "items": { "type": "object", "properties": { "title": {"type": "string"}, "headers": {"type": "array", "items": {"type": "string"}}, "rows": { "type": "array", "items": {"type": "array", "items": {"type": "string"}}, }, "footnote": {"type": "string"}, }, "required": ["headers", "rows"], }, } }, } PROMPT = """请识别图中所有表格，严格按 JSON 输出：{"tables": [{"title": str, "headers": [str...], "rows": [[str...]...], "footnote": str}]}规则：1. 合并单元格保持合并后的实际内容，重复填充到每个子行2. 数字保留原始格式（含千分位、百分号、货币符号）3. 表格标题从上下文提取4. 不要编造数据，看不清就填空字符串""" async def extract(self, img_b64: str, mock: bool = True) -> list[dict]: if mock: return [{ "title": "Q3 财报摘要", "headers": ["指标", "本期", "上期", "同比"], "rows": [ ["营收", "12.8亿", "10.5亿", "+21.9%"], ["净利润", "2.1亿", "1.6亿", "+31.3%"], ], "footnote": "数据来源：公司公告", }] # 生产实现：调用 Vision，用 tool_use 强制 schema raise NotImplementedError
工程上的关键点：不要用自由文本让模型"描述"表格，永远走 tool_use 或 JSON mode，把 schema 写死。这样下游可以直接 json.loads 消费，不需要正则清洗。
图表数据提取：从像素到数值
折线图、柱图这种"图形化数据"，纯文字提取完全没辙。Vision 能做的事情是读取轴标签 + 估算数据点。
class ChartExtractor: """图表数据提取（折线图/柱图 → 数值序列）""" PROMPT = """图中包含图表，请提取为 JSON：{"charts": [{ "type": "line|bar|pie", "title": str, "x_axis": {"label": str, "unit": str}, "y_axis": {"label": str, "unit": str}, "series": [{"name": str, "data": [{"x": ..., "y": number}]}]}]}估算数值时给出你的置信度 confidence: 0-1""" async def extract(self, img_b64: str, mock: bool = True) -> list[dict]: if mock: return [{ "type": "line", "title": "近12月月活", "x_axis": {"label": "月份", "unit": ""}, "y_axis": {"label": "MAU", "unit": "万"}, "series": [{ "name": "iOS", "data": [{"x": f"2026-{m:02d}", "y": 100 + m * 8} for m in range(1, 13)], }], "confidence": 0.82, }] raise NotImplementedError
注意点：图表数据提取永远是估算，不是精确读数。生产中的做法是：Vision 拿到近似值，如果原始报告有配套 Excel/CSV 数据源，用 Vision 结果做校验和索引，实际数据走原始源。别把 Vision 的估算值当真值直接入库。
批量处理：asyncio 并发控制
真实项目里，PDF 往往是批量的——一个客户送来 200 份合同，或者你要索引整个论文库。这时候必须做并发，但又不能把 API 打爆。
核心工具：asyncio.Semaphore 控制并发上限、指数退避重试、进度追踪。
class BatchDocProcessor: """批量 PDF 处理：并发 + 重试 + 进度""" def __init__( self, parser: PDFParser, max_concurrent: int = 4, max_retries: int = 3, ): self.parser = parser self.sem = asyncio.Semaphore(max_concurrent) self.max_retries = max_retries async def process_batch( self, pdf_paths: list[str], pages_per_doc: int = 3, ) -> list[ParseResult]: total = len(pdf_paths) done = 0 results: list[ParseResult | None] = [None] * total progress = ProgressBar(total, width=30) progress.render(0) async def worker(idx: int, path: str): nonlocal done async with self.sem: results[idx] = await self._retry(path, pages_per_doc) done += 1 progress.render(done) await asyncio.gather(*(worker(i, p) for i, p in enumerate(pdf_paths))) print() # 换行 return [r for r in results if r] async def _retry(self, path: str, pages: int) -> ParseResult: for attempt in range(self.max_retries): try: return await self.parser.parse(path, num_pages=pages) except Exception as e: if attempt == self.max_retries - 1: # 最后一次失败，返回空结果而不是抛异常 r = ParseResult(pdf_path=path) r.pages.append(PageResult( page_no=0, strategy_used=Strategy.HYBRID, text=f"[ERROR] {e}", confidence=0.0, )) return r # 指数退避 await asyncio.sleep(2 ** attempt + random.random()) return ParseResult(pdf_path=path)class ProgressBar: """极简进度条，不依赖 tqdm""" def __init__(self, total: int, width: int = 30): self.total = total self.width = width self.t0 = time.time() def render(self, done: int): pct = done / self.total if self.total else 1 filled = int(pct * self.width) bar = "█" * filled + "░" * (self.width - filled) elapsed = time.time() - self.t0 eta = elapsed / done * (self.total - done) if done else 0 print( f"\r[{bar}] {done}/{self.total} ({pct*100:.0f}%) " f"耗时{elapsed:.1f}s ETA{eta:.1f}s", end="", flush=True, )
端到端演示
把上面的组件串起来，跑一个完整的批量处理 demo：
async def main(): # 模拟 6 个不同类型的 PDF pdf_paths = [ "docs/report_native.pdf", # 原生数字文档，走文字路径 "docs/scan_invoice.pdf", # 扫描件，走 Vision "docs/garbled_export.pdf", # 文字层乱码，降级 Vision "docs/native_paper.pdf", "docs/scan_contract.pdf", "docs/native_ppt_export.pdf", ] parser = PDFParser(strategy=Strategy.HYBRID, mock=True) processor = BatchDocProcessor(parser, max_concurrent=3) print(f"批量处理 {len(pdf_paths)} 个 PDF...") results = await processor.process_batch(pdf_paths, pages_per_doc=2) print("\n=== 处理报告 ===") total_tokens = 0 for r in results: strategies = [p.strategy_used.value for p in r.pages] avg_conf = sum(p.confidence for p in r.pages) / max(len(r.pages), 1) print( f"{Path(r.pdf_path).name:30s} " f"策略={strategies} 置信={avg_conf:.2f} " f"耗时={r.total_latency_ms}ms tokens={r.total_cost_tokens}" ) total_tokens += r.total_cost_tokens print(f"\n总 tokens: {total_tokens}（Vision 调用产生）")if __name__ == "__main__": asyncio.run(main())
在 mock 模式下运行，输出类似：
批量处理 6 个 PDF...[██████████████████████████████] 6/6 (100%) 耗时0.3s ETA0.0s=== 处理报告 ===report_native.pdf 策略=['text', 'text'] 置信=0.95 耗时=0ms tokens=0scan_invoice.pdf 策略=['vision', 'vision'] 置信=0.92 耗时=190ms tokens=4800garbled_export.pdf 策略=['vision', 'vision'] 置信=0.92 耗时=180ms tokens=4800native_paper.pdf 策略=['text', 'text'] 置信=0.95 耗时=0ms tokens=0scan_contract.pdf 策略=['vision', 'vision'] 置信=0.92 耗时=195ms tokens=4800native_ppt_export.pdf 策略=['text', 'text'] 置信=0.95 耗时=0ms tokens=0总 tokens: 14400（Vision 调用产生）
一眼看出：3 个原生 PDF 走文字零成本、3 个扫描/乱码走 Vision 花掉 14400 tokens。如果全部一刀切走 Vision，成本会翻倍。
实际使用建议
- 默认走 HYBRID 策略
。永远不要一开始就 all-in Vision，那是把生产成本翻数倍。 - 置信度阈值不是拍脑袋定的
。跑一批真实文档，收集"文字提取成功但内容有问题"的样本，反推阈值。我的经验值是 0.65-0.75。 - 表格永远走 JSON 输出
。自由文本描述表格是灾难，无法程序化消费。 - 图表数据是估算
。别让业务方以为 Vision 能"精确读图"，一定要在 UI 上标注置信度。 - 批量处理必须做并发上限
。主流 Vision API 平台有 RPM/TPM 限制，Semaphore 值一般从 4-8 起步，压测后再调。 - 重试要指数退避
，遇到 429 立刻停一下，别死冲。 - 失败要 fail-soft
。批量任务里一个 PDF 挂掉不能拖垮整批，返回带 ERROR 标记的空结果，让下游决定如何处理。
下篇预告
下一篇进入《多模态AI工程实战》第 04 篇：多模态RAG——图文混合索引与跨模态检索。我们会拆解：
图文混合索引的两种主流范式（图像 embedding vs 图像描述文本） 跨模态检索：用文字 query 召回图像、用图像 query 召回文字 融合排序：多模态结果的重排序策略 完整可运行的多模态 RAG 骨架
多模态 RAG 是把 Vision + 检索 + 生成串起来的关键工程，也是多模态在生产中真正跑起来的形态。别错过。