1. 大模型预训练里数据质量过滤到底在解决什么问题做过大模型预训练的人都有一个共识模型效果的上限很大程度上在数据准备阶段就已经被决定了。你后面用多少张卡、调多细的学习率、换多花哨的并行策略都很难弥补一批脏数据带来的伤害。MindSpore 作为一套全场景深度学习框架在预训练链路上提供了从数据加载、分布式切分到算子加速的完整能力但框架本身不会替你判断“这条数据该不该喂给模型”——数据质量过滤这一环必须由我们自己设计并落地。所谓数据质量过滤说白了就是在海量原始语料进入训练管道之前用一套可复现、可量化、可扩展的规则和模型把低质、重复、有害、无信息量的样本剔除或降权。它要解决的问题非常具体原始爬取语料里充斥着导航栏文本、广告、乱码、模板化段落、机器翻译腔、重复转载内容这些东西如果直接参与预训练轻则浪费算力重则让模型学会一堆无意义的模式表现为生成重复、答非所问、指令跟随能力差。这套方案适合谁参考我认为有三类人一是正在用 MindSpore 做领域大模型预训练、需要自己清洗语料的工程师二是做多模态大模型、需要处理图文配对数据质量的同学三是想把数据工程流程标准化的团队负责人。哪怕你用的是别的框架过滤思路和阈值设计逻辑也是通用的只是实现载体换成了 MindSpore 的 Dataset 管道和 MindRecord 格式。我下面会从整体设计、核心细节、实操落地、问题排查四个层面把一套可复现的数据质量过滤方案讲透。所有参数和阈值都来自实际项目调优经验不是拍脑袋写的你可以直接抄作业也可以按自己的语料分布微调。2. 整体方案设计与过滤层级拆解2.1 为什么采用“多级漏斗”而不是单次过滤很多人第一反应是写一个规则函数遍历所有数据不合格就丢掉。我早期也这么干过结果发现两个致命问题一是规则之间互相干扰比如长度过滤和重复过滤如果放在同一轮你很难判断一条被删的数据到底是因为太短还是因为重复二是算力浪费用模型打分这种重操作如果对全量数据跑成本高到离谱。所以正确的做法是设计成多级漏斗从便宜到昂贵逐级收敛。第一级用极低成本的规则快速砍掉明显垃圾第二级用统计特征做去重和分布过滤第三级才上模型打分。每一级的输出都落盘留痕方便回溯和调参。这套漏斗在 MindSpore 里可以很自然地用mindspore.dataset的map、filter、batch算子串起来配合GeneratorDataset做流式处理内存占用可控。具体分层我一般这么划层级过滤目标典型手段单条成本保留率经验值L1 规则层乱码、超短、超长、特殊符号占比正则、长度统计微秒级70%~85%L2 统计层重复、模板化、语言混杂MinHash、n-gram、语言识别毫秒级60%~75%L3 模型层低信息量、低质量、有害质量分类器、困惑度十毫秒级80%~92%保留率是逐级相乘的所以 L1 千万别设太狠否则后面没数据可用了。我见过有人 L1 就把保留率压到 40%最后整个语料不够训一个 epoch只能回炉重来。2.2 MindSpore 数据管道为什么适合承载这套方案选 MindSpore 做这件事核心原因是它的数据引擎支持算子级并行和流式处理。mindspore.dataset里的map操作可以指定num_parallel_workers过滤这种天然可并行的任务能直接吃满 CPU 多核。另外 MindRecord 格式对大规模语料的顺序读写做了优化比反复读 JSONL 快不少尤其在分布式训练场景下每个卡各自读自己那份分片不会出现 IO 争抢。还有一个容易被忽略的点MindSpore 的filter算子支持传入 Python 可调用对象这意味着你可以把自定义的过滤逻辑直接挂上去不用把数据先导出再处理。对于需要调用外部模型打分的场景可以用map配合output_columns把分数写回样本再用filter按分数阈值筛整条链路是声明式的改起来很清爽。提示如果你的语料规模在 TB 级以上建议先用 Spark 或 Ray 做 L1、L2 的粗筛把数据压到百 GB 量级再进 MindSpore 管道做 L3 精筛。全流程都塞进 MindSpore 不是不行但分布式调度和容错会更麻烦。2.3 过滤策略背后的取舍逻辑这里必须讲清楚一个反直觉的点过滤不是越干净越好。过度过滤会带来两个副作用一是数据分布偏移比如你把所有口语化表达都当低质删了模型在对话场景就会显得很生硬二是长尾知识丢失某些小众领域的内容天然就短、就“不流畅”但它们恰恰是模型稀缺的知识来源。我的经验是给每个过滤维度设一个软阈值和硬阈值。硬阈值直接删软阈值降权或者打标。比如困惑度这个指标超过 500 直接删200 到 500 之间降权到 0.5200 以下正常保留。降权在 MindSpore 里可以通过给样本加loss_weight字段实现训练时在 loss 计算里乘上去这样既不浪费数据又能抑制低质样本的影响。3. 核心过滤维度的细节解析与阈值设计3.1 规则层长度、字符分布与乱码识别规则层是最容易写、也最容易写错的。长度过滤看似简单但阈值怎么定有讲究。中文语料我一般设最小 50 字符、最大 100000 字符。最小 50 是因为低于这个长度的样本很难承载完整语义最大 10 万是为了防止个别超长文档把 batch 撑爆。英文语料最小可以放到 200 字符因为英文信息密度相对低。字符分布这块核心是统计有效字符占比。有效字符指中文、英文、数字、常用标点。如果一条样本里有效字符占比低于 0.7基本可以判定是乱码或者特殊符号堆砌。实现上用一个正则[\u4e00-\u9fa5a-zA-Z0-9。、]去匹配算比例。乱码识别还要看连续重复字符。比如“啊啊啊啊啊啊”或者“。。。。。。。。”这种用正则(.)\1{9,}就能抓出来连续重复超过 10 次直接删。另外还有一类是编码错误产生的“锟斤拷”式乱码这个用字符集白名单过滤最稳。import re def rule_filter(sample): text sample[text] if len(text) 50 or len(text) 100000: return False valid len(re.findall(r[\u4e00-\u9fa5a-zA-Z0-9。、], text)) if valid / max(len(text), 1) 0.7: return False if re.search(r(.)\1{9,}, text): return False return True这段逻辑挂到 MindSpore 的filter算子上就能跑num_parallel_workers设成 CPU 核数的 0.8 倍比较合适留点余量给其他进程。3.2 统计层MinHash 去重与 n-gram 模板检测重复数据是预训练里最隐蔽的杀手。它不会让模型报错但会让模型对重复内容过拟合生成时反复吐同一句话。去重分两种完全重复和近似重复。完全重复用哈希去重就行近似重复必须上 MinHash 或者 SimHash。MinHash 的原理我用大白话解释一下把每条文档切成若干 shingle比如连续的 5 个词对每个 shingle 算多个哈希值取最小值组成签名。两条文档的签名越接近说明它们越相似。Jaccard 相似度超过 0.8 就判定为近似重复保留其中一条即可。MindSpore 本身不提供 MinHash但可以用datasketch库先算好签名把签名作为字段写进 MindRecord训练管道里直接按签名分桶去重。n-gram 模板检测是另一个利器。很多网页正文其实是模板生成的比如“本文由 XX 编辑整理转载请注明出处”。这类文本的 3-gram 或 5-gram 重复率极高。我的做法是统计整个语料的 top-1000 高频 n-gram如果一条样本里高频 n-gram 占比超过 0.5就判定为模板化文本删掉。检测项阈值处理方式备注Jaccard 相似度 0.8保留一条分桶后组内比较高频 3-gram 占比 0.5删除高频表按语料统计语言混杂度主语言占比 0.85删除或分流用 fastText 识别重复行占比 0.3删除按行去重后统计语言混杂度这个维度特别重要中英夹杂严重的样本会让模型的语言建模能力变差。用 fastText 的语言识别模型跑一遍主语言置信度低于 0.85 的直接分流到对应语种别硬塞进中文语料。3.3 模型层质量分类器与困惑度打分到了模型层成本上来了但收益也最明显。这里有两个主流做法一是训一个质量分类器用人工标注的高质/低质样本做二分类二是用一个小语言模型算困惑度困惑度高的样本通常语义混乱。质量分类器我一般用 RoBERTa 或者小号的 BERT 做 backbone标注 5 万条左右就能到不错的水平。标注标准要统一我通常从四个维度打分信息密度、语言流畅度、事实准确性、表达完整性。每条样本让标注员打 1 到 5 分3 分以上算高质。训完之后对全量语料推理输出一个 0 到 1 的质量分。困惑度这条路更省事不需要标注。用一个在干净语料上训好的小模型比如 1 亿参数级别对每条样本算平均负对数似然。困惑度低于 200 算优质200 到 500 降权超过 500 删除。注意这里的小模型必须和预训练主模型的语言分布一致否则困惑度没有参考意义。注意模型层过滤一定要做分数校准。不同批次推理出来的分数分布可能有偏移建议先跑一个 10 万条的采样集画出分数直方图再定阈值。直接拍一个 0.5 的阈值很可能把一半数据误杀。在 MindSpore 里做模型推理过滤可以用mindspore.nn定义分类器加载权重后用model.predict批量推理。为了提速把 batch size 开到 256 以上配合mindspore.dataset的batch算子GPU 利用率能拉满。4. 实操落地从原始语料到 MindRecord 的完整流程4.1 环境准备与依赖清单先把环境搭起来。MindSpore 的安装按官方文档走CPU 版和 GPU 版都行过滤阶段其实 CPU 就够模型推理那步再上 GPU。Python 建议 3.8 到 3.10太新的版本有些依赖库还没跟上。pip install mindspore2.2.0 pip install datasketch fasttext jieba numpy tqdmdatasketch做 MinHashfasttext做语言识别jieba做中文分词辅助 n-gram 统计。这几个库都很轻装起来不费劲。如果你要用质量分类器还得装transformers来加载预训练权重或者直接用 MindSpore 的nn.Cell自己搭。4.2 数据加载与分片策略原始语料一般是 JSONL 或者 Parquet 格式。我习惯先转成 MindRecord因为后续反复读取会快很多。转换的时候按 1GB 左右切一个分片分布式训练时每个卡读一个分片避免单文件过大导致的 IO 瓶颈。import mindspore.dataset as ds from mindspore.mindrecord import FileWriter def jsonl_to_mindrecord(jsonl_path, mindrecord_path, shard_size100000): writer FileWriter(mindrecord_path, shard_num1) schema {text: {type: string}, lang: {type: string}} writer.add_schema(schema, pretrain_data) with open(jsonl_path, r, encodingutf-8) as f: buffer [] for line in f: item json.loads(line) buffer.append({text: item[text], lang: item.get(lang, zh)}) if len(buffer) shard_size: writer.write_raw_data(buffer) buffer [] if buffer: writer.write_raw_data(buffer) writer.commit()分片大小 10 万条是个经验值太小会导致文件数过多元数据管理麻烦太大则单次写入内存压力大。转换完之后用ds.MindDataset加载配合shuffle和num_parallel_workers就能开始过滤了。4.3 多级过滤管道的串联实现把前面讲的 L1、L2、L3 串成一条管道。关键点是每一级过滤后都要落盘方便统计各级保留率和排查问题。def build_filter_pipeline(mindrecord_path): dataset ds.MindDataset(mindrecord_path, columns_list[text, lang], num_parallel_workers8) dataset dataset.map(operationsrule_filter, input_columns[text], num_parallel_workers8) dataset dataset.filter(predicatelambda x: x, input_columns[__filter__]) dataset dataset.map(operationsminhash_dedup, input_columns[text], num_parallel_workers4) dataset dataset.map(operationsquality_score, input_columns[text], num_parallel_workers2) dataset dataset.filter(predicatelambda s: s 0.3, input_columns[quality_score]) return dataset这里有个细节filter算子的predicate接收的是列数据返回布尔值。MinHash 去重因为需要全局信息不能简单用filter得先算签名再分桶桶内排序保留一条。我一般把这一步单独拎出来用 Pandas 或 Spark 做做完再写回 MindRecord。4.4 参数计算与阈值调优实录阈值不是拍出来的是算出来的。以困惑度为例我拿 10 万条人工抽检的样本分别算困惑度然后画分布。发现优质样本的困惑度中位数在 80 左右95 分位在 180低质样本中位数在 600 以上。所以我把硬阈值定在 500软阈值定在 200这个区间能覆盖 90% 以上的优质样本同时砍掉大部分低质样本。长度阈值也是类似逻辑。统计全量语料的长度分布取 1 分位和 99 分位作为上下限这样能砍掉极端值又不误伤正常样本。我实测下来中文语料 1 分位大概在 30 到 60 字符之间99 分位在 8 万到 15 万字符之间具体看语料来源。参数初始值调优后调整依据最小长度30501 分位统计最大长度20000010000099 分位统计有效字符占比0.60.7抽检误杀率困惑度硬阈值800500分布拐点质量分阈值0.50.3保留率平衡调优的核心原则是先松后紧。第一轮跑完看保留率和抽检质量如果保留率太高比如 95%说明阈值太松逐步收紧如果保留率骤降说明某级阈值过狠回退再调。5. 常见问题与排查技巧实录5.1 过滤后数据量骤降怎么办这是最常见的问题。先别慌按级排查。把每一级的输入输出条数打出来看是哪一级掉得最狠。如果是 L1 掉太多大概率是长度阈值或者有效字符占比设太严如果是 L2 掉太多检查 MinHash 的相似度阈值是不是设太低或者高频 n-gram 表统计有误如果是 L3 掉太多看质量分类器的分数分布是不是整体偏低可能是分类器训偏了。我踩过的一个坑是MinHash 的 shingle 大小设成了 3导致短文本之间相似度虚高大量正常样本被误判为重复。后来改成 5问题解决。shingle 大小一般取 5 到 9太短容易误判太长则漏判。5.2 模型打分阶段 GPU 利用率上不去模型推理过滤时如果 batch size 太小或者数据加载是瓶颈GPU 会一直等数据。解决办法有两个一是把num_parallel_workers调大让 CPU 预取数据二是用mindspore.dataset的prefetch算子做流水线重叠。我一般设prefetch_size32配合 batch size 256GPU 利用率能从 30% 拉到 80% 以上。还有一个隐藏问题是动态 shape。如果每条样本长度差异很大padding 到最大长度会浪费大量算力。建议按长度分桶同桶内 batch这样 padding 浪费最小。MindSpore 的bucket_batch_by_length算子就是干这个的直接拿来用。5.3 过滤规则误杀优质长尾数据长尾数据往往短、口语化、格式不规范很容易被规则误杀。我的应对策略是给长尾数据开白名单。具体做法是先用一个宽松的规则跑一遍把被删的样本单独存一份人工抽检 1000 条看有多少是误杀。如果误杀率超过 5%就针对这类样本放宽阈值或者单独走一条过滤链路。另一个技巧是分层过滤。把语料按来源分成若干层比如百科、新闻、论坛、代码每层用不同的阈值。百科类可以严一点论坛类放宽长度和流畅度要求。这样既保证整体质量又不丢长尾。问题现象可能原因排查方法解决手段保留率低于 30%阈值过严逐级统计保留率放宽对应级阈值去重后仍见重复shingle 过大抽检重复样本减小 shingle 到 5GPU 利用率低数据加载瓶颈看 prefetch 队列增大 workers 和 prefetch长尾数据丢失规则一刀切误杀样本抽检分层设阈值或白名单分数分布偏移分类器训偏画分数直方图重新校准或补标注5.4 分布式训练下的数据一致性分布式预训练时每个卡读到的数据必须是不重叠的否则等于变相增加了重复数据。MindSpore 的MindDataset支持shard_id和num_shards参数按卡切分。但要注意过滤阶段如果是在切分之后做的各卡的过滤结果可能不一致导致最终数据量对不齐。正确做法是先全量过滤再切分或者过滤时保证各卡规则完全一致。还有一个坑是随机种子。shuffle的种子如果不固定每次重启训练数据顺序都变实验不可复现。建议在dataset.shuffle(buffer_size, seed42)里显式指定种子并在实验记录里写清楚。6. 我在实际项目里的几点体会这套方案我在两个中文预训练项目里跑过一个 30B 量级一个 100B 量级。最大的体会是数据过滤的收益不是线性的而是先陡后缓。从不过滤到过滤效果提升非常明显但从 90% 质量过滤到 95%投入产出比就下来了。所以别追求极致干净找到性价比拐点更重要。另外过滤规则一定要版本化。每次调阈值都记下来配上对应的模型评测结果。我见过团队改了过滤规则没记录后来模型效果波动查了一周才发现是数据变了。用 Git 管理过滤脚本和配置文件每次实验打 tag这是基本功。最后分享一个小技巧在过滤管道里加一个采样落盘的环节每级过滤随机保留 0.1% 的样本到单独目录。这些样本是排查问题的金矿出问题时直接看这些样本比翻日志快得多。这个习惯帮我省了无数次 debug 时间。