PDF 数据处理实战：从原始文件到高质量 RAG 语料的完整工程指南

PDF 是企业知识库中占比最高、解析难度最大的格式。它不是文本格式，而是一种以"视觉呈现"为目标设计的排版格式——这意味着你看到的页面，和机器读到的字节流，几乎是两个完全不同的世界。

一、先搞清楚你面对的是哪种 PDF

在动手写任何解析代码之前，必须先对数据源做分类诊断。PDF 在技术实现上存在根本性差异，不同类型需要完全不同的处理路径。

快速诊断代码：

import fitz  # PyMuPDF

def diagnose_pdf(pdf_path: str) -> dict:
    doc = fitz.open(pdf_path)
    results = []

    for page_num, page in enumerate(doc):
        text = page.get_text("text")
        blocks = page.get_text("blocks")
        images = page.get_images(full=True)

        # 文本密度：有文本层但字符极少，可能是扫描件
        char_count = len(text.strip())
        image_count = len(images)
        block_count = len(blocks)

        page_type = "unknown"
        if char_count < 50 and image_count > 0:
            page_type = "scanned"
        elif char_count > 100 and block_count > 0:
            page_type = "text_based"
        elif char_count > 0 and image_count > 0:
            page_type = "mixed"

        results.append({
            "page": page_num + 1,
            "type": page_type,
            "chars": char_count,
            "images": image_count,
        })

    doc.close()
    # 以最多页面类型作为文档整体类型
    from collections import Counter
    dominant = Counter(r["type"] for r in results).most_common(1)[0][0]
    return {"dominant_type": dominant, "pages": results}

二、标准文本型 PDF：看似简单，陷阱遍地

2.1 工具选型：pdfplumber vs PyMuPDF

这是处理文本型 PDF 最常用的两个库，各有侧重：

建议策略：双库互为 fallback

import pdfplumber
import fitz

def extract_text_robust(pdf_path: str, page_num: int) -> str:
    """主用 pdfplumber，失败时 fallback 到 PyMuPDF"""
    try:
        with pdfplumber.open(pdf_path) as pdf:
            page = pdf.pages[page_num]
            text = page.extract_text(
                x_tolerance=3,      # 水平方向字符合并容差
                y_tolerance=3,      # 垂直方向行合并容差
                layout=True,        # 保留空白布局
            )
            if text and len(text.strip()) > 20:
                return text
    except Exception:
        pass

    # Fallback: PyMuPDF
    doc = fitz.open(pdf_path)
    page = doc[page_num]
    text = page.get_text("text", sort=True)  # sort=True 按阅读顺序排列
    doc.close()
    return text

2.2 文本乱序问题：最常见的隐性错误

很多 PDF 的内部文本存储顺序与视觉阅读顺序不一致——尤其是使用 InDesign、LaTeX 或特殊 PDF 打印驱动生成的文件。直接提取会得到语义混乱的文本。

解决方案：基于坐标的文本块重排

import fitz
from operator import itemgetter

def extract_ordered_text(pdf_path: str, page_num: int) -> str:
    doc = fitz.open(pdf_path)
    page = doc[page_num]

    # 获取所有文本块及其坐标 (x0, y0, x1, y1, text, block_no, block_type)
    blocks = page.get_text("blocks", sort=False)

    # 过滤图片块（block_type == 1），只保留文本块
    text_blocks = [b for b in blocks if b[6] == 0]

    # 按 y 坐标（从上到下）、再按 x 坐标（从左到右）排序
    # 容差：y 坐标差异在 5pt 以内视为同一行
    def sort_key(block):
        y_rounded = round(block[1] / 5) * 5   # y0 坐标，按 5pt 取整分组
        return (y_rounded, block[0])            # (行组, x0)

    sorted_blocks = sorted(text_blocks, key=sort_key)

    lines = [b[4].strip() for b in sorted_blocks if b[4].strip()]
    doc.close()
    return "\n".join(lines)

2.3 页眉页脚的自动识别与剥离

页眉页脚是 RAG 数据中最常见的噪声来源之一，它们会在每一页都出现，导致语义重复且干扰检索。

核心思路：利用位置频率统计

出现在固定 Y 坐标区域（顶部/底部 10% 范围）且跨页高度重复的文本块，大概率是页眉页脚。

from collections import Counter
import re

def detect_and_remove_header_footer(pdf_path: str) -> dict:
    doc = fitz.open(pdf_path)
    page_height = doc[0].rect.height

    # 收集所有页面顶部和底部区域的文本块内容
    header_candidates = Counter()
    footer_candidates = Counter()

    for page in doc:
        blocks = page.get_text("blocks")
        for b in blocks:
            if b[6] != 0:
                continue
            y0, y1 = b[1], b[3]
            text = b[4].strip()
            if not text:
                continue
            # 规范化：去掉页码数字，保留结构特征
            normalized = re.sub(r'\d+', 'N', text)
            if y0 < page_height * 0.08:       # 顶部 8%
                header_candidates[normalized] += 1
            elif y1 > page_height * 0.92:     # 底部 8%
                footer_candidates[normalized] += 1

    total_pages = len(doc)
    # 出现超过 30% 页面的文本块视为页眉/页脚
    threshold = total_pages * 0.3

    headers_to_remove = {k for k, v in header_candidates.items() if v >= threshold}
    footers_to_remove = {k for k, v in footer_candidates.items() if v >= threshold}

    doc.close()
    return {
        "headers": headers_to_remove,
        "footers": footers_to_remove,
    }

三、多栏布局：最容易被忽视的语义灾难

学术论文、期刊、产品手册普遍采用双栏或三栏布局。如果不做分栏处理，提取出来的文本会把左栏第一行和右栏第一行拼在一起——生成完全无意义的混乱文本。

3.1 基于 X 坐标聚类的分栏检测

import numpy as np
from sklearn.cluster import KMeans

def detect_columns(pdf_path: str, page_num: int) -> list:
    """返回检测到的栏位 x 范围列表，如 [(0, 280), (300, 580)]"""
    doc = fitz.open(pdf_path)
    page = doc[page_num]
    blocks = page.get_text("blocks")

    # 收集所有文本块的 x0 坐标
    x0_positions = [b[0] for b in blocks if b[6] == 0 and len(b[4].strip()) > 10]

    if len(x0_positions) < 4:
        doc.close()
        return [(0, page.rect.width)]  # 单栏

    x0_arr = np.array(x0_positions).reshape(-1, 1)

    # 尝试 K=2（双栏）和 K=3（三栏），选择轮廓系数更高的
    best_k, best_centers = 1, [0]
    for k in [2, 3]:
        if len(x0_positions) < k * 2:
            continue
        km = KMeans(n_clusters=k, random_state=42, n_init=10)
        labels = km.fit_predict(x0_arr)
        centers = sorted(km.cluster_centers_.flatten())
        # 栏间距检验：两栏中心间距需大于页宽的 20%
        page_width = page.rect.width
        if all(centers[i+1] - centers[i] > page_width * 0.2 for i in range(len(centers)-1)):
            best_k = k
            best_centers = centers

    doc.close()
    # 根据聚类中心构建栏位 x 范围
    if best_k == 1:
        return [(0, page.rect.width)]

    # 用相邻中心的中点作为分栏边界
    boundaries = [0]
    for i in range(len(best_centers) - 1):
        boundaries.append((best_centers[i] + best_centers[i+1]) / 2)
    boundaries.append(page.rect.width)

    return [(boundaries[i], boundaries[i+1]) for i in range(len(boundaries)-1)]


def extract_multicolumn_text(pdf_path: str, page_num: int) -> str:
    """按分栏顺序提取多栏文本"""
    columns = detect_columns(pdf_path, page_num)
    doc = fitz.open(pdf_path)
    page = doc[page_num]

    all_text_parts = []
    for (x_start, x_end) in columns:
        # 只提取该栏 x 范围内的文本
        clip_rect = fitz.Rect(x_start, 0, x_end, page.rect.height)
        col_text = page.get_text("text", clip=clip_rect, sort=True)
        if col_text.strip():
            all_text_parts.append(col_text.strip())

    doc.close()
    return "\n\n".join(all_text_parts)

四、表格提取：结构化数据的还原工程

表格是信息密度最高的内容形式，也是解析难度最大的。错误的表格提取不仅丢失信息，还会生成误导性的乱序文本。

4.1 工具对比与选型

4.2 pdfplumber 表格提取最佳配置

import pdfplumber
import pandas as pd

def extract_tables_from_page(pdf_path: str, page_num: int) -> list:
    """返回该页所有表格的 DataFrame 列表"""
    tables = []
    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[page_num]

        # 调整表格检测参数
        table_settings = {
            "vertical_strategy": "lines",      # 使用线条检测竖线
            "horizontal_strategy": "lines",    # 使用线条检测横线
            "explicit_vertical_lines": [],
            "explicit_horizontal_lines": [],
            "snap_tolerance": 3,               # 线条对齐容差（像素）
            "snap_x_tolerance": 3,
            "snap_y_tolerance": 3,
            "join_tolerance": 3,
            "edge_min_length": 3,
            "min_words_vertical": 3,
            "min_words_horizontal": 1,
            "intersection_tolerance": 3,
            "intersection_x_tolerance": 3,
            "intersection_y_tolerance": 3,
            "text_tolerance": 3,
            "text_x_tolerance": 3,
            "text_y_tolerance": 3,
        }

        extracted = page.extract_tables(table_settings)
        for raw_table in extracted:
            if not raw_table or len(raw_table) < 2:
                continue
            # 第一行作为表头
            df = pd.DataFrame(raw_table[1:], columns=raw_table[0])
            # 清洗：去除空列、填充 None
            df = df.dropna(how="all", axis=1)
            df = df.fillna("")
            tables.append(df)

    return tables


def table_to_rag_text(df: pd.DataFrame, context_title: str = "") -> str:
    """将表格转为适合 RAG 索引的文本表示"""
    col_names = "、".join([c for c in df.columns if c])
    row_count = len(df)

    # 自然语言摘要（作为检索入口）
    summary = f"{'【' + context_title + '】' if context_title else ''}以下表格包含 {row_count} 行数据，" \
              f"字段包括：{col_names}。"

    # Markdown 表格（保留结构语义）
    md_table = df.to_markdown(index=False)

    # 逐行自然语言化（可选，提升召回率）
    row_texts = []
    for _, row in df.iterrows():
        pairs = [f"{col}为{val}" for col, val in row.items() if val and col]
        if pairs:
            row_texts.append("、".join(pairs) + "。")

    return "\n\n".join(filter(None, [summary, md_table, "\n".join(row_texts)]))

4.3 无边框表格的处理策略

无边框表格（仅靠空白对齐的表格）是最难处理的情形。推荐切换 Camelot 的 lattice=False 模式（network 模式），利用文字间距规律识别列结构：

import camelot

def extract_borderless_tables(pdf_path: str, page_num: int) -> list:
    tables = camelot.read_pdf(
        pdf_path,
        pages=str(page_num + 1),
        flavor="stream",           # stream 模式处理无边框表格
        edge_tol=50,               # 允许的边缘容差
        row_tol=2,                 # 行间距容差
        column_tol=0,              # 列间距容差
    )
    return [t.df for t in tables if t.accuracy > 70]   # 仅保留置信度 > 70% 的结果

五、扫描型 PDF：OCR 的工程化实践

5.1 完整处理流程

5.2 图像预处理：OCR 精度的前置条件

import cv2
import numpy as np
from PIL import Image
import fitz

def pdf_page_to_image(pdf_path: str, page_num: int, dpi: int = 300) -> np.ndarray:
    """将 PDF 页面渲染为高分辨率图像"""
    doc = fitz.open(pdf_path)
    page = doc[page_num]
    # 计算缩放矩阵（默认 72 DPI，缩放到目标 DPI）
    zoom = dpi / 72
    mat = fitz.Matrix(zoom, zoom)
    pix = page.get_pixmap(matrix=mat, alpha=False)
    img_array = np.frombuffer(pix.samples, dtype=np.uint8)
    img = img_array.reshape(pix.height, pix.width, pix.n)
    doc.close()
    return cv2.cvtColor(img, cv2.COLOR_RGB2BGR)


def preprocess_for_ocr(img: np.ndarray) -> np.ndarray:
    """图像预处理流水线"""
    # 1. 转灰度
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    # 2. 自适应二值化（对光照不均的扫描件更鲁棒）
    binary = cv2.adaptiveThreshold(
        gray, 255,
        cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
        cv2.THRESH_BINARY,
        blockSize=31,    # 邻域大小，需为奇数
        C=10             # 从均值中减去的常数
    )

    # 3. 去噪：形态学开运算去除小噪点
    kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (2, 2))
    denoised = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)

    # 4. 倾斜校正
    coords = np.column_stack(np.where(denoised < 128))
    if len(coords) > 100:
        angle = cv2.minAreaRect(coords)[-1]
        if angle < -45:
            angle = 90 + angle
        if abs(angle) > 0.5:   # 只在倾斜明显时才校正
            (h, w) = denoised.shape
            center = (w // 2, h // 2)
            M = cv2.getRotationMatrix2D(center, angle, 1.0)
            denoised = cv2.warpAffine(
                denoised, M, (w, h),
                flags=cv2.INTER_CUBIC,
                borderMode=cv2.BORDER_REPLICATE
            )

    return denoised

5.3 PaddleOCR 工程化调用

from paddleocr import PaddleOCR, PPStructure

# 初始化（建议全局单例，避免重复加载模型）
ocr_engine = PaddleOCR(
    use_angle_cls=True,     # 启用文本方向分类（处理旋转文本）
    lang="ch",              # 中英混合
    use_gpu=True,
    enable_mkldnn=True,     # CPU 加速
    det_db_thresh=0.3,
    det_db_box_thresh=0.5,
    rec_batch_num=6,
    show_log=False,
)

def ocr_with_confidence_filter(img: np.ndarray, 
                                confidence_threshold: float = 0.8) -> list:
    """
    返回 OCR 结果列表，每项包含文本、置信度和坐标
    过滤低置信度结果并标记
    """
    result = ocr_engine.ocr(img, cls=True)
    if not result or not result[0]:
        return []

    filtered = []
    low_confidence_count = 0

    for line in result[0]:
        box, (text, confidence) = line
        if confidence >= confidence_threshold:
            filtered.append({
                "text": text,
                "confidence": confidence,
                "box": box,           # 四个顶点坐标
                "y_center": (box[0][1] + box[2][1]) / 2,
                "x_center": (box[0][0] + box[2][0]) / 2,
            })
        else:
            low_confidence_count += 1

    # 按阅读顺序（从上到下，从左到右）排序
    filtered.sort(key=lambda x: (round(x["y_center"] / 10) * 10, x["x_center"]))

    if low_confidence_count > len(filtered) * 0.3:
        # 低置信度比例过高，整页标记为低质量
        for item in filtered:
            item["page_quality"] = "low"

    return filtered

六、特殊内容处理：公式、图注与附注

6.1 数学公式的处理策略

PDF 中的数学公式有两种存在形态：

字符编码型：LaTeX 生成的 PDF，公式以特殊字体字符存在，可尝试提取但可读性差 图像型：Word 或其他工具生成，公式以嵌入图像存在，需多模态处理

def handle_formula_regions(page, formula_boxes: list) -> list:
    """
    formula_boxes: 通过版面分析识别出的公式区域 [(x0,y0,x1,y1), ...]
    策略：提取为图像，生成占位符描述，后续可对接公式识别服务
    """
    results = []
    for i, box in enumerate(formula_boxes):
        clip = fitz.Rect(*box)
        pix = page.get_pixmap(clip=clip, matrix=fitz.Matrix(2, 2))
        img_bytes = pix.tobytes("png")

        # 方案A：保留图像，生成占位符（简单）
        placeholder = f"[FORMULA_{i+1}: 数学公式，见原文第{box}区域]"

        # 方案B：调用 pix2tex 或 UniMER 进行公式识别（精确）
        # latex_str = formula_recognizer.predict(img_bytes)
        # placeholder = f"$${latex_str}$$"

        results.append({"type": "formula", "text": placeholder, "box": box})
    return results

6.2 图片说明文字的关联提取

图注（Caption）是重要的语义信息，但很容易被当作普通段落处理而失去其与图片的关联关系。

def extract_figure_captions(page_blocks: list) -> list:
    """
    识别图注并与其所属图片区域关联
    图注特征：通常以 "图X" "Figure X" "Fig." 开头，位于图片正下方
    """
    import re
    caption_pattern = re.compile(
        r'^(图\s*\d+|Figure\s*\d+|Fig\.\s*\d+|表\s*\d+|Table\s*\d+)',
        re.IGNORECASE
    )
    captions = []
    for block in page_blocks:
        text = block.get("text", "").strip()
        if caption_pattern.match(text):
            captions.append({
                "caption": text,
                "y_pos": block.get("y0"),
                "type": "figure_caption"
            })
    return captions

七、后处理：从提取文本到可用语料

经过上述各类提取后，还需要统一的后处理步骤，将不同来源的文本规范化为一致格式。

7.1 文本拼接中的段落重建

PDF 提取出来的文本常常存在错误的换行（因为 PDF 的行是视觉行，不是语义段落）。

import re

def reconstruct_paragraphs(raw_text: str) -> str:
    """
    将 PDF 提取的视觉换行还原为语义段落
    核心规则：
    - 行末有句号/问号等，保留换行（真正的段落边界）
    - 行末无标点且下一行首字母小写/不是数字序号，合并为同一段落
    """
    lines = raw_text.split('\n')
    result = []
    buffer = ""

    sentence_enders = re.compile(r'[。！？.!?：:…]$')
    list_marker = re.compile(r'^[\d一二三四五六七八九十]+[.、）)]\s')
    heading_like = re.compile(r'^[#（(【\[第]|^\d+\s')

    for i, line in enumerate(lines):
        line = line.strip()
        if not line:
            if buffer:
                result.append(buffer)
                buffer = ""
            continue

        # 标题行、列表项：单独成段
        if heading_like.match(line) or list_marker.match(line):
            if buffer:
                result.append(buffer)
                buffer = ""
            result.append(line)
            continue

        if buffer:
            # 上一行有句子结束符：另起段落
            if sentence_enders.search(buffer):
                result.append(buffer)
                buffer = line
            else:
                # 合并：补一个空格（中文不需要，英文需要）
                if buffer and buffer[-1].isascii() and line and line[0].isascii():
                    buffer += " " + line
                else:
                    buffer += line
        else:
            buffer = line

    if buffer:
        result.append(buffer)

    return "\n\n".join(result)

7.2 统一字符清洗

import unicodedata

def clean_pdf_text(text: str) -> str:
    # Unicode 规范化：处理连字（fi、fl 等）、全角字符
    text = unicodedata.normalize("NFKC", text)

    # 清除 PDF 特有的乱码字符（私有区域字符）
    text = re.sub(r'[\uf000-\uf8ff]', '', text)

    # 清除零宽字符
    text = re.sub(r'[\u200b\u200c\u200d\ufeff\u00ad]', '', text)

    # 修复常见 OCR/PDF 错误替换
    ocr_corrections = {
        '０': '0', '１': '1', '２': '2', '３': '3', '４': '4',
        '５': '5', '６': '6', '７': '7', '８': '8', '９': '9',
        '\x0c': '\n',    # Form Feed 转换为换行
        '\xa0': ' ',     # Non-breaking space 转换为普通空格
    }
    for wrong, correct in ocr_corrections.items():
        text = text.replace(wrong, correct)

    # 中英文之间插入空格（改善分词效果）
    text = re.sub(r'([\u4e00-\u9fa5])([A-Za-z0-9])', r'\1 \2', text)
    text = re.sub(r'([A-Za-z0-9])([\u4e00-\u9fa5])', r'\1 \2', text)

    # 合并多余空白
    text = re.sub(r'[ \t]{2,}', ' ', text)
    text = re.sub(r'\n{3,}', '\n\n', text)

    return text.strip()

八、质量评估与异常告警

对每一页、每一个处理结果都应当自动评分，异常页面记录日志供人工复查。

from dataclasses import dataclass
from typing import Optional

@dataclass
class PageQualityReport:
    page_num: int
    char_count: int
    has_tables: bool
    ocr_avg_confidence: Optional[float]
    paragraph_count: int
    quality_score: float
    issues: list

def assess_page_quality(page_num: int, extracted: dict) -> PageQualityReport:
    issues = []
    text = extracted.get("text", "")
    char_count = len(text)

    # 字符数过少
    if char_count < 30:
        issues.append("char_count_too_low")

    # 乱码检测：非常用字符比例
    unusual = len(re.findall(r'[^\u4e00-\u9fa5\u0020-\u007e\n]', text))
    unusual_ratio = unusual / max(char_count, 1)
    if unusual_ratio > 0.05:
        issues.append(f"high_unusual_char_ratio:{unusual_ratio:.2f}")

    # 重复行检测
    lines = [l.strip() for l in text.split('\n') if l.strip()]
    if lines:
        unique_ratio = len(set(lines)) / len(lines)
        if unique_ratio < 0.7:
            issues.append(f"high_duplicate_lines:{1-unique_ratio:.2f}")

    # OCR 置信度
    ocr_conf = extracted.get("avg_confidence")
    if ocr_conf is not None and ocr_conf < 0.75:
        issues.append(f"low_ocr_confidence:{ocr_conf:.2f}")

    # 综合评分
    score = 1.0
    score -= len(issues) * 0.15
    score -= unusual_ratio * 2
    if ocr_conf:
        score *= ocr_conf
    score = max(0.0, min(1.0, score))

    return PageQualityReport(
        page_num=page_num,
        char_count=char_count,
        has_tables=len(extracted.get("tables", [])) > 0,
        ocr_avg_confidence=ocr_conf,
        paragraph_count=len(lines),
        quality_score=round(score, 3),
        issues=issues,
    )

九、整体工程架构：可生产部署的 PDF 处理管道

标准化输出格式

每个 PDF 页面处理完成后，输出为统一的 JSON 结构：

{
  "source_file": "产品手册_v2.pdf",
  "page_num": 5,
  "page_type": "text_based",
  "quality_score": 0.91,
  "paragraphs": [
    {
      "text": "第三章 系统配置与初始化...",
      "type": "heading",
      "level": 2
    },
    {
      "text": "在首次启动系统之前，需要完成以下配置步骤...",
      "type": "paragraph"
    }
  ],
  "tables": [
    {
      "caption": "表3-1 配置参数说明",
      "markdown": "| 参数名 | 类型 | 默认值 | 说明 |\n|...",
      "row_count": 8
    }
  ],
  "metadata": {
    "section_title": "第三章 系统配置",
    "has_formulas": false,
    "has_images": true,
    "language": "zh"
  }
}

十、总结：PDF 处理的工程原则

处理 PDF 的核心挑战在于：它是为人眼设计的，而不是为程序设计的。没有一个工具能处理所有情形，工程上需要的是一套有明确决策树的复合策略。

几条核心原则值得反复强调：

1. 先分类，再处理：在动任何代码之前，诊断 PDF 类型，拒绝"一刀切"

1. 坐标是第一公民：PDF 的布局信息藏在坐标里，要善用坐标而不是依赖字符流顺序

1. 表格单独对待：表格永远不能和正文用同一套逻辑处理，它需要独立的提取和表示策略

1. 质量要有数字：每一页都应当有质量评分，低于阈值的进人工复查，而不是静默地把错误输入向量库

1. 为下游着想：所有输出都应当面向"如何让 embedding 模型理解"来设计，而不仅仅是"把文字提取出来"

PDF 处理做好了，RAG 系统的召回质量通常能提升 30% 以上——这比任何模型升级都来得更快、更经济。

所有代码片段为工程示意，生产环境需根据实际 PDF 特征进行参数调优和异常处理增强。