1. 大模型预训练里，数据质量过滤到底在解决什么问题
做过大模型预训练的人都有一个共识：模型效果的上限，很大程度上不是被网络结构卡住的，而是被数据质量卡住的。我刚开始接触 MindSpore 做预训练任务时，也曾经天真地以为只要把海量文本一股脑喂进去，模型就能自己学会说话。结果跑完一轮 loss 曲线看着还行，实际生成出来的内容却惨不忍睹——重复、乱码、逻辑断裂，甚至把网页导航栏和广告文案都背下来了。后来复盘才发现，问题根本不在模型，而在于我喂进去的数据里，有大量低质、重复、格式混乱的垃圾内容。
所谓数据质量过滤，说白了就是在数据进入训练流程之前，用一套规则加模型的手段，把那些“吃了没用甚至有害”的样本筛掉。它要解决的核心问题有三个：第一，去重，互联网文本重复率极高，同一篇新闻可能被转载几百次，模型反复看同一句话会严重过拟合；第二，去噪，网页里夹杂的 HTML 标签、乱码、无意义符号、机器生成的垃圾文本，都会污染训练信号；第三，筛选有效信息密度，有些文本虽然通顺，但信息量极低，比如“点击这里查看更多”“版权所有”这类模板化内容，对模型学习语言规律几乎没有帮助。
这套方案适合谁参考？如果你正在用 MindSpore 做百亿甚至千亿参数级别的预训练，或者你手上有几十 GB 到几 TB 的原始语料需要清洗，那这篇内容基本可以直接抄作业。哪怕你只是做小规模微调，数据过滤的思路同样适用，只是规模不同而已。我下面会从整体设计、核心细节、实操流程到踩坑排查，完整拆一遍我在 MindSpore 环境下落地这套方案的全过程。
2. 整体方案设计与技术选型思路
2.1 为什么过滤要分层次而不是一刀切
很多人一上来就想用一个模型分类器把所有低质数据干掉，我试过，效果并不好。原因很简单：不同维度的质量问题需要用不同的手段处理，混在一起反而会误杀。比如一篇技术博客可能包含少量 HTML 残留，但内容本身很有价值，如果你用严格的正则一刀切，可能连正文都删了；反过来，一篇纯广告文本虽然格式干净，但信息密度为零，正则根本识别不出来。
所以我的整体设计是四层过滤管线，从粗到细逐层收紧：
- 第一层：格式规范化与基础清洗，处理编码、HTML 标签、特殊符号、空白字符，把原始数据变成统一格式的纯文本。
- 第二层：规则级过滤，用统计指标和启发式规则快速剔除明显低质样本，比如长度过短、重复率过高、符号占比异常。
- 第三层：去重处理，包括精确去重和近似去重，解决互联网文本高度重复的问题。
- 第四层：模型级质量打分，用轻量分类模型对剩余数据做质量评分，保留高分样本。
这个分层逻辑的好处是：前两层成本极低，能干掉大部分垃圾，减轻后面模型打分的压力；后两层精度高，但计算贵，只处理已经比较干净的数据。实测下来，在 MindSpore 上跑一套 TB 级语料，分层过滤比单模型过滤快了三倍以上，而且误杀率明显更低。
2.2 MindSpore 在这个流程里的角色定位
MindSpore 在这套方案里主要承担两个职责：一是数据处理的并行加速，MindSpore 的 Dataset 模块支持多进程、多线程的数据管道，配合map、filter、batch等操作，可以很自然地把过滤逻辑嵌入数据加载流程；二是质量分类模型的训练与推理，我用 MindSpore 训练了一个小型的文本质量分类器，参数量控制在千万级别，推理速度快，适合在预处理阶段大规模跑。
这里有个选型细节值得说：为什么不用 PyTorch 或 TensorFlow 做质量分类模型，而是坚持用 MindSpore？因为整个预训练流程都在 MindSpore 生态里，数据管道和模型推理如果跨框架，中间的数据格式转换和序列化开销会非常大。我试过用外部框架跑分类器再把结果导回 MindSpore Dataset，光磁盘 IO 就多了一倍。统一在 MindSpore 里做，数据可以以内存友好的方式流转，整体吞吐量提升很明显。
2.3 过滤强度与数据保留率的平衡
这是最容易被忽视但最致命的问题：过滤太松，垃圾数据影响模型效果；过滤太严，数据量不够，模型欠拟合。我一开始按网上一些教程设了很严格的阈值，结果 1TB 原始数据过滤完只剩 80GB，模型训练时明显感觉数据多样性不足，生成内容变得非常保守和模板化。
后来我调整策略，采用分级保留的思路：把数据按质量分成高、中、低三档，高质量数据全部保留，中质量数据按比例采样保留，低质量数据直接丢弃。具体比例根据你的总数据量来定，如果原始数据超过 500GB，可以适当收紧；如果只有几十 GB，就要放宽标准，优先保证数据量。这个平衡点没有绝对公式，需要根据你的模型规模和任务目标做几次小规模实验来校准。
3. 核心细节解析与实操要点
3.1 格式规范化：把脏数据变成可处理的纯文本
原始语料最常见的几种脏法：编码混乱（GBK、UTF-8 混在一起）、HTML 标签残留、多余空白和换行、特殊控制字符。这一步的目标不是过滤，而是把所有数据变成统一、干净、可比较的纯文本，为后面的规则和模型处理打基础。
在 MindSpore 的 Dataset 管道里，我通常把这一步写成自定义的map操作。核心处理逻辑包括：
- 编码统一：用
chardet检测编码，统一转成 UTF-8，遇到无法解码的字符用忽略策略而不是报错中断。 - HTML 清洗：用正则或
BeautifulSoup去掉标签，但要注意保留段落结构，不能把<p>直接删掉导致所有文字粘在一起。 - 空白规范化：把连续多个空格、换行、制表符统一成单个空格或换行，去掉首尾空白。
- 控制字符过滤：去掉 ASCII 控制字符和 Unicode 私有区字符，这些通常是乱码来源。
import re import html def normalize_text(text): # 解码 HTML 实体 text = html.unescape(text) # 去掉 HTML 标签但保留换行 text = re.sub(r'<br\s*/?>', '\n', text) text = re.sub(r'<[^>]+>', '', text) # 去掉控制字符 text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text) # 规范化空白 text = re.sub(r'[ \t]+', ' ', text) text = re.sub(r'\n{3,}', '\n\n', text) return text.strip()注意：HTML 清洗不要用过于激进的正则，比如
<.*?>这种非贪婪匹配在嵌套标签场景下会出错。如果数据里 HTML 占比很高，建议用专门的解析库，虽然慢一点但准确率高很多。
3.2 规则级过滤：用统计指标快速筛掉明显垃圾
这一步是整个管线里性价比最高的环节。我常用的规则指标有五个：
| 指标 | 含义 | 典型阈值 | 处理方式 |
|---|---|---|---|
| 文本长度 | 字符数 | 少于 50 或超过 100000 | 丢弃 |
| 平均行长度 | 总字符数/行数 | 少于 10 | 丢弃 |
| 符号占比 | 非字母数字字符比例 | 超过 0.5 | 丢弃 |
| 重复行占比 | 重复行数/总行数 | 超过 0.3 | 丢弃 |
| 停用词占比 | 常见停用词比例 | 低于 0.01 | 丢弃 |
这些阈值不是拍脑袋定的，我是先抽样统计了 10 万条数据，看各项指标的分布，再取分位数作为阈值。比如文本长度，我看了下 5% 分位数是 48 个字符，95% 分位数是 87000，所以把下限设 50、上限设 100000，既能去掉太短的垃圾，又不会误杀长文档。
在 MindSpore 里实现规则过滤，可以直接用 Dataset 的filter操作：
import mindspore.dataset as ds def quality_filter(sample): text = sample['text'] if len(text) < 50 or len(text) > 100000: return False lines = text.split('\n') if len(lines) > 1: avg_line_len = sum(len(l) for l in lines) / len(lines) if avg_line_len < 10: return False repeat_ratio = 1 - len(set(lines)) / len(lines) if repeat_ratio > 0.3: return False symbol_ratio = sum(1 for c in text if not c.isalnum() and not c.isspace()) / len(text) if symbol_ratio > 0.5: return False return True dataset = dataset.filter(quality_filter)实操心得：规则过滤的阈值一定要根据你的数据来源调整。中文数据和英文数据的符号占比分布完全不同，中文里标点占比天然更高，阈值要适当放宽。我一开始用英文数据的阈值套中文，结果误杀了一大批正常文本。
3.3 去重处理：精确去重与近似去重的组合拳
互联网文本的重复率有多高？我实测过一批爬取的新闻数据，精确重复率大约 15%，近似重复率超过 30%。如果不做去重，模型会反复看到同样的内容，导致生成时倾向于复制训练数据，缺乏创造性。
精确去重比较简单，对文本做哈希（比如 SHA256），维护一个哈希集合，遇到已存在的就丢弃。但精确去重只能处理完全一样的文本，对于改了几个词的转载文章无能为力。
近似去重我用的是MinHash + LSH的方案。核心思路是：把每篇文档拆成 n-gram 集合，用多个哈希函数生成 MinHash 签名，再通过 LSH 把可能相似的文档分到同一个桶里，最后在桶内做精确的 Jaccard 相似度计算。这个方案在 MindSpore 里可以用mindspore.dataset配合自定义的 Python 操作实现，也可以先用 Spark 做一轮粗筛，再用 MindSpore 处理剩余数据。
from datasketch import MinHash, MinHashLSH def get_minhash(text, num_perm=128): m = MinHash(num_perm=num_perm) for i in range(len(text) - 5): m.update(text[i:i+5].encode('utf8')) return m lsh = MinHashLSH(threshold=0.8, num_perm=128) for idx, text in enumerate(texts): m = get_minhash(text) if lsh.query(m): continue # 近似重复，丢弃 lsh.insert(idx, m)注意：MinHash 的
num_perm和threshold需要根据数据规模调。数据量越大，num_perm可以适当降低以节省内存；threshold设 0.8 意味着相似度超过 80% 就算重复，这个值在中文场景下比较合适，英文可以设到 0.85。
3.4 模型级质量打分：用轻量分类器做最终把关
经过前三层过滤，剩下的数据已经比较干净了，但仍有部分“格式干净但内容无意义”的样本，比如机器生成的伪原创、关键词堆砌的 SEO 垃圾。这时候就需要一个质量分类模型来做最终判断。
我的做法是：人工标注 5000 条样本，分为高质量、中等、低质量三档，训练一个基于 BERT 的小型分类器。模型结构不用太复杂，用 MindSpore 的nn.BertModel加载预训练权重，接一个三分类头，训练几个 epoch 就能达到 90% 以上的准确率。
推理阶段，把分类器嵌入 MindSpore Dataset 管道，对每条数据打分，只保留高质量和部分中等质量样本。这里有个性能优化点：分类器推理要 batch 化，不要一条一条跑。我一开始用单条推理，吞吐量只有几百条每秒，改成 batch size 64 之后，直接提升到上万条每秒。
import mindspore.nn as nn import mindspore.ops as ops class QualityClassifier(nn.Cell): def __init__(self, bert_model, num_classes=3): super().__init__() self.bert = bert_model self.classifier = nn.Dense(768, num_classes) self.softmax = nn.Softmax(axis=-1) def construct(self, input_ids, attention_mask, token_type_ids): _, pooled = self.bert(input_ids, attention_mask, token_type_ids) logits = self.classifier(pooled) return self.softmax(logits)实操心得：标注数据时，一定要让多个标注员交叉验证，保证标注一致性。我一开始只让一个人标，结果模型学到的标准很偏，后来改成三人标注加投票，模型效果明显更稳。
4. 完整实操流程与关键环节实现
4.1 环境准备与依赖安装
在开始之前，先把环境搭好。我用的组合是 MindSpore 2.x + Python 3.9 + CUDA 11.6，如果你用 Ascend 环境，把 CUDA 相关依赖换成对应的 CANN 包即可。
pip install mindspore-gpu==2.2.0 pip install datasketch pip install beautifulsoup4 pip install chardet pip install tqdm注意：MindSpore 版本和 Python 版本有对应关系，装之前先查一下官方兼容性列表。我有一次用 Python 3.11 装 MindSpore 2.0，结果各种报错，换成 3.9 就顺利了。
4.2 数据加载与管道构建
MindSpore 的 Dataset 支持从多种格式加载数据，我通常把原始语料存成 JSONL 格式，每行一个样本，包含text和source字段。加载时用ds.TextFileDataset或ds.GeneratorDataset，然后依次挂上规范化、规则过滤、去重、质量打分的操作。
import mindspore.dataset as ds import json def gen_samples(file_path): with open(file_path, 'r', encoding='utf-8') as f: for line in f: item = json.loads(line) yield (item['text'],) dataset = ds.GeneratorDataset(gen_samples('raw_data.jsonl'), column_names=['text']) dataset = dataset.map(normalize_text, input_columns=['text']) dataset = dataset.filter(quality_filter) dataset = dataset.batch(64, drop_remainder=True)这里有个细节：map和filter的顺序很重要。一定要先做规范化再做过滤，否则规则过滤会误判。比如原始文本里有 HTML 标签，符号占比会很高，直接过滤会误杀正常内容。
4.3 去重模块的工程实现
去重是整套流程里最耗资源的环节，尤其是近似去重。我的工程实现分两步走：
第一步，精确去重，用哈希集合在内存里做，速度快，内存占用可控。如果数据量超过内存容量，可以用布隆过滤器替代，代价是有极小的误判率。
第二步，近似去重，用 MinHash + LSH。这里的关键是分片处理：把数据按来源或时间分片，每片单独做 LSH，最后合并结果。这样既能控制内存，又能并行加速。
def dedup_pipeline(dataset, lsh_threshold=0.8): seen_hashes = set() lsh = MinHashLSH(threshold=lsh_threshold, num_perm=128) for batch in dataset.create_dict_iterator(): texts = batch['text'].asnumpy().tolist() for text in texts: h = hashlib.sha256(text.encode('utf-8')).hexdigest() if h in seen_hashes: continue m = get_minhash(text) if lsh.query(m): continue seen_hashes.add(h) lsh.insert(h, m) yield (text,)实操心得：MinHash 的 n-gram 大小建议用 5，中文场景下 5-gram 能比较好地捕捉语义相似性。太小容易误判，太大又抓不住改写后的重复。
4.4 质量分类模型的训练与推理
质量分类模型的训练数据来自人工标注，我标了 5000 条，按 8:1:1 划分训练、验证、测试。训练时用 MindSpore 的TrainOneStepCell配合AdamWeightDecay优化器，学习率设 2e-5，batch size 32，跑 5 个 epoch。
from mindspore import nn, Model from mindspore.train.callback import LossMonitor optimizer = nn.AdamWeightDecay(classifier.trainable_params(), learning_rate=2e-5) loss_fn = nn.SoftmaxCrossEntropyWithLogits(sparse=True, reduction='mean') model = Model(classifier, loss_fn, optimizer, metrics={'acc': nn.Accuracy()}) model.train(5, train_dataset, callbacks=[LossMonitor()])推理阶段，把模型设成 eval 模式，对每条数据打分，保留高分样本。这里有个技巧：不要只保留最高分，而是按分数排序后取前 70%，这样能保证数据多样性，避免模型只学到某一类文本。
4.5 过滤效果评估与迭代
过滤完不是就结束了，一定要做效果评估。我的评估方法有两种：
一是人工抽样检查，从过滤后的数据里随机抽 200 条，人工判断质量，看误杀率和漏杀率。我一般要求误杀率低于 5%，漏杀率低于 10%。
二是下游任务验证，用过滤前后的数据分别训练小模型，对比生成质量和 loss 曲线。实测下来，经过完整过滤的数据训练出的模型，在生成连贯性和多样性上都有明显提升。
5. 常见问题与排查技巧实录
5.1 过滤后数据量骤降怎么办
这是最常见的问题。我一开始也遇到过，1TB 数据过滤完只剩 50GB，差点以为代码写错了。排查下来发现两个原因：一是规则阈值设得太严，二是去重把大量正常数据误判为重复。
解决办法：先统计各层过滤的丢弃率，定位是哪一层丢得最多。如果是规则层，放宽阈值；如果是去重层，调高 MinHash 的 threshold。另外，可以引入分级保留机制，中等质量数据不要全丢，按比例采样保留。
5.2 质量分类模型误判严重怎么调
模型误判通常有两个原因：标注数据质量不高，或者模型过拟合。我的处理步骤是：先检查标注数据，看有没有标错的；如果标注没问题，就增加数据量，或者用数据增强扩充样本；还可以尝试用更大的预训练模型做 backbone，比如从 BERT-base 换成 BERT-large。
5.3 MindSpore 数据管道速度慢的优化
数据管道慢通常卡在map和filter操作上。优化手段有几个：提高num_parallel_workers，让多个进程并行处理；把 Python 操作改成 MindSpore 原生算子，比如用ds.transforms里的算子替代自定义 Python 函数；预取数据，用dataset.prefetch提前加载下一批数据。
| 问题 | 排查方向 | 解决方案 |
|---|---|---|
| 过滤后数据量骤降 | 统计各层丢弃率 | 放宽阈值，分级保留 |
| 模型误判严重 | 检查标注数据 | 增加标注量，换更大模型 |
| 管道速度慢 | 看 CPU 利用率 | 提高并行度，用原生算子 |
| 去重内存溢出 | 看数据规模 | 分片处理，用布隆过滤器 |
| 编码错误中断 | 看报错日志 | 用忽略策略，统一转 UTF-8 |
最后分享一个小技巧：过滤流程一定要做日志记录，每条数据在哪一层被丢弃、丢弃原因是什么，都要记下来。这样出问题时能快速定位，也方便后续调参。我一开始没做日志，出了问题只能从头跑一遍，浪费了大量时间。
这套方案我在多个预训练项目里反复打磨过，从几十 GB 到 TB 级数据都跑过，整体稳定性不错。核心就一句话：分层过滤，逐级收紧，规则先行，模型兜底，评估闭环。你如果刚开始做，建议先用小规模数据跑通全流程，再逐步放大，不要一上来就怼 TB 级数据，那样调试成本太高。
