自有文档构建 RAG 与微调数据集：Word/Excel/PPT 数据处理方案

1. 分类 (Classification)

遍历输入目录，根据文件扩展名将文件分为文本类（Word）、表格类（Excel）、演示文稿类（PPT）及图像类（JPG/PNG）。

2. 分别处理 (Extraction)

针对不同类别调用专用解析工具：

• 文本：直接读取内容，按语义或固定长度分块。
• 表格：解析为 CSV 或 DataFrame 形式，转为纯文本描述。
• 图像：使用 OCR 技术提取文字，保留坐标信息。

3. 合并到 JSON (Merging)

将同一文件的所有解析结果合并为一个 JSON 对象列表，确保每个块都有唯一的 ID 和来源标记。

4. 制作数据集 (Dataset Construction)

对合并后的数据进行清洗、去重，并按比例划分为训练集、验证集和测试集。

环境准备

建议使用 Linux 或 WSL 环境，Python 版本推荐 3.10。

1. 创建 Conda 环境

   conda create -n rag_data python=3.10
   conda activate rag_data
   

2. 安装依赖库 主要依赖包括文档解析库、OCR 库及数据处理库：

   pip install python-docx openpyxl pdfplumber pytesseract pandas numpy
   

关键代码实现

以下是一个通用的处理逻辑示例，展示了如何实现上述流程。实际项目中可根据具体需求封装为类或模块。

1. 文件分类与遍历

import os
from pathlib import Path

def get_file_types(directory):
    extensions = {
        'text': ['.docx', '.txt'],
        'table': ['.xlsx', '.xls', '.csv'],
        'ppt': ['.pptx'],
        'image': ['.jpg', '.png', '.pdf']
    }
    files = {'text': [], 'table': [], 'ppt': [], 'image': []}
    
    for root, _, filenames in os.walk(directory):
        for filename in filenames:
            ext = Path(filename).suffix.lower()
            for ftype, exs in extensions.items():
                if ext in exs:
                    files[ftype].append(os.path.join(root, filename))
                    break
    return files

2. 内容提取与分块

文本处理

对于 Word 文档，使用 python-docx 提取段落，并进行智能分块（Chunking），避免切断句子。

from docx import Document

def extract_text_from_docx(file_path, chunk_size=400, max_chunk_size=600):
    doc = Document(file_path)
    full_text = []
    for para in doc.paragraphs:
        text = para.text.strip()
        if text:
            full_text.append(text)
    
    # 简单分块逻辑，实际可结合 NLP 模型进行语义切分
    chunks = []
    current_chunk = ""
    for line in full_text:
        if len(current_chunk) + len(line) <= max_chunk_size:
            current_chunk += line + "\n"
        else:
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = line + "\n"
    if current_chunk:
        chunks.append(current_chunk)
    return chunks

表格处理

Excel 文件需转换为文本描述，保留行列关系。

import pandas as pd

def extract_tables_from_excel(file_path):
    tables = []
    xls = pd.ExcelFile(file_path)
    for sheet_name in xls.sheet_names:
        df = pd.read_excel(xls, sheet_name=sheet_name)
        # 将 DataFrame 转为文本行
        for idx, row in df.iterrows():
            tables.append(" | ".join([str(v) for v in row]))
    return tables

图像 OCR 处理

对于 PPT 中的截图或 JPG 图片，使用 OCR 提取文字。

import pytesseract
from PIL import Image

def run_ocr_on_image(image_path):
    img = Image.open(image_path)
    try:
        text = pytesseract.image_to_string(img, lang='chi_sim+eng')
        return text.strip()
    except Exception as e:
        print(f"OCR failed for {image_path}: {e}")
        return ""

3. 数据合并与导出

将所有提取的内容组装成字典列表，并保存为 JSONL 文件。

import json
import uuid

def build_dataset(files_dict, output_path):
    dataset = []
    block_id = 0
    
    for ftype, paths in files_dict.items():
        for path in paths:
            file_name = os.path.basename(path)
            content_list = []
            
            if ftype == 'text':
                content_list = extract_text_from_docx(path)
            elif ftype == 'table':
                content_list = extract_tables_from_excel(path)
            elif ftype == 'image':
                content_list = [run_ocr_on_image(path)]
            
            for content in content_list:
                if not content:
                    continue
                record = {
                    "block_ID": str(uuid.uuid4()),
                    "doc_ID": os.path.splitext(file_name)[0],
                    "content_type": ftype,
                    "file_type": Path(path).suffix,
                    "file_source": file_name,
                    "text": content
                }
                dataset.append(record)
                block_id += 1
    
    with open(output_path, 'w', encoding='utf-8') as f:
        for item in dataset:
            f.write(json.dumps(item, ensure_ascii=False) + '\n')
    
    print(f"Dataset saved to {output_path}, total records: {len(dataset)}")

4. 数据集划分

使用 Datasets 类或手动划分来生成训练集、验证集和测试集。

import random

def split_dataset(data, test_ratio=0.1, val_ratio=0.1):
    random.shuffle(data)
    n = len(data)
    test_end = int(n * test_ratio)
    val_end = int(n * (test_ratio + val_ratio))
    
    test_set = data[:test_end]
    val_set = data[test_end:val_end]
    train_set = data[val_end:]
    
    return train_set, val_set, test_set

最佳实践建议

1. 元数据完整性：尽量保留文件的创建时间、修改者等信息，有助于后续溯源。
2. 分块策略：RAG 场景下，分块大小建议在 400-600 tokens 之间，并设置重叠区（Overlap）以保证上下文连贯。
3. 质量校验：定期抽检生成的 JSONL 文件，检查是否有乱码或截断情况。
4. 隐私脱敏：若文档包含敏感信息，需在提取前进行正则替换或脱敏处理。

总结

通过自动化脚本将异构办公文档转化为标准化的 JSONL 数据集，可以显著提升 RAG 系统的召回率和微调模型的效果。本方案提供了从环境搭建、文件分类、内容解析到数据集生成的完整流程，开发者可根据实际业务需求调整解析逻辑和分块参数。掌握这一数据处理能力，是构建高质量 AI 应用的基础。