一、摘要

大语言模型、多模态大模型的预训练效果，数据质量是决定性因素。高质量、高纯净的训练语料能够显著提升模型语义理解、逻辑推理、知识表达能力；而低质数据、重复文本、违规内容、噪声字符、残缺语句会直接导致模型收敛变慢、生成内容错乱、偏见与错误知识泛滥，甚至引发安全风险。因此，在基于昇腾硬件与昇思 MindSpore 开展大模型预训练前，数据集质量过滤是数据预处理环节中不可或缺的核心步骤。

MindSpore 作为面向全场景的深度学习框架，原生支持大规模分布式数据加载、流水线预处理、算子加速与硬件协同调度，非常适配 TB 级、PB 级海量预训练语料的清洗与过滤工作。结合文本规则过滤、统计特征过滤、模型打分过滤、相似度去重四大主流技术，可构建一套完整、高效、可分布式并行的数据质量过滤链路。围绕大模型预训练数据质量过滤的技术体系、分层过滤策略、基于 MindSpore 的工程实现、分布式部署及落地优化展开讲解，并提供完整可运行代码，从源头保障大模型训练质量。

二、大模型预训练数据质量问题与过滤体系

2.1 常见低质数据类型

公开爬取、开源汇聚的预训练语料普遍存在多类缺陷，也是过滤工作的主要目标：

1. 噪声数据：乱码、特殊符号、HTML 标签、URL、代码碎片、表情符号、无意义重复字符、空白文本；
2. 残缺文本：句子截断、单字、短语、语序混乱、语法错误严重的短句；
3. 重复数据：段落重复、整文重复、局部片段抄袭，重复样本会造成梯度冗余、模型过拟合；
4. 低信息密度数据：大量语气词、口水话、纯标点、篇幅极短、无有效语义的内容；
5. 违规与风险数据：涉敏、低俗、侵权内容，需在过滤阶段提前拦截，满足内容安全要求；
6. 语种混杂数据：中英混杂、多语种乱掺，干扰模型语言建模能力。

单纯依靠单一规则无法实现全面净化，行业内普遍采用分层递进式过滤思路，由浅至深逐步淘汰低质样本。

2.2 分层质量过滤整体架构

结合 MindSpore 分布式数据处理能力，将过滤流程分为四层，逐层提升过滤精度，兼顾处理效率与过滤效果：

1. 第一层：规则粗过滤。基于正则表达式、字符统计、长度阈值做快速筛选，剔除明显噪声、空白文本、超长 / 超短文本，该层计算量极小，适合全量数据首轮清洗；
2. 第二层：统计特征过滤。统计字符分布、词汇密度、语种占比、重复度等特征，量化判定文本信息密度，过滤低价值内容；
3. 第三层：相似度去重。基于文本向量、局部哈希算法识别重复段落与相似文本，解决数据冗余问题；
4. 第四层：模型细过滤。使用轻量打分模型对文本质量、语义流畅度、合规性进行打分，保留高分优质样本，是精细化筛选的核心环节。

四层链路可与 MindSpore 数据加载流水线结合，实现边读取、边过滤、边缓存，无需将全量数据落地中转，大幅节省磁盘 IO 与存储开销。

2.3 MindSpore 适配优势

传统单机脚本清洗海量数据存在速度慢、内存溢出、无法并行等问题，MindSpore 针对该场景提供多项能力支撑：

• 内置Dataset数据流水线，支持多进程、多线程并行预处理，充分利用 CPU 与昇腾 NPU 算力；
• 支持离线预处理与训练时在线过滤两种模式，灵活适配不同业务流程；
• 算子化封装统计计算、向量编码等逻辑，相比原生 Python 脚本运算速度提升数倍；
• 原生支持分布式场景，多卡集群可分片处理数据集，线性提升清洗吞吐；
• 兼容主流数据格式（JSON、JSONL、TXT、Parquet），适配大模型主流语料格式。

三、核心过滤规则与算法设计

3.1 规则粗过滤规则集

规则过滤以快速拦截为目标，定义硬阈值条件，不依赖复杂模型：

• 文本长度过滤：设置最小字符数与最大字符数，过滤极短碎片文本和异常超长文本；
• 空白与噪声过滤：清除空白行、纯标点、控制字符、HTML 标签、链接、特殊符号；
• 语种过滤：通过字符编码与词表判断主语种，过滤多语种严重混杂样本；
• 字符占比过滤：统计中文字符、英文字母、数字、符号占比，剔除符号 / 数字占比过高的内容。

3.2 统计特征过滤指标

通过量化指标衡量文本信息密度，常用指标如下：

1. 有效字符占比：有效语义字符总数 / 全文总长度，占比过低判定为低质；
2. 行重复率：统计文本内部重复行、重复短句数量，识别灌水内容；
3. 平均句长：短句过多说明内容碎片化，统一阈值进行过滤。

3.3 文本去重算法

针对大规模语料，采用局部敏感哈希（LSH）+ SimHash方案，对文本生成指纹，快速比对相似度，相比全文比对复杂度大幅降低，可支撑千万级以上文本去重。

3.4 模型打分过滤

选用轻量预训练模型对文本流畅度、语义质量打分，将分数作为最终筛选依据。该环节可利用 MindSpore 推理能力部署打分模型，在 NPU 上加速运算。

四、基于 MindSpore 的数据质量过滤代码实现

下文完整实现规则过滤、统计特征过滤、去重、流水线集成全流程代码，基于 MindSpore Dataset 构建预处理链路，支持单机多进程加速，适配大模型 JSONL 格式预训练语料。代码基于 MindSpore 2.2 + 版本开发，可直接在昇腾服务器、Linux 环境运行。

4.1 环境依赖与前置说明

# 安装依赖
pip install mindspore sentence-transformers simhash-py regex tqdm

数据集采用大模型通用的 JSONL 格式，每行一条样本，格式示例：{"text": "预训练文本内容"}。

4.2 完整代码实现

import re
import string
import simhash
from tqdm import tqdm
import mindspore as ms
from mindspore.dataset import TextDataset, GeneratorDataset
from mindspore.dataset.transforms import py_transforms

# ===================== 1. 全局过滤规则与常量定义 =====================
# 文本长度阈值
MIN_LEN = 30
MAX_LEN = 2000
# 噪声正则：HTML标签、URL、特殊符号、控制字符
NOISE_PATTERN = re.compile(r"<.*?>|http[s]?://\S+|[\x00-\x1F\x7F]")
PUNCTUATION = set(string.punctuation + "，。、；：？！‘’“”（）《》【】……——")

# 初始化SimHash去重对象
def get_simhash(text: str) -> int:
    """生成文本指纹，用于重复检测"""
    words = re.findall(r"[\u4e00-\u9fa5a-zA-Z0-9]+", text)
    if not words:
        return 0
    sh = simhash.SimHash(words)
    return sh.value

# ===================== 2. 分层过滤函数 =====================
def rule_filter(text: str) -> bool:
    """第一层：规则粗过滤，返回True表示保留，False表示过滤"""
    # 去除首尾空白
    text = text.strip()
    # 长度过滤
    if len(text) < MIN_LEN or len(text) > MAX_LEN:
        return False
    # 清除噪声内容
    clean_text = NOISE_PATTERN.sub("", text)
    # 纯标点/空白过滤
    valid_chars = [c for c in clean_text if c not in PUNCTUATION and not c.isspace()]
    if len(valid_chars) < MIN_LEN / 2:
        return False
    return True

def stat_filter(text: str) -> bool:
    """第二层：统计特征过滤，判断信息密度"""
    total_len = len(text)
    # 统计有效语义字符
    char_cn = len(re.findall(r"[\u4e00-\u9fa5]", text))
    char_en = len(re.findall(r"[a-zA-Z]", text))
    valid_total = char_cn + char_en
    # 有效字符占比阈值
    if valid_total / total_len < 0.4:
        return False
    return True

def full_filter_func(sample):
    """整合多层过滤，适配MindSpore数据预处理"""
    text = sample[0]
    # 逐层过滤
    if not rule_filter(text):
        return None
    if not stat_filter(text):
        return None
    # 返回清洗后文本
    return (text,)

# ===================== 3. 基于MindSpore Dataset构建过滤流水线 =====================
def create_filter_dataset(data_path: str, num_workers: int = 8):
    """
    构建MindSpore分布式过滤数据集
    :param data_path: 原始JSONL语料路径
    :param num_workers: 并行进程数
    """
    # 读取JSONL文本数据集
    dataset = TextDataset(
        data_files=data_path,
        column_names=["text"],
        file_format="jsonl",
        num_parallel_workers=num_workers
    )
    # 应用过滤变换，丢弃过滤后的空样本
    dataset = dataset.map(
        operations=full_filter_func,
        output_columns=["text"],
        num_parallel_workers=num_workers
    )
    dataset = dataset.filter(
        predicate=lambda x: x is not None,
        num_parallel_workers=num_workers
    )
    return dataset

# ===================== 4. 全局去重处理 =====================
def deduplicate_dataset(dataset, output_path: str):
    """第三层：基于SimHash全局去重，并输出清洗后语料"""
    hash_set = set()
    total_count = 0
    dup_count = 0
    with open(output_path, "w", encoding="utf-8") as f_out:
        for data in tqdm(dataset.create_dict_iterator(num_epochs=1)):
            text = data["text"].asnumpy().item().decode("utf-8")
            text_hash = get_simhash(text)
            total_count += 1
            if text_hash in hash_set:
                dup_count += 1
                continue
            hash_set.add(text_hash)
            # 写入清洗后的JSONL文件
            f_out.write(f'{{"text": "{text}"}}\n')
    print(f"原始样本总数: {total_count}")
    print(f"重复样本数量: {dup_count}")
    print(f"最终有效样本数量: {total_count - dup_count}")

# ===================== 5. 主函数执行全流程过滤 =====================
if __name__ == "__main__":
    # 配置参数
    ms.set_context(mode=ms.PYNATIVE_MODE)
    RAW_DATA = "./raw_pretrain_data.jsonl"
    CLEAN_DATA = "./clean_pretrain_data.jsonl"
    WORKER_NUM = 8

    print("===== 开始MindSpore大模型预训练数据质量过滤 =====")
    # 1. 构建过滤流水线
    filter_ds = create_filter_dataset(RAW_DATA, num_workers=WORKER_NUM)
    # 2. 执行过滤+去重并输出结果
    deduplicate_dataset(filter_ds, CLEAN_DATA)
    print("===== 数据质量过滤完成，清洗后数据已输出 =====")

五、代码解析与功能扩展

5.1 代码核心模块说明

1. 规则过滤模块：通过正则、长度、字符占比完成首轮快速清洗，计算量极低，多进程下吞吐极高；
2. 统计特征模块：量化文本有效信息占比，过滤低语义密度内容，弥补单纯规则过滤的不足；
3. MindSpore Dataset 流水线：利用TextDataset原生支持 JSONL 格式，map与filter算子实现并行预处理，num_parallel_workers配置多进程加速，充分利用多核 CPU；
4. SimHash 去重模块：对清洗后的文本生成哈希指纹，全局比对剔除重复内容，解决数据集冗余问题。

5.2 功能扩展：接入模型打分过滤

在现有四层架构基础上，可接入 MindSpore 部署的文本质量打分模型，在去重之后增加模型打分环节。将清洗后的文本送入轻量分类模型，输出质量分数，设定分数阈值筛选优质样本，进一步提升数据集纯度。该部分可复用 MindSpore 推理能力，将打分逻辑嵌入数据流水线，实现端到端一体化处理。

5.3 分布式扩展

面对 PB 级超大规模语料，可使用 MindSpore 分布式能力，通过set_auto_parallel开启数据并行，将数据集分片分发至多台服务器 / 多块昇腾卡，分片并行过滤后再合并结果，实现集群级海量数据清洗。

六、性能优化与工程落地建议

6.1 运行效率优化

1. 合理设置并行进程数：num_parallel_workers建议设置为 CPU 核心数的 0.8 倍，避免进程争抢导致效率下降；
2. 分级存储：原始数据、中间数据、最终清洗数据分盘存放，降低磁盘 IO 瓶颈；
3. 离线预处理优先：正式预训练前完成全量数据清洗，避免训练阶段在线过滤占用算力。

6.2 阈值调优策略

不同领域预训练语料（通用文本、专业文献、对话数据）特征差异较大，需要针对性调整阈值：专业文献可适当提高最小长度、提高有效字符占比；对话数据可适度放宽长度限制。建议先抽样分析数据分布，再确定过滤参数。

6.3 异常处理与日志

工程化落地时，可增加异常捕获、过滤日志输出，记录每一类低质数据的剔除数量，便于统计数据分布、迭代过滤规则。同时对特殊编码、生僻字符做兼容处理，避免样本报错中断流程。

6.4 安全合规补充

针对涉敏、违规内容，可在过滤链路中接入关键词词库、内容审核模型，构建安全过滤层，确保预训练数据集符合合规要求。

七、总结

大模型预训练是数据驱动的工程体系，数据质量直接决定模型上限。基于 MindSpore 构建规则过滤、统计过滤、相似度去重、模型打分的四层质量过滤链路，结合框架原生的并行数据流水线能力，能够高效完成海量预训练语料的清洗净化工作。

本文从数据问题分析、分层过滤架构、算法设计、完整代码实现、性能优化等维度，完整阐述了 MindSpore 生态下大模型预训练数据质量过滤全流程。