做过大模型预训练的朋友应该都有类似体验：代码写得再漂亮，显卡堆得再多，只要喂进去的数据是乱的，loss 曲线就会用各种方式教你做人。我去年用 MindSpore 跑一个大模型预训练项目，前期花在数据上的时间比真正跑训练的时间还长，而其中最关键、也最容易翻车的一环，就是数据质量过滤方案。所谓数据质量过滤，就是在语料进入模型之前，把重复、乱码、广告、非目标语言、毒性内容和隐私信息筛掉，只给模型喂干净、多样、有用的文本。这篇内容适合准备自己从零整理预训练语料、或者正在优化现有数据管线的团队参考，我会把设计思路、实操代码和踩坑经验一次性讲清楚，尽量少让你走弯路。

数据质量过滤听起来不像模型结构那么“有技术含量”，但它直接决定预训练的天花板。同样的参数量、同样的训练卡数，数据干净程度不同，最终效果可以拉开一个身位。所以这篇文章不只是给一套脚本，而是帮你想清楚：为什么要筛、按什么维度筛、在 MindSpore 上怎么高效落地。

1. 大模型预训练的第一步，是给数据“体检”

我见过不少团队上来就急着调模型结构、跑 benchmark，结果语料还是爬虫抓完没清洗过的原始文件。第一版模型训出来，生成文本里夹着广告链接、乱码和重复片段，这时才回头补数据，白白烧掉一大笔训练预算。

1.1 脏数据会怎么反噬训练效果

脏数据对预训练的影响不是“有一点干扰”，而是会直接改变模型学到的概率分布。第一种典型危害是重复文本导致的记忆坍塌。自回归模型的核心任务是预测下一个 token，某句话如果在语料中出现了一万次，模型就会把它当成一个极高频的先验，生成时动不动就复读这句话。我之前跑过一个测试，原始语料里有一条论坛帖子被爬虫重复抓了 4000 多遍，模型训完之后，只要上下文稍微沾点边，就整段整段往外吐这段内容，看起来像是在“背诵”而不是在“生成”。

第二种危害是噪声语料占用训练预算。你规划好训 1000 亿 token，如果里面 10% 是乱码、广告导航、口水话，那等于有 100 亿 token 的算力被浪费在垃圾上。更麻烦的是，这些噪声并不是均匀分布的，它们往往集中在某些域名或某些采集来源里，模型会在局部领域学出一堆错误映射。

第三种危害是错误知识被固化。模型预训练阶段没有人工纠错机制，喂进去多少错误信息，它就老老实实记多少。等到下游指令微调阶段，再想去掉这些错误知识非常困难，因为它们的权重已经深深写进参数里了。数据质量过滤，本质上就是在大规模训练之前，把模型将来可能犯的“低级错误”提前掐死在摇篮里。

1.2 数据质量问题的典型形态与危害等级

很多人以为数据清洗就是去掉 HTML 标签，其实真实语料里的问题种类多到能列一长串。我整理了一个常用的问题形态表，方便你做初检时对号入座。

需要注意的是，这些问题经常叠加出现。一条样本可能同时是乱码文本加广告导流，也可能前半段是正常新闻、后半段被拼进了论坛灌水回复。所以过滤不能只做一道，最好是多条规则并行、层层设卡。

1.3 质量过滤在 MindSpore 预训练流水线里的位置

一条完整的预训练数据管线大致是这样：原始语料采集 → 格式清洗 → 质量过滤 → 去重 → Tokenization → 组装训练样本 → 数据并行加载 → 模型训练。质量过滤放在 Tokenization 之前，原因很简单：token 化之后再做过滤，很多内容已经被切碎重组，原始文本信号反而丢了。而且同样的过滤规则，在 token 化前只需要处理一次，在 token 化后却可能涉及样本拼接逻辑，排查问题时非常痛苦。

在 MindSpore 生态里，模型训练时数据是通过mindspore.dataset系列 API 流入 Trainer 的。你可以把质量过滤逻辑放在 Dataset 流水线里用filter算子做，也可以提前离线处理完毕再转成 MindRecord 格式。我的建议是：重活、脏活全部离线做，训练时的 Dataset 只保留最后一层轻量兜底过滤。因为在训练阶段反复执行昂贵的 Python 过滤函数，不但拖慢数据加载速度，还会在每个 epoch 重复计算，纯粹是浪费。

2. 拆开数据质量过滤的六个技术维度

“数据质量”是个很宽泛的词，真要落地必须拆成可计算的维度。我习惯把过滤方案拆成六块：语言识别、质量打分、去重、毒性过滤、隐私过滤、格式清洗。每块解决一类问题，互相之间也有重叠，但边界尽量清晰，这样出问题时能快速定位是哪一环失守了。

2.1 语言识别：先把混入的杂音找出来

预训练语料如果目标语言是中文，第一道关就是把非中文内容筛掉。有些人觉得这一步无所谓，反正 tokenizer 会自动处理，但实际上混入的大量英文、日文、韩文文本会浪费 tokenizer 的词表容量，还会让语料分布变得不伦不类。

做语言识别不需要一上来就上大模型。对中文语料，最简单的“汉字占比法”就能解决 90% 以上的问题。我常用的做法是统计文本里中文字符[\u4e00-\u9fff]的占比，低于某个阈值就淘汰。代码只有几行：

import re HANZI_RE = re.compile(r'[\u4e00-\u9fff]') def hanzi_ratio(text: str) -> float: if not text: return 0.0 return len(HANZI_RE.findall(text)) / len(text)

中文新闻类语料的汉字占比通常在 0.7 以上，如果低于 0.5，基本可以断定是外文或者乱码。但要注意一个特殊情况：代码和数学公式占比高的文本，汉字占比会显著下降，比如一篇技术文档里大量出现英文变量名和代码块，汉字占比可能只有 0.3。这种文本不能一棍子打死，建议单独走“代码语料白名单”通道，或者把语言识别的阈值调低，放到质量打分环节再判断。

如果你处理的是多语言语料，可以用 fastText 的lid.176.bin这类轻量语言识别模型，它输出每个语言的置信度。实际操作时记得设置置信度阈值，比如 0.5 以上才认定为目标语言，避免把语种模棱两可的短文本误判。

2.2 质量打分：规则裁判和模型评测员怎么配合

语言识别只能判断“是什么语言”，判断不了“内容质量好不好”。真正决定过滤质量的核心环节，是给每条样本打一个质量分。我常用的方案是“规则裁判 + 模型评测员”双层结构，两层配合而不是只用一层。

规则裁判是一组加权特征：文本长度、平均句长、标点密度、特殊符号占比、重复 n-gram 比例、URL 数量。它的优点是执行快、可解释性强，每条规则都能单独看统计。我给自己项目的打分器设计了这样的核心特征：太短的文本直接给低分，因为单句口水话信息量有限；URL 占比太高说明大概率是导航页或采集站；标点密度异常低可能是纯符号堆叠；连续重复字符比例高则可能是乱码或刷屏。把这些特征标准化到 0 到 1 区间，再按经验权重加权求和，就得到一个可用的quality_score。

模型评测员则负责那些“看起来正常、但读起来很别扭”的文本。主流做法是用困惑度（Perplexity）衡量：训练一个小规模语言模型，给每条文本算困惑度，模型越“意外”的序列困惑度越高。这些高分困惑度样本里确实藏了大量低质文本，但也误伤了代码、古文、数学公式这类“高信息熵”内容。我印象最深的一次，就是把困惑度最高的 10% 全删了，结果古文语料全没了，模型后来在文言文相关的下游任务上直接崩盘。

所以我的建议是：模型评测员不要做“一刀切删除”，而是做“分层降采样”。把困惑度分位数算出来，最高的 10% 样本按 30% 概率保留，中间的 30% 按 70% 概率保留，低困惑度的全部保留。这样既压制了坏样本，又保留了多样性。

2.3 去重：让每一条样本都有新鲜感

质量过滤解决的是“每一条样本自身质量”的问题，去重解决的是“样本之间互相重复”的问题。预训练语料里重复现象极其普遍，尤其是资讯站采集的文章，经常出现完全相同的段落被复制到不同页面。

精确去重没什么好说的，对整篇文本做 MD5 或 SHA1，完全相同就去掉，这是第一道廉价的拦网。但真正麻烦的是近似重复：两篇文章 90% 内容相同，只改了几个段落首尾，精确哈希完全失效。

模糊去重我推荐 MinHash + LSH。核心思路是：把文本切成一堆 shingle（一般是 5 到 8 个字符的滑动窗口），把它们转成哈希集合，再用 MinHash 把集合压缩成固定长度的签名，最后通过局部敏感哈希（LSH）加速查找相似签名。这样你能以较高召回率找出 Jaccard 相似度超过阈值的文本对。实操时可以直接用datasketch库：

from datasketch import MinHashLSH, MinHash def build_minhash(text: str, num_perm: int = 128, shingle_size: int = 6): mh = MinHash(num_perm=num_perm) if len(text) < shingle_size: return mh for i in range(len(text) - shingle_size + 1): mh.update(text[i:i+shingle_size].encode("utf-8")) return mh

在大规模场景下，比如几十亿条样本，直接用 MinHashLSH 查询内存可能扛不住。可以先用 Bloom Filter 统计高频 n-gram，把明显撞车的样本过滤掉，再对剩下的候选集做 MinHash。数据量级越大，越要遵循“先粗筛、再细查”的思路，否则方案再好也跑不动。

2.4 毒性、隐私与格式清洗：安全底线类的过滤

质量打分管“好不好”，去重管“重复不重复”，还有一类过滤管的是“能不能用”。毒性过滤解决语料里可能存在的攻击性、暴力、色情等不合规内容。技术上通常用两类手段并行：一类是维护一个覆盖常见词汇的词表，做正则匹配；另一类是训练一个轻量文本分类器，对整条样本打标签。词表匹配速度快但容易误伤，分类器更准但需要标注数据。我的经验是：先词表粗筛，命中可疑样本再交给分类器复核，这样性能与准确率都能兼顾。

隐私过滤处理的是个人敏感信息，比如身份证号、手机号、邮箱、具体家庭住址。这类信息一旦进入模型，训练完之后可能被模型在生成时“复述”出来，这是非常严重的安全隐患。先写正则做第一轮匹配，手机号和邮箱这类有明确格式的很好处理，地址和姓名要靠命名实体识别补漏。对于预训练语料，建议匹配到就直接整段删除，不要抱着“只有一小段泄露了没关系”的侥幸心理。

格式清洗则是把 HTML 标签、控制字符、零宽空格、异常 Unicode 清掉。这里要提醒一句：不要过度清洗。如果你把文本里的括号、引号、代码符号全部干掉，语料的可读性会大打折扣。特别是代码类语料，特殊符号本身就是语义的一部分，盲目清理等于把代码阉割了。格式清洗的目标是把“噪音”清掉，而不是把“特征”清掉。

2.5 过滤维度的取舍：不是过滤得越狠越好

很多团队第一次做质量过滤，容易走上另一个极端：规则叠加了一堆，恨不得把可疑文本全干掉，结果原本 500GB 的语料过滤完只剩 100GB。数据量骤降带来的直接问题就是多样性受损，模型学习的知识面变窄，在长尾任务上表现明显变差。

我自己的原则是“硬过滤”与“软过滤”分开。硬过滤针对的是绝对不可用的内容，比如乱码、隐私信息、涉暴涉恐、重复度接近 100% 的文本，这些直接删除。软过滤针对的是“质量不太好但还有信息量”的文本，比如口水话夹杂少量有效信息、广告周边带一点正文，这些按质量分降采样保留一部分。

阈值怎么定，不是拍脑袋拍出来的。我的做法是每次调参前，从数据里随机抽 400 条，人工打标分成“可用”“一般”“垃圾”三类，再画特征分布图，看哪些分数段能把“垃圾”和“可用”明显分开。没有这步人工标注，所有阈值都只是猜测。

3. MindSpore 上落地数据质量过滤 Pipeline 的完整过程

维度想清楚了，接下来就是工程实现。这一章我会完整展示在 MindSpore 上搭建过滤管线的过程，从开发环境、总体架构、核心代码到分布式并行都会写到，你可以直接照着搭。

3.1 开发环境准备：VSCode 配 MindSpore 内核，调试效率翻倍

我强烈建议你不要在终端里黑灯瞎火地写数据处理脚本。我自己的习惯是在本地用 VSCode 做 Remote-SSH 远程开发，连到训练服务器上，把 Python 解释器切到安装了 MindSpore 的 conda 环境。这样在 VSCode 里新建 notebook 时，就能选择对应的 MindSpore 内核，直接在 Jupyter 环境里逐行调试数据处理代码。

别小看这个环境配置的差别。处理数据时你经常想看看某条样本被过滤后到底长什么样、某个特征算出来是多少，如果只能靠print打日志，效率会低很多。用 notebook 加断点，你可以在内存里直接检查text变量的内容、查看 Dataset 的 shape、单步跑过滤函数，整个调试体验是质的提升。特别是 MindSpore 的 Dataset API 链式调用很长，你可以在每一步之后查看输出，确认shuffle、batch、repeat的顺序有没有写错。

这里有个小提醒：在共享训练服务器上调试时，别直接在登录节点跑大规模for循环，尽量用交互开发节点或者先取一小块数据来跑。我见过有人在登录节点跑全量过滤，把整个服务器的 CPU 占满，导致其他同事的作业全部排队，差点引发事故。

3.2 Pipeline 总体架构：从原始语料到可训练数据集

我建议把过滤管线拆成离线预处理和训练时加载两个阶段。离线预处理阶段用 Python 多进程脚本处理原始语料，产出的中间格式是 JSONL；然后再把清洗后的 JSONL 转成 MindRecord，供训练时的高效加载。整个流程长这样：

原始语料（JSONL / TXT） → Python 多进程预处理（语言识别、质量打分、去重、敏感词过滤） → 清洗后 JSONL → 转换为 MindRecord 分片 → MindSpore Dataset 加载 → 按卡号切分数据 → shuffle / batch / repeat → 进入训练

为什么中间格式用 JSONL？因为一行就是一个样本，天然支持大文件流式读写；清洗过程中出问题，可以用grep抽查任意一行；断点续跑时只需要记录处理到哪个文件了。它最大的缺点是读取速度慢，但那是训练阶段的事，不是预处理阶段的事。

为什么训练前要转 MindRecord？因为 MindRecord 是 MindSpore 的原生数据格式，采用二进制存储并支持按 shard 随机读取。同样的 20GB 数据，用 JSONL 在训练时每次读取都要做字符串解析，转换成 MindRecord 之后能明显降低数据加载的 CPU 开销。更关键的是，MindRecord 天然支持num_shards切分，方便你在分布式训练时让每张卡只读属于自己的那部分数据。

3.3 核心代码实现：质量打分器 + Dataset 过滤

先给出一份可以直接改的规则打分器代码。我只保留最核心的特征，避免把函数写得过于臃肿，你可以在实际项目中往里面继续加特征。

import re HANZI_RE = re.compile(r'[\u4e00-\u9fff]') URL_RE = re.compile(r'https?://[^\s]+|www\.[^\s]+') CONTROL_RE = re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f]') PUNCT_SET = set('，。！？；：、,.!?;:') def quality_score(text: str) -> float: if not text: return 0.0 text = CONTROL_RE.sub('', text) text_len = len(text) if text_len < 30: return 0.0 hanzi_ratio_value = len(HANZI_RE.findall(text)) / text_len url_count = len(URL_RE.findall(text)) url_ratio = url_count / text_len punct_count = sum(1 for ch in text if ch in PUNCT_SET) punct_ratio = punct_count / text_len # 连续重复字符比例，用于识别乱码与刷屏 repeat_count = 0 for i in range(len(text) - 1): if text[i] == text[i + 1]: repeat_count += 1 repeat_ratio = repeat_count / max(text_len - 1, 1) score = ( 0.35 * min(hanzi_ratio_value, 1.0) + 0.25 * min(punct_ratio * 4, 1.0) + 0.25 * (1.0 - min(url_ratio * 20, 1.0)) + 0.15 * (1.0 - min(repeat_ratio * 10, 1.0)) ) return score

这个打分器的输出在 0 到 1 之间。0.6 这个阈值是我抽了 400 条人工标注之后画分布图找出来的，不是随手定的。你可以照这个流程在自己数据上调。

接下来是离线并行过滤脚本。注意这里用multiprocessing.Pool按文件粒度并行，避免多个进程同时写一个输出文件造成的锁竞争。

import json from multiprocessing import Pool from pathlib import Path def filter_one_line(line: str): line = line.strip() if not line: return None try: obj = json.loads(line) except json.JSONDecodeError: return None text = obj.get("text", "") if not text or quality_score(text) < 0.6: return None return json.dumps(obj, ensure_ascii=False) def process_file(src_path: Path, dst_path: Path): with src_path.open("r", encoding="utf-8") as fin, \ dst_path.open("w", encoding="utf-8") as fout: for line in fin: filtered = filter_one_line(line) if filtered: fout.write(filtered + "\n") if __name__ == "__main__": src_files = list(Path("raw_chunk").glob("*.jsonl")) tasks = [] for src in src_files: dst = Path("clean_chunk") / src.name tasks.append((src, dst)) with Pool(16) as pool: pool.starmap(process_file, tasks)

清洗完之后，需要把 JSONL 转成 MindRecord。这里我写了一个生成器，避免一次性把所有数据读进内存。

import json from pathlib import Path from mindspore.mindrecord import FileWriter def data_generator(clean_dir: str): for file in Path(clean_dir).glob("*.jsonl"): with file.open("r", encoding="utf-8") as fin: for line in fin: obj = json.loads(line) text = obj.get("text", "") score = obj.get("score", 0.0) if text: yield {"text": text, "score": float(score)} schema = {"text": {"type": "string"}, "score": {"type": "float32"}} writer = FileWriter(file_name="data/pretrain.mindrecord", shard_num=8) writer.add_schema(schema, "filtered pretrain corpus") writer.write_raw_data(data_generator("clean_chunk")) writer.commit()

最后是训练时的 Dataset 加载和兜底过滤。这时 MindRecord 已经按score字段离线筛过一遍了，训练时再过滤一次是为了挡住那些离线阶段可能漏过的极端短文本。

import mindspore.dataset as ds def is_good_record(record): text = record["text"] # 兜底规则：长度太短的文本直接丢弃 if len(text) < 50: return False return True dataset_files = ["data/pretrain.mindrecord"] full_ds = ds.MindDataset(dataset_files, num_parallel_workers=8) full_ds = full_ds.filter(predicate=is_good_record) full_ds = full_ds.shuffle(buffer_size=10000) full_ds = full_ds.batch(batch_size=256, drop_remainder=True)

这里要特别说明：不要把昂贵的大模型评分逻辑写进训练时的filter，因为filter会在每个 epoch 被重复执行，大模型跑一次就要命。所有重计算都应该离线完成，训练时只留最轻量的规则。

3.4 分布式并行：让每张卡拿到不同但均衡的数据

大模型预训练基本都是分布式训练，MindSpore 里通过MindDataset的num_shards和shard_id参数来给每张卡分数据。示例代码如下：

from mindspore.communication import init, get_rank, get_group_size init() rank_id = get_rank() world_size = get_group_size() train_ds = ds.MindDataset( dataset_files="data/pretrain.mindrecord", num_shards=world_size, shard_id=rank_id, num_parallel_workers=8, )

有一个细节容易被忽略：分片之后的数据均衡问题。如果 MindRecord 各个分片的样本数差太多，训练时某些卡的梯度计算会快，某些卡会慢，整个集群都在等最慢的那张卡。所以你在FileWriter转格式时，shard_num不要设得比卡数少，最好设成卡数的整数倍，并且写入数据时尽量打散后再写，避免某个分片恰好集中了某个来源的语料。

另外，shuffle在分布式训练里要特别注意。每条数据在加载时已经按num_shards切给了固定卡号，如果shuffle只在数据加载初期做一次，整个训练过程每张卡见到的数据组合就固定了。我建议每个 epoch 都换一下随机种子，或者直接开启shuffle_each_epoch，保证跨 epoch 的数据顺序不同，防止模型在固定顺序上过拟合。

3.5 性能优化：CPU 才是这场战役的主战场

不少人会误以为数据过滤是 GPU 的活，真跑起来才发现瓶颈全在 CPU 和磁盘 IO。质量过滤本质上是一个 CPU 密集加 IO 密集的任务，优化顺序非常重要。

第一优先级的优化是“把重活离线化”，不要在训练循环里反复做文本正则和大模型推理。第二优先级是“并行化”，用multiprocessing.Pool按文件粒度并行，我实测过同样对 1GB 文本做规则过滤，单进程跑要接近 20 分钟，16 进程并行后压到 3 分钟以内。第三优先级是“减少重复计算”，正则表达式统一预编译，不要在每行样本里重复re.compile；对大文件用缓冲读取，不要一行一行做系统调用。

如果你的过滤规则里还包含大模型打分，那就要引入“先粗后细”的策略。先用廉价的规则打分器把所有样本跑一遍，只有落在“灰色地带”的样本才交给大模型做二次判断。比如规则分在 0.4 到 0.7 之间，才让大模型重新打分；规则分很高或很低的样本，直接放行或删除。这样既保住了精度，又不会让大模型成为整个管线的吞吐瓶颈。

MindSpore 的 Dataset API 虽然提供num_parallel_workers参数，但 Python 层回调在并行时会有 GIL 和跨语言调用开销。如果确实需要在 Dataset 里跑自定义过滤函数，建议把函数写得足够轻量，并且用num_parallel_workers和prefetch_size配合调参，找到你机器上性价比最高的参数组合。

4. 数据过滤实战中的常见问题与排查技巧

数据过滤这套系统做完之后，真正的挑战才刚刚开始。你在真实数据上跑几轮，一定会遇到过滤过猛、漏网、性能崩盘这类问题。我把实际踩过的坑和排查思路整理出来，希望能帮你省掉几个晚上的排查时间。

4.1 过滤过猛，训练集缩水了怎么办

最常见的翻车现场是：规则叠加得很爽，语料量却肉眼可见地缩水了。我有一次把“符号占比”“URL 占比”“最短长度”三条规则一起收紧，结果原本 300GB 的语料只剩 80GB。当时第一反应是“数据太脏了”，后来抽查被删样本才发现，大量包含数学符号和代码片段的优质技术文章被误杀了。

排查方法其实不复杂：给每条过滤规则单独加一个计数器，统计它分别淘汰了多少样本。我会把过滤结果分成“已通过”“已删除”“灰色待定”三个输出目录，并定时对“已删除”目录做抽样人工检查。看到底是规则本身太激进，还是阈值设得不对。

修复时不要只调一个阈值，我建议把“硬删除”改为“分层降采样”。质量分低于 0.2 的直接删；0.2 到 0.5 之间的按 20% 概率保留；0.5 以上的全部保留。这样即使某类文本被某个规则误伤，至少还有一部分能留下来，不至于整个语料类型消失。

4.2 坏样本漏网，如何科学评估过滤器

过滤器上线后，你还需要一套评估机制，否则很难回答一个关键问题：到底漏掉了多少坏样本。我自己的做法是构造一个“合成坏样本集”：拿干净文本，人为插入广告句式、重复段落、乱码字符、敏感词、隐私信息，然后跑过滤器看召回率。

举个例子，你可以生成 1000 条测试样本，其中 500 条是干净样本，500 条是伪造的坏样本。跑完过滤器之后算两个指标：坏样本召回率（坏样本被删掉的比例）和干净样本误杀率（干净样本被误删的比例）。对硬过滤规则，我要求坏样本召回率不低于 99%；对软过滤降采样规则，则要求误杀率足够低，否则会伤害语料多样性。

这套测试可以沉淀成 pytest 用例，每次改过滤规则之后先跑一遍，再上全量数据。否则你很难知道自己加了一条规则之后，到底改善了效果还是制造了新的误杀。

4.3 流水线跑得太慢，先定位瓶颈再动手

数据过滤跑得慢，人人都想优化，但很多人一上来就盲目加进程数，结果 CPU 早就满了，瓶颈其实在磁盘 IO。先做一次分阶段计时，把处理单个文件的时间拆成“读文件时间”“逐行处理时间”“写文件时间”三部分。

如果是逐行处理时间占比高，先检查是不是正则没有预编译，或者每条样本都调了一次大模型。如果是写文件时间占比高，考虑用更大的写入缓冲，或者让每个进程只写一个独立输出文件，最后再合并。如果是读取时间占比高，检查是不是大量的样本集中在一个超大文件里导致无法并行读取，提前把大文件切成多个小分片就能解决。

我还习惯用py-spy这类工具在工作进程上 dump 一下调用栈，看看 Python 层到底卡在哪个函数里。有时候你以为卡在quality_score，实际是日志打印太频繁，print成了性能杀手。把日志改成按比例采样，流水线速度能快不少。

4.4 参数速查表与我的默认配置

不同语料的最优参数肯定不同，但你可以从下面这套“起始配置”出发，再根据自己数据的情况调整。它是我在中英文混合语料上验证过的一版，作为起点足够安全。

这版配置不一定适合所有场景。如果你的语料是垂直领域（比如代码、医学、法律），建议把领域特有的质量特征加进打分器，并且重新抽样本做人工标注。迁移到一个新领域时，最忌讳直接照搬上一套阈值而不做校验。

最后说一点个人体会。数据质量过滤方案不是一个做完就固定的脚本，而是要跟着模型评测结果不断迭代的工程系统。我第一次把 300GB 语料过滤到 80GB，觉得数据非常干净了，结果模型在代码生成任务上明显变笨，后来才发现是代码和数学语料被误杀。调整阈值、把硬删除改成降采样，第二轮就把效果拉了回来。现在我的原则很简单：过滤 Pipeline 一定要可观测、可回滚、可抽查。每条规则过滤了多少样本要有统计，每次调整阈值之前要有备份，每个被删的样本都要有地方能翻出来复查。能做到这三点，数据质量过滤这套方案就不会成为你预训练项目里的定时炸弹。