在 RAG 应用中，数据清洗和预处理是提升检索精度的关键环节。以下是详细的最佳实践：

1. 数据清洗流程

1.1 基础清洗

import re
from typing import List, Dict
import html

def basic_cleaning(text: str) -> str:
    """基础文本清洗"""
    # 去除HTML标签
    text = re.sub(r'<[^>]+>', '', text)
    # 解码HTML实体
    text = html.unescape(text)
    # 去除多余空白
    text = re.sub(r'\s+', ' ', text)
    # 去除特殊字符（保留中文、英文、数字、基本标点）
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s.,!?;:()""''【】（）。，！？；：]', '', text)
    # 去除控制字符
    text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
    
    return text.strip()

1.2 去除噪声

def remove_noise(text: str) -> str:
    """去除噪声内容"""
    noise_patterns = [
        r'Copyright\s+\d{4}.*',  # 版权信息
        r'All rights reserved',   # 版权声明
        r'点击这里.*',            # 点击提示
        r'更多内容.*',            # 更多提示
        r'广告',                  # 广告
        r'推荐阅读',              # 推荐内容
        r'相关文章',              # 相关文章
        r'分享到.*',              # 分享按钮
        r'上一篇|下一篇',         # 导航链接
        r'返回.*',                # 返回链接
    ]
    
    for pattern in noise_patterns:
        text = re.sub(pattern, '', text, flags=re.IGNORECASE)
    
    return text

1.3 格式标准化

def normalize_format(text: str) -> str:
    """格式标准化"""
    # 统一引号
    text = text.replace('"', '"').replace('"', '"')
    text = text.replace(''', "'").replace(''', "'")
    
    # 统一省略号
    text = re.sub(r'\.{2,}', '……', text)
    
    # 统一破折号
    text = re.sub(r'[-—]{2,}', '——', text)
    
    # 数字标准化（中文数字转阿拉伯数字）
    chinese_nums = {'零': '0', '一': '1', '二': '2', '三': '3', '四': '4',
                    '五': '5', '六': '6', '七': '7', '八': '8', '九': '9'}
    for cn, num in chinese_nums.items():
        text = text.replace(cn, num)
    
    return text

2. 文档结构化处理

2.1 提取结构化信息

from bs4 import BeautifulSoup
import pdfplumber
from docx import Document

def extract_structure(file_path: str, file_type: str) -> Dict:
    """提取文档结构"""
    structure = {
        'title': '',
        'headings': [],
        'paragraphs': [],
        'tables': [],
        'lists': [],
        'metadata': {}
    }
    
    if file_type == 'html':
        structure = extract_html_structure(file_path)
    elif file_type == 'pdf':
        structure = extract_pdf_structure(file_path)
    elif file_type == 'docx':
        structure = extract_docx_structure(file_path)
    
    return structure

def extract_html_structure(file_path: str) -> Dict:
    """提取HTML文档结构"""
    with open(file_path, 'r', encoding='utf-8') as f:
        soup = BeautifulSoup(f.read(), 'html.parser')
    
    structure = {
        'title': soup.title.string if soup.title else '',
        'headings': [],
        'paragraphs': [],
        'tables': [],
        'lists': []
    }
    
    # 提取标题
    for i in range(1, 7):
        for heading in soup.find_all(f'h{i}'):
            structure['headings'].append({
                'level': i,
                'text': heading.get_text().strip()
            })
    
    # 提取段落
    for p in soup.find_all('p'):
        text = p.get_text().strip()
        if text:
            structure['paragraphs'].append(text)
    
    # 提取表格
    for table in soup.find_all('table'):
        table_data = []
        for row in table.find_all('tr'):
            row_data = [cell.get_text().strip() for cell in row.find_all(['td', 'th'])]
            if row_data:
                table_data.append(row_data)
        if table_data:
            structure['tables'].append(table_data)
    
    return structure

2.2 保留元数据

def enrich_with_metadata(text: str, metadata: Dict) -> Dict:
    """为文本添加元数据"""
    enriched = {
        'text': text,
        'metadata': {
            'source': metadata.get('source', ''),
            'author': metadata.get('author', ''),
            'publish_date': metadata.get('publish_date', ''),
            'category': metadata.get('category', ''),
            'tags': metadata.get('tags', []),
            'language': detect_language(text),
            'length': len(text),
            'word_count': len(text.split())
        }
    }
    return enriched

def detect_language(text: str) -> str:
    """检测文本语言"""
    # 简单实现，实际可使用 langdetect 等库
    chinese_chars = len(re.findall(r'[\u4e00-\u9fa5]', text))
    english_chars = len(re.findall(r'[a-zA-Z]', text))
    
    if chinese_chars > english_chars:
        return 'zh'
    elif english_chars > chinese_chars:
        return 'en'
    else:
        return 'mixed'

3. 智能分块策略

3.1 语义分块

import nltk
from typing import List
import numpy as np
from sentence_transformers import SentenceTransformer

class SemanticChunker:
    def __init__(self, model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2'):
        self.model = SentenceTransformer(model_name)
    
    def semantic_chunk(self, text: str, max_chunk_size: int = 512, 
                      similarity_threshold: float = 0.7) -> List[str]:
        """基于语义相似度的智能分块"""
        # 先按句子分割
        sentences = self._split_sentences(text)
        
        if len(sentences) <= 1:
            return [text]
        
        # 计算句子嵌入
        embeddings = self.model.encode(sentences)
        
        # 计算相邻句子的相似度
        similarities = self._calculate_similarities(embeddings)
        
        # 根据相似度确定分割点
        chunks = self._create_chunks(sentences, similarities, 
                                    max_chunk_size, similarity_threshold)
        
        return chunks
    
    def _split_sentences(self, text: str) -> List[str]:
        """分割句子"""
        # 使用nltk或自定义规则
        sentences = re.split(r'[。！？.!?]+', text)
        return [s.strip() for s in sentences if s.strip()]
    
    def _calculate_similarities(self, embeddings: np.ndarray) -> List[float]:
        """计算相邻句子相似度"""
        similarities = []
        for i in range(len(embeddings) - 1):
            sim = np.dot(embeddings[i], embeddings[i+1]) / (
                np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i+1])
            )
            similarities.append(sim)
        return similarities
    
    def _create_chunks(self, sentences: List[str], similarities: List[float],
                      max_size: int, threshold: float) -> List[str]:
        """创建chunks"""
        chunks = []
        current_chunk = []
        current_length = 0
        
        for i, sentence in enumerate(sentences):
            # 检查是否应该开始新的chunk
            should_split = False
            
            if current_length + len(sentence) > max_size:
                should_split = True
            elif i > 0 and similarities[i-1] < threshold:
                should_split = True
            
            if should_split and current_chunk:
                chunks.append(''.join(current_chunk))
                current_chunk = []
                current_length = 0
            
            current_chunk.append(sentence)
            current_length += len(sentence)
        
        if current_chunk:
            chunks.append(''.join(current_chunk))
        
        return chunks

3.2 递归分块

class RecursiveChunker:
    def __init__(self, separators: List[str] = None):
        self.separators = separators or [
            '\n\n',  # 段落
            '\n',    # 行
            '。',    # 中文句号
            '！',    # 中文感叹号
            '？',    # 中文问号
            '.',     # 英文句号
            '!',     # 英文感叹号
            '?',     # 英文问号
            '，',    # 中文逗号
            ',',     # 英文逗号
            ' '      # 空格
        ]
    
    def recursive_split(self, text: str, max_size: int = 512, 
                       overlap: int = 50) -> List[str]:
        """递归分块"""
        return self._split_recursive(text, max_size, overlap, 0)
    
    def _split_recursive(self, text: str, max_size: int, 
                        overlap: int, separator_index: int) -> List[str]:
        """递归分割实现"""
        # 如果文本足够小，直接返回
        if len(text) <= max_size:
            return [text]
        
        # 如果已经尝试了所有分隔符，强制分割
        if separator_index >= len(self.separators):
            return self._force_split(text, max_size, overlap)
        
        separator = self.separators[separator_index]
        
        # 使用当前分隔符分割
        splits = text.split(separator)
        
        # 合并分割结果
        chunks = []
        current_chunk = ""
        
        for split in splits:
            if len(current_chunk) + len(split) <= max_size:
                current_chunk += split + separator
            else:
                if current_chunk:
                    chunks.append(current_chunk.rstrip(separator))
                current_chunk = split + separator
        
        if current_chunk:
            chunks.append(current_chunk.rstrip(separator))
        
        # 检查是否还有过大的chunk
        final_chunks = []
        for chunk in chunks:
            if len(chunk) > max_size:
                # 递归使用下一个分隔符
                sub_chunks = self._split_recursive(
                    chunk, max_size, overlap, separator_index + 1
                )
                final_chunks.extend(sub_chunks)
            else:
                final_chunks.append(chunk)
        
        # 添加重叠
        return self._add_overlap(final_chunks, overlap)
    
    def _force_split(self, text: str, max_size: int, overlap: int) -> List[str]:
        """强制分割"""
        chunks = []
        for i in range(0, len(text), max_size - overlap):
            chunk = text[i:i + max_size]
            chunks.append(chunk)
        return chunks
    
    def _add_overlap(self, chunks: List[str], overlap: int) -> List[str]:
        """添加重叠"""
        if overlap <= 0:
            return chunks
        
        overlapped_chunks = []
        for i, chunk in enumerate(chunks):
            if i > 0:
                # 添加前一个chunk的末尾
                prev_chunk = chunks[i-1]
                overlap_text = prev_chunk[-overlap:] if len(prev_chunk) > overlap else prev_chunk
                chunk = overlap_text + chunk
            overlapped_chunks.append(chunk)
        
        return overlapped_chunks

3.3 结构感知分块

def structure_aware_chunk(structure: Dict, max_chunk_size: int = 512) -> List[Dict]:
    """基于文档结构的分块"""
    chunks = []
    
    # 按标题层级组织内容
    current_section = {
        'heading': '',
        'level': 0,
        'content': '',
        'metadata': {}
    }
    
    # 处理标题和段落
    for heading in structure['headings']:
        # 保存当前section
        if current_section['content']:
            chunks.append(create_chunk(current_section))
        
        # 开始新section
        current_section = {
            'heading': heading['text'],
            'level': heading['level'],
            'content': '',
            'metadata': {
                'heading_level': heading['level'],
                'heading_text': heading['text']
            }
        }
    
    # 添加段落内容
    for paragraph in structure['paragraphs']:
        if len(current_section['content']) + len(paragraph) > max_chunk_size:
            chunks.append(create_chunk(current_section))
            current_section['content'] = paragraph
        else:
            current_section['content'] += paragraph + '\n'
    
    # 保存最后一个section
    if current_section['content']:
        chunks.append(create_chunk(current_section))
    
    return chunks

def create_chunk(section: Dict) -> Dict:
    """创建chunk"""
    chunk_text = section['heading'] + '\n' + section['content'] if section['heading'] else section['content']
    
    return {
        'text': chunk_text.strip(),
        'metadata': section['metadata']
    }

4. 数据增强

4.1 查询扩展

def query_expansion(query: str, synonyms: Dict[str, List[str]] = None) -> List[str]:
    """查询扩展"""
    expanded_queries = [query]
    
    if synonyms:
        for word, syn_list in synonyms.items():
            if word in query:
                for syn in syn_list:
                    expanded_query = query.replace(word, syn)
                    expanded_queries.append(expanded_query)
    
    return expanded_queries

# 同义词词典示例
SYNONYMS = {
    '人工智能': ['AI', '机器智能', '智能计算'],
    '机器学习': ['ML', '自动学习'],
    '深度学习': ['DL', '深层学习'],
    '神经网络': ['神经网', 'NN'],
}

4.2 文档摘要

def generate_summary(text: str, max_length: int = 200) -> str:
    """生成文档摘要"""
    # 简单实现：提取关键句子
    sentences = re.split(r'[。！？.!?]+', text)
    
    # 按句子长度和位置评分
    scored_sentences = []
    for i, sent in enumerate(sentences):
        score = len(sent) * (1 - i / len(sentences))  # 偏向前面的句子
        scored_sentences.append((score, sent))
    
    # 选择得分最高的句子
    scored_sentences.sort(reverse=True)
    summary_sentences = [sent for score, sent in scored_sentences[:3]]
    
    summary = '。'.join(summary_sentences)
    return summary[:max_length]

5. 质量控制

5.1 内容质量评估

def assess_quality(text: str) -> Dict:
    """评估内容质量"""
    quality_metrics = {
        'length_score': 0,
        'readability_score': 0,
        'informativeness_score': 0,
        'overall_score': 0
    }
    
    # 长度评分（适中为佳）
    length = len(text)
    if 100 <= length <= 1000:
        quality_metrics['length_score'] = 1.0
    elif 50 <= length < 100 or 1000 < length <= 2000:
        quality_metrics['length_score'] = 0.7
    else:
        quality_metrics['length_score'] = 0.3
    
    # 可读性评分
    avg_sentence_length = len(text.split()) / max(1, len(re.split(r'[。！？.!?]+', text)))
    if avg_sentence_length < 20:
        quality_metrics['readability_score'] = 1.0
    elif avg_sentence_length < 30:
        quality_metrics['readability_score'] = 0.7
    else:
        quality_metrics['readability_score'] = 0.4
    
    # 信息量评分（基于关键词密度）
    keywords = ['重要', '关键', '核心', '主要', '必须', '应该', '注意']
    keyword_count = sum(1 for kw in keywords if kw in text)
    quality_metrics['informativeness_score'] = min(1.0, keyword_count / 3)
    
    # 综合评分
    quality_metrics['overall_score'] = (
        quality_metrics['length_score'] * 0.3 +
        quality_metrics['readability_score'] * 0.4 +
        quality_metrics['informativeness_score'] * 0.3
    )
    
    return quality_metrics

5.2 去重处理

from difflib import SequenceMatcher

def remove_duplicates(chunks: List[Dict], threshold: float = 0.85) -> List[Dict]:
    """去除重复内容"""
    unique_chunks = []
    
    for chunk in chunks:
        is_duplicate = False
        for existing_chunk in unique_chunks:
            similarity = SequenceMatcher(
                None, chunk['text'], existing_chunk['text']
            ).ratio()
            
            if similarity > threshold:
                is_duplicate = True
                break
        
        if not is_duplicate:
            unique_chunks.append(chunk)
    
    return unique_chunks

6. 完整预处理流程

class RAGPreprocessor:
    def __init__(self):
        self.semantic_chunker = SemanticChunker()
        self.recursive_chunker = RecursiveChunker()
    
    def preprocess(self, file_path: str, file_type: str) -> List[Dict]:
        """完整的预处理流程"""
        # 1. 提取文档结构
        structure = extract_structure(file_path, file_type)
        
        # 2. 清洗文本
        cleaned_text = self._clean_document(structure)
        
        # 3. 智能分块
        chunks = self._chunk_document(cleaned_text, structure)
        
        # 4. 质量控制
        chunks = self._quality_control(chunks)
        
        # 5. 去重
        chunks = remove_duplicates(chunks)
        
        return chunks
    
    def _clean_document(self, structure: Dict) -> str:
        """清洗文档"""
        cleaned_paragraphs = []
        
        for para in structure['paragraphs']:
            # 基础清洗
            cleaned = basic_cleaning(para)
            # 去除噪声
            cleaned = remove_noise(cleaned)
            # 格式标准化
            cleaned = normalize_format(cleaned)
            
            if cleaned:
                cleaned_paragraphs.append(cleaned)
        
        return '\n'.join(cleaned_paragraphs)
    
    def _chunk_document(self, text: str, structure: Dict) -> List[Dict]:
        """文档分块"""
        # 优先使用结构感知分块
        if structure['headings']:
            chunks = structure_aware_chunk(structure)
        else:
            # 使用语义分块
            chunks_text = self.semantic_chunker.semantic_chunk(text)
            chunks = [{'text': chunk, 'metadata': {}} for chunk in chunks_text]
        
        return chunks
    
    def _quality_control(self, chunks: List[Dict]) -> List[Dict]:
        """质量控制"""
        filtered_chunks = []
        
        for chunk in chunks:
            quality = assess_quality(chunk['text'])
            
            # 只保留质量较高的chunk
            if quality['overall_score'] >= 0.5:
                chunk['quality_score'] = quality
                filtered_chunks.append(chunk)
        
        return filtered_chunks

7. 最佳实践总结

7.1 分块策略选择

• 结构化文档：使用结构感知分块
• 长文档：使用语义分块
• 通用场景：使用递归分块

7.2 参数调优

• chunk大小：512-1024 tokens
• 重叠大小：50-100 tokens
• 相似度阈值：0.7-0.85

7.3 质量指标

• 完整性：保留关键信息
• 连贯性：chunk内部语义连贯
• 可检索性：包含足够上下文
• 多样性：避免重复内容

通过以上数据清洗和预处理流程，可以显著提升 RAG 应用的检索精度和整体性能。