1. 项目概述为什么RAG的第一步如此关键如果你正在构建一个基于大语言模型LLM的问答系统、智能客服或者企业知识库那么“RAG”这个词对你来说一定不陌生。RAG即检索增强生成它通过从外部知识库中检索相关信息来“喂”给大模型从而生成更准确、更可靠的回答有效缓解了大模型的“幻觉”问题。听起来很美好对吧但很多朋友在兴致勃勃地开始搭建RAG系统时往往一头扎进向量数据库选型、检索算法优化这些“高大上”的环节却忽略了最基础、也最容易出问题的一步多格式文档的加载与文本预处理。我见过太多项目因为前期文档处理不当导致后续的检索召回率极低生成的答案牛头不对马嘴整个系统效果大打折扣。这就像盖房子地基没打牢上面的装修再豪华也是白搭。今天我们就来彻底拆解RAG实战的第一步聚焦于如何将五花八门的文档——PDF、Word、PPT、Excel、TXT、Markdown甚至网页——可靠地转换成干净、结构化的纯文本为后续的向量化和检索打下坚实基础。这个过程直接决定了你的知识库“原料”质量是RAG项目成败的生命线。2. 核心思路与工具选型构建稳健的文档处理流水线一个稳健的文档处理流水线其核心目标就两个兼容性和保真度。兼容性是指能处理尽可能多的文档格式保真度是指转换后的文本要尽可能保留原文的结构、语义和关键信息如表格、列表。基于这个思路我们的技术选型需要分层考虑。2.1 文档加载层按格式分而治之没有一种工具能完美处理所有格式。因此我们需要一个“工具箱”策略针对不同格式选用最合适的解析器。PDF文档这是最复杂也最常见的格式。我推荐使用PyMuPDF和pdfplumber的组合。PyMuPDF提取速度和文本保真度极高能很好地处理由文字构成的PDF。但对于扫描件或图片型PDF它无能为力。pdfplumber它的强项在于精确提取表格数据和维持文本的物理布局如多栏排版。对于包含复杂表格的PDF它是首选。实战心得对于重要文档我通常会先用PyMuPDF提取一遍再用pdfplumber专门提取表格然后将结果融合。虽然pypdf和pdf2imageOCR也是选项但前者功能较弱后者如pytesseract处理速度慢、依赖环境复杂非扫描件场景下不推荐作为首选。Office文档Word (.docx)python-docx库是绝对的主流。它能完美解析段落、标题、列表、表格甚至图片标注。对于老旧的.doc格式可以先用libreoffice命令行工具将其转换为.docx再处理。Excel (.xlsx)pandas是不二之选。关键在于如何处理多个sheet。我的做法是将每个sheet转换成一个格式化的文本块例如用|分隔符模拟表格并注明sheet名这样能最大程度保留表格的二维信息。PowerPoint (.pptx)python-pptx可以提取每页幻灯片的文本框内容。需要注意的是PPT内容通常是碎片化的需要将每页的内容合理拼接并保留幻灯片标题作为上下文。纯文本与标记语言TXT/CSV直接用Python内置的open函数读取即可注意编码问题优先使用utf-8。Markdown/HTMLBeautifulSoup4可以处理HTML提取正文并去除标签。对于Markdownmarkdown库可以将其转换为HTML再用BeautifulSoup处理或者直接用正则表达式提取关键部分。目标是保留标题层级和列表结构。网页内容除了BeautifulSoup4newspaper3k是一个更高级的选择它能自动识别文章主体内容过滤导航栏、广告等噪音提取效果非常好。注意在实际项目中手动为每种格式写加载器非常繁琐。我强烈建议使用像LangChain或LlamaIndex这样的框架。它们提供了统一的DocumentLoader接口背后集成了上述各种解析器。例如LangChain的PyMuPDFLoader、UnstructuredWordDocumentLoader、UnstructuredExcelLoader等能极大提升开发效率。但了解其底层原理对于排查解析错误至关重要。2.2 文本预处理层从原始文本到干净语料加载得到原始文本后里面充满了对LLM无用的“噪音”预处理的目的就是去除噪音保留精华。清洗无用字符去除不可见字符如\x00、乱码、过多的空白符将连续的换行、空格标准化。页眉页脚/页码这是PDF处理中的老大难问题。简单的正则表达式如匹配“第X页”或页眉特定文字可以解决一部分但对于复杂的版面可能需要结合文本位置信息PyMuPDF可以提供坐标进行过滤。冗余信息去除文档中的版权声明、网址链接有时需要保留、电子邮件等模板化内容。标准化编码统一确保所有文本都是UTF-8编码。全半角转换将全角字符中文标点、字母、数字转换为半角保持一致性。日期/数字格式将“2023年12月1日”统一为“2023-12-01”有助于后续的实体识别和检索。结构化信息提取与保留标题与章节在解析时就应识别并标记H1, H2, H3等标题。这不仅是文本结构在后续的“文本分割”步骤中标题是重要的分割边界能防止一个章节被生生切断。列表与表格将列表项用“-”或数字明确标出。表格尽量转换为Markdown表格格式或结构化的描述文本如“表格12023年销售数据列包括地区、Q1、Q2、Q3、Q4各行数据如下...”。元数据保留文档来源、文件名、作者、最后修改时间等信息作为Document对象的元数据。这些元数据在后续的混合检索结合关键词过滤中非常有用。3. 实战构建一个可复用的多格式文档处理模块理论说再多不如一行代码。下面我将展示如何用Python构建一个核心处理模块。这里我们不依赖LangChain的高层API而是从底层实现以便你彻底理解每个环节。3.1 环境准备与依赖安装首先创建一个新的虚拟环境并安装核心依赖。这些库覆盖了我们前面讨论的主要格式。# 创建并激活虚拟环境以conda为例 conda create -n rag_preprocess python3.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(fUnsupported 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(fPyMuPDF读取{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(fpdfplumber读取{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(indexFalse, sep|) sheet_texts.append(sheet_str) return \n.join(sheet_texts), metadata # 其他格式的加载方法_load_pptx, _load_text等遵循类似模式此处省略详细实现以节省篇幅。 # 它们会提取核心文本并返回统一的字典结构。3.3 文本分割策略为向量化做准备加载并清洗后的文本可能很长比如一本书直接向量化会丢失细节且检索效率低。因此我们需要进行“文本分割”。这里的关键不是简单按固定长度切分而是基于语义和结构进行智能分块。递归字符分割这是最基本的方法按固定字符数如500字分割重叠一部分如50字以保持上下文。LangChain的RecursiveCharacterTextSplitter在此基础上做了优化它会优先尝试按段落、句子、单词等自然分隔符来分割避免在单词或句子中间切断。基于标记的分割直接使用大模型的Tokenizer如tiktokenfor GPT按Token数分割能更精确地控制输入LLM的上下文长度。语义分割这是更高级的方法利用小型模型如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_onheaders_to_split_on) md_header_splits markdown_splitter.split_text(markdown_text) # 然后对每个标题块内部进行更细粒度的分割 final_splits [] text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, 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提取时丢失大量文本或顺序完全错乱。排查先用Adobe Acrobat或Foxit等专业PDF阅读器检查文档属性看是“文本型PDF”还是“图像型PDF”。对于图像型PDF必须使用OCR。可以先用pdf2image将PDF转为图片再用pytesseract或效果更好的PaddleOCR、EasyOCR进行识别。LangChain的UnstructuredPDFLoader在modeocr模式下整合了这个流程。对于顺序错乱通常是多栏排版导致的。pdfplumber可以通过extract_text()方法的layoutTrue参数尝试保持布局或者用PyMuPDF获取每个文本块的坐标然后按y坐标排序后拼接。心得对于非常重要的文档手动校对前几页的提取结果是值得的。可以写一个简单的脚本将提取的文本块和坐标打印出来对比原PDF查看问题所在。4.2 表格和特殊格式处理失败问题表格被提取成一团乱麻的文本公式、流程图等完全丢失。解决方案表格优先使用pdfplumber或camelot对于更复杂的表格。将表格转换为结构化数据如列表的列表后再决定如何文本化。对于RAG我倾向于将其转换为描述性文字例如“下表展示了2023年各季度销售数据华东地区Q1为100万Q2为120万...”这比原始的|分隔符更利于语义理解。公式/图表目前没有完美的文本化方案。一个折中办法是在预处理时识别出这些区域可以通过PyMuPDF检测Drawings或特定类型的对象并在文本中插入一个标记如[公式1]或[图2: 销售趋势图]。同时将对应的图片区域单独保存下来。在RAG检索后如果需要可以再将这个图片作为上下文的一部分提供给LLM多模态RAG。4.3 处理速度慢内存占用高问题处理上千个PDF时脚本运行缓慢甚至崩溃。优化策略并发处理使用concurrent.futures.ThreadPoolExecutor或multiprocessing池。注意PyMuPDF等库可能不是完全线程安全的建议使用进程池。流式处理对于超大文档如数百页的PDF不要一次性读入内存。PyMuPDF可以逐页处理对于文本文件可以按行或按块读取。选择性处理不是所有页面都需要。可以先提取目录或分析页面布局只处理正文页。缓存中间结果将清洗和分割后的文本块序列化如pickle或jsonl保存到磁盘。这样在调整向量化或检索策略时无需重新解析文档。4.4 编码与语言问题问题处理中文、日文等非拉丁语系文档时出现乱码。解决方案对于未知编码的文本文件使用chardet库检测。确保你的Python环境和终端/编辑器都使用UTF-8编码。某些旧的PDF或DOC文件可能使用特定字体编码。对于PDFPyMuPDF的get_text(“text”)通常能较好处理。如果不行尝试get_text(“dict”)或get_text(“html”)查看更原始的数据。考虑使用专门针对中文优化的OCR引擎如PaddleOCR。5. 工程化与质量评估当处理海量文档时需要一个工程化的流水线和质量检查机制。构建处理流水线使用工作流引擎如Apache Airflow、Prefect或简单的脚本调度将流程模块化文件监听 - 格式识别 - 分发解析 - 文本清洗 - 智能分割 - 存储文本块元数据。日志与监控为每个文档处理步骤记录详细的日志包括成功/失败状态、耗时、提取的字符数、遇到的警告如编码猜测、表格识别置信度低等。这有助于快速定位问题文档。质量评估指标完整性提取的文本长度与原文档预估长度是否匹配例如一页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系统就成功了一半。它决定了知识库的“原料”质量后续无论用多么精妙的检索和重排序算法都难以弥补源头上的信息缺失或噪声。我的建议是在项目初期花足够的时间来打磨这个预处理流水线建立严格的质量检查点这会为你省下后期大量的调试和返工时间。记住高质量的输入是高质量输出的第一前提。
1. 项目概述为什么RAG的第一步如此关键如果你正在构建一个基于大语言模型LLM的问答系统、智能客服或者企业知识库那么“RAG”这个词对你来说一定不陌生。RAG即检索增强生成它通过从外部知识库中检索相关信息来“喂”给大模型从而生成更准确、更可靠的回答有效缓解了大模型的“幻觉”问题。听起来很美好对吧但很多朋友在兴致勃勃地开始搭建RAG系统时往往一头扎进向量数据库选型、检索算法优化这些“高大上”的环节却忽略了最基础、也最容易出问题的一步多格式文档的加载与文本预处理。我见过太多项目因为前期文档处理不当导致后续的检索召回率极低生成的答案牛头不对马嘴整个系统效果大打折扣。这就像盖房子地基没打牢上面的装修再豪华也是白搭。今天我们就来彻底拆解RAG实战的第一步聚焦于如何将五花八门的文档——PDF、Word、PPT、Excel、TXT、Markdown甚至网页——可靠地转换成干净、结构化的纯文本为后续的向量化和检索打下坚实基础。这个过程直接决定了你的知识库“原料”质量是RAG项目成败的生命线。2. 核心思路与工具选型构建稳健的文档处理流水线一个稳健的文档处理流水线其核心目标就两个兼容性和保真度。兼容性是指能处理尽可能多的文档格式保真度是指转换后的文本要尽可能保留原文的结构、语义和关键信息如表格、列表。基于这个思路我们的技术选型需要分层考虑。2.1 文档加载层按格式分而治之没有一种工具能完美处理所有格式。因此我们需要一个“工具箱”策略针对不同格式选用最合适的解析器。PDF文档这是最复杂也最常见的格式。我推荐使用PyMuPDF和pdfplumber的组合。PyMuPDF提取速度和文本保真度极高能很好地处理由文字构成的PDF。但对于扫描件或图片型PDF它无能为力。pdfplumber它的强项在于精确提取表格数据和维持文本的物理布局如多栏排版。对于包含复杂表格的PDF它是首选。实战心得对于重要文档我通常会先用PyMuPDF提取一遍再用pdfplumber专门提取表格然后将结果融合。虽然pypdf和pdf2imageOCR也是选项但前者功能较弱后者如pytesseract处理速度慢、依赖环境复杂非扫描件场景下不推荐作为首选。Office文档Word (.docx)python-docx库是绝对的主流。它能完美解析段落、标题、列表、表格甚至图片标注。对于老旧的.doc格式可以先用libreoffice命令行工具将其转换为.docx再处理。Excel (.xlsx)pandas是不二之选。关键在于如何处理多个sheet。我的做法是将每个sheet转换成一个格式化的文本块例如用|分隔符模拟表格并注明sheet名这样能最大程度保留表格的二维信息。PowerPoint (.pptx)python-pptx可以提取每页幻灯片的文本框内容。需要注意的是PPT内容通常是碎片化的需要将每页的内容合理拼接并保留幻灯片标题作为上下文。纯文本与标记语言TXT/CSV直接用Python内置的open函数读取即可注意编码问题优先使用utf-8。Markdown/HTMLBeautifulSoup4可以处理HTML提取正文并去除标签。对于Markdownmarkdown库可以将其转换为HTML再用BeautifulSoup处理或者直接用正则表达式提取关键部分。目标是保留标题层级和列表结构。网页内容除了BeautifulSoup4newspaper3k是一个更高级的选择它能自动识别文章主体内容过滤导航栏、广告等噪音提取效果非常好。注意在实际项目中手动为每种格式写加载器非常繁琐。我强烈建议使用像LangChain或LlamaIndex这样的框架。它们提供了统一的DocumentLoader接口背后集成了上述各种解析器。例如LangChain的PyMuPDFLoader、UnstructuredWordDocumentLoader、UnstructuredExcelLoader等能极大提升开发效率。但了解其底层原理对于排查解析错误至关重要。2.2 文本预处理层从原始文本到干净语料加载得到原始文本后里面充满了对LLM无用的“噪音”预处理的目的就是去除噪音保留精华。清洗无用字符去除不可见字符如\x00、乱码、过多的空白符将连续的换行、空格标准化。页眉页脚/页码这是PDF处理中的老大难问题。简单的正则表达式如匹配“第X页”或页眉特定文字可以解决一部分但对于复杂的版面可能需要结合文本位置信息PyMuPDF可以提供坐标进行过滤。冗余信息去除文档中的版权声明、网址链接有时需要保留、电子邮件等模板化内容。标准化编码统一确保所有文本都是UTF-8编码。全半角转换将全角字符中文标点、字母、数字转换为半角保持一致性。日期/数字格式将“2023年12月1日”统一为“2023-12-01”有助于后续的实体识别和检索。结构化信息提取与保留标题与章节在解析时就应识别并标记H1, H2, H3等标题。这不仅是文本结构在后续的“文本分割”步骤中标题是重要的分割边界能防止一个章节被生生切断。列表与表格将列表项用“-”或数字明确标出。表格尽量转换为Markdown表格格式或结构化的描述文本如“表格12023年销售数据列包括地区、Q1、Q2、Q3、Q4各行数据如下...”。元数据保留文档来源、文件名、作者、最后修改时间等信息作为Document对象的元数据。这些元数据在后续的混合检索结合关键词过滤中非常有用。3. 实战构建一个可复用的多格式文档处理模块理论说再多不如一行代码。下面我将展示如何用Python构建一个核心处理模块。这里我们不依赖LangChain的高层API而是从底层实现以便你彻底理解每个环节。3.1 环境准备与依赖安装首先创建一个新的虚拟环境并安装核心依赖。这些库覆盖了我们前面讨论的主要格式。# 创建并激活虚拟环境以conda为例 conda create -n rag_preprocess python3.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(fUnsupported 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(fPyMuPDF读取{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(fpdfplumber读取{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(indexFalse, sep|) sheet_texts.append(sheet_str) return \n.join(sheet_texts), metadata # 其他格式的加载方法_load_pptx, _load_text等遵循类似模式此处省略详细实现以节省篇幅。 # 它们会提取核心文本并返回统一的字典结构。3.3 文本分割策略为向量化做准备加载并清洗后的文本可能很长比如一本书直接向量化会丢失细节且检索效率低。因此我们需要进行“文本分割”。这里的关键不是简单按固定长度切分而是基于语义和结构进行智能分块。递归字符分割这是最基本的方法按固定字符数如500字分割重叠一部分如50字以保持上下文。LangChain的RecursiveCharacterTextSplitter在此基础上做了优化它会优先尝试按段落、句子、单词等自然分隔符来分割避免在单词或句子中间切断。基于标记的分割直接使用大模型的Tokenizer如tiktokenfor GPT按Token数分割能更精确地控制输入LLM的上下文长度。语义分割这是更高级的方法利用小型模型如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_onheaders_to_split_on) md_header_splits markdown_splitter.split_text(markdown_text) # 然后对每个标题块内部进行更细粒度的分割 final_splits [] text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, 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提取时丢失大量文本或顺序完全错乱。排查先用Adobe Acrobat或Foxit等专业PDF阅读器检查文档属性看是“文本型PDF”还是“图像型PDF”。对于图像型PDF必须使用OCR。可以先用pdf2image将PDF转为图片再用pytesseract或效果更好的PaddleOCR、EasyOCR进行识别。LangChain的UnstructuredPDFLoader在modeocr模式下整合了这个流程。对于顺序错乱通常是多栏排版导致的。pdfplumber可以通过extract_text()方法的layoutTrue参数尝试保持布局或者用PyMuPDF获取每个文本块的坐标然后按y坐标排序后拼接。心得对于非常重要的文档手动校对前几页的提取结果是值得的。可以写一个简单的脚本将提取的文本块和坐标打印出来对比原PDF查看问题所在。4.2 表格和特殊格式处理失败问题表格被提取成一团乱麻的文本公式、流程图等完全丢失。解决方案表格优先使用pdfplumber或camelot对于更复杂的表格。将表格转换为结构化数据如列表的列表后再决定如何文本化。对于RAG我倾向于将其转换为描述性文字例如“下表展示了2023年各季度销售数据华东地区Q1为100万Q2为120万...”这比原始的|分隔符更利于语义理解。公式/图表目前没有完美的文本化方案。一个折中办法是在预处理时识别出这些区域可以通过PyMuPDF检测Drawings或特定类型的对象并在文本中插入一个标记如[公式1]或[图2: 销售趋势图]。同时将对应的图片区域单独保存下来。在RAG检索后如果需要可以再将这个图片作为上下文的一部分提供给LLM多模态RAG。4.3 处理速度慢内存占用高问题处理上千个PDF时脚本运行缓慢甚至崩溃。优化策略并发处理使用concurrent.futures.ThreadPoolExecutor或multiprocessing池。注意PyMuPDF等库可能不是完全线程安全的建议使用进程池。流式处理对于超大文档如数百页的PDF不要一次性读入内存。PyMuPDF可以逐页处理对于文本文件可以按行或按块读取。选择性处理不是所有页面都需要。可以先提取目录或分析页面布局只处理正文页。缓存中间结果将清洗和分割后的文本块序列化如pickle或jsonl保存到磁盘。这样在调整向量化或检索策略时无需重新解析文档。4.4 编码与语言问题问题处理中文、日文等非拉丁语系文档时出现乱码。解决方案对于未知编码的文本文件使用chardet库检测。确保你的Python环境和终端/编辑器都使用UTF-8编码。某些旧的PDF或DOC文件可能使用特定字体编码。对于PDFPyMuPDF的get_text(“text”)通常能较好处理。如果不行尝试get_text(“dict”)或get_text(“html”)查看更原始的数据。考虑使用专门针对中文优化的OCR引擎如PaddleOCR。5. 工程化与质量评估当处理海量文档时需要一个工程化的流水线和质量检查机制。构建处理流水线使用工作流引擎如Apache Airflow、Prefect或简单的脚本调度将流程模块化文件监听 - 格式识别 - 分发解析 - 文本清洗 - 智能分割 - 存储文本块元数据。日志与监控为每个文档处理步骤记录详细的日志包括成功/失败状态、耗时、提取的字符数、遇到的警告如编码猜测、表格识别置信度低等。这有助于快速定位问题文档。质量评估指标完整性提取的文本长度与原文档预估长度是否匹配例如一页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系统就成功了一半。它决定了知识库的“原料”质量后续无论用多么精妙的检索和重排序算法都难以弥补源头上的信息缺失或噪声。我的建议是在项目初期花足够的时间来打磨这个预处理流水线建立严格的质量检查点这会为你省下后期大量的调试和返工时间。记住高质量的输入是高质量输出的第一前提。
)
