RAG 文件解析：PDF / Word / Excel / HTML 全格式文本提取

本文是【AI 专题精讲】系列第 01 篇。 下一篇：RAG 文档切片策略：固定长度 vs 递归 vs 语义切分

---

这篇文章你会得到什么

你想搭一个知识库，让 AI 能基于你的文档来回答问题（也就是 RAG）。第一步是什么？

不是选向量数据库，不是调 Embedding 模型——第一步是把文件变成纯文本。

听起来简单，但实际做的时候坑多到超出想象：PDF 里的表格解析出来全是乱的、Word 的嵌套样式丢失关键信息、Excel 多 sheet 合并单元格一团糟、HTML 满屏广告和导航栏混在正文里……

今天的目标：用 Python 实现一个统一的 FileParser 类，传入任意 PDF / Word / Excel / HTML 文件，输出结构化的纯文本。这个模块会是后续所有 RAG 文章的基础。

---

为什么文件解析是 RAG 的第一关

RAG 的完整链路是这样的：

文件 → 解析成文本 → 切片 → Embedding → 存入向量库 → 检索 → 喂给 LLM → 生成回答

解析质量决定了后面每一步的上限。如果这一步就丢了关键内容、格式全乱了，后面 Embedding 再好、模型再强也没用——garbage in, garbage out。

实际项目里，用户上传的文件五花八门：

格式	典型场景	解析难度
PDF	技术文档、合同、论文	★★★★★（最难）
Word (.docx)	内部文档、SOP、方案	★★★
Excel (.xlsx)	数据报表、配置表	★★★
HTML	网页内容、帮助文档	★★

我在自己的项目里踩过的坑：一开始图省事用了一个”万能”解析库，结果 PDF 里的表格全丢了，用户问”Q3 营收多少”，AI 回答”文档中未找到相关信息”。后来才意识到——每种格式都需要专门的解析策略。

---

统一输出格式设计

在开始解析之前，先定义统一的输出结构。不管什么格式进来，出来的数据结构必须一致，这样后续切片和入库才不用做格式适配。

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ParsedDocument:
    filename: str
    content: str
    file_type: str
    metadata: dict = field(default_factory=dict)
    pages: list[str] = field(default_factory=list)

    @property
    def total_chars(self) -> int:
        return len(self.content)

    @property
    def page_count(self) -> int:
        return len(self.pages)

关键字段：

字段	说明
filename	原始文件名
content	合并后的全文纯文本
file_type	pdf / docx / xlsx / html
metadata	元数据（作者、创建时间、页数等）
pages	按页/sheet 分隔的文本列表

pages 字段很重要——后续切片时可以用页码信息做定位，回答时能告诉用户”来源：第 3 页”。

---

PDF 解析：最难啃的骨头

PDF 是 RAG 场景中最常见也最难解析的格式。难在哪？PDF 本质上是一个排版描述文件，它存储的是”在坐标 (x, y) 画一个字符 ‘A‘“这种信息，不是结构化的段落和表格。

工具选型

库	特点	适合场景
PyMuPDF (fitz)	速度快，支持图片提取，轻量	通用文档，首选
pdfplumber	表格解析能力强	含大量表格的文档
PyPDF2	纯 Python，无 C 依赖	简单文本提取
unstructured	一站式方案，支持多格式	不想自己组合的场景

我的推荐：PyMuPDF 作为主力，pdfplumber 处理表格。

安装依赖

pip install PyMuPDF pdfplumber

基础文本提取

import fitz  # PyMuPDF

def parse_pdf_basic(filepath: str) -> ParsedDocument:
    doc = fitz.open(filepath)
    pages = []
    metadata = {
        "author": doc.metadata.get("author", ""),
        "title": doc.metadata.get("title", ""),
        "page_count": len(doc),
    }

    for page in doc:
        text = page.get_text("text")
        if text.strip():
            pages.append(text.strip())

    doc.close()

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content="\n\n".join(pages),
        file_type="pdf",
        metadata=metadata,
        pages=pages,
    )

page.get_text("text") 是最基础的提取方式，按阅读顺序输出纯文本。对于纯文字的 PDF，这就够了。

表格处理：pdfplumber 上场

纯文字提取碰到表格会变成一堆错位的文字。这时候需要 pdfplumber 的表格识别能力：

import pdfplumber

def extract_tables_from_pdf(filepath: str) -> list[str]:
    """提取 PDF 中所有表格，转成 Markdown 格式"""
    tables_text = []

    with pdfplumber.open(filepath) as pdf:
        for i, page in enumerate(pdf.pages):
            tables = page.extract_tables()
            for table in tables:
                if not table:
                    continue
                md = table_to_markdown(table)
                tables_text.append(f"[表格 - 第{i+1}页]\n{md}")

    return tables_text


def table_to_markdown(table: list[list]) -> str:
    """将二维数组转成 Markdown 表格"""
    if not table or not table[0]:
        return ""

    # 清理 None 值
    cleaned = [[str(cell) if cell else "" for cell in row] for row in table]

    header = "| " + " | ".join(cleaned[0]) + " |"
    separator = "| " + " | ".join(["---"] * len(cleaned[0])) + " |"
    rows = [
        "| " + " | ".join(row) + " |"
        for row in cleaned[1:]
    ]

    return "\n".join([header, separator] + rows)

把表格转成 Markdown 格式是个小技巧——LLM 对 Markdown 表格的理解能力远强于纯文本拼接。

完整 PDF 解析器

把文本和表格合在一起：

def parse_pdf(filepath: str) -> ParsedDocument:
    """完整 PDF 解析：文本 + 表格"""
    doc = fitz.open(filepath)
    pages = []
    metadata = {
        "author": doc.metadata.get("author", ""),
        "title": doc.metadata.get("title", ""),
        "page_count": len(doc),
    }

    for page in doc:
        text = page.get_text("text").strip()
        if text:
            pages.append(text)

    doc.close()

    # 提取表格并追加
    tables = extract_tables_from_pdf(filepath)
    if tables:
        pages.append("\n\n".join(tables))

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content="\n\n".join(pages),
        file_type="pdf",
        metadata=metadata,
        pages=pages,
    )

扫描件 PDF 怎么办

如果 PDF 是扫描件（图片形式），get_text() 会返回空字符串。这时候需要 OCR：

def parse_scanned_pdf(filepath: str) -> ParsedDocument:
    """扫描件 PDF：转图片后 OCR"""
    doc = fitz.open(filepath)
    pages = []

    for page in doc:
        # 将页面渲染成图片
        pix = page.get_pixmap(dpi=200)
        img_bytes = pix.tobytes("png")

        # 调用 OCR（这里用 Tesseract 示意）
        from PIL import Image
        import pytesseract
        import io

        image = Image.open(io.BytesIO(img_bytes))
        text = pytesseract.image_to_string(image, lang="chi_sim+eng")
        if text.strip():
            pages.append(text.strip())

    doc.close()

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content="\n\n".join(pages),
        file_type="pdf",
        metadata={"ocr": True, "page_count": len(pages)},
        pages=pages,
    )

实际项目中，可以先尝试 get_text()，如果结果为空再走 OCR 分支，避免不必要的 OCR 开销。

---

Word 解析：比想象中简单

.docx 文件底层是 XML，结构比 PDF 清晰得多。用 python-docx 可以很方便地提取。

安装

pip install python-docx

实现

from docx import Document

def parse_docx(filepath: str) -> ParsedDocument:
    doc = Document(filepath)

    metadata = {
        "author": doc.core_properties.author or "",
        "title": doc.core_properties.title or "",
    }

    paragraphs = []
    for para in doc.paragraphs:
        text = para.text.strip()
        if not text:
            continue

        # 保留标题层级信息
        if para.style.name.startswith("Heading"):
            level = para.style.name.replace("Heading ", "")
            try:
                prefix = "#" * int(level)
                text = f"{prefix} {text}"
            except ValueError:
                pass

        paragraphs.append(text)

    # 提取表格
    for table in doc.tables:
        rows = []
        for row in table.rows:
            cells = [cell.text.strip() for cell in row.cells]
            rows.append(cells)

        if rows:
            md = table_to_markdown(rows)
            paragraphs.append(md)

    content = "\n\n".join(paragraphs)

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content=content,
        file_type="docx",
        metadata=metadata,
        pages=[content],
    )

几个要点：

• 标题样式转 Markdown：Word 的 Heading 1 / Heading 2 转成 # / ##，保留文档结构，对后续 LLM 理解非常有帮助。
• 表格提取：python-docx 的表格 API 比 PDF 好用太多，直接遍历 table.rows 和 row.cells。
• 页码问题：Word 没有物理页的概念（页码是渲染出来的），所以 pages 只放整篇内容。

---

Excel 解析：结构化数据的特殊处理

Excel 和前面的文档不一样——它本身就是结构化数据，不需要”提取文本”，而是需要把表格数据转成 LLM 能理解的文本格式。

安装

pip install openpyxl

实现

from openpyxl import load_workbook

def parse_xlsx(filepath: str) -> ParsedDocument:
    wb = load_workbook(filepath, read_only=True, data_only=True)
    pages = []

    metadata = {
        "sheet_names": wb.sheetnames,
        "sheet_count": len(wb.sheetnames),
    }

    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
        rows = []

        for row in ws.iter_rows(values_only=True):
            # 跳过全空行
            if all(cell is None for cell in row):
                continue
            cells = [str(cell) if cell is not None else "" for cell in row]
            rows.append(cells)

        if not rows:
            continue

        # 转成 Markdown 表格
        md = table_to_markdown(rows)
        sheet_text = f"## Sheet: {sheet_name}\n\n{md}"
        pages.append(sheet_text)

    wb.close()

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content="\n\n".join(pages),
        file_type="xlsx",
        metadata=metadata,
        pages=pages,
    )

合并单元格处理

合并单元格是 Excel 解析的常见坑——合并区域只有左上角的单元格有值，其余是 None：

def parse_xlsx_with_merged(filepath: str) -> ParsedDocument:
    wb = load_workbook(filepath, data_only=True)
    pages = []

    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]

        # 构建合并单元格映射：被合并的单元格 → 左上角的值
        merged_values = {}
        for merged_range in ws.merged_cells.ranges:
            top_left_value = ws.cell(
                merged_range.min_row, merged_range.min_col
            ).value
            for row in range(merged_range.min_row, merged_range.max_row + 1):
                for col in range(merged_range.min_col, merged_range.max_col + 1):
                    merged_values[(row, col)] = top_left_value

        rows = []
        for row_idx, row in enumerate(ws.iter_rows(), start=1):
            cells = []
            for col_idx, cell in enumerate(row, start=1):
                value = cell.value
                if value is None:
                    value = merged_values.get((row_idx, col_idx), "")
                cells.append(str(value) if value else "")
            if any(c for c in cells):
                rows.append(cells)

        if rows:
            md = table_to_markdown(rows)
            pages.append(f"## Sheet: {sheet_name}\n\n{md}")

    wb.close()

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content="\n\n".join(pages),
        file_type="xlsx",
        metadata={"sheet_names": wb.sheetnames},
        pages=pages,
    )

Excel 解析的经验

• 大文件用 read_only=True：openpyxl 默认会把整个文件加载到内存，大文件（10 万行以上）必须开 read_only 模式。
• data_only=True：读取公式计算后的值，而不是公式本身。不然你会得到 =SUM(A1:A10) 这种东西。
• 每个 Sheet 独立输出：用 ## Sheet: xxx 标记，方便后续切片时按 Sheet 分割。

---

HTML 解析：去噪是关键

HTML 解析的难点不在于提取文本——而在于去掉噪声。一个网页里真正有价值的”正文”可能只占 HTML 的 20%，其余全是导航、广告、侧边栏、页脚。

安装

pip install beautifulsoup4 readability-lxml lxml

方案一：BeautifulSoup 手动清理

from bs4 import BeautifulSoup

def parse_html_bs4(filepath: str) -> ParsedDocument:
    with open(filepath, "r", encoding="utf-8") as f:
        html = f.read()

    soup = BeautifulSoup(html, "lxml")

    # 移除无用标签
    for tag in soup.find_all(["script", "style", "nav", "footer", "header", "aside"]):
        tag.decompose()

    # 提取标题
    title = soup.title.string if soup.title else ""

    # 提取正文
    text = soup.get_text(separator="\n", strip=True)

    # 清理多余空行
    lines = [line for line in text.split("\n") if line.strip()]
    content = "\n".join(lines)

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content=content,
        file_type="html",
        metadata={"title": title},
        pages=[content],
    )

方案二：readability 自动提取正文（推荐）

readability-lxml 是 Firefox 阅读模式用的算法的 Python 实现，自动识别正文区域：

from readability import Document as ReadabilityDocument

def parse_html_readability(filepath: str) -> ParsedDocument:
    with open(filepath, "r", encoding="utf-8") as f:
        html = f.read()

    doc = ReadabilityDocument(html)
    title = doc.title()
    summary_html = doc.summary()

    # 再用 BeautifulSoup 把提取出的正文 HTML 转成纯文本
    soup = BeautifulSoup(summary_html, "lxml")
    content = soup.get_text(separator="\n", strip=True)

    lines = [line for line in content.split("\n") if line.strip()]
    content = "\n".join(lines)

    return ParsedDocument(
        filename=filepath.split("/")[-1],
        content=content,
        file_type="html",
        metadata={"title": title},
        pages=[content],
    )

readability 的效果比手动清理好很多，它能准确定位文章主体，自动过滤导航、广告等。

在线网页解析

如果你的 RAG 需要解析在线网页而不是本地文件，加个 HTTP 请求就行：

import httpx

def parse_url(url: str) -> ParsedDocument:
    response = httpx.get(url, timeout=15, follow_redirects=True)
    response.raise_for_status()

    doc = ReadabilityDocument(response.text)
    title = doc.title()
    soup = BeautifulSoup(doc.summary(), "lxml")
    content = soup.get_text(separator="\n", strip=True)

    lines = [line for line in content.split("\n") if line.strip()]
    content = "\n".join(lines)

    return ParsedDocument(
        filename=url,
        content=content,
        file_type="html",
        metadata={"title": title, "url": url},
        pages=[content],
    )

---

统一入口：FileParser 类

现在把四种格式的解析器统一到一个类里：

import os

class FileParser:
    """统一文件解析器，支持 PDF / Word / Excel / HTML"""

    SUPPORTED_TYPES = {
        ".pdf": "pdf",
        ".docx": "docx",
        ".xlsx": "xlsx",
        ".html": "html",
        ".htm": "html",
    }

    def parse(self, filepath: str) -> ParsedDocument:
        ext = os.path.splitext(filepath)[1].lower()
        file_type = self.SUPPORTED_TYPES.get(ext)

        if not file_type:
            raise ValueError(
                f"不支持的文件格式: {ext}，"
                f"支持: {', '.join(self.SUPPORTED_TYPES.keys())}"
            )

        parser_map = {
            "pdf": parse_pdf,
            "docx": parse_docx,
            "xlsx": parse_xlsx,
            "html": parse_html_readability,
        }

        result = parser_map[file_type](filepath)
        return self._post_process(result)

    def _post_process(self, doc: ParsedDocument) -> ParsedDocument:
        """统一后处理：清理多余空白、标准化换行"""
        import re

        content = doc.content
        content = re.sub(r"\n{3,}", "\n\n", content)  # 最多两个连续换行
        content = re.sub(r"[ \t]+", " ", content)       # 合并多余空格
        content = content.strip()

        doc.content = content
        doc.pages = [
            re.sub(r"\n{3,}", "\n\n", p).strip()
            for p in doc.pages
            if p.strip()
        ]

        return doc

    def parse_batch(self, filepaths: list[str]) -> list[ParsedDocument]:
        """批量解析"""
        results = []
        for fp in filepaths:
            try:
                results.append(self.parse(fp))
            except Exception as e:
                print(f"解析失败 [{fp}]: {e}")
        return results

使用起来非常简单：

parser = FileParser()

# 单个文件
doc = parser.parse("./docs/技术方案.pdf")
print(f"文件: {doc.filename}")
print(f"类型: {doc.file_type}")
print(f"字数: {doc.total_chars}")
print(f"页数: {doc.page_count}")
print(f"内容预览: {doc.content[:200]}...")

# 批量解析
docs = parser.parse_batch([
    "./docs/技术方案.pdf",
    "./docs/需求文档.docx",
    "./docs/数据报表.xlsx",
    "./docs/帮助中心.html",
])
print(f"成功解析 {len(docs)} 个文件")

---

和 FastAPI 集成：文件上传解析接口

实际项目中，文件解析通常做成一个 API 接口。用 FastAPI 包一层：

from fastapi import FastAPI, UploadFile, HTTPException
import tempfile
import os

app = FastAPI()
parser = FileParser()

@app.post("/api/parse")
async def parse_file(file: UploadFile):
    ext = os.path.splitext(file.filename)[1].lower()

    if ext not in FileParser.SUPPORTED_TYPES:
        raise HTTPException(
            status_code=400,
            detail=f"不支持的文件格式: {ext}",
        )

    # 保存到临时文件
    with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name

    try:
        doc = parser.parse(tmp_path)
        return {
            "filename": file.filename,
            "file_type": doc.file_type,
            "total_chars": doc.total_chars,
            "page_count": doc.page_count,
            "content": doc.content,
            "metadata": doc.metadata,
        }
    finally:
        os.unlink(tmp_path)

前端调用：

async function parseFile(file) {
  const formData = new FormData();
  formData.append('file', file);

  const response = await fetch('/api/parse', {
    method: 'POST',
    body: formData,
  });

  if (!response.ok) {
    const error = await response.json();
    throw new Error(error.detail);
  }

  return response.json();
}

// 使用
const input = document.querySelector('input[type="file"]');
input.addEventListener('change', async (e) => {
  const file = e.target.files[0];
  const result = await parseFile(file);
  console.log(`解析完成：${result.total_chars} 字，${result.page_count} 页`);
});

---

生产环境的注意事项

1. 文件大小限制

大文件会吃内存，需要设上限：

MAX_FILE_SIZE = 50 * 1024 * 1024  # 50MB

@app.post("/api/parse")
async def parse_file(file: UploadFile):
    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(413, "文件大小超过 50MB 限制")
    # ...

2. 超时控制

某些 PDF 解析可能卡住（比如超大扫描件做 OCR），需要加超时：

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

@app.post("/api/parse")
async def parse_file(file: UploadFile):
    # ...
    try:
        doc = await asyncio.wait_for(
            asyncio.get_event_loop().run_in_executor(
                executor, parser.parse, tmp_path
            ),
            timeout=60,  # 60 秒超时
        )
    except asyncio.TimeoutError:
        raise HTTPException(408, "文件解析超时，请尝试较小的文件")

3. 编码检测

HTML 文件的编码不一定是 UTF-8，需要检测：

import chardet

def detect_encoding(filepath: str) -> str:
    with open(filepath, "rb") as f:
        raw = f.read(10000)
    result = chardet.detect(raw)
    return result["encoding"] or "utf-8"

4. 安全检查

用户上传的文件可能包含恶意内容（比如 PDF 里的 JavaScript）。生产环境要：

• 限制文件扩展名白名单
• 不在服务器上执行任何文件内容（只读取）
• 使用 tempfile 并及时清理
• 考虑用沙箱环境运行解析

---

完整依赖清单

# requirements.txt
PyMuPDF>=1.24.0
pdfplumber>=0.11.0
python-docx>=1.1.0
openpyxl>=3.1.0
beautifulsoup4>=4.12.0
readability-lxml>=0.8.0
lxml>=5.0.0
chardet>=5.0.0
httpx>=0.27.0
fastapi>=0.115.0
uvicorn>=0.30.0

---

总结

1. 文件解析是 RAG 的第一关——解析质量决定了检索和回答的上限。
2. 每种格式都有专门的坑：PDF 最难（表格、扫描件），Word 和 Excel 相对好处理，HTML 重在去噪。
3. 统一输出格式（ParsedDocument）是关键设计——后续切片、入库、检索都基于这个结构。
4. 工具选择：PDF 用 PyMuPDF + pdfplumber，Word 用 python-docx，Excel 用 openpyxl，HTML 用 readability-lxml。
5. 表格转 Markdown：LLM 对 Markdown 表格的理解能力远强于纯文本拼接。
6. 生产环境要注意：文件大小限制、解析超时、编码检测、安全检查。

下一篇我们解决解析之后的下一个问题：拿到纯文本后怎么切片？ 切片策略直接决定了检索的准确率——切错了，AI 回答就跑偏。

---

下一篇预告：02 | RAG 文档切片策略：固定长度 vs 递归 vs 语义切分

---

讨论话题：你在做 RAG 的时候，遇到过什么奇葩的文件解析问题？PDF 里的表格你是怎么处理的？评论区聊聊。