RAG实战：多格式文档处理完全指南
这是RAG系统最核心的实战问题！让我给你一个完整的、可直接使用的解决方案。
📚 一、文档格式处理全览
处理流程总览
原始文档 (PDF/Word/Excel/PPT)↓
【步骤1】文档加载与解析├─ PDF → pdfplumber (文字) + OCR (扫描版)├─ Word → python-docx├─ Excel → pandas└─ PPT → python-pptx↓
【步骤2】内容分类提取├─ 纯文本 → 直接提取├─ 表格 → 结构化解析 + 文字描述├─ 图片 → OCR + 多模态理解└─ 混合内容 → 分别处理后组合↓
【步骤3】智能分块 (Chunking)├─ 语义完整性优先├─ 保留结构信息└─ 添加元数据标记↓
【步骤4】向量化 (Embedding)├─ 文本向量化├─ 表格内容向量化└─ 多模态内容处理↓
【步骤5】存储到向量数据库├─ 向量 + 原始内容├─ 元数据（来源、类型、页码）└─ 建立索引
🔧 二、统一文档处理器（实战代码）
核心处理类
import os
from pathlib import Path
from typing import List, Dict, Any
import logging# 文档处理库
import pdfplumber
from docx import Document
import pandas as pd
from pptx import Presentation
from PIL import Image# AI处理库
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document as LangChainDocument
import pytesseract # OCR
from paddleocr import PaddleOCR # 中文OCR更好logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class UniversalDocumentProcessor:"""统一文档处理器 - 支持所有常见格式"""def __init__(self):self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=100,separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""])# 初始化OCR（中文）self.ocr = PaddleOCR(use_angle_cls=True, lang='ch')def process_document(self, file_path: str) -> List[LangChainDocument]:"""处理任意格式的文档返回：分块后的文档列表"""file_path = Path(file_path)suffix = file_path.suffix.lower()logger.info(f"开始处理文档: {file_path.name}")# 根据文件类型调用不同的处理器processors = {'.pdf': self.process_pdf,'.docx': self.process_word,'.doc': self.process_word,'.xlsx': self.process_excel,'.xls': self.process_excel,'.pptx': self.process_ppt,'.txt': self.process_text,'.md': self.process_text,}if suffix not in processors:raise ValueError(f"不支持的文件格式: {suffix}")# 处理文档documents = processors[suffix](file_path)logger.info(f"处理完成: 生成 {len(documents)} 个文档块")return documents
📄 三、PDF处理（最复杂）
完整的PDF处理方案
def process_pdf(self, file_path: Path) -> List[LangChainDocument]:"""处理PDF - 支持文字版和扫描版"""documents = []with pdfplumber.open(file_path) as pdf:for page_num, page in enumerate(pdf.pages, start=1):logger.info(f"处理PDF第 {page_num}/{len(pdf.pages)} 页")# 1. 提取文本text = page.extract_text()# 2. 提取表格tables = page.extract_tables()# 3. 检测是否有图片images = page.images# === 场景1：纯文本页面 ===if text and not tables and not images:chunks = self._chunk_text(text, metadata={"source": file_path.name,"page": page_num,"type": "text"})documents.extend(chunks)# === 场景2：有表格的页面 ===elif tables:# 文本部分if text:text_without_table = self._remove_table_from_text(text, tables)if text_without_table.strip():chunks = self._chunk_text(text_without_table,metadata={"source": file_path.name,"page": page_num,"type": "text"})documents.extend(chunks)# 表格部分（重点！）for table_idx, table in enumerate(tables):table_docs = self._process_table(table,metadata={"source": file_path.name,"page": page_num,"type": "table","table_index": table_idx})documents.extend(table_docs)# === 场景3：扫描版（OCR）===elif not text and images:logger.info(f"第{page_num}页是扫描版，使用OCR")# 转为图片img = page.to_image(resolution=300)pil_img = img.original# OCR识别ocr_result = self.ocr.ocr(pil_img, cls=True)ocr_text = '\n'.join([line[1][0] for line in ocr_result[0]])if ocr_text.strip():chunks = self._chunk_text(ocr_text,metadata={"source": file_path.name,"page": page_num,"type": "text","ocr": True})documents.extend(chunks)# === 场景4：混合内容 ===else:# 文本if text:chunks = self._chunk_text(text, {"source": file_path.name,"page": page_num,"type": "mixed"})documents.extend(chunks)# 图片中的文字（OCR）for img_idx, img in enumerate(images):try:img_text = self._extract_text_from_image(img)if img_text:documents.append(LangChainDocument(page_content=img_text,metadata={"source": file_path.name,"page": page_num,"type": "image_text","image_index": img_idx}))except Exception as e:logger.warning(f"图片{img_idx}处理失败: {e}")return documents
📊 四、表格处理（核心难点）
为什么表格难处理？
问题：直接向量化表格，效果很差示例表格：
| 科目 | 及格分数 | 优秀分数 |
|------|---------|---------|
| 语文 | 60 | 90 |
| 数学 | 60 | 90 |直接向量化："科目 及格分数 优秀分数 语文 60 90 数学 60 90"
→ 结构信息丢失，语义不清学生提问："语文多少分及格？"
→ 检索不到准确答案
表格处理的3种策略
def _process_table(self, table: List[List], metadata: Dict) -> List[LangChainDocument]:"""表格处理 - 3种策略组合"""if not table or not table[0]:return []documents = []# === 策略1：结构化存储（保留原始格式）===df = pd.DataFrame(table[1:], columns=table[0])# 转为Markdown格式（更易向量化）markdown_table = df.to_markdown(index=False)documents.append(LangChainDocument(page_content=markdown_table,metadata={**metadata,"representation": "structured"}))# === 策略2：生成自然语言描述 ⭐⭐⭐⭐⭐ ===# 这是最重要的！让表格可以被语义检索table_description = self._table_to_natural_language(df, table[0])documents.append(LangChainDocument(page_content=table_description,metadata={**metadata,"representation": "natural_language"}))# === 策略3：逐行拆分（适合大表格）===if len(df) > 5: # 大表格才拆分for idx, row in df.iterrows():row_text = self._row_to_text(row, table[0])documents.append(LangChainDocument(page_content=row_text,metadata={**metadata,"representation": "row","row_index": idx}))return documentsdef _table_to_natural_language(self, df: pd.DataFrame, headers: List) -> str:"""表格转自然语言描述（关键技术！）"""descriptions = []# 1. 表格整体描述descriptions.append(f"这是一个关于{headers[0]}的表格，包含{len(df)}行数据。")# 2. 列的描述descriptions.append(f"表格有以下列：{', '.join(headers)}")# 3. 逐行转换为自然语言for idx, row in df.iterrows():# 根据列名生成语句# 例如："语文的及格分数是60分，优秀分数是90分"sentences = []subject = row[headers[0]] # 第一列通常是主语for col in headers[1:]:value = row[col]# 生成类似"语文的及格分数是60分"的句子sentences.append(f"{subject}的{col}是{value}")descriptions.append("；".join(sentences) + "。")# 4. 如果有数值，生成统计信息numeric_cols = df.select_dtypes(include=['number']).columnsif len(numeric_cols) > 0:for col in numeric_cols:try:descriptions.append(f"{col}的范围是{df[col].min()}到{df[col].max()}")except:passreturn "\n".join(descriptions)def _row_to_text(self, row: pd.Series, headers: List) -> str:"""单行转文本"""parts = []subject = row[headers[0]]for col in headers[1:]:parts.append(f"{col}：{row[col]}")return f"{subject}，{', '.join(parts)}"
表格处理示例
# 输入表格：
"""
| 科目 | 及格分数 | 优秀分数 | 满分 |
|------|---------|---------|------|
| 语文 | 60 | 90 | 100 |
| 数学 | 60 | 90 | 100 |
| 英语 | 60 | 90 | 100 |
"""# 策略1输出（Markdown）：
"""
| 科目 | 及格分数 | 优秀分数 | 满分 |
|------|---------|---------|------|
| 语文 | 60 | 90 | 100 |
| 数学 | 60 | 90 | 100 |
| 英语 | 60 | 90 | 100 |
"""# 策略2输出（自然语言）⭐ 最重要：
"""
这是一个关于科目的表格，包含3行数据。
表格有以下列：科目, 及格分数, 优秀分数, 满分语文的及格分数是60；语文的优秀分数是90；语文的满分是100。
数学的及格分数是60；数学的优秀分数是90；数学的满分是100。
英语的及格分数是60；英语的优秀分数是90；英语的满分是100。及格分数的范围是60到60
优秀分数的范围是90到90
满分的范围是100到100
"""# 策略3输出（逐行）：
"""
语文，及格分数：60, 优秀分数：90, 满分：100
数学，及格分数：60, 优秀分数：90, 满分：100
英语，及格分数：60, 优秀分数：90, 满分：100
"""# 现在学生问："语文多少分及格？"
# → 可以准确检索到"语文的及格分数是60"！✅
📝 五、Word文档处理
def process_word(self, file_path: Path) -> List[LangChainDocument]:"""处理Word文档"""doc = Document(file_path)documents = []current_section = ""current_content = []for element in doc.element.body:# 处理段落if element.tag.endswith('p'):para = elementtext = ''.join(node.text for node in para.iter() if node.text)# 检测是否是标题if self._is_heading(para):# 保存之前的内容if current_content:documents.extend(self._chunk_text('\n'.join(current_content),metadata={"source": file_path.name,"section": current_section,"type": "text"}))# 开始新章节current_section = textcurrent_content = []else:current_content.append(text)# 处理表格elif element.tag.endswith('tbl'):# 保存之前的文本if current_content:documents.extend(self._chunk_text('\n'.join(current_content),metadata={"source": file_path.name,"section": current_section,"type": "text"}))current_content = []# 提取表格table_data = self._extract_word_table(element)table_docs = self._process_table(table_data,metadata={"source": file_path.name,"section": current_section,"type": "table"})documents.extend(table_docs)# 处理最后的内容if current_content:documents.extend(self._chunk_text('\n'.join(current_content),metadata={"source": file_path.name,"section": current_section,"type": "text"}))return documentsdef _extract_word_table(self, table_element) -> List[List]:"""从Word中提取表格数据"""from docx.table import Tabletable = Table(table_element, None)data = []for row in table.rows:row_data = [cell.text.strip() for cell in row.cells]data.append(row_data)return data
📈 六、Excel处理
def process_excel(self, file_path: Path) -> List[LangChainDocument]:"""处理Excel - 支持多Sheet"""documents = []# 读取所有sheetexcel_file = pd.ExcelFile(file_path)for sheet_name in excel_file.sheet_names:logger.info(f"处理Excel Sheet: {sheet_name}")df = pd.read_excel(file_path, sheet_name=sheet_name)# 跳过空sheetif df.empty:continue# === 策略1：整个Sheet作为一个表格 ===if len(df) <= 20: # 小表格table_data = [df.columns.tolist()] + df.values.tolist()table_docs = self._process_table(table_data,metadata={"source": file_path.name,"sheet": sheet_name,"type": "table"})documents.extend(table_docs)# === 策略2：大表格分块处理 ===else:# 每10行一组chunk_size = 10for i in range(0, len(df), chunk_size):chunk_df = df.iloc[i:i+chunk_size]table_data = [df.columns.tolist()] + chunk_df.values.tolist()table_docs = self._process_table(table_data,metadata={"source": file_path.name,"sheet": sheet_name,"type": "table","chunk_start_row": i,"chunk_end_row": min(i+chunk_size, len(df))})documents.extend(table_docs)return documents
🎨 七、图片和多模态内容处理
def _extract_text_from_image(self, image) -> str:"""从图片中提取文字（OCR）"""try:# 使用PaddleOCR（中文效果更好）result = self.ocr.ocr(image, cls=True)if result and result[0]:text = '\n'.join([line[1][0] for line in result[0]])return textexcept Exception as e:logger.warning(f"OCR失败: {e}")return ""def process_image_with_multimodal(self, image_path: str) -> Dict:"""使用多模态模型理解图片（可选，高级功能）适用场景：- 课本插图需要理解图意- 流程图、架构图等- 包含关键信息的图表"""from transformers import Qwen2VLForConditionalGenerationmodel = Qwen2VLForConditionalGeneration.from_pretrained("Qwen/Qwen2-VL-7B")prompt = """请详细描述这张图片的内容，包括：1. 图片类型（插图/流程图/表格/公式等）2. 主要内容和信息3. 图中的文字4. 这张图想要表达的核心概念请用教育性的语言描述，便于学生理解。"""response = model.generate(image_path, prompt)return {"description": response,"ocr_text": self._extract_text_from_image(image_path),"image_path": image_path}
✂️ 八、智能分块策略（Chunking）
为什么分块很重要？
❌ 糟糕的分块：
Chunk 1: "光合作用是植物利用光能，将二氧化碳和"
Chunk 2: "水转化为有机物的过程。这个过程需要三个"
→ 语义被破坏，检索效果差✅ 好的分块：
Chunk 1: "光合作用是植物利用光能，将二氧化碳和水转化为有机物的过程。"
Chunk 2: "光合作用需要三个条件：光照、叶绿素和原料（二氧化碳和水）。"
→ 语义完整，检索准确
智能分块实现
def _chunk_text(self, text: str, metadata: Dict) -> List[LangChainDocument]:"""智能分块 - 保持语义完整性"""# 基础分块chunks = self.text_splitter.split_text(text)# 增强元数据documents = []for i, chunk in enumerate(chunks):# 提取关键信息keywords = self._extract_keywords(chunk)enhanced_metadata = {**metadata,"chunk_index": i,"total_chunks": len(chunks),"keywords": keywords,"char_count": len(chunk),"has_formula": self._contains_formula(chunk),"has_number": self._contains_number(chunk),}documents.append(LangChainDocument(page_content=chunk,metadata=enhanced_metadata))return documentsdef _extract_keywords(self, text: str, top_k: int = 5) -> List[str]:"""提取关键词"""import jieba.analysekeywords = jieba.analyse.extract_tags(text, topK=top_k, withWeight=False)return keywordsdef _contains_formula(self, text: str) -> bool:"""检测是否包含公式"""# 简单检测：包含化学符号或数学符号formula_indicators = ['=', '→', '↓', '↑', '₂', '²', '³', '+', '-', '×', '÷']return any(indicator in text for indicator in formula_indicators)def _contains_number(self, text: str) -> bool:"""检测是否包含数字"""import rereturn bool(re.search(r'\d', text))
🗄️ 九、向量化与存储
完整的向量化流程
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chromaclass VectorStoreManager:"""向量数据库管理器"""def __init__(self, persist_directory: str = "./chroma_db"):# 使用BGE中文嵌入模型self.embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5",model_kwargs={'device': 'cuda'}, # 或 'cpu'encode_kwargs={'normalize_embeddings': True})self.vectorstore = Chroma(persist_directory=persist_directory,embedding_function=self.embeddings)def add_documents(self, documents: List[LangChainDocument], batch_size: int = 100):"""批量添加文档到向量数据库"""total = len(documents)logger.info(f"开始向量化 {total} 个文档")for i in range(0, total, batch_size):batch = documents[i:i+batch_size]# 向量化并存储self.vectorstore.add_documents(batch)logger.info(f"已处理 {min(i+batch_size, total)}/{total}")# 持久化self.vectorstore.persist()logger.info("向量化完成")def search(self, query: str, k: int = 3, filter_dict: Dict = None):"""检索相关文档"""if filter_dict:# 带过滤的检索results = self.vectorstore.similarity_search(query, k=k,filter=filter_dict)else:results = self.vectorstore.similarity_search(query, k=k)return results
🚀 十、完整的使用示例
实战代码
def main():"""完整的文档处理流程"""# 1. 初始化处理器processor = UniversalDocumentProcessor()vector_manager = VectorStoreManager()# 2. 处理文档目录document_folder = Path("./textbooks")all_documents = []for file_path in document_folder.glob("*"):if file_path.suffix.lower() in ['.pdf', '.docx', '.xlsx', '.pptx']:try:logger.info(f"\n{'='*60}")logger.info(f"处理文件: {file_path.name}")logger.info(f"{'='*60}")# 处理文档docs = processor.process_document(file_path)all_documents.extend(docs)logger.info(f"✅ {file_path.name} 处理完成，生成 {len(docs)} 个文档块")except Exception as e:logger.error(f"❌ 处理失败: {file_path.name}, 错误: {e}")continue# 3. 向量化并存储logger.info(f"\n{'='*60}")logger.info(f"开始向量化，总文档数: {len(all_documents)}")logger.info(f"{'='*60}")vector_manager.add_documents(all_documents)# 4. 测试检索logger.info(f"\n{'='*60}")logger.info("测试检索功能")logger.info(f"{'='*60}")test_queries = ["什么是光合作用？","语文多少分及格？","细胞的结构有哪些？"]for query in test_queries:logger.info(f"\n问题: {query}")results = vector_manager.search(query, k=2)for i, doc in enumerate(results, 1):logger.info(f" 结果{i}:")logger.info(f" 内容: {doc.page_content[:100]}...")logger.info(f" 来源: {doc.metadata.get('source')}")logger.info(f" 类型: {doc.metadata.get('type')}")logger.info("\n✅ 全部完成！")if __name__ == "__main__":main()
运行结果示例
============================================================
处理文件: 初中生物九年级.pdf
============================================================
INFO:__main__:处理PDF第 1/120 页
INFO:__main__:处理PDF第 2/120 页
...
INFO:__main__:发现表格，使用表格处理策略
INFO:__main__:生成自然语言描述：这是一个关于科目的表格...
✅ 初中生物九年级.pdf 处理完成，生成 458 个文档块============================================================
开始向量化，总文档数: 1234
============================================================
INFO:__main__:已处理 100/1234
INFO:__main__:已处理 200/1234
...
INFO:__main__:向量化完成============================================================
测试检索功能
============================================================问题: 语文多少分及格？结果1:内容: 这是一个关于科目的表格，包含3行数据。语文的及格分数是60；语文的优秀分数是90...来源: 成绩标准.xlsx类型: table结果2:内容: 语文，及格分数：60, 优秀分数：90, 满分：100来源: 成绩标准.xlsx类型: table✅ 全部完成！
📋 十一、最佳实践检查清单
文档处理清单
□ PDF处理✓ 支持文字版PDF提取✓ 支持扫描版OCR识别✓ 表格单独处理✓ 图片文字提取□ Word处理✓ 段落提取✓ 表格提取✓ 标题识别✓ 保留章节结构□ Excel处理✓ 多Sheet支持✓ 表格转自然语言✓ 大表格分块✓ 数值统计□ 通用处理✓ 元数据完整✓ 关键词提取✓ 公式识别✓ 图片描述
性能优化建议
# 1. 批量处理
batch_size = 100 # 一次处理100个文档块# 2. 并行处理
from concurrent.futures import ThreadPoolExecutorwith ThreadPoolExecutor(max_workers=4) as executor:results = executor.map(processor.process_document, file_list)# 3. 缓存重复计算
@lru_cache(maxsize=1000)
def extract_keywords(text):# 缓存关键词提取结果pass# 4. 增量更新
# 只处理新文档或修改过的文档
def process_incremental(files, db):for file in files:if file.mtime > db.get_last_update(file):# 只处理修改过的process_document(file)
🎯 十二、常见问题解决
Q1: 表格检索不准确？
# 解决方案：多表示策略
# 同时存储3种形式# 1. 原始Markdown（精确匹配）
# 2. 自然语言（语义检索）⭐
# 3. 逐行拆分（细粒度检索）
Q2: 公式向量化效果差？
# 解决方案：公式 + 文字描述def process_formula(formula):return {"original": "6CO₂ + 6H₂O → C₆H₁₂O₆ + 6O₂","description": "六分子二氧化碳加六分子水生成一分子葡萄糖和六分子氧气","keywords": ["二氧化碳", "水", "葡萄糖", "氧气"]}
Q3: 图片信息丢失？
# 解决方案：OCR + 多模态模型 + 人工标注# 自动：OCR提取文字
# 自动：多模态模型生成描述
# 人工：关键图片手动标注（最准确）
Q4: 大文件处理慢？
# 解决方案：
# 1. 流式处理（不要一次加载全部）
# 2. 并行处理
# 3. GPU加速向量化
# 4. 分批存储