📚 预训练数据准备与清洗：大模型的"粮食"是怎样炼成的 数据是 AI 的"石油"。但 Common Crawl 原始数据中，90% 以上是垃圾。从 300TB 的网页乱炖中提炼出 15TB 的高质量训练数据，再从 15TB 中筛选出几十万条高价值样本——决定模型最终能力的，不是参数量，而是你喂进去的食物质量。
📑 目录 数据决定模型上限
2026年主流预训练数据集全景
数据采集：从哪里来
质量过滤：去粗取精
去重：别让模型背同样的课文
隐私与安全过滤
Tokenization：把文本变成数字
数据配比：烹饪秘方
北大DataFlow：2026年数据工程新范式
实战：构建完整的数据清洗Pipeline
总结
一、数据决定模型上限 1.1 Scaling Law 的转折 2024-2026 年，AI 社区经历了一个关键认知转变：
旧共识：更多参数 + 更多算力 + 更多数据 = 更强模型 ↑ ↑ 边际递减 边际递减 新共识：高质量数据 × 高效架构 × 适量算力 = 最强模型 ↑ 现在最关键的因素！ 标志性事件：
Phi-4（14B）超越 GPT-4o — 用高质量合成数据，14B 参数在 GPQA 上击败了万亿参数模型 [1]
FineWeb-Edu 挑战 — HuggingFace 证明：只用 1.3T 高质量 token 训练的模型，超过了 15T 未筛选数据的 2.7 倍 [2]
DataFlow 实验（2026） — 15K 高质量 SFT 样本 > 50K 未筛选数据，10K 领域数据接近官方 Instruct 版本 [3]
1.2 数据管线的完整流程
原始数据 → URL过滤 → 语言识别 → 质量过滤 → 去重 → 隐私过滤 → Tokenize → 数据配比 → 训练 ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ Common 屏蔽低 只保留 启发式+ 精确+ PII BPE/ 代码:数学: Crawl 质量网站 目标语言 分类器 模糊 去除 BBPE 英文 = 3:1:6 规模参考（以训练一个 70B 模型为例）：
阶段 数据量 保留率 Common Crawl 原始数据 300 TB 100% URL 过滤 + 语言识别 150 TB 50% 质量过滤（启发式） 45 TB 15% 质量过滤（分类器） 15 TB 5% 去重（精确 + 模糊） 8 TB 2.7% 最终训练数据 5-8 TB 2-3% 二、2026年主流预训练数据集全景 2.1 开源数据集对比 数据集 发布方 大小 数据源 质量方法 特点 FineWeb HuggingFace 15T tokens Common Crawl × 96 多层过滤+去重 开源最强基线 [2] FineWeb-Edu HuggingFace 1.3T tokens FineWeb子集 FastText分类器 小数据高性能 DCLM 多机构 300T+ Common Crawl DataComp比赛框架 标准化评估 [4] DCLM-Baseline 多机构 4T tokens Common Crawl 高质量流水线 7B模型训实验 RedPajama-V2 Together 30T tokens Common Crawl 多质量信号 质量评分元数据 CulturaX 多机构 6T tokens mC4+OSCAR 多语言对齐 涵盖167种语言 SkyPile 昆仑万维 1T tokens 中文互联网 中文专用过滤 高质量中文数据 DataFlow-30B 北大DCAI 30B tokens 多源 系统化治理 优于FineWeb-Edu [3] 2.2 数据集质量与模型性能 FineWeb-Edu 的发现震撼了整个社区 [2]：
模型参数量: 7B 预训练token数: 训练到相同loss 数据集 Token量 下游平均分 Common Crawl (原版) 15T 基线 FineWeb (全量) 15T +5.2% FineWeb-Edu (筛选) 1.3T +14.1% ← 只有1/10的数据，但效果更好！ 结论：质量 > 数量，1.3T 高质量 >> 15T 原始数据 💡 面试加分点：FineWeb-Edu 使用的不是复杂模型，而是一个轻量级的 FastText 分类器（在 Wikipedia 参考数据上训练），对每个网页打分。这说明有时候简单的机器学习方法比高大上的模型更高效。
三、数据采集：从哪里来 3.1 数据源分布 以 DeepSeek V4（1.6T MoE）和 LLaMA 4 的训练数据为例：
数据源 占比 用途 举例 Web 抓取 60-70% 世界知识、常识 Common Crawl 学术论文 5-10% 科学推理、专业知识 ArXiv, PubMed, PMC 代码仓库 10-15% 代码生成、逻辑推理 GitHub, StackOverflow 书籍 5-8% 长文本理解、文学素养 Books3, PG-19 维基百科 3-5% 事实性知识、结构化信息 Wikipedia, Wikidata 社交媒体 2-5% 对话能力、现代用语 Reddit, Twitter 合成数据 5-15% ↑ 推理能力、特定技能 GPT-4/mixtral生成 2026年关键变化：
合成数据占比快速增长：从 2024 年的 2% 增长到 2026 年的 5-15%，Phi-4 和 Gemma 4 都大量使用合成数据
代码数据持续增加：从 5% 增长到 10-15%（平衡"语言能力"和"逻辑推理"）
视频/多模态数据开始进入预训练阶段（Gemma 4 原生多模态）
3.2 中文数据生态 数据源 规模 质量 获取难度 百度百科 ~1亿词条 较高 有限开放 知乎 ~10亿回答 高 API限制 微信公众号 ~PB级 中高 封闭 微博 ~TB级 低中 API限制 中文维基百科 ~500万条目 极高 完全开放 法律文书 ~TB级 高 部分开放 中文代码(GitHub) ~TB级 高 完全开放 百度贴吧 ~PB级 低 限制 注意：中文高质量数据的可获得性远低于英文。Qwen3 和 GLM-5 通过自建爬虫和合成数据来弥补这个差距。
四、质量过滤：去粗取精 4.1 启发式过滤（第一道防线） 最粗糙但也最高效的过滤方法，通过简单规则去除明显无用的内容：
“”“启发式质量过滤”“” import re from typing import List, Dict class HeuristicFilter: “”“基于规则的快速过滤”“”
# 低质量信号
MIN_DOC_LENGTH = 200 # 最小字符数
MAX_DOC_LENGTH = 100000 # 最大字符数
MAX_PUNCTUATION_RATIO = 0.3 # 标点符号比例上限
MAX_BULLET_RATIO = 0.9 # 列表项比例上限
MAX_STOP_WORD_RATIO = 0.5 # 停用词比例上限
MIN_ALPHA_RATIO = 0.6 # 字母比例下限
# 黑名单模式
BLOCKED_PATTERNS = [
r'\\bclick\\s+here\\b',
r'\\bsubscribe\\s+now\\b',
r'\\bjoin\\s+now\\b',
r'\\{.*?\\}', # 大量CSS/代码残留
r'javascript:',
r'window\\.',
]
# 低质量关键词比例
LOW_QUALITY_KEYWORDS = [
"advertisement", "sponsored", "promotion",
"click here", "subscribe", "follow us",
"广告", "推广", "赞助",
]
@classmethod
def filter(cls, doc: str) -> bool:
"""返回True表示文档保留，False表示丢弃"""
# 1. 长度过滤
if len(doc) < cls.MIN_DOC_LENGTH:
return False
if len(doc) > cls.MAX_DOC_LENGTH:
return False
# 2. 标点符号比例
punct_count = sum(1 for c in doc if c in '"\\'.,!?;:，。！？；：""''')
if punct_count / len(doc) > cls.MAX_PUNCTUATION_RATIO:
return False
# 3. 列表/子弹点比例
bullet_count = doc.count('\\n- ') + doc.count('\\n* ') + doc.count('\\n1. ')
if bullet_count * 50 / len(doc) > cls.MAX_BULLET_RATIO:
return False
# 4. 字母比例（英文文档）
alpha_count = sum(1 for c in doc if c.isalpha())
if alpha_count / max(len(doc), 1) < cls.MIN_ALPHA_RATIO:
return False
# 5. 黑名单模式
for pattern in cls.BLOCKED_PATTERNS:
if re.search(pattern, doc, re.IGNORECASE):
return False
# 6. 低质量关键词
keyword_count = sum(
doc.lower().count(kw.lower())
for kw in cls.LOW_QUALITY_KEYWORDS
)
if keyword_count > 5:
return False
return True
FineWeb 使用的 5 个核心启发式指标 [2]
class FineWebHeuristics: “”“FineWeb 启发式过滤指标”“”
@staticmethod
def compute_signals(doc: str) -> Dict[str, float]:
return {
"line_count": doc.count('\\n'),
"char_len": len(doc),
"word_count": len(doc.split()),
"bullet_ratio": doc.count('- ') / max(len(doc), 1),
"stop_word_ratio": sum(
doc.lower().count(w) for w in ["the", "a", "an", "is", "are"]
) / max(len(doc.split()), 1),
"caps_ratio": sum(1 for c in doc if c.isupper()) / max(len(doc), 1),
"ellipsis_count": doc.count('…'),
"repetition_ratio": cls._compute_ngram_repetition(doc, 5),
}
@staticmethod
def _compute_ngram_repetition(text: str, n: int = 5) -> float:
"""5-gram 重复率：检测重复内容"""
words = text.split()
ngrams = set()
total = 0
for i in range(len(words) – n + 1):
gram = ' '.join(words[i:i+n])
if gram in ngrams:
total += 1
else:
ngrams.add(gram)
return total / max(len(words), 1)
4.2 分类器过滤（第二道防线） 启发式过滤后，用分类器进行语义层面的质量评估：
FineWeb-Edu 的质量分类方法 [2]： Step 1: 收集"高质量"标准
- 正样本：Wikipedia、ArXiv、Books、教育网站
- 负样本：广告页、SEO垃圾、机器生成 Step 2: 训练 FastText 分类器
- 模型：FastText (轻量级)
- 特征：n-gram
- 训练数据：50K 标注样本 Step 3: 对每个网页打分 score = classifier.predict(webpage_text) → 得分 > 3 为 “educational” → 保留 → 得分 < 1 为 “low quality” → 丢弃 结果：15T → 1.3T (保留 8.7%)
“”“FastText 质量分类器”“” from sklearn.linear_model import LogisticRegression from sklearn.feature_extraction.text import TfidfVectorizer import numpy as np class QualityClassifier: “”“网页质量二元分类器”“”
def __init__(self):
self.vectorizer = TfidfVectorizer(
max_features=50000,
ngram_range=(1, 3), # unigram + bigram + trigram
max_df=0.8, # 去除太常见
min_df=0.001, # 去除太罕见
sublinear_tf=True # 用 1+log(tf)
)
self.classifier = LogisticRegression(max_iter=1000, C=0.1)
def train(self, positive_docs: List[str], negative_docs: List[str]):
"""训练质量分类器"""
docs = positive_docs + negative_docs
labels = [1] * len(positive_docs) + [0] * len(negative_docs)
# 特征提取
X = self.vectorizer.fit_transform(docs)
# 训练
self.classifier.fit(X, labels)
# 输出最重要的特征
feature_names = self.vectorizer.get_feature_names_out()
coefs = self.classifier.coef_[0]
top_positive = np.argsort(coefs)[-20:] # 高质量信号词
top_negative = np.argsort(coefs)[:20] # 低质量信号词
print("高质量信号词:", [feature_names[i] for i in top_positive])
print("低质量信号词:", [feature_names[i] for i in top_negative])
def predict_quality(self, doc: str) -> float:
"""返回质量分数 [0, 1]"""
X = self.vectorizer.transform([doc])
return self.classifier.predict_proba(X)[0][1]
def batch_filter(self, docs: List[str], threshold: float = 0.6) -> List[str]:
"""批量过滤"""
X = self.vectorizer.transform(docs)
scores = self.classifier.predict_proba(X)[:, 1]
return [
doc for doc, score in zip(docs, scores)
if score >= threshold
]
2026年新趋势：直接用 LLM 做质量评估（Model-in-the-Loop）
“”“使用LLM做数据质量评估”“” import openai # 或 vLLM 本地 def quality_score_with_llm(text: str, model=“Qwen3-8B”) -> int: “”“LLM评估文本质量，返回1-5分”“” response = openai.ChatCompletion.create( model=model, messages=[ {“role”: “system”, “content”: “”“评估网页内容的质量，给出1-5分： 5 – 教育/学术/专业：Wikipedia、ArXiv 4 – 高质量博客/新闻 3 – 普通内容 2 – 低质量/SEO/稀疏 1 – 垃圾/广告/机器生成 只返回数字，不要解释。”“”}, {“role”: “user”, “content”: text[:2000]} # 只用前2000字符 ], max_tokens=1, temperature=0 ) return int(response.choices[0].message.content) DataFlow 框架的核心创新之一：200+ 内置算子覆盖从启发式到 LLM-in-the-loop 的所有过滤需求 [3]。
五、去重：别让模型背同样的课文 5.1 为什么需要去重？ 训练数据中的重复内容的危害 [5]：
假设你的训练数据中有 5% 的重复内容：
- 每个重复样本看到的次数多 → 过拟合
- 损失函数被重复内容主导 → 长尾知识被忽视
- 记忆而不是理解 → 泛化能力下降
- 创造性降低 → 相似输出 实测：去重后的 7B 模型在 MMLU 上 +2.3% 5.2 精确去重（Exact Dedup） 最简单的去重方法：用 MinHash + LSH 判断文档级别重复。
“”“MinHash + LSH 文档去重”“” from datasketch import MinHash, MinHashLSH import hashlib class DocumentDeduplicator: “”“基于 MinHash 的文档去重”“”
def __init__(self, num_perm=128, threshold=0.8):
self.num_perm = num_perm
self.threshold = threshold
self.lsh = MinHashLSH(
threshold=threshold,
num_perm=num_perm
)
self.documents = {}
def _shingle(self, text: str, k: int = 13) -> List[str]:
"""生成 k-shingle"""
words = text.split()
shingles = set()
for i in range(len(words) – k + 1):
shingle = ' '.join(words[i:i+k])
shingles.add(shingle.encode('utf-8'))
return shingles
def _compute_minhash(self, text: str) -> MinHash:
"""计算文档的 MinHash"""
m = MinHash(num_perm=self.num_perm)
shingles = self._shingle(text)
for shingle in shingles:
m.update(shingle)
return m
def add_document(self, doc_id: str, text: str):
"""添加文档到 LSH 索引"""
m = self._compute_minhash(text)
self.lsh.insert(doc_id, m)
self.documents[doc_id] = text
def is_duplicate(self, doc_id: str, text: str) -> bool:
"""判断文档是否重复"""
m = self._compute_minhash(text)
results = self.lsh.query(m)
# 去掉自身
others = [r for r in results if r != doc_id]
return len(others) > 0
def deduplicate(self, docs: List[Dict]) -> List[Dict]:
"""对文档列表去重"""
unique_docs = []
for doc in docs:
doc_id = hashlib.md5(doc["text"][:100].encode()).hexdigest()
if not self.is_duplicate(doc_id, doc["text"]):
self.add_document(doc_id, doc["text"])
unique_docs.append(doc)
return unique_docs
FineWeb 使用的 URL 级去重 [2]
class URLDeduplicator: “”“URL标准化去重”“”
@staticmethod
def normalize_url(url: str) -> str:
"""URL标准化"""
from urllib.parse import urlparse, urlunparse
parsed = urlparse(url.lower())
# 移除 www 和默认端口
netloc = parsed.netloc.replace('www.', '')
# 移除尾部 /
path = parsed.path.rstrip('/')
# 移除 tracking 参数
query = '&'.join([
q for q in parsed.query.split('&')
if not any(t in q for t in ['utm_', 'ref=', 'source='])
])
return urlunparse((
parsed.scheme,
netloc,
path,
parsed.params,
query,
parsed.fragment
))
@staticmethod
def deduplicate_urls(records: List[Dict]) -> List[Dict]:
"""保留每个规范URL的最新版本"""
seen = set()
unique = []
for record in records:
url = URLDeduplicator.normalize_url(record.get("url", ""))
if url not in seen:
seen.add(url)
unique.append(record)
return unique
去重层次 [5]：
层次 方法 去除比例 效果 URL 级 规范URL + 黑名单 10-20% 基础 文档级 MinHash + LSH 15-30% 核心 段落级 SimHash 5-10% 精细 N-gram 级 Exact n-gram dedup 2-5% 最后 5.3 模糊去重（Fuzzy Dedup） CCNet Pipeline 的灵魂：不仅仅是精确重复，还要检测高度相似（近重复，Near-duplicate）：
“”“CCNet 风格的模糊去重 [5]”“” from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity class FuzzyDeduplicator: “”“基于 embeddings 的语义去重”“”
def __init__(self, model_name="BAAI/bge-small-zh-v1.5"):
self.encoder = SentenceTransformer(model_name)
self.embeddings = []
self.texts = []
self.threshold = 0.92 # 余弦相似度阈值
def is_near_duplicate(self, text: str) -> bool:
"""判断文本是否是已有文本的近重复"""
emb = self.encoder.encode(text, normalize_embeddings=True)
if len(self.embeddings) > 0:
all_embs = np.stack(self.embeddings)
similarities = all_embs @ emb # 归一化后的余弦
if similarities.max() > self.threshold:
return True
self.embeddings.append(emb)
self.texts.append(text)
return False
六、隐私与安全过滤 6.1 PII 检测与移除
“”“PII（个人身份信息）检测与移除”“” import re class PIIFilter: “”“检测并移除训练数据中的个人信息”“”
PATTERNS = {
"email": r'\\b[\\w\\.-]+@[\\w\\.-]+\\.\\w{2,}\\b',
"phone_cn": r'1[3-9]\\d{9}', # 中国手机号
"phone_us": r'\\b\\d{3}[-.]?\\d{3}[-.]?\\d{4}\\b',
"id_card": r'\\b[1-9]\\d{5}(19|20)\\d{2}(0[1-9]|1[0-2])\\d{2}\\d{3}[\\dXx]\\b',
"ip_address": r'\\b\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\b',
"credit_card": r'\\b\\d{4}[- ]?\\d{4}[- ]?\\d{4}[- ]?\\d{4}\\b',
"wechat_id": r'[Ww][Xx][a-zA-Z]\\w{5,19}',
"qq_number": r'\\b[1-9]\\d{4,11}\\b',
}
@classmethod
def detect_pii(cls, text: str) -> Dict[str, List[str]]:
"""检测文本中的PII"""
found = {}
for name, pattern in cls.PATTERNS.items():
matches = re.findall(pattern, text)
if matches:
found[name] = matches[:5] # 只显示前5个
return found
@classmethod
def redact_pii(cls, text: str) -> str:
"""替换PII为占位符"""
replacements = {
"email": "[EMAIL]",
"phone_cn": "[PHONE]",
"phone_us": "[PHONE]",
"id_card": "[ID_CARD]",
"ip_address": "[IP]",
"credit_card": "[CC]",
}
for name, pattern in cls.PATTERNS.items():
replacement = replacements.get(name, "[PII]")
text = re.sub(pattern, replacement, text)
return text
2026年各实验室的PII处理实践
class PIIScanner: “”“Industrial PII detection with LLM-assisted”“”
@staticmethod
def scan_with_rules(text: str) -> List[Dict]:
"""基于规则的PII扫描"""
results = []
for name, pattern in PIIFilter.PATTERNS.items():
for match in re.finditer(pattern, text):
results.append({
"type": name,
"value": match.group(),
"start": match.start(),
"end": match.end()
})
return results
@staticmethod
def scan_with_llm(text: str) -> List[Dict]:
"""使用LLM辅助检测规则难以捕获的PII"""
# 使用小模型做 PII 标注
# 比如：Qwen3-4B 在本地检测
pass
七、Tokenization：把文本变成数字 7.1 Tokenizer 的选择 Tokenizer 词表大小 适用场景 中文效率 代表模型 BPE (GPT) 50K-200K 英文为主 中 GPT系列 BBPE (SentencePiece) 32K-256K 多语言 好 LLaMA, Qwen Unigram (SentencePiece) 32K-256K 多语言 最好 DeepSeek, GLM TikToken 100K-200K 通用 好 OpenAI, Claude 7.2 中文 Tokenization 的挑战
“”“同一个意思，不同tokenizer的效率对比”“” texts = [ “机器学习是大语言模型的核心”, “Machine learning is the core of LLM” ]
BPE (LLaMA 3 tokenizer, vocab=128K) 的中文效率
LLaMA 3: 每个中文字约 1.5-2 tokens
“机器学习是大语言模型的核心”
→ BPE: 约24 tokens (编码)
→ BBPE: 约14 tokens
→ Unigram: 约12 tokens
效率对比（相同的语义内容）
text_cn = “人工智能正在深刻改变世界” text_en = “Artificial intelligence is profoundly changing the world” “”" tokenizer | 中文 | 英文 | 中文/英文比例 GPT-4o | 8 | 7 | 1.14 ← 优化的中文tokenizer LLaMA 3 | 14 | 8 | 1.75 ← 中文效率一般 Qwen3 | 7 | 7 | 1.0 ← 中英文1:1 DeepSeek V4 | 6 | 6 | 1.0 ← 中英文1:1, 最优 结论：单语言训练中，tokenizer 效率直接影响模型容量 如果一个中文词=2个token，那模型的有效上下文就少了一半 “”" 7.3 训练 Tokenizer
“”“使用 SentencePiece 训练中文/多语言 Tokenizer”“” import sentencepiece as spm
训练参数
params = { ‘input’: ‘train_corpus.txt’, # 训练文本（建议10GB+） ‘model_prefix’: ‘qwen3_tokenizer’, # 输出前缀 ‘vocab_size’: 151936, # Qwen3 的词表大小 ‘character_coverage’: 0.9995, # 字符覆盖率 ‘model_type’: ‘bpe’, # BPE/Unigram/BPE ‘max_sentence_length’: 4192, # 最大句子长度 ‘pad_id’: 0, ‘unk_id’: 1, ‘bos_id’: 2, ‘eos_id’: 3, ‘split_digits’: True, # 拆分数字 ‘split_by_unicode_script’: True, # 按unicode脚本拆分 ‘byte_fallback’: True, # UNK时回退到字节 ‘allow_whitespace_only_pieces’: True, ‘remove_extra_whitespaces’: False, ‘normalization_rule_name’: ‘nfkc’, # 标准化 }
训练
spm.SentencePieceTrainer.train(**params)
测试
sp = spm.SentencePieceProcessor() sp.load(‘qwen3_tokenizer.model’) print(“词表大小:”, sp.vocab_size()) print(“中文’人工智能’:”, sp.encode(“人工智能”, out_type=str)) print(“Token数:”, len(sp.encode(“人工智能”)))
输出: Token数: 1 (好的tokenizer)
八、数据配比：烹饪秘方 8.1 常见数据配比 模型 配比（代码:数学:书籍:网页:其他） 总token数 LLaMA 1 4.5% : 2% : 4.5% : 80% : 9% 1.4T LLaMA 3 12% : 3% : 8% : 65% : 12% 15T DeepSeek V4 15% : 5% : 10% : 55% : 15% 25T Qwen3 12% : 4% : 8% : 60% : 16% 18T Phi-4 30% 合成 : 20% 代码 : 50% 文本 10T 2026 年趋势：
代码占比从 4.5% → 15%：逻辑推理能力依赖代码
合成数据从 0% → 30%：高质量合成正在替代低质量爬虫数据
数学数据独立成类：链式推理训练需要更高比例的数学
8.2 数据配比优化
“”“动态数据配比调度器”“” import numpy as np from collections import defaultdict class DataMixScheduler: “”“动态数据配比：在训练过程中调整各领域占比”“”
def __init__(self, base_mix: Dict[str, float]):
"""
base_mix: {"code": 0.12, "math": 0.03, "book": 0.08, "web": 0.65, "other": 0.12}
"""
self.base_mix = base_mix
self.loss_history = defaultdict(list)
self.step = 0
# 各领域的"重要性权重"
self.domain_weights = {
"code": 0.25, # 高权重：代码训练提升推理
"math": 0.20, # 高权重：数学训练提升逻辑
"book": 0.10,
"web": 0.05, # 低权？但量大
"other": 0.15,
}
def update_domain_loss(self, domain: str, loss: float):
"""记录各领域的loss"""
self.loss_history[domain].append(loss)
if len(self.loss_history[domain]) > 100:
self.loss_history[domain].pop(0)
def get_mix(self, global_step: int) -> Dict[str, float]:
"""根据loss趋势调整数据配比"""
mix = dict(self.base_mix)
if global_step < 1000:
# 预热阶段：使用base mix
return mix
# 根据loss变化调整
# 如果某个领域loss降得慢 → 增加该领域数据
for domain in mix:
if len(self.loss_history[domain]) >= 20:
recent_loss = np.mean(self.loss_history[domain][-20:])
early_loss = np.mean(self.loss_history[domain][:20])
ratio = recent_loss / max(early_loss, 1e-8)
if ratio > 1.1: # loss降得慢
mix[domain] *= 1.05 # 增加
elif ratio < 0.95: # loss降得太快
mix[domain] *= 0.95 # 减少
# 归一化
total = sum(mix.values())
for domain in mix:
mix[domain] /= total
return mix
===== 标准 DataMix 构建 =====
def build_dataloader_mix( datasets: Dict[str, ‘Dataset’], mix_ratios: Dict[str, float], batch_size: int = 256, seed: int = 42 ) -> ‘DataLoader’: “”“构建带配比的 DataLoader”“”
rng = np.random.RandomState(seed)
class MixedIterator:
def __init__(self, datasets, mix_ratios):
self.datasets = datasets
self.mix_ratios = mix_ratios
self.iterators = {
k: iter(d) for k, d in datasets.items()
}
def __next__(self):
# 按 mix_ratio 采样
domain = rng.choice(
list(self.mix_ratios.keys()),
p=list(self.mix_ratios.values())
)
try:
batch = next(self.iterators[domain])
except StopIteration:
self.iterators[domain] = iter(self.datasets[domain])
batch = next(self.iterators[domain])
return batch, domain
return MixedIterator(datasets, mix_ratios)
九、北大DataFlow：2026年数据工程新范式 2026 年，北京大学 DCAI 团队发布的 DataFlow 框架，彻底改变了数据准备的方式 [3]。
9.1 三大核心理念
DataFlow 处理 SFT 数据
15K 高质量 DataFlow 样本 = 50K 传统筛选数据
数学能力
DataFlow-Reasoning-10K → Qwen2.5-32B: 55.7 avg Open-R1 (10K): 54.2
代码能力
DataFlow-Code-10K → 7B: 46.2 avg LiveCodeBench: 21.9 → 33.2 (+52%)
Text-to-SQL
DataFlow-Text2SQL-90K → Qwen2.5-Coder-7B: Spider: 73.4% → 82.0% (+8.6%) BIRD: 50.9% → 59.2% (+8.3%) EHRSQL: 24.3% → 56.1% (+31.8%)
统一微调
DataFlow-Instruct-10K: 已接近官方Instruct版本 → 10K > 50K !! 核心启示：10K 精心构造的 DataFlow 数据 ≈ 50K 传统筛选 ≈ 200K 原始数据 [3]
十、实战：构建完整的数据清洗Pipeline 10.1 CCNet 风格 Pipeline
“”“完整的CCNet风格数据清洗Pipeline”“” import argparse import json import logging from pathlib import Path from typing import Iterator, Dict, List from urllib.parse import urlparse logging.basicConfig(level=logging.INFO) logger = logging.getLogger(name) class CCNetPipeline: “”" CCNet Pipeline 的 Python 实现 流程: 原始Common Crawl → URL过滤 → 语言识别 → 质量过滤 → 去重 → 输出 “”"
def __init__(self, config: Dict):
self.config = config
self.stats = {
"total_input": 0,
"after_url_filter": 0,
"after_lang_filter": 0,
"after_quality_filter": 0,
"after_dedup": 0,
}
def url_filter(self, record: Dict) -> bool:
"""URL过滤"""
url = record.get("url", "")
domain = urlparse(url).netloc.lower()
# 黑名单域名
blocked_domains = {
"example.com", "spam.com",
"ad.doubleclick.net", "analytics.*",
}
for blocked in blocked_domains:
if blocked in domain:
return False
# 文件类型过滤
blocked_extensions = {".pdf", ".zip", ".rar", ".exe", ".dmg"}
for ext in blocked_extensions:
if url.lower().endswith(ext):
return False
return True
def language_filter(self, text: str, target_lang="en") -> bool:
"""语言识别 – 使用 fastText"""
# 简化：通过Unicode范围判断
if target_lang == "zh":
chinese_chars = sum(1 for c in text if '\\u4e00' <= c <= '\\u9fff')
return chinese_chars / max(len(text), 1) > 0.1
elif target_lang == "en":
alpha_chars = sum(1 for c in text if c.isalpha())
return alpha_chars / max(len(text), 1) > 0.5
return True
def quality_filter(self, text: str) -> bool:
"""质量过滤"""
if len(text) < 200:
return False
# 重复检测
lines = text.split('\\n')
unique_lines = set(lines)
if len(unique_lines) / max(len(lines), 1) < 0.3:
return False
# 标点比例
punct = sum(1 for c in text if c in '.,!?;:')
if punct / max(len(text), 1) < 0.01:
return False
return True
def exact_dedup(self, text: str, hash_set: set) -> bool:
"""精确重复检测"""
text_hash = hash(text)
if text_hash in hash_set:
return False
hash_set.add(text_hash)
return True
def process(
self,
input_stream: Iterator[Dict],
output_path: str
) -> Dict:
"""执行完整Pipeline"""
output_file = open(output_path, 'w', encoding='utf-8')
doc_hash_set = set()
for record in input_stream:
self.stats["total_input"] += 1
text = record.get("text", "")
# Step 1: URL过滤
if not self.url_filter(record):
continue
self.stats["after_url_filter"] += 1
# Step 2: 语言过滤
if not self.language_filter(text, self.config.get("language", "en")):
continue
self.stats["after_lang_filter"] += 1
# Step 3: 质量过滤
if not self.quality_filter(text):
continue
self.stats["after_quality_filter"] += 1
# Step 4: 去重
if not self.exact_dedup(text, doc_hash_set):
continue
self.stats["after_dedup"] += 1
# 写入
output_file.write(json.dumps(record, ensure_ascii=False) + '\\n')
if self.stats["after_dedup"] % 10000 == 0:
logger.info(f"Processed {self.stats['after_dedup']:,} docs…")
output_file.close()
# 打印保留率
for stage, count in self.stats.items():
if stage == "total_input":
continue
retention = count / max(self.stats["total_input"], 1) * 100
logger.info(f"{stage}: {count:,} ({retention:.1f}%)")
return self.stats
10.2 工业级 Pipeline 配置
data_pipeline_config.yaml
pipeline: url_filter: enabled: true blocklist: “blocked_domains.txt” block_extensions: [“.pdf”, “.zip”, “.exe”, “.dmg”]
language: target: [“zh”, “en”, “ja”, “ko”] model: “fasttext-langdetect” confidence: 0.8
quality: heuristics: – min_length: 200 – max_length: 100000 – min_alpha_ratio: 0.3 – max_repetition_ratio: 0.3 – min_stop_word_ratio: 0.05 – removed: [“gibberish”, “seo_spam”, “ad_only”] classifier: model: “fasttext-quality” threshold: 3.0 llm_assisted: enabled: true # 2026新增 model: “Qwen3-4B” # 用小模型标注 sample_rate: 0.01 # 只抽1%让LLM复审
deduplication: url_level: true document_level: method: “minhash” threshold: 0.8 num_perm: 128 paragraph_level: method: “simhash” threshold: 0.9 ngram_level: n: 13 threshold: 0.95
privacy: enabled: true rules: [“email”, “phone”, “id_card”, “ip”, “credit_card”] llm_assisted: true # 检测非结构化PII
output: format: “jsonl” compression: “zstd” shard_size: 1024 # MB per shard 总结 核心要点 质量 > 数量 — FineWeb-Edu 证明 1.3T 高质量 > 15T 原始，Phi-4 证明好数据>大参数
多层过滤 — URL → 语言 → 质量(启发式+分类器) → 去重(精确+模糊) → PII → Tokenize
数据配比是秘方 — 代码从 4.5% 涨到 15%，合成数据从 0 涨到 30%+
2026 三大利器 — DataFlow (北大)、FineWeb (HF)、DCLM (多机构)
DataFlow 核心启示 — 10K 精心构造 > 50K 粗筛 > 200K 原始
2026 数据工程趋势 趋势 状态 影响 合成数据占比 > 20% ✅ 已确认 小型团队也能产高质量数据 Model-in-the-Loop 成为标配 ✅ 已确认 LLM 参与数据生成/评估 Agent 自动数据工程 ⚠️ 发展中 DataFlow-Agent 初现 数据观测性工具化 ✅ 已确认 DataFlow-WebUI 数据配比自动调整 ⚠️ 研究中 动态 Loss-based 调度 面试高频问题 Q: FineWeb-Edu 只用了 1.3T token 却超过了 15T 的 Common Crawl，原理是什么？ A: 核心在于数据密度。1.3T edu 数据的"有效信息"密度远超 15T 未筛选数据。模型在高质量数据上少token也能学到更多，because you fewer “junk” tokens wasted.
Q: 处理中文数据的特殊难点？ A: 1) 中文高质量数据源远少于英文 2) 中文 token 效率偏低（每个字 1-2 tokens）3) 中文网页噪音大（SEO 严重）4) 中文 PII 格式多样，难以穷举
Q: 2026年合成数据为什么这么火？ A: Phi-4 证明了高质量合成 ≈ 甚至 > 真实数据。合成数据的优势：1) 可控质量与难度 2) 针对弱点生成 3) 无限量 4) 无隐私问题。DataFlow 实验进一步证明 10K 合成数据效果接近官方 Instruct。
参考资料
[1] Microsoft Research. Phi-4: Technical Report. 2025.12 [2] HuggingFace. FineWeb: Large-Scale Web Text Dataset for LLM Pretraining. 2024 [3] Peking University DCAI. DataFlow: Systematic Data Engineering for LLMs. arXiv:2512.16676. 2026 [4] DataComp-LM. DCLM: A Framework for LLM Data-Centric AI. 2024 [5] CCNet Pipeline. Wenzek et al. CCNet: Extracting High Quality Monolingual Datasets from Web Crawl Data. 2020 [6] LLaMA 3 Team. The Llama 3 Herd of Models. 2024 [7] DeepSeek. DeepSeek V4 Technical Report. 2026
