数据清洗与预处理

在文档切分之前，对文本内容进行数据清洗和预处理，可以显著提高切分质量和检索效果。清洗环节投入的工作量，能在后续每个环节节省大量调试时间，同时降低向量存储成本、提升检索速度和答案质量。

清洗目标与原则

• 去除无关内容和噪声：HTML/XML 标签、Markdown 符号、无意义代码片段等
• 标准化文本格式：修复 PDF 断行、统一换行符（\r\n → \n）、合并短行、标准化空格
• 提高切分边界识别的准确性：保留文档结构和层次关系，让切分算法更精准地定位分割点

不同类型数据的清洗方法

数据类型	清洗要点	核心操作
纯文本	去除多余空白、修复断词、标准化引号	re.sub 正则清洗
表格	去空行列、填充 NaN、去重	pandas DataFrame 操作
图像	二值化、降噪、倾斜校正	OpenCV 滤波处理
代码	移除行尾空白、合并空行	逐行清洗
混合文档	分类处理、统一入口	路由到对应清洗函数

纯文本文档清洗

纯文本清洗包括普通文本、HTML 和 Markdown 三种场景，分别处理空白字符、标签和格式符号：

import re


def clean_text(text):
    """
    清洗普通文本，去除多余空白、修复断开的单词、标准化引号

    处理流程：
    1. 将多个空白字符（空格、换行、制表符等）压缩为单个空格
    2. 将断开的单词重新连接（如 "word- \nnew" -> "wordnew"）
    3. 将弯引号转换为直引号

    参数:
        text (str): 待清洗的原始文本

    返回:
        str: 清洗后的文本
    """
    # 去除多余空白：将连续的空白的字符替换为单个空格
    text = re.sub(r'\s+', ' ', text)

    # 修复断开的单词：匹配 "单词- +换行/空格 +单词" 的模式，合并为 "单词单词"
    text = re.sub(r'(\w+)-\s+(\w+)', r'\1\2', text)

    # 标准化引号：将全角引号转换为半角引号
    text = text.replace('"', '"').replace('"', '"')
    return text.strip()


def clean_html(text):
    """
    清洗HTML文本，移除所有HTML标签但保留标签内的文字内容

    参数:
        text (str): 包含HTML标签的原始文本

    返回:
        str: 移除HTML标签后的纯文本
    """
    return re.sub(r'<.*?>', '', text)  # 移除HTML标签


def clean_markdown(text):
    """
    清洗Markdown文本，移除格式标记但保留文字内容

    处理：
    1. 移除粗体标记 **text** 但保留文字
    2. 移除链接标记 [text](url) 但保留文字

    参数:
        text (str): 包含Markdown格式的原始文本

    返回:
        str: 移除格式标记后的纯文本
    """
    text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)  # 移除粗体标记但保留文字
    text = re.sub(r'\[(.*?)\]\(.*?\)', r'\1',
                  text)  # 移除链接标记但保留文字
    return text

    # ==================== 调用示例 ====================


if __name__ == "__main__":
    # 示例1: clean_text - 普通文本清洗
    dirty_text = "这是    一段   含有\n\n多余空白\n和断开单词的-text。"
    print("原始文本:", repr(dirty_text))
    print("清洗后:", clean_text(dirty_text))
    # 输出: 这是 一段 含有 多余空白和断开单词的text。

    # 示例2: clean_html - HTML标签移除
    html_text = "<p>这是 <strong>一段</strong> 含 <a href='#'>链接</a> 的文本</p>"
    print("\n原始HTML:", html_text)
    print("清洗后:", clean_html(html_text))
    # 输出: 这是 一段 含 链接 的文本

    # 示例3: clean_markdown - Markdown格式移除
    md_text = "这是 **粗体文字** 和 [链接文字](https://example.com) 的文本"
    print("\n原始Markdown:", md_text)
    print("清洗后:", clean_markdown(md_text))
    # 输出: 这是 粗体文字 和 链接文字 的文本

表格数据清洗

表格数据需要特别注意结构完整性，使用 pandas 进行标准化处理：

# 安装pandas
# pip install pandas

import pandas as pd


def clean_table(df):
    """
    清洗 pandas DataFrame 数据表，统一格式并去除无效数据

    处理流程：
    1. 确保输入是 DataFrame 类型（如果不是则转换）
    2. 删除完全空白的行和列（所有值都为NaN）
    3. 将剩余的 NaN 值填充为空字符串
    4. 删除完全重复的行
    5. 标准化列名（去除首尾空白并转为字符串）

    参数:
        df (pd.DataFrame 或 list/dict): 原始数据表

    返回:
        pd.DataFrame: 清洗后的数据表
    """
    # 类型检查：确保输入是 DataFrame，如果不是则自动转换
    # 支持 list（嵌套列表）或 dict（字典）格式的输入
    if type(df) != pd.DataFrame:
        df = pd.DataFrame(df)

        # 去除完全空白的行：dropna(how='all') 删除所有值都为NaN的行
    # 去除完全空白的列：dropna(axis=1, how='all') 删除所有值都为NaN的列
    df = df.dropna(how='all').dropna(axis=1, how='all')

    # 填充NaN值：将剩余的 NaN 替换为空字符串 ''
    # 这样可以避免后续处理时出现 NaN 相关的错误
    df = df.fillna('')

    # 删除完全重复的行：drop_duplicates() 保留首次出现的行
    # 注意：不要使用 inplace=True，它会返回 None 而非处理后的 DataFrame
    df = df.drop_duplicates()

    # 标准化列名：
    # 1. str(col) 确保列名是字符串类型
    # 2. strip() 去除列名首尾的空白字符
    df.columns = [str(col).strip() for col in df.columns]

    return df

    # ==================== 调用示例 ====================


if __name__ == "__main__":
    # 示例1: 从字典创建 DataFrame
    data = {
        '姓名': ['张三', '李四', '王五', '张三'],
        '年龄': [25, None, 30, 25],
        '城市': ['', '', '', ''],
        # 完全空白的列
    }
    df1 = pd.DataFrame(data)
    print("=" * 50)
    print("示例1 - 从字典创建 DataFrame")
    print("清洗前:")
    print(df1)
    print("\n清洗后:")
    print(clean_table(df1))
    # 姓名  年龄  城市
    # 0  张三  25
    # 1  李四
    # 2  王五  30
    # 3  张三  25

    # 示例2: 从嵌套列表创建 DataFrame（包含重复行和NaN）
    data2 = [
        ['产品', '价格', '数量'],
        ['苹果', 5.0, 10],
        ['香蕉', None, 20],
        ['苹果', 5.0, 10],
        # 重复行
        ['  苹果  ', 5.0, 10],
        # 列名含空白
    ]
    print("\n" + "=" * 50)
    print("示例2 - 从嵌套列表创建 DataFrame")
    print("清洗前:")
    df2 = pd.DataFrame(data2[1:], columns=data2[0])
    print(df2)
    print("\n清洗后:")
    print(clean_table(df2))

    # 示例3: 直接传入字典列表（会被自动转换）
    data3 = [
        {'name': 'Alice', 'score': 85},
        {'name': 'Bob', 'score': None},
        {'name': 'Alice', 'score': 85},
        # 重复
    ]
    print("\n" + "=" * 50)
    print("示例3 - 直接传入字典列表（自动转换）")
    print("清洗后:")
    print(clean_table(data3))

图像文档清洗

图像文档（如扫描的 PDF）在 OCR 识别前需要进行预处理，以提高文字识别准确率：

处理步骤	方法	OpenCV 函数
灰度转换	将彩色图像转为灰度图	cv2.cvtColor()
二值化	将灰度图转为黑白图	cv2.threshold()
高斯模糊去噪	平滑噪声点	cv2.GaussianBlur()
中值滤波去噪	对椒盐噪声效果好	cv2.medianBlur()
双边滤波	去噪同时保持边缘	cv2.bilateralFilter()
倾斜校正	矫正扫描歪斜	cv2.getRotationMatrix2D()

import cv2
import numpy as np


def clean_image(img_path, output_dir="output"):
    """
    图像降噪 - 去除扫描文档中的噪声点

    Args:
        img_path: 图像文件路径（建议使用绝对路径）
        output_dir: 输出目录路径（默认为当前目录下的 output 文件夹）

    Returns:
        dict: 包含灰度图和三种降噪结果的字典
    """
    import os
    # ========== 0. 创建输出目录 ==========
    os.makedirs(output_dir, exist_ok=True)

    # ========== 1. 读取图像 ==========
    # cv2.imread() 默认以 BGR 格式加载图像（非 RGB）
    img = cv2.imread(img_path)
    if img is None:
        raise FileNotFoundError(f"无法读取图像: {img_path}")

    # BGR 转 RGB（OpenCV 显示图像时需要正确的颜色通道顺序）
    img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

    # ========== 2. 转换为灰度图 ==========
    # 灰度图处理更快，且去噪效果在灰度图上更明显
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    # ========== 3. 保存原始图像 ==========
    print("保存原始图像...")
    cv2.imwrite(f"{output_dir}/01_original.png", img)
    cv2.imwrite(f"{output_dir}/02_gray.png", gray)

    # ========== 4. 图像降噪处理 ==========

    # 方法1: 高斯模糊去噪
    # - 使用 (5, 5) 卷积核进行高斯滤波
    # - 对高斯噪声（服从正态分布的噪声）效果好
    # - sigmaX=0 表示自动计算标准差
    gaussian_denoised = cv2.GaussianBlur(gray, (5, 5), 0)
    print("保存高斯模糊去噪结果...")
    cv2.imwrite(f"{output_dir}/03_gaussian_denoised.png", gaussian_denoised)

    # 方法2: 中值滤波去噪
    # - 卷积核大小为 3，使用邻域像素的中值替代中心像素
    # - 对椒盐噪声（随机出现的黑白点）效果最好
    # - 能较好保留边缘细节
    median_denoised = cv2.medianBlur(gray, 3)
    print("保存中值滤波去噪结果...")
    cv2.imwrite(f"{output_dir}/04_median_denoised.png", median_denoised)

    # 方法3: 双边滤波去噪
    # - 在去除噪声的同时保持边缘清晰（区别于高斯模糊的关键点）
    # - 参数: d=9(邻域直径), sigmaColor=75(颜色空间标准差), sigmaSpace=75(坐标空间标准差)
    # - 适合处理文档扫描件，能去除噪声但保持文字边缘锐利
    bilateral_denoised = cv2.bilateralFilter(gray, 9, 75, 75)
    print("保存双边滤波去噪结果...")
    cv2.imwrite(f"{output_dir}/05_bilateral_denoised.png", bilateral_denoised)

    # 返回结果字典，方便后续使用
    return {
        "gray": gray,
        "gaussian": gaussian_denoised,
        "median": median_denoised,
        "bilateral": bilateral_denoised
    }

    # ========== 使用示例 ==========


if __name__ == "__main__":
    # 使用绝对路径确保文件能被正确读取
    img_path = "/Users/wangkaiqi/Documents/my_project/ai-agent/langchain/RAG/文档切分/tmp_233483b12be534b25ef8bef1a0d05135.jpg"
    # 指定输出目录，图片会保存到这个文件夹中
    output_dir = "/Users/wangkaiqi/Documents/my_project/ai-agent/langchain/RAG/文档切分/output"
    results = clean_image(img_path, output_dir)
    print(f"\n所有图片已保存到: {output_dir}")

注意

图片表格需要先经过 OCR 识别转为结构化数据，再使用表格清洗流程处理。

代码块清洗

代码块需要保持格式和缩进，主要处理行尾空白和多余空行：

import re

def clean_code(code_text, remove_comments=False):
    """增强版代码清洗"""
    if not code_text:
        return ""

    cleaned_lines = []
    in_multiline_comment = False

    for line in code_text.split('\n'):
        # 移除行尾空白
        clean_line = line.rstrip()

        # 可选：移除注释
        if remove_comments:
            if not in_multiline_comment:
                if '"""' in clean_line or "'''" in clean_line:
                    in_multiline_comment = not in_multiline_comment
                    continue
                clean_line = re.sub(r'#.*$', '', clean_line)
            else:
                if '"""' in clean_line or "'''" in clean_line:
                    in_multiline_comment = False
                continue

        # 保留内容行和单个空行
        if clean_line or (cleaned_lines and not cleaned_lines[-1]):
            cleaned_lines.append(clean_line)

    result = '\n'.join(cleaned_lines).strip()
    if result and not result.endswith('\n'):
        result += '\n'
    return result

混合文档清洗

混合文档通过统一入口路由到对应的清洗函数，支持批量处理：

def clean_mixed_content(content_type, content):
    """
    统一清洗入口
    content_type: 'text', 'table', 'image', 'code'
    """
    try:
        if content_type == 'text':
            text = clean_text(content)
            text = clean_html(text)
            text = clean_markdown(text)
            return text
        elif content_type == 'table':
            return clean_table(content)
        elif content_type == 'image':
            return clean_image(content)
        elif content_type == 'code':
            return clean_code(content)
        else:
            print(f"未知内容类型: {content_type}")
            return content
    except Exception as e:
        print(f"清洗 {content_type} 时出错: {e}")
        return content

def batch_clean(documents):
    """
    批量清洗文档
    documents: 列表，每个元素是 (content_type, content) 元组
    """
    results = []
    for i, (doc_type, content) in enumerate(documents):
        print(f"清洗文档 {i+1}: {doc_type}")
        cleaned = clean_mixed_content(doc_type, content)
        results.append((doc_type, cleaned))
    return results

最佳实践

数据清洗五项原则

1. 保持原始结构：清洗过程中尽量保留文档的原始结构和层次关系
2. 最小化信息损失：只去除明确的噪声，避免删除可能有用的内容
3. 标准化格式：统一标点符号、引号、连字符等格式元素
4. 处理特殊字符：转义或替换可能影响后续处理的特殊字符
5. 版本控制：保留原始文档副本，以便需要时回滚