三亩地 三亩地SAN MU DI · CODE DIARY
交流投稿
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天，欢迎挑你感兴趣的翻一翻。

首页/学习日记/日记详情

RAG系统构建：多格式文档加载与文本预处理实战指南

发布时间：2026/8/8 4:45:15 三亩地 编程学习日记
RAG系统构建：多格式文档加载与文本预处理实战指南

1. 项目概述：为什么RAG的第一步如此关键？

如果你正在构建一个基于大语言模型（LLM）的问答系统、智能客服或者企业知识库，那么“RAG”这个词对你来说一定不陌生。RAG，即检索增强生成，它通过从外部知识库中检索相关信息来“喂”给大模型，从而生成更准确、更可靠的回答，有效缓解了大模型的“幻觉”问题。听起来很美好，对吧？但很多朋友在兴致勃勃地开始搭建RAG系统时，往往一头扎进向量数据库选型、检索算法优化这些“高大上”的环节，却忽略了最基础、也最容易出问题的一步：多格式文档的加载与文本预处理。

我见过太多项目，因为前期文档处理不当，导致后续的检索召回率极低，生成的答案牛头不对马嘴，整个系统效果大打折扣。这就像盖房子，地基没打牢，上面的装修再豪华也是白搭。今天，我们就来彻底拆解RAG实战的第一步，聚焦于如何将五花八门的文档——PDF、Word、PPT、Excel、TXT、Markdown，甚至网页——可靠地转换成干净、结构化的纯文本，为后续的向量化和检索打下坚实基础。这个过程，直接决定了你的知识库“原料”质量，是RAG项目成败的生命线。

2. 核心思路与工具选型：构建稳健的文档处理流水线

一个稳健的文档处理流水线，其核心目标就两个：兼容性和保真度。兼容性是指能处理尽可能多的文档格式；保真度是指转换后的文本要尽可能保留原文的结构、语义和关键信息（如表格、列表）。基于这个思路，我们的技术选型需要分层考虑。

2.1 文档加载层：按格式分而治之

没有一种工具能完美处理所有格式。因此，我们需要一个“工具箱”策略，针对不同格式选用最合适的解析器。

  1. PDF文档：这是最复杂也最常见的格式。我推荐使用PyMuPDF和pdfplumber的组合。

    • PyMuPDF：提取速度和文本保真度极高，能很好地处理由文字构成的PDF。但对于扫描件或图片型PDF，它无能为力。
    • pdfplumber：它的强项在于精确提取表格数据和维持文本的物理布局（如多栏排版）。对于包含复杂表格的PDF，它是首选。
    • 实战心得：对于重要文档，我通常会先用PyMuPDF提取一遍，再用pdfplumber专门提取表格，然后将结果融合。虽然pypdf和pdf2image+OCR也是选项，但前者功能较弱，后者（如pytesseract）处理速度慢、依赖环境复杂，非扫描件场景下不推荐作为首选。

  2. Office文档：

    • Word (.docx)：python-docx库是绝对的主流。它能完美解析段落、标题、列表、表格甚至图片标注。对于老旧的.doc格式，可以先用libreoffice命令行工具将其转换为.docx再处理。
    • Excel (.xlsx)：pandas是不二之选。关键在于如何处理多个sheet。我的做法是，将每个sheet转换成一个格式化的文本块，例如用|分隔符模拟表格，并注明sheet名，这样能最大程度保留表格的二维信息。
    • PowerPoint (.pptx)：python-pptx可以提取每页幻灯片的文本框内容。需要注意的是，PPT内容通常是碎片化的，需要将每页的内容合理拼接，并保留幻灯片标题作为上下文。

  3. 纯文本与标记语言：

    • TXT/CSV：直接用Python内置的open函数读取即可，注意编码问题（优先使用utf-8）。
    • Markdown/HTML：BeautifulSoup4可以处理HTML，提取正文并去除标签。对于Markdown，markdown库可以将其转换为HTML再用BeautifulSoup处理，或者直接用正则表达式提取关键部分。目标是保留标题层级和列表结构。

  4. 网页内容：除了BeautifulSoup4，newspaper3k是一个更高级的选择，它能自动识别文章主体内容，过滤导航栏、广告等噪音，提取效果非常好。

注意：在实际项目中，手动为每种格式写加载器非常繁琐。我强烈建议使用像LangChain或LlamaIndex这样的框架。它们提供了统一的DocumentLoader接口，背后集成了上述各种解析器。例如，LangChain的PyMuPDFLoader、UnstructuredWordDocumentLoader、UnstructuredExcelLoader等，能极大提升开发效率。但了解其底层原理，对于排查解析错误至关重要。

2.2 文本预处理层：从原始文本到干净语料

加载得到原始文本后，里面充满了对LLM无用的“噪音”，预处理的目的就是去除噪音，保留精华。

  1. 清洗：

    • 无用字符：去除不可见字符（如\x00）、乱码、过多的空白符（将连续的换行、空格标准化）。
    • 页眉页脚/页码：这是PDF处理中的老大难问题。简单的正则表达式（如匹配“第X页”或页眉特定文字）可以解决一部分，但对于复杂的版面，可能需要结合文本位置信息（PyMuPDF可以提供坐标）进行过滤。
    • 冗余信息：去除文档中的版权声明、网址链接（有时需要保留）、电子邮件等模板化内容。

  2. 标准化：

    • 编码统一：确保所有文本都是UTF-8编码。
    • 全半角转换：将全角字符（中文标点、字母、数字）转换为半角，保持一致性。
    • 日期/数字格式：将“2023年12月1日”统一为“2023-12-01”，有助于后续的实体识别和检索。

  3. 结构化信息提取与保留：

    • 标题与章节：在解析时，就应识别并标记H1, H2, H3等标题。这不仅是文本结构，在后续的“文本分割”步骤中，标题是重要的分割边界，能防止一个章节被生生切断。
    • 列表与表格：将列表项用“-”或数字明确标出。表格尽量转换为Markdown表格格式或结构化的描述文本（如“表格1：2023年销售数据，列包括：地区、Q1、Q2、Q3、Q4，各行数据如下...”）。
    • 元数据：保留文档来源、文件名、作者、最后修改时间等信息，作为Document对象的元数据。这些元数据在后续的混合检索（结合关键词过滤）中非常有用。

3. 实战：构建一个可复用的多格式文档处理模块

理论说再多，不如一行代码。下面，我将展示如何用Python构建一个核心处理模块。这里我们不依赖LangChain的高层API，而是从底层实现，以便你彻底理解每个环节。

3.1 环境准备与依赖安装

首先，创建一个新的虚拟环境并安装核心依赖。这些库覆盖了我们前面讨论的主要格式。

# 创建并激活虚拟环境（以conda为例） conda create -n rag_preprocess python=3.10 conda activate rag_preprocess # 安装核心依赖 pip install pymupdf pdfplumber python-docx pandas openpyxl python-pptx beautifulsoup4 newspaper3k pip install markdown unstructured[md] # 用于Markdown和更复杂的文档处理 pip install chardet # 用于检测文件编码

3.2 核心加载器类的实现

我们将创建一个DocumentProcessor类，它根据文件扩展名自动分派到对应的加载方法。

import os import fitz # PyMuPDF import pdfplumber from docx import Document import pandas as pd from bs4 import BeautifulSoup import markdown import chardet from typing import List, Dict, Any, Optional import re class DocumentProcessor: def __init__(self): self.supported_extensions = { '.pdf': self._load_pdf, '.docx': self._load_docx, '.xlsx': self._load_excel, '.pptx': self._load_pptx, '.txt': self._load_text, '.md': self._load_markdown, '.html': self._load_html, '.csv': self._load_csv, } def load_document(self, file_path: str) -> Dict[str, Any]: """主加载函数，根据后缀名调用对应方法""" ext = os.path.splitext(file_path)[1].lower() if ext not in self.supported_extensions: raise ValueError(f"Unsupported file format: {ext}") loader = self.supported_extensions[ext] raw_text, metadata = loader(file_path) # 基础清洗（所有格式通用） cleaned_text = self._basic_clean(raw_text) # 增强元数据 metadata.update({ 'file_path': file_path, 'file_name': os.path.basename(file_path), 'file_size': os.path.getsize(file_path), 'extension': ext }) return {'text': cleaned_text, 'metadata': metadata} def _basic_clean(self, text: str) -> str: """基础文本清洗""" # 替换多种空白字符为单个空格 text = re.sub(r'\s+', ' ', text) # 去除首尾空白 text = text.strip() # 简单的全角转半角（示例，仅处理字母数字和部分标点） text = text.translate(str.maketrans('，。！？；：“”‘’（）【】', ',.!?;:\"\"\'\'()[]')) return text def _load_pdf(self, file_path: str): """加载PDF文件，结合PyMuPDF和pdfplumber""" full_text = "" metadata = {'type': 'pdf'} tables_text = [] # 1. 使用PyMuPDF提取主要文本和元信息 try: doc = fitz.open(file_path) metadata['page_count'] = doc.page_count metadata['author'] = doc.metadata.get('author', '') metadata['title'] = doc.metadata.get('title', '') for page_num, page in enumerate(doc): # 提取文本，保留简单的布局信息 text = page.get_text("text") # 简单的页眉页脚过滤（启发式规则） lines = text.split('\n') filtered_lines = [line for line in lines if not self._is_header_footer(line, page_num, len(lines))] full_text += '\n'.join(filtered_lines) + '\n--- Page Break ---\n' except Exception as e: print(f"PyMuPDF读取{file_path}失败: {e}") full_text = "" # 2. 使用pdfplumber尝试提取表格 try: with pdfplumber.open(file_path) as pdf: for page in pdf.pages: tables = page.extract_tables() for table in tables: if table: # 将表格转换为Markdown格式字符串 md_table = self._table_to_markdown(table) tables_text.append(f"\n[Extracted Table]:\n{md_table}\n") except Exception as e: print(f"pdfplumber读取{file_path}表格失败: {e}") # 合并文本和表格 combined_text = full_text if tables_text: combined_text += "\n\n--- Extracted Tables ---\n" + "\n".join(tables_text) return combined_text, metadata def _is_header_footer(self, line: str, page_num: int, total_lines: int) -> bool: """简单的页眉页脚判断（启发式规则）""" line_lower = line.strip().lower() # 规则1：匹配页码模式 if re.match(r'^第?\d+页$', line_lower) or re.match(r'^\d+/\d+$', line_lower): return True # 规则2：出现在第一行或最后一行，且长度较短（可能是页眉页脚） if (page_num == 0 and total_lines < 3) or (page_num == total_lines - 1 and len(line) < 50): # 可以进一步加入黑名单关键词，如“保密”、“公司名称”等 blacklist = ['confidential', '内部文件', '草案'] if any(word in line_lower for word in blacklist): return True return False def _table_to_markdown(self, table_data): """将二维列表转换为Markdown表格字符串""" if not table_data: return "" md_lines = [] # 处理表头 headers = table_data[0] md_lines.append('| ' + ' | '.join(str(h) for h in headers) + ' |') md_lines.append('|' + ' --- |' * len(headers)) # 处理数据行 for row in table_data[1:]: md_lines.append('| ' + ' | '.join(str(cell) for cell in row) + ' |') return '\n'.join(md_lines) def _load_docx(self, file_path: str): """加载Word文档""" doc = Document(file_path) full_text = [] metadata = {'type': 'docx'} # 提取段落 for para in doc.paragraphs: if para.text.strip(): full_text.append(para.text) # 提取表格 for table in doc.tables: table_data = [] for row in table.rows: row_data = [cell.text.strip() for cell in row.cells] table_data.append(row_data) if table_data: md_table = self._table_to_markdown(table_data) full_text.append(f"\n[Document Table]:\n{md_table}\n") return '\n'.join(full_text), metadata def _load_excel(self, file_path: str): """加载Excel文件，处理所有sheet""" xl = pd.ExcelFile(file_path) sheet_texts = [] metadata = {'type': 'excel', 'sheets': xl.sheet_names} for sheet_name in xl.sheet_names: df = xl.parse(sheet_name) # 将DataFrame转换为格式化的文本表示 sheet_str = f"\n--- Sheet: {sheet_name} ---\n" # 简单处理：转换为CSV格式的字符串，保留表头 sheet_str += df.to_csv(index=False, sep='|') sheet_texts.append(sheet_str) return '\n'.join(sheet_texts), metadata # 其他格式的加载方法（_load_pptx, _load_text等）遵循类似模式，此处省略详细实现以节省篇幅。 # 它们会提取核心文本，并返回统一的字典结构。

3.3 文本分割策略：为向量化做准备

加载并清洗后的文本可能很长（比如一本书），直接向量化会丢失细节，且检索效率低。因此，我们需要进行“文本分割”。这里的关键不是简单按固定长度切分，而是基于语义和结构进行智能分块。

  1. 递归字符分割：这是最基本的方法，按固定字符数（如500字）分割，重叠一部分（如50字）以保持上下文。LangChain的RecursiveCharacterTextSplitter在此基础上做了优化，它会优先尝试按段落、句子、单词等自然分隔符来分割，避免在单词或句子中间切断。
  2. 基于标记的分割：直接使用大模型的Tokenizer（如tiktokenfor GPT）按Token数分割，能更精确地控制输入LLM的上下文长度。
  3. 语义分割：这是更高级的方法，利用小型模型（如sentence-transformers）计算句子间的相似度，在语义变化大的地方进行分割。效果更好，但计算成本高。

我的实战策略是分层分割：

  • 第一层：按文档结构分割。利用加载时保留的标题信息（H1, H2），将文档分成若干个大节。
  • 第二层：在大节内部进行递归字符分割。设置一个较大的块大小（如1000字符）和重叠（200字符）。
  • 第三层：为每个块提取关键句或生成摘要，作为该块的“元描述”，便于后续的元数据检索。
from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter # 假设我们已经从文档中提取出了Markdown格式的文本，并保留了标题 markdown_text = "# 第一章 引言\n\n这是引言内容...\n\n## 1.1 背景\n\n背景描述..." # 首先，按Markdown标题分割 headers_to_split_on = [("#", "H1"), ("##", "H2"), ("###", "H3")] markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) md_header_splits = markdown_splitter.split_text(markdown_text) # 然后，对每个标题块内部进行更细粒度的分割 final_splits = [] text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) for chunk in md_header_splits: # chunk.metadata 包含了标题信息，如 {'H1': '第一章 引言', 'H2': '1.1 背景'} sub_splits = text_splitter.split_text(chunk.page_content) for sub in sub_splits: # 将上级标题的元数据继承下来 final_splits.append({ 'text': sub, 'metadata': {**chunk.metadata, 'chunk_id': len(final_splits)} })

4. 避坑指南与性能优化

在实际操作中，你会遇到各种各样的问题。下面是我总结的常见“坑”及解决方案。

4.1 内容提取不完整或错乱

  • 问题：PDF提取时丢失大量文本，或顺序完全错乱。
  • 排查：
    1. 先用Adobe Acrobat或Foxit等专业PDF阅读器检查文档属性，看是“文本型PDF”还是“图像型PDF”。
    2. 对于图像型PDF，必须使用OCR。可以先用pdf2image将PDF转为图片，再用pytesseract或效果更好的PaddleOCR、EasyOCR进行识别。LangChain的UnstructuredPDFLoader在mode="ocr"模式下整合了这个流程。
    3. 对于顺序错乱，通常是多栏排版导致的。pdfplumber可以通过extract_text()方法的layout=True参数尝试保持布局，或者用PyMuPDF获取每个文本块的坐标，然后按y坐标排序后拼接。
  • 心得：对于非常重要的文档，手动校对前几页的提取结果是值得的。可以写一个简单的脚本，将提取的文本块和坐标打印出来，对比原PDF查看问题所在。

4.2 表格和特殊格式处理失败

  • 问题：表格被提取成一团乱麻的文本，公式、流程图等完全丢失。
  • 解决方案：
    • 表格：优先使用pdfplumber或camelot（对于更复杂的表格）。将表格转换为结构化数据（如列表的列表）后，再决定如何文本化。对于RAG，我倾向于将其转换为描述性文字，例如：“下表展示了2023年各季度销售数据：华东地区Q1为100万，Q2为120万...”，这比原始的|分隔符更利于语义理解。
    • 公式/图表：目前没有完美的文本化方案。一个折中办法是，在预处理时识别出这些区域（可以通过PyMuPDF检测Drawings或特定类型的对象），并在文本中插入一个标记，如[公式1]或[图2: 销售趋势图]。同时，将对应的图片区域单独保存下来。在RAG检索后，如果需要，可以再将这个图片作为上下文的一部分提供给LLM（多模态RAG）。

4.3 处理速度慢，内存占用高

  • 问题：处理上千个PDF时，脚本运行缓慢甚至崩溃。
  • 优化策略：
    1. 并发处理：使用concurrent.futures.ThreadPoolExecutor或multiprocessing池。注意，PyMuPDF等库可能不是完全线程安全的，建议使用进程池。
    2. 流式处理：对于超大文档（如数百页的PDF），不要一次性读入内存。PyMuPDF可以逐页处理；对于文本文件，可以按行或按块读取。
    3. 选择性处理：不是所有页面都需要。可以先提取目录或分析页面布局，只处理正文页。
    4. 缓存中间结果：将清洗和分割后的文本块序列化（如pickle或jsonl）保存到磁盘。这样，在调整向量化或检索策略时，无需重新解析文档。

4.4 编码与语言问题

  • 问题：处理中文、日文等非拉丁语系文档时出现乱码。
  • 解决方案：
    1. 对于未知编码的文本文件，使用chardet库检测。
    2. 确保你的Python环境和终端/编辑器都使用UTF-8编码。
    3. 某些旧的PDF或DOC文件可能使用特定字体编码。对于PDF，PyMuPDF的get_text(“text”)通常能较好处理。如果不行，尝试get_text(“dict”)或get_text(“html”)查看更原始的数据。
    4. 考虑使用专门针对中文优化的OCR引擎，如PaddleOCR。

5. 工程化与质量评估

当处理海量文档时，需要一个工程化的流水线和质量检查机制。

  1. 构建处理流水线：使用工作流引擎（如Apache Airflow、Prefect）或简单的脚本调度，将流程模块化：文件监听 -> 格式识别 -> 分发解析 -> 文本清洗 -> 智能分割 -> 存储（文本块+元数据）。
  2. 日志与监控：为每个文档处理步骤记录详细的日志，包括成功/失败状态、耗时、提取的字符数、遇到的警告（如编码猜测、表格识别置信度低等）。这有助于快速定位问题文档。
  3. 质量评估指标：
    • 完整性：提取的文本长度与原文档预估长度是否匹配？（例如，一页A4纸大约500-800字）。
    • 保真度：随机抽样若干文本块，人工检查其是否准确反映了原文内容，表格、列表结构是否保留。
    • 可用性：将处理后的文本块，用简单的关键词检索或向量检索进行测试，看是否能准确召回相关信息。这是最直接的验收标准。

一个简单的质量检查脚本可以这样写：

def quality_check(original_file_path, processed_chunks): """简单的质量检查""" issues = [] total_processed_text = "".join([chunk['text'] for chunk in processed_chunks]) # 检查1：文本长度是否异常短？（可能解析失败） if len(total_processed_text) < 100: # 假设文档至少100字符 issues.append(f"警告：{original_file_path} 提取的文本过短，可能解析失败。") # 检查2：是否包含大量无意义的乱码或占位符？ garbage_patterns = [r'�{5,}', r'\.{10,}'] # 连续的特殊字符或句点 for pattern in garbage_patterns: if re.search(pattern, total_processed_text): issues.append(f"警告：{original_file_path} 文本中包含疑似乱码。") break # 检查3：关键信息是否存在？（例如，文档标题或已知关键词） # 这里需要根据业务定义关键词 expected_keywords = ['摘要', '引言', '结论'] for keyword in expected_keywords: if keyword not in total_processed_text: issues.append(f"提示：{original_file_path} 中未找到常见章节“{keyword}”，请确认文档类型。") return issues

把多格式文档加载和预处理这一步做扎实了，你的RAG系统就成功了一半。它决定了知识库的“原料”质量，后续无论用多么精妙的检索和重排序算法，都难以弥补源头上的信息缺失或噪声。我的建议是，在项目初期，花足够的时间来打磨这个预处理流水线，建立严格的质量检查点，这会为你省下后期大量的调试和返工时间。记住，高质量的输入是高质量输出的第一前提。

编程学习 学习日记 实战经验
← 返回列表