第2章：数据准备与处理

2.1 文档加载器设计与选择

文档加载器是RAG系统的数据入口，负责从各种数据源读取和解析文档。一个优秀的文档加载器需要具备高效性、稳定性和可扩展性。

文档加载器核心功能

1. 多格式支持

• PDF文档：技术文档、报告、论文
• Word文档：办公文档、手册
• HTML页面：网页内容、技术博客
• Markdown文件：技术文档、笔记
• 纯文本文件：日志文件、配置文件
• 数据库内容：结构化数据查询

2. 解析能力

• 文本提取：准确获取文档内容
• 格式保留：保持原文档结构
• 元数据提取：标题、作者、创建时间等
• 错误处理：异常格式容错处理

3. 性能优化

• 批量处理：支持多文件并行处理
• 缓存机制：避免重复解析
• 增量更新：支持文档变化检测
• 内存管理：大数据量内存优化

主流加载器对比

1. LangChain加载器

PyPDFLoader

from langchain.document_loaders import PyPDFLoader

loader = PyPDFLoader("document.pdf")
documents = loader.load()

# 优点：轻量级、简单易用
# 缺点：不支持复杂表格、OCR能力有限
# 适用场景：简单PDF文档、技术手册
​

PyMuPDFLoader

from langchain.document_loaders import PyMuPDFLoader

loader = PyMuPDFLoader("document.pdf")
documents = loader.load()

# 优点：性能高、支持复杂格式
# 缺点：依赖较多、内存占用大
# 适用场景：大型PDF、包含表格的文档
​

TextLoader

from langchain.document_loaders import TextLoader

loader = TextLoader("document.txt")
documents = loader.load()

# 优点：简单高效
# 缺点：格式单一
# 适用场景：纯文本文件、日志文件
​

2. 专业加载器

PDFPlumberLoader

from langchain.document_loaders import PDFPlumberLoader

loader = PDFPlumberLoader("document.pdf")
documents = loader.load()

# 优点：表格处理能力强、文本质量高
# 缺点：速度相对较慢
# 适用场景：包含表格的PDF文档
​

UnstructuredLoader

from langchain.document_loaders import UnstructuredLoader

loader = UnstructuredLoader("document.html")
documents = loader.load()

# 优点：支持多种格式、智能分割
# 缺点：需要额外依赖
# 适用场景：混合格式文档
​

加载器选择指南

选择标准

场景化选择

1. 企业文档环境

推荐配置:
  加载器: PDFPlumberLoader + python-docx加载器
  特点: 企业级稳定、支持办公格式
  适用: 内部文档、报告、手册
  优势: 格式兼容性好、处理稳定
​

2. 技术文档环境

推荐配置:
  加载器: PyMuPDFLoader + MarkdownLoader
  特点: 技术文档优化、代码支持
  适用: API文档、技术博客、代码注释
  优势: 代码高亮、结构保留
​

3. 多源数据环境

推荐配置:
  加载器: UnstructuredLoader + 自定义加载器
  特点: 格式多样、智能处理
  适用: 混合格式、网络爬取
  优势: 自动识别、批量处理
​

最佳实践

1. 性能优化

批量加载

from langchain.document_loaders import DirectoryLoader

# 批量加载PDF文件
loader = DirectoryLoader(
    './documents',
    glob='**/*.pdf',
    loader_cls=PyPDFLoader,
    recursive=True,
    show_progress=True
)
documents = loader.load()
​

并行处理

from concurrent.futures import ThreadPoolExecutor
from langchain.document_loaders import PyPDFLoader

def load_pdf_batch(file_paths):
    """批量加载PDF文件"""
    documents = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        loaders = [PyPDFLoader(path) for path in file_paths]
        results = executor.map(lambda loader: loader.load(), loaders)
        documents.extend([doc for result in results for doc in result])
    return documents
​

2. 错误处理

异常捕获

import logging
from langchain.document_loaders import PyPDFLoader

def safe_load_document(file_path):
    """安全加载文档"""
    try:
        loader = PyPDFLoader(file_path)
        documents = loader.load()
        return documents
    except Exception as e:
        logging.error(f"加载文档失败 {file_path}: {str(e)}")
        return []
​

3. 元数据管理

元数据提取

import os
from datetime import datetime

def extract_metadata(file_path):
    """提取文档元数据"""
    stat = os.stat(file_path)
    
    metadata = {
        'source': file_path,
        'file_name': os.path.basename(file_path),
        'file_size': stat.st_size,
        'created_at': datetime.fromtimestamp(stat.st_ctime),
        'modified_at': datetime.fromtimestamp(stat.st_mtime),
        'document_type': os.path.splitext(file_path)[1].lower(),
        'loading_timestamp': datetime.now()
    }
    
    return metadata
​

元数据增强

def enhance_documents_metadata(documents, source_metadata):
    """增强文档元数据"""
    for doc in documents:
        doc.metadata.update(source_metadata)
        doc.metadata['chunk_id'] = hash(doc.page_content) % 10000
        doc.metadata['content_length'] = len(doc.page_content)
    
    return documents
​

实际应用示例

1. 企业知识库加载器

class EnterpriseKnowledgeLoader:
    """企业知识库文档加载器"""
    
    def __init__(self, base_path: str):
        self.base_path = base_path
        self.logger = logging.getLogger(__name__)
    
    def load_all_documents(self):
        """加载所有文档"""
        all_documents = []
        
        # 定义加载器映射
        loader_mapping = {
            '.pdf': PDFPlumberLoader,
            '.docx': self._load_word_document,
            '.txt': TextLoader,
            '.md': MarkdownLoader,
            '.html': BSHTMLLoader
        }
        
        for root, dirs, files in os.walk(self.base_path):
            for file in files:
                file_path = os.path.join(root, file)
                file_extension = os.path.splitext(file)[1].lower()
                
                if file_extension in loader_mapping:
                    try:
                        loader_class = loader_mapping[file_extension]
                        documents = self._load_single_document(loader_class, file_path)
                        all_documents.extend(documents)
                    except Exception as e:
                        self.logger.error(f"加载文档失败 {file_path}: {str(e)}")
        
        return all_documents
    
    def _load_single_document(self, loader_class, file_path):
        """加载单个文档"""
        metadata = extract_metadata(file_path)
        
        if loader_class == self._load_word_document:
            documents = self._load_word_document(file_path)
        else:
            loader = loader_class(file_path)
            documents = loader.load()
        
        # 增强元数据
        documents = enhance_documents_metadata(documents, metadata)
        return documents
​

2.2 文本预处理技术

文本预处理是RAG系统数据质量的关键环节。高质量的预处理能够显著提升检索准确性和生成质量。

预处理流程概述

文本清洗

1. 基础清洗

去除特殊字符

import re

def clean_special_characters(text: str) -> str:
    """去除特殊字符"""
    # 保留中文、英文、数字、基本标点
    pattern = r'[^\u4e00-\u9fa5a-zA-Z0-9\s，。！？；：""''（）【】《》、…—]'
    cleaned_text = re.sub(pattern, '', text)
    return cleaned_text

# 使用示例
text = "这是一个测试文本！包含特殊字符# $ % & * ( )。"
cleaned = clean_special_characters(text)
print(cleaned)  # "这是一个测试文本包含特殊字符。"
​

去除HTML标签

from bs4 import BeautifulSoup

def remove_html_tags(text: str) -> str:
    """去除HTML标签"""
    soup = BeautifulSoup(text, 'html.parser')
    return soup.get_text()

# 使用示例
html_text = "<p>这是一个<b>HTML</b>文本。</p>"
cleaned = remove_html_tags(html_text)
print(cleaned)  # "这是一个HTML文本。"
​

去除多余空格

import re

def normalize_whitespace(text: str) -> str:
    """标准化空格"""
    # 去除多余空格，保留单个空格
    text = re.sub(r'\s+', ' ', text)
    # 去除行首行尾空格
    text = text.strip()
    return text

# 使用示例
text = "这是一个   文本   包含   多个   空格  "
cleaned = normalize_whitespace(text)
print(cleaned)  # "这是一个文本包含多个空格"
​

文本标准化

1. 字符标准化

繁简转换

# 需要安装opencc库
# pip install opencc

from opencc import OpenCC

def simplified_to_traditional(text: str) -> str:
    """简体转繁体"""
    converter = OpenCC('s2t')
    return converter.convert(text)

def traditional_to_simplified(text: str) -> str:
    """繁体转简体"""
    converter = OpenCC('t2s')
    return converter.convert(text)

# 使用示例
simplified_text = "这是一个简体中文文本"
traditional_text = simplified_to_traditional(simplified_text)
print(traditional_text)  # "這是一個簡體中文文本"
​

大小写标准化

def normalize_case(text: str, language: str = 'zh') -> str:
    """文本大小写标准化"""
    if language == 'zh':
        # 中文不需要处理大小写
        return text
    elif language == 'en':
        # 英文统一转为小写
        return text.lower()
    else:
        return text

# 使用示例
english_text = "This is an English TEXT"
normalized = normalize_case(english_text, 'en')
print(normalized)  # "this is an english text"
​

分词处理

1. 中文分词

jieba分词

import jieba

def chinese_word_segmentation(text: str, method: str = 'default') -> list:
    """中文分词"""
    if method == 'default':
        words = jieba.lcut(text)
    elif method == 'accurate':
        words = jieba.lcut(text, cut_all=False)
    elif method == 'full':
        words = jieba.lcut(text, cut_all=True)
    elif method == 'search':
        words = jieba.lcut_for_search(text)
    else:
        words = jieba.lcut(text)
    
    return words

# 使用示例
text = "这是一个中文分词的示例"
words = chinese_word_segmentation(text)
print(words)  # ['这是', '一个', '中文', '分词', '的', '示例']
​

停用词过滤

1. 停用词表准备

def load_stopwords(file_path: str = None) -> set:
    """加载停用词表"""
    default_stopwords = {
        # 中文停用词
        '的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个',
        '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好',
        '自己', '这', '那', '现在', '可以', '但是', '还是', '因为', '什么', '如果',
        # 英文停用词
        'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of',
        'with', 'by', 'from', 'up', 'about', 'into', 'through', 'during', 'before',
        'after', 'above', 'below', 'between', 'among', 'is', 'are', 'was', 'were',
        'be', 'been', 'being', 'have', 'has', 'had', 'having', 'do', 'does', 'did',
        'doing', 'will', 'would', 'shall', 'should', 'may', 'might', 'must', 'can',
        'could', 'i', 'you', 'he', 'she', 'it', 'we', 'they', 'them', 'their', 'what',
        'which', 'who', 'when', 'where', 'why', 'how', 'all', 'each', 'every', 'some',
        'any', 'no', 'not', 'only', 'own', 'same', 'so', 'than', 'too', 'very'
    }
    
    if file_path:
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                custom_stopwords = set(line.strip() for line in f)
                default_stopwords.update(custom_stopwords)
        except FileNotFoundError:
            print(f"停用词文件不存在: {file_path}")
    
    return default_stopwords
​

2. 停用词过滤

def remove_stopwords(words: list, stopwords: set = None, custom_words: list = None) -> list:
    """过滤停用词"""
    if stopwords is None:
        stopwords = load_stopwords()
    
    # 添加自定义停用词
    if custom_words:
        stopwords.update(custom_words)
    
    # 过滤停用词
    filtered_words = [word for word in words if word.lower() not in stopwords]
    
    return filtered_words

# 使用示例
words = ['这是', '一个', '测试', '的', '示例', 'test', 'example']
filtered_words = remove_stopwords(words)
print(filtered_words)  # ['测试', '示例', 'test', 'example']
​

预处理管道

1. 完整预处理管道

class TextPreprocessor:
    """文本预处理管道"""
    
    def __init__(self, config: dict = None):
        self.config = config or self.default_config
        self.stopwords = load_stopwords()
        
    @property
    def default_config(self):
        return {
            'remove_special_chars': True,
            'normalize_whitespace': True,
            'normalize_punctuation': True,
            'segmentation': True,
            'remove_stopwords': True,
            'lemmatization': True,
            'min_word_length': 2,
            'max_word_length': 50,
            'custom_stopwords': []
        }
    
    def preprocess_text(self, text: str) -> list:
        """预处理单个文本"""
        original_text = text
        
        # 1. 文本清洗
        if self.config['remove_special_chars']:
            text = clean_special_characters(text)
        
        if self.config['normalize_whitespace']:
            text = normalize_whitespace(text)
        
        if self.config['normalize_punctuation']:
            text = normalize_punctuation(text)
        
        # 2. 分词
        if self.config['segmentation']:
            if self._is_chinese_text(text):
                words = chinese_word_segmentation(text)
            else:
                words = english_word_segmentation(text)
        else:
            words = [text]
        
        # 3. 过滤停用词
        if self.config['remove_stopwords']:
            words = remove_stopwords(words, self.stopwords, self.config['custom_stopwords'])
        
        # 4. 过滤长度
        words = [word for word in words 
                if len(word) >= self.config['min_word_length'] 
                and len(word) <= self.config['max_word_length']]
        
        return words
    
    def _is_chinese_text(self, text: str) -> bool:
        """判断是否为中文文本"""
        chinese_chars = len(re.findall(r'[\u4e00-\u9fa5]', text))
        total_chars = len(text)
        return chinese_chars / total_chars > 0.3
    
    def preprocess_batch(self, texts: list) -> list:
        """批量预处理文本"""
        return [self.preprocess_text(text) for text in texts]
​

实际应用示例

1. 企业文档预处理系统

class EnterpriseDocumentPreprocessor:
    """企业文档预处理系统"""
    
    def __init__(self, base_path: str, config: dict = None):
        self.base_path = base_path
        self.preprocessor = TextPreprocessor(config)
        self.logger = logging.getLogger(__name__)
    
    def process_all_documents(self) -> list:
        """处理所有文档"""
        processed_documents = []
        
        for root, dirs, files in os.walk(self.base_path):
            for file in files:
                file_path = os.path.join(root, file)
                
                try:
                    # 加载文档
                    documents = self._load_document(file_path)
                    
                    # 预处理每个文档
                    for doc in documents:
                        processed_text = self.preprocessor.preprocess_text(doc.page_content)
                        doc.page_content = ' '.join(processed_text)
                        processed_documents.append(doc)
                        
                except Exception as e:
                    self.logger.error(f"处理文档失败 {file_path}: {str(e)}")
        
        return processed_documents
    
    def _load_document(self, file_path: str) -> list:
        """加载文档"""
        file_extension = os.path.splitext(file_path)[1].lower()
        
        if file_extension == '.pdf':
            loader = PyPDFLoader(file_path)
        elif file_extension == '.txt':
            loader = TextLoader(file_path)
        else:
            raise ValueError(f"不支持的文件格式: {file_extension}")
        
        return loader.load()
​

总结

文本预处理是RAG系统质量保障的关键环节。通过系统化的预处理流程，可以显著提升后续检索和生成的质量。主要要点包括：

1. 文档加载：选择合适的加载器确保文档格式支持
2. 文本清洗：去除噪声和无关信息
3. 文本标准化：统一格式和编码
4. 分词处理：准确切分文本
5. 停用词过滤：去除无意义词汇
6. 特征提取：提取重要特征

在实际应用中，建议根据具体需求调整预处理参数，并通过持续优化提升预处理效果。预处理的质量直接关系到RAG系统的性能和用户体验，需要给予足够的重视和投入。