大模型预训练数据质量过滤实战：基于MindSpore的完整方案与避坑指南

1. 大模型预训练里数据质量过滤为什么是生死线做过大模型预训练的人都有一个共识模型效果的上限很大程度上在数据准备阶段就已经被决定了。你后面用多少张卡、跑多少天、调多少超参都只是在逼近这个上限而已。而数据质量过滤就是决定这个上限的核心环节。我参与过几个从零开始的预训练项目也帮朋友排查过不少“loss 降不下去”“模型胡说八道”“训练到一半梯度爆炸”的问题最后追根溯源十有八九都指向同一件事——喂进去的数据太脏了。重复文本、乱码、广告、模板化内容、低信息密度的口水话这些东西混在语料里模型学到的就是这些垃圾的分布。你指望它输出高质量内容等于指望一个天天看地摊文学的人写出学术论文。MindSpore 作为国产深度学习框架在大模型预训练这条链路上已经相当成熟。但框架本身不会帮你解决数据质量问题它只提供计算能力。数据质量过滤方案需要你自己设计、自己实现、自己迭代。这篇文章我就把我在 MindSpore 环境下做大模型预训练数据过滤的完整思路和实操细节拆开讲包括规则设计、阈值选择、并行加速、常见坑以及我踩过的那些血泪教训。适合谁看如果你正在用 MindSpore 做预训练或者微调手头有一堆原始语料不知道从哪下手清洗或者你已经做了一版过滤但效果不理想那这篇内容应该能帮你省下不少试错时间。如果你还没接触过预训练但想了解数据工程在大模型里的实际分量也可以顺着看下去我会尽量用大白话把原理讲清楚。2. 数据质量过滤的整体设计与思路拆解2.1 为什么不能只靠“人工看一眼”很多人第一次做预训练数据准备时直觉反应是“我抽几百条看看不就行了”。这个思路在小规模场景下勉强能用但大模型预训练的语料量级通常是 TB 级别动辄几十亿到上百亿条文本。你人工抽一万条覆盖率连百万分之一都不到根本代表不了整体分布。更关键的是数据质量问题不是均匀分布的。它往往集中在某些来源、某些时间段、某些采集渠道。你随机抽样很可能恰好避开了最脏的那部分。我见过一个案例团队抽检时觉得数据还行结果训练出来的模型在特定领域表现极差回头一查发现那个领域的语料里有大量机器翻译的残留文本格式混乱、语义断裂但占比只有百分之几随机抽样很难命中。所以数据质量过滤必须走自动化、多维度、可量化的路线。核心思路是设计一组可计算的指标对每条文本打分然后根据分数决定保留、降权还是丢弃。2.2 过滤方案的分层设计我在实际项目中通常把过滤分成三层从粗到细逐层收紧第一层是格式与编码层。这一层最便宜计算量最小主要解决“这条数据能不能用”的问题。比如编码错误、HTML 标签残留、超长无意义字符、纯数字或纯符号、乱码等。这一层过滤完通常能去掉 5% 到 15% 的原始数据。第二层是规则与统计层。这一层针对文本的统计特征做判断比如重复率、句子长度分布、标点符号比例、停用词占比、特殊字符密度等。这一层是过滤的主力能去掉大部分低质量内容通常再去掉 20% 到 40%。第三层是模型与语义层。这一层成本最高但能解决前两层搞不定的问题比如语义重复、机器生成的低质文本、模板化内容、信息密度极低的段落。常用手段包括用小型分类模型打分、用困惑度筛选、用嵌入向量做去重等。这一层通常再去掉 10% 到 20%。三层叠加下来最终保留的数据量大概是原始的 40% 到 60%。听起来很浪费但你要这么想用 50% 的高质量数据训练效果远好于用 100% 的混合数据。这一点我在多个项目里反复验证过不是玄学是实打实的实验结论。2.3 MindSpore 在这个流程里的角色MindSpore 本身提供了mindspore.dataset模块里面有丰富的文本处理算子比如TextFileDataset、Tokenizer、BucketBatch等。但说实话数据过滤阶段我很少直接用 MindSpore 的 dataset 做主力原因很简单过滤阶段的核心瓶颈是 CPU 密集型的文本处理而不是 GPU 计算。用 MindSpore 的图模式跑这些逻辑反而绕远了。我的做法是过滤阶段用 Python 多进程 高效文本处理库如 regex、datasketch完成输出干净的数据文件然后用 MindSpore 的 dataset 接口读取干净数据做 tokenization 和 batch 组装。这样分工明确各取所长。当然如果你想把过滤逻辑也纳入 MindSpore 计算图技术上可行但调试成本和开发效率都不划算。提示不要为了“全流程统一框架”而强行把所有环节塞进 MindSpore。工具是拿来解决问题的不是拿来供着的。3. 核心细节解析与实操要点3.1 格式与编码层的具体规则这一层的目标是快速剔除“根本没法用”的数据。我常用的规则包括编码检测用chardet或ftfy检测文本编码如果无法识别或修复后仍然乱码直接丢弃。HTML/XML 标签残留用正则匹配[^]如果标签密度超过阈值比如每百字超过 3 个说明这条数据是从网页粗暴抓取的丢弃或重新清洗。控制字符ASCII 码 0-31 中除了\n、\t之外的字符出现即标记。超长无空格字符串连续超过 100 个字符没有空格或标点通常是编码错误或爬虫拼接问题。纯符号/纯数字整条文本去掉标点和数字后长度为 0直接丢弃。这些规则实现起来很简单但顺序很重要。我一般先做编码检测再做标签清洗最后做字符级检查。因为编码错误会导致后续所有正则匹配失效必须先解决。import re import ftfy def basic_clean(text): # 修复编码问题 text ftfy.fix_text(text) # 去除 HTML 标签 text re.sub(r[^], , text) # 去除控制字符 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f], , text) return text.strip() def is_valid_format(text, min_len50, max_len100000): if len(text) min_len or len(text) max_len: return False # 检查是否有连续超长无空格串 if re.search(r\S{100,}, text): return False # 检查是否纯符号 if not re.search(r[\u4e00-\u9fa5a-zA-Z], text): return False return True这段代码看着简单但实际跑起来ftfy.fix_text是性能大户。在 TB 级数据上单进程跑根本来不及。我的做法是用multiprocessing.Pool开满 CPU 核心数每个进程处理一个数据分片。实测下来32 核机器处理 100GB 文本大概需要 2 到 3 小时可以接受。3.2 规则与统计层的指标设计这一层是过滤的核心战场指标设计得好不好直接决定最终数据质量。我通常关注以下几类指标重复率指标。重复是大模型预训练的头号敌人。重复文本会让模型过度拟合某些模式导致输出多样性下降严重时甚至引发训练不稳定。我一般做两个粒度的去重句子级去重用 MinHash 或 SimHash 对句子做指纹相似度超过 0.8 的视为重复。文档级去重对整篇文档做 MinHash相似度超过 0.7 的保留一篇。这里有个经验值中文语料的重复率通常比英文高因为中文互联网内容搬运现象更严重。我做过的一个中文项目原始语料去重后直接少了 35%。所以中文预训练一定要把去重做扎实。长度分布指标。文本长度不是越长越好也不是越短越好。太短的文本信息量不足太长的文本可能包含多个主题影响模型学习。我一般会统计语料的长度分布然后掐掉两端的长尾。比如保留 50 到 5000 字符之间的文本具体阈值根据语料特点调整。标点与符号比例。正常文本的标点比例大概在 5% 到 15% 之间。如果标点比例过低可能是机器生成的连续文本如果过高可能是代码或表格残留。特殊符号如#%……*密度过高也要警惕通常是广告或垃圾内容。停用词与实词比例。这个指标稍微复杂一点但很有用。正常文本里停用词占比大概 30% 到 50%如果停用词占比异常高说明文本可能是“的的的的”这种无意义重复如果异常低可能是关键词堆砌的 SEO 垃圾。下面是一个统计指标的计算示例import re from collections import Counter def compute_stats(text): total_chars len(text) if total_chars 0: return None # 标点比例 punctuation len(re.findall(r[。、,.!?;:], text)) punct_ratio punctuation / total_chars # 特殊符号比例 special len(re.findall(r[#%……*()\[\]{}], text)) special_ratio special / total_chars # 数字比例 digits len(re.findall(r\d, text)) digit_ratio digits / total_chars # 平均句子长度 sentences re.split(r[。.!?], text) sentences [s for s in sentences if len(s.strip()) 0] avg_sent_len sum(len(s) for s in sentences) / max(len(sentences), 1) return { punct_ratio: punct_ratio, special_ratio: special_ratio, digit_ratio: digit_ratio, avg_sent_len: avg_sent_len, num_sentences: len(sentences) }拿到这些统计量之后怎么定阈值我的经验是先统计全量数据的分布然后看分位数。比如标点比例我一般保留 5% 到 95% 分位之间的数据两端的各 5% 丢弃。这样既去掉了异常值又不会误杀太多正常数据。3.3 模型与语义层的筛选策略前两层做完数据已经干净很多了但还有一些“看起来正常、实际上没营养”的内容。比如机器翻译生成的生硬文本模板化的新闻通稿问答社区里“顶”“沙发”“路过”这类无意义回复内容重复但表述略有不同的洗稿文章这些问题靠规则很难解决需要上模型。我常用的方案有两种方案一用小分类模型打分。训练一个二分类模型高质量 vs 低质量用人工标注的几千条数据做训练然后对全量语料打分。模型不用很大BERT-base 级别就够了。在 MindSpore 里可以用mindspore.nn.BertModel加载预训练权重接一个分类头训练几个 epoch 就能用。方案二用困惑度筛选。用一个在高质量语料上训练好的语言模型计算每条文本的困惑度。困惑度特别高的文本通常是语法混乱或语义不通的困惑度特别低的可能是重复或模板化内容。掐掉两端保留中间部分。这两种方案各有优劣。分类模型更准但需要标注数据困惑度不需要标注但阈值不好定。我一般先用困惑度做粗筛再用分类模型做精筛两者结合效果最好。注意语义层过滤的计算成本很高建议放在流程最后只对前两层保留下来的数据做处理。否则你会浪费大量算力在明显该丢弃的数据上。4. 实操过程与核心环节实现4.1 数据分片与并行处理框架TB 级数据不可能单机单进程处理必须做分片和并行。我的标准做法是按文件大小分片。把原始数据切成 1GB 左右的分片文件每个分片独立处理。用多进程池并行。每个进程处理一个分片进程数设为 CPU 核心数的 1.5 倍考虑 I/O 等待。每个进程内部用流式处理。不要一次性把整个分片读进内存用生成器逐行读取、逐行处理、逐行写出。import os import json from multiprocessing import Pool def process_shard(shard_path, output_dir): output_path os.path.join(output_dir, os.path.basename(shard_path)) with open(shard_path, r, encodingutf-8) as fin, \ open(output_path, w, encodingutf-8) as fout: for line in fin: try: record json.loads(line) text record.get(text, ) # 第一层格式清洗 text basic_clean(text) if not is_valid_format(text): continue # 第二层统计过滤 stats compute_stats(text) if not pass_stat_filter(stats): continue # 写出干净数据 fout.write(json.dumps({text: text}, ensure_asciiFalse) \n) except Exception as e: # 记录错误但不中断 continue def run_parallel(input_dir, output_dir, num_workers32): shards [os.path.join(input_dir, f) for f in os.listdir(input_dir)] os.makedirs(output_dir, exist_okTrue) with Pool(num_workers) as pool: pool.starmap(process_shard, [(s, output_dir) for s in shards])这个框架跑起来很稳但有几个细节要注意异常处理要细。单条数据出错不能影响整个分片所以try-except要包在循环内部。输出要及时 flush。多进程写文件时缓冲区可能积压定期 flush 避免数据丢失。进度要可见。我一般用tqdm加在每个分片的处理循环上方便观察进度。4.2 去重环节的 MinHash 实现去重是过滤流程里最耗时的环节之一。我试过几种方案最终稳定用的是MinHash LSH。原理不复杂把每条文本转换成一组哈希签名相似文本的签名会有很高概率相同然后通过分桶快速找到候选重复对。在 Python 里可以用datasketch库但那个库在超大规模数据上性能一般。我的做法是自己实现一个简化版用numpy做向量化计算import numpy as np import hashlib def minhash_signature(text, num_hashes128, shingle_size5): # 生成 shingles tokens text.split() shingles set() for i in range(len(tokens) - shingle_size 1): shingles.add( .join(tokens[i:ishingle_size])) if not shingles: return None # 计算 MinHash 签名 signature np.full(num_hashes, np.inf) for shingle in shingles: for i in range(num_hashes): h int(hashlib.md5(f{i}_{shingle}.encode()).hexdigest(), 16) if h signature[i]: signature[i] h return signature def jaccard_estimate(sig1, sig2): return np.mean(sig1 sig2)这个实现比datasketch快不少但内存占用也更大。128 个哈希、每条签名 128 个 int64一亿条数据大概需要 100GB 内存。如果内存不够可以降到 64 个哈希精度略降但可接受。实际去重时我一般分两步先做精确去重完全相同的文本直接删再做近似去重MinHash 相似度超过阈值的保留一条。精确去重可以用set或数据库唯一索引速度极快。近似去重才是 MinHash 的主场。4.3 与 MindSpore 数据管道的对接过滤完的数据最终要喂给 MindSpore 训练。我的做法是把干净数据存成 JSONL 格式然后用mindspore.dataset.TextFileDataset读取import mindspore.dataset as ds import mindspore.dataset.text as text def create_dataset(data_path, batch_size32, max_length512): dataset ds.TextFileDataset(data_path, shuffleTrue) # 分词 tokenizer text.BertTokenizer(vocab_filevocab.txt) dataset dataset.map(operationstokenizer, input_columns[text]) # 截断和填充 dataset dataset.map( operationstext.TruncateSequencePair(max_length), input_columns[text_ids] ) dataset dataset.batch(batch_size, drop_remainderTrue) return dataset这里有个坑MindSpore 的 dataset 默认是单线程的如果数据量大读取会成为瓶颈。可以在map里加num_parallel_workers参数开多线程dataset dataset.map( operationstokenizer, input_columns[text], num_parallel_workers8 )另外如果过滤后的数据还是很大建议先转成 MindSpore 的MindRecord格式读取效率会高很多。转换用mindspore.mindrecord.FileWriter就行官方文档有详细说明。5. 常见问题与排查技巧实录5.1 过滤后数据量骤降怎么办这是最常见的问题。很多人第一次跑完过滤发现只剩 20% 的数据直接慌了。我的建议是先别急着调阈值先分析丢弃原因。我一般会在过滤代码里加一个统计模块记录每条数据被哪条规则丢弃。跑完之后汇总看看哪个规则的丢弃率最高。如果某条规则丢弃了 50% 以上的数据那大概率是阈值设得太严了。比如我有一次发现“标点比例”规则丢弃了 40% 的数据回头一看阈值设的是 5% 到 15%但实际语料里很多技术文档的标点比例只有 3% 左右。把下限调到 2% 之后丢弃率降到 8%数据质量也没明显下降。提示过滤阈值没有绝对标准一定要结合你的语料特点来定。先统计再定阈值不要拍脑袋。5.2 去重后仍然有大量相似内容MinHash 去重有个特点它对“整体相似”敏感但对“局部重复”不敏感。比如两篇文章开头一样、结尾一样中间不同MinHash 可能认为它们不重复。但实际训练时这种局部重复也会让模型学到重复模式。解决方法是做段落级去重。把文档切成段落对每个段落做 MinHash然后统计文档内重复段落的占比。如果占比超过阈值比如 30%整篇文档降权或丢弃。另外模板化内容也是去重的难点。比如“XX公司成立于XX年总部位于XX”这种句式换几个词就是一条新数据。MinHash 很难识别这种。我的做法是额外做一个n-gram 频率统计如果某些 n-gram 在全量数据里出现频率异常高就把包含这些 n-gram 的文本标记为模板化内容降权处理。5.3 过滤流程跑得太慢TB 级数据过滤跑一两天很正常。但如果你发现跑了一周还没完那肯定有问题。常见的性能瓶颈和优化手段瓶颈环节表现优化手段编码检测CPU 占用高速度慢用ftfy替代chardet或先做快速 ASCII 检测正则匹配单条处理耗时长预编译正则合并多个正则为一次匹配MinHash内存占用高速度慢减少哈希数量用 numpy 向量化文件 I/O磁盘读写慢用 SSD输出用缓冲写入多进程通信进程间等待减少进程间数据传递每个进程独立读写我实测下来预编译正则能提升 30% 以上的速度用 numpy 做向量化能提升 MinHash 5 到 10 倍。这两个优化最值得做。5.4 过滤后的数据分布偏移这个问题比较隐蔽。你过滤完之后数据量少了但数据分布可能也变了。比如你过滤掉了大量短文本结果剩下的都是长文本模型训练时就会偏向长文本的分布。我的做法是过滤前后都做分布统计对比长度分布、领域分布、来源分布。如果发现某个维度偏移超过 10%就要考虑是不是过滤规则太激进或者需要做重采样。举个例子我有一次过滤后发现法律领域的语料少了 60%原因是法律文本里有很多条款编号和特殊符号被“特殊符号比例”规则误杀了。后来我针对法律领域单独放宽了阈值问题就解决了。5.5 常见问题速查表问题可能原因排查方法解决方案数据量骤降阈值过严统计各规则丢弃率放宽高丢弃率规则的阈值仍有重复内容局部重复未识别抽样检查重复类型增加段落级去重处理速度慢单进程瓶颈监控 CPU 和 I/O多进程 预编译正则分布偏移过滤规则有偏对比过滤前后分布分领域调整阈值内存溢出MinHash 签名过大监控内存占用减少哈希数量或分批处理输出文件损坏多进程写冲突检查文件完整性每个进程独立输出文件6. 我踩过的坑和最后分享几个实用技巧第一个坑是过度依赖单一指标。我早期做过滤时特别迷信“困惑度”这个指标觉得困惑度低就是好数据。结果发现很多重复文本的困惑度极低因为模型见过太多次了。后来我把困惑度和去重结合使用才解决了这个问题。任何单一指标都有盲区多指标交叉验证才是正道。第二个坑是忽略数据来源的差异性。不同来源的数据质量分布完全不同。网页抓取的数据噪声大书籍扫描的数据格式乱论坛数据口语化严重。我一开始用同一套阈值处理所有来源结果要么网页数据过滤不干净要么书籍数据被误杀太多。后来改成按来源分组每组独立定阈值效果好了很多。第三个坑是过滤完不做验证。过滤不是跑完就完事了一定要做验证。我的验证方法是从过滤后的数据里随机抽 1000 条人工标注质量算一下准确率。如果准确率低于 90%说明过滤还不够如果高于 98%可能过滤太狠了误杀了不少好数据。这个验证步骤花不了多少时间但能帮你避免很多后续问题。最后分享几个实用技巧先用小样本调参。不要一上来就跑全量数据先拿 1% 的数据做实验把阈值调好再上全量。这样能省下大量时间。保留过滤日志。每条数据被丢弃的原因都记下来后续分析问题时有据可查。定期更新过滤规则。数据在变过滤规则也要跟着变。我一般每季度重新审视一次规则根据新数据的特点做调整。不要追求完美。过滤的目标是提升数据质量不是做到零噪声。保留少量噪声反而能提升模型的鲁棒性。我见过有人把过滤做到极致结果模型在真实场景下表现很差因为训练数据太“干净”了和真实分布脱节。数据质量过滤这件事说到底是个工程活 经验活。工具和框架只是基础真正决定效果的是你对数据的理解和对业务场景的把握。多跑、多看、多总结慢慢就能找到适合自己项目的节奏。