做预训练的朋友应该都有同感大模型的训练代码其实不难搭难的是把喂进去的数据收拾干净。标题里这三个词连在一起——MindSpore、大模型预训练、数据质量过滤——意味着我们要在国产AI框架里把从原始语料到可训练语料的“清洗流水线”真正跑起来。这篇文章不是讲理论而是我自己在MindSpore上做数据治理时沉淀下来的一套方案从为什么要过滤、过滤分哪几层到算子怎么写、阈值怎么调、踩过哪些坑尽量都讲到。适合两类人看一类是刚接触预训练、正被“数据不够干净”折磨的同学另一类是已经在跑清洗流程但想优化过滤效率和效果的数据工程师。1. 为什么预训练数据质量过滤在MindSpore里是个绕不开的课题1.1 脏数据对预训练的影响远比想象中大先说结论大模型预训练七分靠数据三分靠训练。数据质量决定了模型能力的上限训练只是让模型尽量逼近这个上限。最典型的问题是重复数据。爬虫拿到的网页里同一篇文章经常被门户网站、转载号、内容农场反复发布正文可能一字不差。训练时模型反复看到同样的token序列会把这些片段“背”得特别牢。这不是好事它会让模型在其他相关的、但表达不同的文本上泛化变差。更直接的表现是训练集里重复数据的比例越高模型越容易出现“熟读背诵、换个说法就懵”的现象。其次是噪声数据。网页里混着导航栏、版权声明、广告、乱码字符、空壳页面。这些内容如果不过滤会拉高训练的loss造成loss曲线反复震荡。尤其是一些控制字符、异常编码对tokenizer非常不友好轻则增加词表负担重则让分词结果完全错乱相当于给模型喂了毒药。再有就是安全性问题。公开语料的来源很杂用户生成内容里可能夹杂个人隐私、敏感信息、冒犯性表述。这类内容一旦进入预训练语料模型就会在生成时“有样学样”。后期想靠对齐来抹掉这些影响成本极高效果也不见得彻底。所以数据质量过滤不只是“让loss更好看”它是预训练上线前必须完成的一步也是影响最终模型可靠性的关键一关。1.2 为什么直接在MindSpore里做而不是事先在外面处理好很多团队习惯先在离线Spark或者Hadoop上把数据洗好再拷给训练集群。这种思路没错但到了MindSpore场景下我更推荐在训练侧的数据pipeline里再叠加一层过滤算子。原因有几个。第一是分片语义统一。MindSpore的mindspore.dataset天然支持多卡数据并行下的自动分片shard参数能保证每张卡拿到的数据不重叠。如果你在外部洗数据还得自己处理“到底哪些数据该分给哪张卡”的逻辑一旦分片和训练并行配置对不上就会出现数据重复或漏训。第二是过滤规则可以复用到微调和持续训练阶段。预训练结束后做领域增量训练或大模型微调时同样需要清洗数据。把过滤逻辑写成dataset算子同一个项目仓库里维护后续在微调脚本里直接复用就行不用再开一套离线脚本。第三是性能上有便宜占。map、filter这些算子底层是C实现的天然支持多线程并行。相比之下在Python里面写一个for循环逐条处理几千万条文本速度差距是数量级的。数据越大这个优势越明显。所以我的做法是离线做重活大规模去重、语言筛选MindSpore侧做轻量级拦截质量阈值、格式修补两层配合各管一段。2. 数据质量过滤方案怎么拆我把它分成四层2.1 第一层格式清洗与字符级噪声处理这一层的目标是解决“文本不像人写的”问题。常见问题包括乱码、控制字符、全角半角混杂、多余空白、HTML标签、URL、emoji符号残留等。我常用的一组基础正则大概是这样的import re def clean_text(text: str) - str: # 去掉控制字符 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f], , text) # 去掉HTML标签 text re.sub(r[^], , text) # 去掉链接 text re.sub(rhttps?://\S, , text) # 压缩连续空白 text re.sub(r\s, , text) return text.strip()这里有个容易被忽略的点清洗规则别太贪心。比如去掉所有URL会导致代码文档里的链接、论文里的DOI引用一并被删掉把标点全杀光又会让文本失去边界信息。我用过一个很激进的“清理非字母数字”规则结果一段正常的Python代码被洗成了一堆残缺单词模型在代码任务上的表现明显变差。这个阶段最好像择菜烂叶子要去掉但别把能吃的菜帮也削了。特别是代码、公式、表格类的文本本身结构性强过度清洗等于白白扔掉高价值数据。建议针对不同的语料类型配置不同的清洗模板代码类语料保留换行和缩进自然语言类语料做强清洗。2.2 第二层内容层面的质量过滤格式层做完文本仍然可能是“干干净净的废话”比如只有几个字的页面、纯符号堆砌、重复刷屏的句子。这一层要解决的就是“内容信息量太低”的问题。我一般会组合三个手段长度过滤长度低于某个阈值的文本直接丢弃。中英文预训练语料里我通常会把下限设在30~50个字符。太短的文本没有足够的上下文模型学不到有效语法特征还容易引入碎片化噪声。语言过滤预训练目标是中文就直接统计字符落在\u4e00-\u9fff区间的比例。如果一段文本里中文字符占比不到20%大概率不是中文内容。想更严谨可以接入语言检测模型但大规模场景下用Unicode区间做初筛已经很划算。启发式规则纯标点、大量重复字符比如aaaaaa、疑似广告模板“点击查看详情”“限时优惠”、全英文符号混合异常大小写等都可以通过规则秒杀。这些规则组合起来可以变成一个打分函数每个文本打一个0到1的质量分。我实战里用一个很朴素的公式def quality_score(text: str) - float: if not text: return 0.0 total len(text) chinese_chars len(re.findall(r[\u4e00-\u9fff], text)) alpha_chars len(re.findall(r[A-Za-z0-9], text)) coverage (chinese_chars alpha_chars) / total # 连续重复字符惩罚 repeat_count len(re.findall(r(.)\1{9,}, text)) repeat_penalty max(0.0, 1.0 - repeat_count * 0.2) score 0.7 * coverage 0.3 * repeat_penalty return min(1.0, max(0.0, score))权重不是拍脑袋定的。coverage权重高是因为清洗后的正常文本应该以有意义的文字为主连续重复字符是典型的低质信号所以用乘性惩罚拉低整体分数。阈值怎么选建议先别设死。跑一个小规模抽样把打分结果打印出来看分布。比如抽取5000条样本按分数从低到高排序人工翻一翻低分段到底是些什么文本。如果目标是保留70%的数据就从分布上找70分位对应的分数作为初始阈值再迭代调整。这个步骤很重要直接跳过的话后面很可能出现“过滤后数据量不足”或“低质文本漏网”两个极端。2.3 第三层去重与近重复检测网页类语料里转帖、镜像站、聚合站带来的重复问题非常严重。我见过一个公开爬虫语料做完精确去重后数据量直接掉了20%。而近重复也就是标题不同、正文几乎一样的内容比精确重复更隐蔽影响也更大。先做精确去重这是成本最低的一步。把每个样本的全文做MD5或SHA1哈希用一个集合记录已经出现过的哈希值重复的直接丢弃。这里的“全文”建议只取正文文本不包含标题、时间、作者等元信息否则同一条新闻在不同来源里会因标题改写而漏去重。再做近重复检测我推荐用MinHash或SimHash的思路。核心是把文本切分成连续的n-gram集合比如每5个字符切成一个token然后映射成哈希值。两篇文本的相似程度近似等于这两个哈希集合的Jaccard相似度。大规模场景下不需要两两全比较用局部敏感哈希做分桶把可能相似的文本分到同一个桶里再在桶内做精确相似度计算能省掉大量计算。实战里有个细节去重这种依赖全局状态的逻辑不要放在并行map算子内部做。因为并行算子各线程之间没法同步维护同一个“已见集合”要么多算要么错删。我现在的做法是离线阶段先把重复文本的ID集合算好进入MindSpore的filter算子后只做一次集合查询安全也高效。2.4 第四层安全与合规过滤这一层很多人会漏掉但预训练语料是迟早要面对的。公开语料里混着手机号、身份证号、银行卡号、家庭住址等个人敏感信息也可能包含有害内容。如果模型把这些学到生成阶段就可能在无意识间“背出”隐私内容后果非常严重。我的做法是关键词规则和模型打分双轨制规则层用正则匹配手机号、邮箱、身份证号等模式。这类目标明确规则稳定直接过滤或者用脱敏符替换都可以。模型层训练一个轻量文本分类器对文本做风险分级。高风险样本进人工复检中风险样本降权或剔除。对超大规模语料先用规则把明显问题过滤掉剩下的可疑样本才送模型不然全量推理成本太高。这里也要把握度。过度安全过滤会误伤正常的中性讨论比如“如何防止隐私泄露”这类话题里一样会出现“手机号”“身份证”字样。我的经验是规则层宁可窄不可宽只对明确的隐私模式下手语义层面的判断交给分类模型不要为了省事写一个宽泛的敏感词表一棍子打死。3. MindSpore实操一条可落地的过滤流水线3.1 先搭一个最小可用的Dataset流程以MindSpore 2.x为例核心API是mindspore.dataset里的GeneratorDataset、map、filter。最小流程是这样的import re from mindspore.dataset import GeneratorDataset def read_samples(file_path): with open(file_path, r, encodingutf-8) as f: for line in f: line line.strip() if line: yield {text: line} def clean_text(text): text re.sub(r[^], , text) text re.sub(rhttps?://\S, , text) text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f], , text) text re.sub(r\s, , text).strip() return text def quality_filter(text): if len(text) 30: return False if len(re.findall(r[\u4e00-\u9fff], text)) / len(text) 0.3: return False return True dataset GeneratorDataset( sourceread_samples(raw_data.txt), column_names[text] ) dataset dataset.map(operationsclean_text, input_columns[text], num_parallel_workers8) dataset dataset.filter(predicatequality_filter, input_columns[text], num_parallel_workers8) dataset dataset.shuffle(buffer_size10000)read_samples是一个生成器函数按行迭代读取文件这样就不会把整个文件一次性塞进内存。遇到几百GB的原始语料这一点很关键。map负责对每一条数据执行清洗函数filter根据predicate函数的返回结果决定保留还是丢弃。num_parallel_workers8的意思是同时用8个线程并行处理能明显提升吞吐。但这里也提醒一句并行度不是越大越好开太多线程会带来调度开销甚至跟训练抢占CPU资源。我一般在4到16之间做一次小实验看吞吐不再明显提升时就停住。3.2 把质量打分做成可配置的过滤算子上一节里的quality_filter是硬编码规则不好维护。实际项目中我更推荐把打分函数和阈值拆开做成配置项。举个例子我把上面那个quality_score做成独立的模块然后在配置里控制阈值# config.yaml quality: min_length: 30 min_score: 0.45 coverage_weight: 0.7 repeat_penalty_weight: 0.3过滤算子变成def make_quality_filter(cfg): min_length cfg[quality][min_length] min_score cfg[quality][min_score] def quality_filter(text): if len(text) min_length: return False score quality_score(text) return score min_score return quality_filter dataset dataset.filter( predicatemake_quality_filter(cfg), input_columns[text], num_parallel_workers8 )什么时候调阈值我的习惯是每调整一次参数先在一个抽样出的子集上跑一遍统计看淘汰率和保留文本的样例。直接把阈值从0.4改成0.7看起来只是改一个数字但如果数据分布变化就可能把有用文本大量误杀。调参数之前一定要先知道当前数据集的分数分布长什么样。from collections import Counter import random score_counter Counter() for i, data in enumerate(dataset): score quality_score(data[text]) score_counter[int(score * 10) / 10] 1 if i 20000: break print(score_counter)这个统计逻辑不要在全量数据上跑2万条样本足够看出大致分布了。3.3 分布式场景下的过滤策略预训练一般都在多卡环境下跑。MindSpore里并行训练时每个rank只处理数据的一个分片。最简单的用法是手动分片from mindspore.communication import init, get_rank, get_group_size init() rank_id get_rank() world_size get_group_size() dataset GeneratorDataset(source..., column_names[text]) dataset dataset.shard(num_shardsworld_size, shard_idrank_id)shard要在shuffle之前做原因是先洗牌再分片能打散数据分布先分片再洗牌会让每张卡的数据都集中在某个区间影响训练效果。另外固定的随机种子很重要。多卡场景下如果不固定shuffle的seed每张卡之间可能产生不确定的数据顺序最后你都没法复现实验结果。再提醒一次filter里千万别做带全局状态的逻辑。比如seen set() # 这样写在并行filter里是有问题的 def dedup_filter(text): if text in seen: return False seen.add(text) return True这种写法在单线程小样本下能跑一旦num_parallel_workers开大或者多卡各跑各的seen集合之间互相不感知去重等于没做。真想在MindSpore里做在线去重也得先把去重ID集合预先算好然后filter里只查集合。清洗和训练最好解耦。我自己更推荐的做法是先跑一遍清洗流程把过滤结果写回一份“干净语料”训练脚本直接读这份干净语料而不是每次训练都把原始语料从头洗一遍。好处是规则调整时只需重跑清洗流程训练代码保持干净稳定坏处是多一次磁盘读写但在预训练这种按天计的场景里这点IO成本完全可接受。3.4 清洗后的数据接入预训练脚本清洗完的数据最好存成MindRecord或者TFRecord格式再交给训练脚本读。纯文本文件每行一条虽然直观但读取效率不高尤其是多卡训练时容易成为IO瓶颈。我习惯在清洗完成后把数据写成MindRecordfrom mindspore.mindrecord import FileWriter writer FileWriter(file_namefiltered.mindrecord, shard_num8) schema {text: {type: string}} writer.add_schema(schema) writer.add_index([text]) for text in filtered_texts: writer.write_raw_data([{text: text}]) writer.commit()这里filtered_texts是清洗完成后的生成器实际可以边读边写避免把所有文本攒在内存里。不同MindSpore版本的FileWriter接口略有差异用的时候以官方API文档为准。训练时直接读from mindspore.dataset import MindRecordDataset dataset MindRecordDataset( dataset_filesfiltered.mindrecord, columns_list[text], shard_equal_rowsFalse )如果只是少量微调或领域增量训练用GeneratorDataset也没问题灵活度高写起来也直观。但大规模预训练阶段我还是建议用MindRecord毕竟底层存储格式对IO的优化还是实打实的。4. 常见问题与排查技巧实录4.1 过滤完之后数据量急剧缩减这事我踩过不止一次。原始语料500GB过滤完只剩50GB直接导致训练数据不足。排查顺序是这样的先看是不是某些filter规则重复叠加了。比如长度过滤在map里做了一次又在filter里做了一次逻辑就会比预期严格一倍。其次看是不是规则写得太宽比如“去除所有包含URL的文本”会把大量正常网页正文连坐。我的建议是给每个过滤规则加上独立的计数日志在单线程、小样本模式下先看一眼各规则的淘汰比例再决定要不要全量跑。实际做法是stats {too_short: 0, low_score: 0, bad_char: 0} def debug_filter(text): if len(text) 30: stats[too_short] 1 return False score quality_score(text) if score 0.45: stats[low_score] 1 return False return True注意一旦开启num_parallel_workers这种Python字典的计数会在线程间乱跳所以只在单线程调试时用。线上版本不要依赖这个统计建议将每条样本的命中规则记录成元数据落到日志文件里再做聚合分析。4.2 Hash去重撞车与误删纯用MD5做全文去重理论上存在哈希碰撞但概率极低。真正的问题是近重复文本。两篇转载文章可能只有标题或首段不一样全文哈希完全不同但训练价值几乎一样。我用的是两层策略。第一层精确哈希去重成本低先干掉完全一样的。第二层对剩下的文本做SimHash。SimHash会把文本映射成一个64位指纹两篇文本的相似度可以用指纹的汉明距离衡量。通常汉明距离小于等于3的文本可视作近重复。这个阈值不是绝对的中文语料上我会从3开始试观察误删比例再调整。这个步骤我建议放到离线阶段做不要在MindSpore的训练pipeline里实时算SimHash不然每张卡都算一遍计算量直接翻倍。4.3 分布式数据重复或丢失表现是多卡训练时有的数据被两个卡都读到了有的数据一张卡都没读到。最常见原因是分片时机和洗牌顺序搞错。正确顺序是先shuffle再shard或者先shard再shuffle但要固定种子。另一个原因是filter里的随机性。如果过滤函数里用了random.random()这类不固定种子的函数在多卡场景下不同卡对同一条数据可能做出不同保留判断导致各卡数据量不一致。解决办法是过滤逻辑不允许用非确定性函数或者给随机逻辑也固定seed。排查方法很简单打印每个rank的数据条数和前几条样本人工对比一下有没有明显重叠。一旦发现不一致优先检查分片顺序和filter的确定性。4.4 正则误伤正常内容这块我最有发言权因为真干过这种傻事。为了过滤“纯数字短信语料”写过一条近似“匹配两个数字之间的所有内容”的正则结果把正常年份、软件版本号、论文引用编号全给删干净了模型在涉及时间表达的任务上直接变傻。核心教训是正则规则要窄不要为了省事扩大匹配范围。比如过滤URL只匹配带http://或https://的完整链接不要匹配“www”开头的片段因为“www”可能是正文里的正常缩写。再比如过滤HTML标签[^]只匹配尖括号包裹的内容但如果语料是代码a b and c d这种比较表达式就会误伤。代码类语料建议直接走独立的清洗模板不套用通用清洗规则。上线新规则之前强烈建议准备一份200条左右的人工标注“干净样本”新规则在这份样本上跑一遍如果命中率过高说明规则有问题。4.5 性能瓶颈与内存优化用GeneratorDataset读纯文本时如果source函数本身写得不好比如一次性把整个文件读成列表内存会直接爆炸。正确做法是生成器逐行读取这一点前面已经强调过了。还有一个隐蔽的性能问题统计数据集条数时用了len(list(dataset))。这个操作会把全量数据先加载到Python列表再数长度相当于把数据处理流程跑了整整一遍。如果只是做抽样统计用dataset.take(n)配合循环就好。num_parallel_workers的调优也值得单独说。开8个线程和开32个线程有时吞吐差异并不大反而因为CPU争抢导致训练变慢。我的经验是如果清洗和训练在同一台机器上跑清洗的并行度不要超过物理核数的一半。我把前面提到的几个问题整理成一张速查表方便排查现象可能原因排查方向过滤后数据量骤减规则叠加、阈值过高、正则过宽单条规则计数、抽样检查分数分布去重不彻底只做了精确去重近重复漏网加SimHash或MinHash近重复检测多卡数据重复/丢失分片顺序错误、filter包含随机逻辑先shuffle再shard、固定种子正常文本被误删正则匹配范围过宽白名单样本集验证、收窄规则清洗速度慢/内存高一次性读文件、并行度过高改生成器逐行读、降并行度5. 实战过程中的几点体会个人看法数据质量过滤不是一次性工程它更像是一个持续迭代的过程。数据分布会变模型任务会变过滤规则也要跟着变。所以我在项目里始终坚持把过滤流水线跟训练代码解耦清洗规则配置化去掉硬编码。另外一个很有价值的习惯是把过滤命中的原因记录下来。比如一条文本是因为长度被过滤还是因为质量分低被过滤这个信息保存下来之后既能用来自查规则是否合理也能在后续做数据配比分析时派上用场。看起来多写了几个字段实际上省掉了很多排查时间。最后分享一个后续可以扩展的方向用困惑度模型打分。规则能过滤掉“明显的脏数据”但对于“通顺但信息量极低”的长文本规则很难感知。一个小规模语言模型就能计算文本的困惑度把高困惑度的拼凑文本、语序混乱文本筛出来。这个可以作为第二阶段的增强方案等数据规模做大、规则收益见顶之后再上。