数据清洗与预处理：构建高质量RAG知识库

引言

在 RAG 系统中，有一个广为人知的原则：Garbage In, Garbage Out（GIGO）。无论检索模型多先进、生成模型多强大，如果输入的知识库数据质量低下，最终的回答质量必然受限。

数据质量 → 分块质量 → Embedding质量 → 检索质量 → 生成质量
    │           │           │            │           │
    ▼           ▼           ▼            ▼           ▼
  数据清洗   合适分块     准确向量    高召回率    可信回答

本文将系统讲解 RAG 系统中的数据清洗与预处理技术，帮助你构建高质量的文档语料库。

一、为什么数据质量至关重要

1.1 低质量数据的典型表现

1.2 数据清洗在 RAG 流水线中的位置

原始文档 → 格式解析 → 内容提取 → 数据清洗 → 文本分块 → Embedding
                              ↑
                         我们在这里

二、HTML/XML 文档清洗

2.1 HTML 标签剥离

from bs4 import BeautifulSoup

def clean_html(html_content: str) -> str:
    soup = BeautifulSoup(html_content, 'html.parser')

    # 移除无用元素
    for tag in soup(['script', 'style', 'nav', 'footer', 'header',
                     'iframe', 'noscript', 'meta', 'link']):
        tag.decompose()

    # 提取纯文本
    text = soup.get_text(separator='\n', strip=True)

    # 清理多余空行
    lines = [line.strip() for line in text.split('\n') if line.strip()]
    return '\n'.join(lines)

2.2 使用 trafilatura 提取正文

对于网页文章，trafilatura 比 BeautifulSoup 更智能，能自动识别正文区域：

import trafilatura

def extract_article(url_or_html: str) -> str:
    # 从 URL 下载并提取
    downloaded = trafilatura.fetch_url(url_or_html)

    # 提取主内容（自动过滤导航、侧栏、页脚）
    text = trafilatura.extract(
        downloaded,
        include_links=False,
        include_images=False,
        include_tables=True,
        no_fallback=False
    )
    return text or ""

三、PDF 文本提取

3.1 PyMuPDF（推荐用于结构化PDF）

import fitz  # PyMuPDF

def extract_pdf_text(pdf_path: str) -> str:
    doc = fitz.open(pdf_path)
    text_pages = []

    for page_num in range(doc.page_count):
        page = doc[page_num]
        # 提取文本块，保留布局信息
        blocks = page.get_text("blocks")
        for block in blocks:
            # block: (x0, y0, x1, y1, text, block_type, block_no)
            if block[5] == 0:  # 0 = 文本块
                text_pages.append(block[4].strip())

    doc.close()
    return '\n'.join(text_pages)

3.2 marker-pdf（更准确的PDF转Markdown）

对于复杂排版（多栏、图片说明、表格），marker-pdf 效果更好：

from marker.converters.pdf import PdfConverter
from marker.models import create_model_dict

converter = PdfConverter(
    artifact_dict=create_model_dict()
)
rendered = converter("document.pdf")
markdown_text = rendered.markdown

PDF提取方法对比

四、OCR 识别

对于扫描件或图片型 PDF，需要 OCR 文字识别：

from paddleocr import PaddleOCR

ocr = PaddleOCR(use_angle_cls=True, lang='ch')

def ocr_image(image_path: str) -> str:
    result = ocr.ocr(image_path, cls=True)
    text_lines = []
    for line in result[0]:
        text = line[1][0]        # 识别文本
        confidence = line[1][1]   # 置信度
        if confidence > 0.5:      # 过滤低置信度结果
            text_lines.append(text)
    return '\n'.join(text_lines)

五、噪声去除

5.1 常见噪声类型

5.2 自动化噪声检测

import re

NOISE_PATTERNS = [
    r'©\s*\d{4}.*?(?:Inc\.|Ltd\.|公司|版权)',
    r'(?:分享到|分享至|转发到).*(?:微信|微博|QQ|朋友圈)',
    r'(?:首页|关于我们|产品中心|联系我们|招贤纳士)',
    r'(?:点击|查看)(?:更多|详情|原文)',
    r'^(?:欢迎访问|欢迎来到).*$',
    r'本文(?:来源|转载自|来源于).*',
]

def remove_noise(text: str) -> str:
    lines = text.split('\n')
    cleaned = []

    for line in lines:
        stripped = line.strip()
        # 跳过空行或过短行
        if len(stripped) < 5:
            continue
        # 跳过匹配噪声模式的行
        is_noise = any(
            re.search(pattern, stripped)
            for pattern in NOISE_PATTERNS
        )
        if not is_noise:
            cleaned.append(stripped)

    return '\n'.join(cleaned)

六、去重策略

6.1 精确去重

def exact_deduplicate(documents: list[str]) -> list[str]:
    seen = set()
    unique = []
    for doc in documents:
        # 使用内容的哈希值去重
        content_hash = hash(doc.strip())
        if content_hash not in seen:
            seen.add(content_hash)
            unique.append(doc)
    return unique

6.2 模糊去重（MinHash）

from datasketch import MinHash, MinHashLSH

def create_minhash(text: str) -> MinHash:
    m = MinHash(num_perm=128)
    # 使用 shingles（连续 N-gram 字符）
    for shingle in [text[i:i+5] for i in range(len(text)-4)]:
        m.update(shingle.encode('utf-8'))
    return m

# 构建 LSH 索引
lsh = MinHashLSH(threshold=0.8, num_perm=128)
documents = [...]  # 文档列表

for i, doc in enumerate(documents):
    m = create_minhash(doc)
    lsh.insert(f"doc_{i}", m)

# 查询相似文档
def find_near_duplicates(query_text: str):
    m = create_minhash(query_text)
    return lsh.query(m)

去重方法对比

七、文本规范化

7.1 编码修复

import unicodedata

def normalize_text(text: str) -> str:
    # Unicode 规范化（NFC: 组合形式）
    text = unicodedata.normalize('NFC', text)

    # 统一引号
    text = text.replace('"', '"').replace('"', '"')
    text = text.replace(''', "'").replace(''', "'")

    # 统一破折号
    text = text.replace('—', '—').replace('–', '-')

    # 去除控制字符（保留换行和制表符）
    text = ''.join(
        ch for ch in text
        if ch == '\n' or ch == '\t' or ch == '\r'
        or (ch.isprintable() or ord(ch) > 127)
    )

    return text

7.2 空白字符处理

import re

def clean_whitespace(text: str) -> str:
    # 合并多余空格
    text = re.sub(r' +', ' ', text)
    # 合并多余空行
    text = re.sub(r'\n{3,}', '\n\n', text)
    # 去除行首尾空格
    text = '\n'.join(line.strip() for line in text.split('\n'))
    # 去除全角空格
    text = text.replace('\u3000', ' ')
    return text.strip()

八、完整的预处理流水线

from typing import List, Callable

class DataPreprocessingPipeline:
    def __init__(self):
        self.steps: List[Callable] = []

    def add_step(self, step: Callable):
        self.steps.append(step)

    def process(self, text: str) -> str:
        for step in self.steps:
            text = step(text)
        return text

# 构建流水线
pipeline = DataPreprocessingPipeline()
pipeline.add_step(clean_html)           # 1. HTML清洗
pipeline.add_step(normalize_text)        # 2. Unicode规范化
pipeline.add_step(clean_whitespace)      # 3. 空白处理
pipeline.add_step(remove_noise)          # 4. 噪声去除

# 批量处理
raw_documents = [ ... ]
cleaned_documents = [
    pipeline.process(doc) for doc in raw_documents
]

# 去重
deduplicated = exact_deduplicate(cleaned_documents)

九、数据质量评估

9.1 评估指标

9.2 质量检查示例

def quality_check(text: str) -> dict:
    checks = {
        'total_chars': len(text),
        'total_lines': len(text.split('\n')),
        'avg_line_length': len(text) / max(len(text.split('\n')), 1),
        'contains_html': bool(re.search(r'<[^>]+>', text)),
        'encoding_errors': len(re.findall(r'[\ufffd\ufffe]', text)),
        'empty_lines_ratio': text.count('\n\n') / max(text.count('\n'), 1),
        'unique_chars': len(set(text)),
    }
    return checks

十、最佳实践总结

1. 先清洗后分块 — 清洗后再分块，减少噪声对分块的影响
2. 保留元数据 — 清洗时不要丢失文档标题、来源、日期等元数据
3. 渐进式处理 — 先轻量清洗，再根据效果决定是否需要更深入的清洗
4. 定期评估 — 定期抽样检查清洗效果，建立质量监控体系
5. 保留原始副本 — 始终保留原始文档，以便回溯和对比

相关资源：

• trafilatura 文档
• datasketch MinHash LSH
• marker-pdf 项目