大模型预训练数据管线：从原始语料到高质量训练集
摘要: 预训练的质量七成取决于数据。本文讲数据管线里的每一道工序怎么落地：清洗、质量过滤、去重、混合配比、tokenizer 训练——全部给可运行脚本与量化指标。架构（MLA/MoE）决定模型上限，数据决定它能否摸到上限。
0. 实验目标
- 用一套可运行的 Python 流程，把"原始语料"处理成"可直接训练的 token 流"
- 掌握五道关键工序：清洗 → 质量过滤 → 去重 → 混合配比 → tokenizer
- 理解每道工序的量化指标（去重率、质量分数、混合比、tokenizer 效率）
- 给出预训练数据工程的几条铁律
1. 第一道工序：清洗（去杂质）
python
import re
def clean_text(text: str) -> str:
"""基础清洗：控制字符、多余空白、噪声符号。"""
# 1) 去掉 Unicode 控制字符（零宽字符、BOM、私用区）
text = "".join(ch for ch in text if not (
ord(ch) < 32 and ch not in "\n\t" or 0xE000 <= ord(ch) <= 0xF8FF
))
# 2) 压缩连续空白（保留换行）
text = re.sub(r"[ \t]+", " ", text)
# 3) 去掉连续重复标点（营销噪声常见："!!!!!!"、"????"）
text = re.sub(r"([!?。！？]){3,}", r"\1\1", text)
# 4) 去除 URL 与邮箱（按需，代码语料可能想保留 URL）
# text = re.sub(r"https?://\S+", "<URL>", text)
return text.strip()原理讲解：
- 清洗的目标是降低信噪比，不是"美化"：控制字符会让 tokenizer 产出异常 token，连续标点是网页噪声的标志。
- 清洗顺序有讲究：先去掉不可见字符，再压缩空白，最后处理标点——顺序反了会漏掉藏在空白里的控制符。
- 清洗要可审计：每个规则对应一条计数日志（"移除零宽字符 N 个"），上线后能快速定位"某类语料被误杀"的问题。
2. 第二道工序：质量过滤（分类器打分）
python
# 用轻量启发式规则做初筛（无需模型），规则分数 0-1
def quality_heuristic(text: str) -> float:
n = len(text)
if n < 200:
return 0.0 # 太短：多为碎片/无意义
# 标点密度：太低说明不是自然语言
punct = sum(1 for ch in text if ch in "，。！？；：,.!?;:")
punct_ratio = punct / n
if punct_ratio < 0.02:
return 0.1
# 去重行占比：高重复（如广告、评论区复制）质量差
lines = [l.strip() for l in text.split("\n") if l.strip()]
dup_ratio = 1 - len(set(lines)) / max(len(lines), 1)
if dup_ratio > 0.5:
return 0.15
return min(1.0, 0.3 + punct_ratio * 20) # 经验加权
# 正式流水线：规则初筛（低成本）→ 模型精筛（高成本，只对候选）
# 推荐：fastText / BERT 分类器给候选打分，按 top-k 保留实验结果（示例）：
| 语料类型 | 规则初筛通过率 | 模型精筛后保留率 | 说明 |
|---|---|---|---|
| 维基百科 | 95% | 92% | 高质量基准 |
| 新闻爬虫 | 78% | 71% | 广告/转载噪声多 |
| 论坛帖 | 55% | 41% | 需要强过滤 |
| 代码仓库 | 88% | 85% | 走独立 code 管线 |
关键：两级过滤。先用规则把 99% 的垃圾挡掉（快、便宜），再对通过的语料跑模型打分（贵、准）。全量跑模型分类器是常见的算力浪费。
3. 第三道工序：去重（语义 + 近似）
python
import hashlib
from collections import defaultdict
def minhash_dedup(texts, n_grams=5, n_hashes=64):
"""MinHash + LSH 近似去重：百万级文档量级的标配方案。"""
buckets = defaultdict(list)
for idx, text in enumerate(texts):
# 文档切 5-gram，对每个 gram 取哈希，文档签名 = 最小哈希集合
grams = [text[i:i+n_grams] for i in range(0, max(1, len(text)-n_grams+1), n_grams)]
sig = tuple(min(hashlib.md5(g.encode()).hexdigest() for g in grams)
for _ in range(n_hashes))
buckets[sig].append(idx)
return [indices for indices in buckets.values() if len(indices) > 1]
# 精确去重（数据量小时兜底）
seen = set()
unique = []
for doc in texts:
h = hashlib.md5(doc.encode()).hexdigest()
if h not in seen:
seen.add(h)
unique.append(doc)原理讲解：
- 精确去重（MD5 全文本）只能去掉完全相同的副本；网页转码、模板化生成的"伪不同"文档必须靠近似去重。
- MinHash + LSH 是预训练数据去重的工业标准：把文档切成 shingle（n-gram），用一组最小哈希近似"文档相似度"，相似度高于阈值的聚成一簇再精确比对。
- 还有更省算力的后缀数组去重（如 Google 的 Suffix Array 方案），适合在超大规模语料上做"跨文档子串去重"，防止测试集泄漏进训练集——这直接影响 benchmark 可信度。
4. 第四道工序：混合配比（数据配方）
python
# 典型配方示例（按 token 占比）：
DATA_MIX = {
"wikipedia": 0.35, # 高质通用知识
"books": 0.15, # 长文本连贯性
"code": 0.15, # 代码与推理
"web": 0.25, # 开放世界知识
"math_science": 0.10, # 数学与科学推理
}原理讲解：
- 混合比例是最容易被低估的"超参数"：通用语料占比过高 → 模型"博而不精"；代码/数学占比过高 → 常识缺失。业界经验：代码 10-20%、数学 5-15% 是常见区间，具体靠小规模消融实验定。
- 采样（sample）的粒度是 token，不是文档——不同来源文档长度差异大，按文档采样会无意中过采样长文档类型。
- 训练中后期常做配比退火（mixing-ratio annealing）：逐步提高高质量来源（books/math/code）占比，类似"冲刺阶段上强度"。
5. 第五道工序：tokenizer 训练与检查
python
# 用 tokenizers 库训练 BPE/Unigram tokenizer（中文场景建议 Unigram + 字符回退）
from tokenizers import Tokenizer, models, trainers, pre_tokenizers
tok = Tokenizer(models.Unigram())
tok.pre_tokenizer = pre_tokenizers.Metaspace()
trainer = trainers.UnigramTrainer(
vocab_size=65536,
special_tokens=["<unk>", "<s>", "</s>", "<pad>"],
unk_token="<unk>",
show_progress=True,
)
tok.train(files=["corpus_zh.txt", "corpus_code.txt"], trainer=trainer)
tok.save("tokenizer.json")
# 三个必查指标：
# 1) 中文单字回退率：< 2% 正常，过高说明语料不够或 vocab 太小
# 2) 平均 token 数/字符：压缩率越高的语言建模越好
# 3) 代码语料里注释/缩进的 token 形态：异常说明 pre-tokenizer 有问题工程启示：
- tokenizer 决定"词汇预算"：中文场景 vocab 32K-64K 常见，太小中文压缩率差（长序列贵），太大 embedding 参数浪费。
- pre-tokenizer 对代码语料影响巨大：缩进、注释、字符串的字面 token 形态会显著影响代码补全质量。
- tokenizer 要和语料同源训练：直接复用 Llama 的 tokenizer 训练中文模型，中文 token 效率天然差一截（这是中文模型"vocab 中文占比"被反复讨论的原因）。
6. 全流程编排（一条命令跑完）
bash
# 流水线：raw/ → clean/ → filter/ → dedup/ → mix/ → tokens/
python3 pipeline.py \
--input raw/ \
--output tokens/ \
--clean --filter --dedup --mix \
--mix-config data_mix.json \
--tokenizer tokenizer.json \
--shards 256 # 分片并行，Spark/Ray 可水平扩展工程启示：
- 分片 + 幂等：每个阶段输出不可变分片（
shard-0001.parquet），断点重跑只处理未完成分片——预训练数据量是 TB 级，全量重跑不可接受。 - 数据版本化：每个数据集带版本号与处理参数哈希，训练日志里记录"用了哪个版本的数据"，保证可复现（这是可复现训练的基本功）。
- 质量监控闭环：每阶段输出统计报告（token 数、质量分分布、去重率），训练时如果 loss 异常，先回去查数据报告。
7. 工程启示汇总
- 清洗可审计：每条规则有计数日志，误杀可定位。
- 两级过滤：规则初筛挡 99% 垃圾，模型精筛只跑候选。
- 去重必做近似：MD5 只防完全重复，MinHash+LSH 防模板化伪重复；注意测试集泄漏。
- 按 token 配比：采样粒度是 token；中后期做配比退火。
- tokenizer 与语料同源：中文场景检查单字回退率与压缩率。
- 分片幂等 + 数据版本化：TB 级数据可断点续跑、训练可复现。
- 质量报告进训练日志：loss 异常先查数据，再查模型。