数据工程
15 万亿个 token——LLaMA 3 的训练数据量。这个数字听起来很大，但它从哪来？不可能从天上掉下来。
这一节从互联网原始 HTML 出发，走完数据清洗的完整 pipeline：文本提取、质量过滤、去重、数据混合，每一步为什么做、怎么做。
数据工程的起点是 Common Crawl——一个每月爬取整个互联网的非营利项目，原始数据量约 500TB/月。这些数据是 HTML 格式，里面混着导航栏、广告、JavaScript 代码和评论区内容。
直接喂给模型等于喂垃圾，需要一套系统的方法把原始 HTML 变成干净、可训练的文本。整个过程分为五步：文本提取 → 质量过滤 → 去重 → 数据混合 → Tokenize。每一步的筛选标准都会直接影响最终模型的质量，因此每一步都需要明确的判断规则。
这一节按顺序拆解这五个步骤，每步用真实数据片段展示筛选前后的变化。
0. 数据 Pipeline 总览
先看全景，再逐个击破：
Common Crawl (原始 HTML，~500TB/月)
│
▼
┌─────────────┐
│ 1. 文本提取 │ HTML → 纯文本，去掉导航/广告/脚本
└──────┬──────┘
▼
┌───────────── ┐
│ 2. 语言过滤 │ 只要中文/英文等目标语言
└──────┬──────┘
▼
┌─────────────┐
│ 3. 质量过滤 │ 去广告/导航/乱码/太短/太长
└──────┬──────┘
▼
┌─────────────┐
│ 4. 去重 │ 精确去重 + MinHash 近似去重
└──────┬──────┘
▼
┌─────────────┐
│ 5. 数据混合 │ Common Crawl + Wiki + Books + Code
└──────┬──────┘
▼
干净训练数据 ✨
1. 文本提取：从 HTML 里提取正文
1.1 Common Crawl 里有什么？
Common Crawl 每个月爬取数十亿个网页，存储为 WARC（Web ARChive）格式。每个 WARC 文件里是一个完整的 HTML 页面——包含正文，也包含大量和正文无关的内容：
- 导航栏：「首页 | 关于 | 联系我们」
- 广告弹窗：「限时优惠！点击购买！」
- JavaScript 代码：「function trackUser(){...}」
- CSS 样式：「.sidebar { float: right; }」
- 页脚：「Copyright 2024. All rights reserved.」
- 评论区：「沙发！好文！顶！」
一篇正常的文章被这些东西包裹着。文本提取的任务就是识别并保留正文，丢弃其余部分。这个过程不完全自动化——现代工具（如 trafilatura、resiliparse）结合了 HTML 标签分析和机器学习模型来判断哪些内容是正文。
1.2 用一个模拟的 HTML 演示提取过程
# === 模拟: 一个典型的 Common Crawl HTML 页面 ===
raw_html = """
<!DOCTYPE html>
<html>
<head>
<title>What is Machine Learning? - AI Blog</title>
<meta name="description" content="Learn ML basics">
<script src="tracking.js"></script>
<style>.ad { color: red; }</style>
</head>
<body>
<nav>
<a href="/">Home</a> |
<a href="/about">About</a> |
<a href="/contact">Contact</a>
</nav>
<div class="sidebar">
<div class="ad">
<h3>Sponsored</h3>
<p>Buy the best AI course! 50% off today only!</p>
<button>Click Here!</button>
</div>
<script type="text/javascript">
var _gaq = _gaq || [];
_gaq.push(['_setAccount', 'UA-12345-6']);
_gaq.push(['_trackPageview']);
</script>
</div>
<article>
<h1>What is Machine Learning?</h1>
<p>Machine learning is a subset of artificial intelligence
that enables systems to learn and improve from experience
without being explicitly programmed.</p>
<p>The process of learning begins with observations or data,
such as examples, direct experience, or instruction.</p>
<p>Machine learning algorithms build a mathematical model
based on sample data, known as "training data".</p>
</article>
<div class="comments">
<p>User123: Great article!</p>
<p>Bot456: Buy followers at cheap-followers.com!</p>
</div>
<footer>
<p>Copyright 2024 AI Blog. All rights reserved.</p>
<p>Terms of Service | Privacy Policy | Cookie Settings</p>
</footer>
</body>
</html>
"""
print("=== 原始 HTML ===")
print(raw_html[:500])
print("...")
print()
print("↑ 里面混了：导航栏、广告、JS跟踪代码、评论区垃圾、页脚")
=== 原始 HTML ===
<!DOCTYPE html>
<html>
<head>
<title>What is Machine Learning? - AI Blog</title>
<meta name="description" content="Learn ML basics">
<script src="tracking.js"></script>
<style>.ad { color: red; }</style>
</head>
<body>
<nav>
<a href="/">Home</a> |
<a href="/about">About</a> |
<a href="/contact">Contact</a>
</nav>
<div class="sidebar">
<div class="ad">
<h3>Sponsored</h3>
<p>Buy the best AI course! 50% off today only!</p>
<button>Click Here!</butto
...
↑ 里面混了：导航栏、广告、JS跟踪代码、评论区垃圾、页脚
import re
# === 手工模拟文本提取过程 ===
print("=== 文本提取：HTML → 纯文本 ===")
print()
# Step 1: 去掉 <script> 和 <style> 标签及其内容
def remove_scripts_styles(html):
"""去掉 script 和 style 标签（包括内容）"""
html = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.DOTALL | re.IGNORECASE)
html = re.sub(r'<style[^>]*>.*?</style>', '', html, flags=re.DOTALL | re.IGNORECASE)
return html
# Step 2: 去掉所有 HTML 标签
def strip_tags(html):
"""去掉所有 <...> 标签"""
return re.sub(r'<[^>]+>', '', html)
# Step 3: 清理空白
def clean_whitespace(text):
"""合并多空行、去首尾空白"""
text = re.sub(r'[ \t]+', ' ', text) # 合并连续空格/tab
text = re.sub(r'\n\s*\n\s*\n+', '\n\n', text) # 多空行 → 一个空行
return text.strip()
# 一步一步走
print("Step 1 — 去掉 script/style 标签:")
no_scripts = remove_scripts_styles(raw_html)
print(f" 原始: {len(raw_html)} 字符 → 去掉后: {len(no_scripts)} 字符")
print()
print("Step 2 — 去掉所有 HTML 标签:")
plain_text = strip_tags(no_scripts)
print(f" 去掉标签后: {len(plain_text)} 字符")
print()
print("Step 3 — 清理空白:")
clean_text = clean_whitespace(plain_text)
print(f" 清理空白后: {len(clean_text)} 字符")
print()
print("=" * 60)
print("提取结果:")
print("=" * 60)
print(clean_text)
print("=" * 60)
print()
print("注意：导航栏(Home|About|Contact)还在、广告还在、评论区还在")
print(" → 这些需要在后续的「质量过滤」步骤中清除")
=== 文本提取：HTML → 纯文本 ===
Step 1 — 去掉 script/style 标签:
原始: 1454 字符 → 去掉后: 1226 字符
Step 2 — 去掉所有 HTML 标签:
去掉标签后: 831 字符
Step 3 — 清理空白:
清理空白后: 707 字符
============================================================
提取结果:
============================================================
What is Machine Learning? - AI Blog
Home |
About |
Contact
Sponsored
Buy the best AI course! 50% off today only!
Click Here!
What is Machine Learning?
Machine learning is a subset of artificial intelligence
that enables systems to learn and improve from experience
without being explicitly programmed.
The process of learning begins with observations or data,
such as examples, direct experience, or instruction.
Machine learning algorithms build a mathematical model
based on sample data, known as "training data".
User123: Great article!
Bot456: Buy followers at cheap-followers.com!
Copyright 2024 AI Blog. All rights reserved.
Terms of Service | Privacy Policy | Cookie Settings
============================================================
注意：导航栏(Home|About|Contact)还在、广告还在、评论区还在
→ 这些需要在后续的「质量过滤」步骤中清除
2. 质量过滤
文本提取出来后，大部分内容仍然是低质量的。想象一下 Common Crawl 里有什么：自动生成的 SEO 页面、乱码、只有导航栏没有正文的页面、重复上千次的广告文案……这些内容如果直接拿去训练，模型学到的东西还不如不学。
因此质量过滤是整个 pipeline 中最关键的一步：宁可丢掉一些好数据，也不能让大量垃圾混入训练集。常用的方法分两个层次：
2.1 启发式规则（快速筛掉明显垃圾）
这些规则不需要模型，直接检查文本的统计特征。每条规则针对一类常见垃圾：
import numpy as np
np.random.seed(42)
# === 质量过滤规则：逐条演示 ===
print("=== 质量过滤规则 ===")
print()
# 准备几条「待审」文本
samples = [
{"name": "好文章", "text": "Machine learning is a subset of artificial intelligence that enables systems to learn and improve from experience without being explicitly programmed. The field has grown rapidly since the 2010s, driven by advances in deep learning and the availability of large datasets."},
{"name": "广告", "text": "BUY NOW!!! Click here!!! Limited time offer!!! Subscribe today and get 50% OFF!!! Don't miss this opportunity!!!"},
{"name": "目录页", "text": "Chapter 1. Introduction. Chapter 2. Methods. Chapter 3. Results. Chapter 4. Discussion. Chapter 5. Conclusion. Appendix A. Appendix B. References."},
{"name": "太短", "text": "Hello world."},
{"name": "随机字符", "text": "asdfjkl; qwerty zxcvbnm @#$%@# 123456789 !!!!!!!"},
{"name": "重复模板", "text": "This is a blog post.\n" * 30 + "unique content here"},
]
def quality_check(text):
"""返回 (是否保留, 丢弃原因)"""
# 规则 1: 长度过滤
words = text.split()
if len(words) < 5:
return False, f"太短 ({len(words)} 个词 < 5)"
if len(text) > 5000:
return False, f"太长 ({len(text)} 字符 > 5000)"
# 规则 2: 平均词长 — 正常英文词长 3-10，乱码词很长
avg_word_len = np.mean([len(w) for w in words])
if avg_word_len > 12:
return False, f"平均词长异常 ({avg_word_len:.1f} > 12)"
# 规则 3: 特殊字符占比 — 正常文章标点不会超过 15%
special_count = sum(1 for c in text if not c.isalnum() and not c.isspace())
special_ratio = special_count / max(len(text), 1)
if special_ratio > 0.25:
return False, f"特殊字符太多 ({special_ratio:.1%} > 25%)"
# 规则 4: 行重复率 — 模板页面有很多重复行
lines = [l.strip() for l in text.split('\n') if l.strip()]
if len(lines) > 3:
unique_ratio = len(set(lines)) / len(lines)
if unique_ratio < 0.4:
return False, f"行重复率过高 ({unique_ratio:.1%} < 40%)"
# 规则 5: 大写字母占比 — 全是 CAPS LOCK 的通常是垃圾
if len(text) > 50:
upper_ratio = sum(1 for c in text if c.isupper()) / sum(1 for c in text if c.isalpha())
if upper_ratio > 0.5:
return False, f"大写字母太多 ({upper_ratio:.1%} > 50%)"
return True, "通过"
print(f"{'文本':<12s} {'结果':>8s} {'原因'}")
print("-" * 55)
for sample in samples:
passed, reason = quality_check(sample['text'])
status = "✅ 保留" if passed else "❌ 丢弃"
print(f"{sample['name']:<12s} {status:>8s} {reason}")
print()
print("真实系统通常有 20-50 条这样的规则。")
print("这能过滤掉约 60-80% 的 Common Crawl 文本。")
=== 质量过滤规则 ===
文本 结果 原因
-------------------------------------------------------
好文章 ✅ 保留 通过
广告 ✅ 保留 通过
目录页 ✅ 保留 通过
太短 ❌ 丢弃 太短 (2 个词 < 5)
随机字符 ❌ 丢弃 特殊字符太多 (29.2% > 25%)
重复模板 ❌ 丢弃 行重复率过高 (6.5% < 40%)
真实系统通常有 20-50 条这样的规则。
这能过滤掉约 60-80% 的 Common Crawl 文本。
2.2 基于模型的过滤 — 请「语文老师」来评分
除了人工规则，还可以用一个已经训练好的轻量级语言模型（如 KenLM）来给每篇文章打分。
思路和做英语阅读一样：老师看一眼文章就能判断「这是人写的」还是「这是随机生成的」。
用语言模型计算文章的「困惑度」（Perplexity, PPL）：
PPL 很低 (< 10): 文章太简单，像「a a a a a...」→ 扔
PPL 正常 (10-1000): 像正常人类写的 → 留
PPL 很高 (> 1000): 乱码 → 扔
有些系统还训练一个二分类器：Wikipedia 文章 = 好（正例），Common Crawl 随机文章 = 坏（负例）。训练后让它给每篇文章打分。
关键洞察：Wikipedia = 「教科书级别」的好文本，用维基当尺子来量其他文本。
2.3 PII 和安全过滤：不是所有“正常文字”都适合训练
前面的规则会筛掉乱码、广告和模板页。但还有一类内容看起来很正常，却不应该直接喂给模型：个人信息和危险字符串。
PII（Personally Identifiable Information）指能识别到具体人的信息，比如邮箱、手机号、身份证号、家庭地址。模型如果在训练中反复见到这些内容，可能会在生成时复述出来。安全过滤还会检查 API key、私钥、密码、恶意脚本、明显仇恨/色情/暴力文本等。
这一步和“文章写得好不好”不是一回事。一篇文章可以很通顺，但里面夹着邮箱、电话、密钥；质量过滤会放过它，PII/安全过滤必须单独处理。
工业系统通常有两种做法：
| 做法 | 适合场景 | 例子 |
|---|---|---|
| 删除整篇文档 | 命中严重风险 | 泄露密钥、身份证、恶意代码 |
| 脱敏后保留 | 正文有价值，但夹杂少量 PII | 邮箱替换成 |
关键不是“看到邮箱就一定扔掉”，而是先判断风险等级：风险高的丢，风险低且正文有价值的可以脱敏。
import re
# === PII 和安全过滤演示 ===
print("=== PII / 安全过滤 ===")
print()
records = [
{
"name": "普通教程",
"text": "This tutorial explains gradient descent with a small example.",
},
{
"name": "含邮箱",
"text": "Contact me at [email protected] for the dataset notes.",
},
{
"name": "含手机号",
"text": "我的手机号是 13812345678，训练日志在这里。",
},
{
"name": "疑似密钥",
"text": "AWS_SECRET_ACCESS_KEY = ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890AB",
},
]
patterns = {
"EMAIL": r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}",
"PHONE_CN": r"1[3-9]\d{9}",
"SECRET": r"(api[_-]?key|secret[_a-z0-9-]*|token|password)\s*[=:]\s*[^\s]+",
}
def redact_or_drop(text):
"""检测 PII / 密钥；返回处理动作和处理后的文本"""
lowered = text.lower()
if re.search(patterns["SECRET"], lowered):
return "丢弃", "命中疑似密钥"
cleaned = text
hit = False
for label, pattern in patterns.items():
if label == "SECRET":
continue
if re.search(pattern, cleaned):
hit = True
cleaned = re.sub(pattern, f"<{label}>", cleaned)
if hit:
return "脱敏保留", cleaned
return "保留", cleaned
for record in records:
action, result = redact_or_drop(record["text"])
print(f"{record['name']:<8s} -> {action}")
print(f" {result}")
print()
print("关键观察：PII/安全过滤解决的是泄露风险，不是文章质量问题。")
3. 去重
3.1 为什么去重很重要？
互联网上大量内容是重复的：
- 同一篇新闻被 50 个网站转载（Google News syndication）
- 同一个代码片段在无数博客里出现
- 同一个「Lorem ipsum」占位文本
- 同一个 Cookie 声明（「This website uses cookies...」）
如果不去重：
- 模型花了宝贵的训练计算学重复内容
- 重复内容会让模型「背」而不是「理解」
- 训练数据统计失真（某段话的权重是它本来应有的一百倍）
3.2 两层去重
import hashlib
# === 精确去重 ===
print("=== 第一层：精确去重 (Exact Dedup) ===")
print()
# 模拟 5 篇文章，其中 2 篇是重复的
docs = [
"Machine learning is a subset of artificial intelligence.",
"Deep learning uses neural networks with many layers.",
"Machine learning is a subset of artificial intelligence.", # 和 doc 1 完全一样
"Natural language processing deals with text data.",
"Deep learning uses neural networks with many layers.", # 和 doc 2 完全一样
]
print(f"总共 {len(docs)} 篇文章")
print()
# 哈希去重
seen = set()
unique_docs = []
for i, doc in enumerate(docs):
# SHA256 哈希：把文章变成唯一指纹
fingerprint = hashlib.sha256(doc.encode()).hexdigest()[:16] # 只取前16位展示
is_new = fingerprint not in seen
print(f"文档 {i+1}: hash={fingerprint} {'✅ 保留' if is_new else '❌ 重复，丢弃'}")
if is_new:
seen.add(fingerprint)
unique_docs.append(doc)
print()
print(f"去重后: {len(unique_docs)} 篇 ({len(docs) - len(unique_docs)} 篇被删除)")
print()
print("精确去重能去掉约 5-15% 的 Common Crawl 数据")
print("但还不够——大部分重复是「改写」而非「逐字照抄」")
=== 第一层：精确去重 (Exact Dedup) ===
总共 5 篇文章
文档 1: hash=a5c7f6a65fc1e1e9 ✅ 保留
文档 2: hash=b85a4f441b9eb02a ✅ 保留
文档 3: hash=a5c7f6a65fc1e1e9 ❌ 重复，丢弃
文档 4: hash=90b6942d04cd6a5c ✅ 保留
文档 5: hash=b85a4f441b9eb02a ❌ 重复，丢弃
去重后: 3 篇 (2 篇被删除)
精确去重能去掉约 5-15% 的 Common Crawl 数据
但还不够——大部分重复是「改写」而非「逐字照抄」
import hashlib
# === MinHash 近似去重：原理手算 ===
print("=== 第二层：MinHash 近似去重 ===")
print()
print("问题：100 亿篇文章，怎么快速找出相似的？")
print(" 逐对比较 = 100亿² = 不可能")
print()
print("MinHash 的思路：给每篇文章算一个「指纹」，指纹相似 → 文章相似")
print()
# === 手工 MinHash 演示 ===
print("=== MinHash 手算 ===")
print()
# 三篇文章
doc_A = "the cat sat on the mat and looked at the dog"
doc_B = "the cat sat on the mat and watched the dog" # 和 A 只差一个词
doc_C = "quantum mechanics describes behavior of subatomic particles" # 完全不同的主题
print(f"文档 A: {doc_A}")
print(f"文档 B: {doc_B}")
print(f"文档 C: {doc_C}")
print()
# Step 1: 把每篇文章切成 n-gram 集合（连续 3 个词为一组）
def get_ngrams(text, n=3):
words = text.lower().split()
return set(' '.join(words[i:i+n]) for i in range(len(words) - n + 1))
A_ngrams = get_ngrams(doc_A, 3)
B_ngrams = get_ngrams(doc_B, 3)
C_ngrams = get_ngrams(doc_C, 3)
print(f"文档 A 的 n-gram ({len(A_ngrams)} 个): {A_ngrams}")
print()
print(f"文档 B 的 n-gram ({len(B_ngrams)} 个): {B_ngrams}")
print()
# Step 2: 算 Jaccard 相似度 (交集/并集)
def jaccard(s1, s2):
inter = len(s1 & s2)
union = len(s1 | s2)
return inter / union if union > 0 else 0
j_AB = jaccard(A_ngrams, B_ngrams)
j_AC = jaccard(A_ngrams, C_ngrams)
print(f"A ∩ B 交集大小: {len(A_ngrams & B_ngrams)}")
print(f"A ∪ B 并集大小: {len(A_ngrams | B_ngrams)}")
print(f"Jaccard(A, B) = {len(A_ngrams & B_ngrams)}/{len(A_ngrams | B_ngrams)} = {j_AB:.2%}")
print()
print(f"Jaccard(A, C) = {j_AC:.2%}")
print()
# Step 3: MinHash 指纹（超级简化版——用随机哈希模拟）
print("=== MinHash 签名计算 ===")
print()
# 模拟：把每个 n-gram 哈希到一个整数，取每篇文章的最小 K 个哈希值作为签名
def minhash_signature(ngrams, num_hashes=4):
"""
为 n-gram 集合生成 MinHash 签名
实际 MinHash 用随机排列模拟，这里用不同种子的哈希来模拟
"""
sig = []
for i in range(num_hashes):
# 对每个 n-gram 用 seed=i 算哈希，取最小值
min_val = float('inf')
for ng in ngrams:
raw = (ng + str(i)).encode()
h = int(hashlib.sha256(raw).hexdigest(), 16) % 100000
min_val = min(min_val, h)
sig.append(min_val) if min_val != float('inf') else sig.append(0)
return sig
sig_A = minhash_signature(A_ngrams)
sig_B = minhash_signature(B_ngrams)
sig_C = minhash_signature(C_ngrams)
print(f"A 的 MinHash 签名: {sig_A}")
print(f"B 的 MinHash 签名: {sig_B}")
print(f"C 的 MinHash 签名: {sig_C}")
print()
# MinHash 相似度 = 签名一致的比例
def minhash_sim(s1, s2):
matches = sum(1 for a, b in zip(s1, s2) if a == b)
return matches / len(s1)
mh_AB = minhash_sim(sig_A, sig_B)
mh_AC = minhash_sim(sig_A, sig_C)
print(f"MinHash 近似相似度 A-B: {mh_AB:.2%} (精确 Jaccard: {j_AB:.2%})")
print(f"MinHash 近似相似度 A-C: {mh_AC:.2%} (精确 Jaccard: {j_AC:.2%})")
print()
print("→ MinHash 把「逐对比较 n-gram」变成了「比较 4 个数字」")
print("→ 速度快了几个数量级！相似度 > 阈值（如 80%）→ 只保留一篇")
=== 第二层：MinHash 近似去重 ===
问题：100 亿篇文章，怎么快速找出相似的？
逐对比较 = 100亿² = 不可能
MinHash 的思路：给每篇文章算一个「指纹」，指纹相似 → 文章相似
=== MinHash 手算 ===
文档 A: the cat sat on the mat and looked at the dog
文档 B: the cat sat on the mat and watched the dog
文档 C: quantum mechanics describes behavior of subatomic particles
文档 A 的 n-gram (9 个): {'the mat and', 'at the dog', 'looked at the', 'sat on the', 'on the mat', 'cat sat on', 'and looked at', 'mat and looked', 'the cat sat'}
文档 B 的 n-gram (8 个): {'and watched the', 'the mat and', 'sat on the', 'on the mat', 'cat sat on', 'watched the dog', 'mat and watched', 'the cat sat'}
A ∩ B 交集大小: 5
A ∪ B 并集大小: 12
Jaccard(A, B) = 5/12 = 41.67%
Jaccard(A, C) = 0.00%
=== MinHash 签名计算 ===
A 的 MinHash 签名: [3222, 9218, 1325, 5504]
B 的 MinHash 签名: [11955, 5250, 1325, 8548]
C 的 MinHash 签名: [5570, 14068, 24500, 15543]
MinHash 近似相似度 A-B: 25.00% (精确 Jaccard: 41.67%)
MinHash 近似相似度 A-C: 0.00% (精确 Jaccard: 0.00%)
→ MinHash 把「逐对比较 n-gram」变成了「比较 4 个数字」
→ 速度快了几个数量级！相似度 > 阈值（如 80%）→ 只保留一篇
3.3 去污染：评测题不能混进训练集
去重解决的是“训练集内部有重复”。但还有一个更隐蔽的问题：训练集里混进了评测题。
比如你用 HumanEval 评测代码能力，结果训练数据里早就有 HumanEval 的题目和答案。模型分数会很高，但这不是能力强，而是考试前看过答案。这叫 decontamination（去污染）。
所以工业 pipeline 通常会做两类检查：
| 检查对象 | 目的 | 常见方法 |
|---|---|---|
| 训练集内部 | 少学重复内容 | exact dedup、MinHash |
| 训练集 vs 评测集 | 防止 benchmark 泄露 | exact match、n-gram overlap、embedding 相似度 |
去污染不是为了让数据更“干净好看”，而是为了让评测分数可信。
# === 去污染演示：训练数据里是否混入评测题 ===
print("=== Decontamination：训练集 vs 评测集 ===")
print()
train_docs = [
"To reverse a string in Python, use s[::-1]. This is a common interview trick.",
"The capital of France is Paris, and it is known for the Eiffel Tower.",
"Write a function that returns the sum of two numbers. def add(a, b): return a + b",
]
eval_items = [
"Write a function that returns the sum of two numbers.",
"What is the capital of France?",
]
def word_ngrams(text, n=6):
"""把文本切成连续 n 个词的片段，用于粗略检测重叠"""
words = re.findall(r"[a-zA-Z]+", text.lower())
return set(tuple(words[i:i + n]) for i in range(len(words) - n + 1))
def max_overlap(eval_text, train_texts):
"""返回某道评测题和训练集的最大 n-gram 重叠比例"""
eval_grams = word_ngrams(eval_text, n=4)
best = 0
best_doc = None
for doc in train_texts:
train_grams = word_ngrams(doc, n=4)
if not eval_grams:
continue
overlap = len(eval_grams & train_grams) / len(eval_grams)
if overlap > best:
best = overlap
best_doc = doc
return best, best_doc
for item in eval_items:
score, doc = max_overlap(item, train_docs)
status = "疑似污染" if score >= 0.6 else "未命中"
print(f"评测题: {item}")
print(f"最大重叠: {score:.0%} -> {status}")
if doc:
print(f"命中文档: {doc[:80]}...")
print()
print("关键观察：去污染检查的是训练集和评测集之间的重叠。")
4. 数据混合
假设你已经有了各种数据来源：
| 来源 | 数量 | 质量 | 作用 |
|---|---|---|---|
| Common Crawl（过滤后） | 10T tokens | 中等 | 基础知识、多样性 |
| Wikipedia | 0.1T tokens | 很高 | 事实准确性 |
| Books | 0.5T tokens | 高 | 长文连贯性 |
| Code (GitHub) | 1T tokens | 中等 | 逻辑推理能力 |
| Academic Papers | 0.05T tokens | 很高 | 科学推理 |
4.1 怎么混合？不是按原始数量直接倒在一起
策略：高质量数据可以多学几遍（多 epoch），低质量数据少学几遍。
# === 数据混合策略 ===
print("=== 数据混合：如何配比 ===")
print()
sources = [
("Common Crawl (过滤后)", 10000, 0.6, 1),
("Wikipedia", 100, 0.95, 4),
("Books", 500, 0.85, 2),
("Code (GitHub)", 1000, 0.75, 2),
("ArXiv Papers", 50, 0.9, 4),
("News", 300, 0.7, 1),
]
print(f"{'来源':<25s} {'原始量':>10s} {'质量':>6s} {'Epoch':>6s} {'有效量':>10s} {'占比':>8s}")
print("-" * 72)
total_effective = 0
results = []
for name, size, quality, epochs in sources:
effective = size * epochs
total_effective += effective
results.append((name, size, quality, epochs, effective))
for name, size, quality, epochs, effective in results:
ratio = effective / total_effective * 100
print(f"{name:<25s} {size:>6.0f}B {quality:>5.0%} {epochs:>4d}x {effective:>8.0f}B {ratio:>6.1f}%")
print()
print(f"总有效数据: {total_effective:.0f}B tokens")
print()
print("关键决策:")
print(" • Wikipedia 虽然只有 100B，但 epoch=4 → 实际喂了 400B (高质量多学)")
print(" • Common Crawl 虽然 10T，但 epoch=1 → 只学一遍 (避免垃圾)")
print(" • ArXiv 少量但质量高，epoch=4 → 放大科学推理能力")
print()
print("注意: 多 epoch ≠ 把文章复制 4 份")
print(" 而是训练 4 轮后再重新 shuffle → 让模型每次看到不同的顺序")
=== 数据混合：如何配比 ===
来源 原始量 质量 Epoch 有效量 占比
------------------------------------------------------------------------
Common Crawl (过滤后) 10000B 60% 1x 10000B 71.9%
Wikipedia 100B 95% 4x 400B 2.9%
Books 500B 85% 2x 1000B 7.2%
Code (GitHub) 1000B 75% 2x 2000B 14.4%
ArXiv Papers 50B 90% 4x 200B 1.4%
News 300B 70% 1x 300B 2.2%
总有效数据: 13900B tokens
关键决策:
• Wikipedia 虽然只有 100B，但 epoch=4 → 实际喂了 400B (高质量多学)
• Common Crawl 虽然 10T，但 epoch=1 → 只学一遍 (避免垃圾)
• ArXiv 少量但质量高，epoch=4 → 放大科学推理能力
注意: 多 epoch ≠ 把文章复制 4 份
而是训练 4 轮后再重新 shuffle → 让模型每次看到不同的顺序
4.2 从“配比”到 Data Recipe
前面说“高质量数据多学几遍”，这是直觉版。工业里会把它写成 data recipe：一份可执行的配方，说明每类数据怎么抽样、怎么过滤、怎么混合。
为什么要叫 recipe？因为数据不是一次性洗完就结束。你会不断试：
Recipe A：多放网页，少放代码
Recipe B：多放代码和数学，少放低质量网页
Recipe C： 只保留高分网页，但减少总 token 数
然后用小模型或 proxy training 快速验证：哪个 recipe 的 loss 更低？哪个在 MMLU、GSM8K、HumanEval 上更好？如果 B 的代码能力更强但常识下降，就说明“代码加多了，但通用文本可能不够”。
这就是 data-centric LLM 的核心：不是只调模型结构，也调数据配方。
# === Data Recipe A/B 对比演示 ===
print("=== Data Recipe：数据配方对比 ===")
print()
recipes = {
"A_web_heavy": {
"web": 0.70,
"wiki": 0.10,
"books": 0.10,
"code": 0.05,
"math": 0.05,
},
"B_code_math": {
"web": 0.45,
"wiki": 0.10,
"books": 0.10,
"code": 0.25,
"math": 0.10,
},
"C_quality_web": {
"web": 0.50,
"wiki": 0.20,
"books": 0.15,
"code": 0.10,
"math": 0.05,
},
}
# 这些分数是教学模拟：展示如何比较 recipe，不代表真实模型结果。
eval_scores = {
"A_web_heavy": {"MMLU": 54, "GSM8K": 28, "HumanEval": 12},
"B_code_math": {"MMLU": 52, "GSM8K": 39, "HumanEval": 24},
"C_quality_web": {"MMLU": 57, "GSM8K": 31, "HumanEval": 16},
}
header = (
f"{'Recipe':<15s} {'web':>5s} {'wiki':>5s} {'books':>6s} "
f"{'code':>6s} {'math':>6s} {'MMLU':>6s} {'GSM8K':>6s} {'HEval':>6s}"
)
print(header)
print("-" * len(header))
for name, mix in recipes.items():
scores = eval_scores[name]
print(
f"{name:<15s} "
f"{mix['web']:>4.0%} {mix['wiki']:>5.0%} {mix['books']:>6.0%} "
f"{mix['code']:>6.0%} {mix['math']:>6.0%} "
f"{scores['MMLU']:>6.1f} {scores['GSM8K']:>6.1f} {scores['HumanEval']:>6.1f}"
)
print()
print("关键观察：recipe 没有绝对最优，要看你更在意通用能力、数学还是代码。")
4.3 常见 Benchmark：data recipe 要用什么尺子评？
前面说用 MMLU、GSM8K、HumanEval 看 data recipe 的效果。那这些名字到底是什么？
先记住一个边界：benchmark 通常不是预训练数据来源，而是评测尺子。它们像考试卷，用来观察模型在哪些能力上变强或变弱。训练数据工程里关注 benchmark，有两个目的：
- 判断 data recipe 的方向。加代码数据后，HumanEval / MBPP 有没有涨？加数学数据后，GSM8K / MATH 有没有涨？过滤网页后，MMLU / C-Eval 有没有掉？
- 做去污染检查。训练集里不能混进评测题，否则分数会虚高。
常见 benchmark 可以按能力分类：
| 能力类型 | 常见 benchmark | 它大概在测什么 |
|---|---|---|
| 通识/学科知识 | MMLU, MMLU-Pro, C-Eval, CMMLU, AGIEval | 多学科考试题，覆盖历史、法律、医学、理工等 |
| 数学推理 | GSM8K, MATH, AIME | 从小学应用题到竞赛数学，重点看多步推理 |
| 代码能力 | HumanEval, HumanEval+, MBPP, LiveCodeBench | 写函数、补代码、通过单元测试 |
| 复杂推理 | BBH, GPQA, ARC-Challenge | 难题、多步推理、科学知识和抽象推理 |
| 常识/语言理解 | HellaSwag, Winogrande, ARC-Easy | 常识续写、指代消解、基础问答 |
| 指令遵循/对话 | IFEval, MT-Bench, AlpacaEval | 是否按用户要求的格式、约束和风格回答 |
| 检索/RAG | HotpotQA, Natural Questions, RAGAS 自建集 | 能不能基于外部资料回答，而不是凭记忆乱猜 |
| Agent/工具使用 | GAIA, SWE-bench, WebArena | 会不会查资料、改代码、操作 工具并完成任务 |
| 多模态 | MMMU, MMBench, MathVista, VQAv2 | 看图答题、图表理解、视觉数学推理 |
几个容易混淆的名字：
- MMLU / C-Eval / CMMLU / AGIEval：都偏“考试型知识”，但语言和题源不同。英文模型只看 MMLU 不够，中文模型要看 C-Eval / CMMLU。
- GSM8K vs MATH：GSM8K 是小学应用题，很多模型已经接近饱和；MATH 更难，更接近竞赛题。
- HumanEval vs MBPP：都看代码，但题目集合不同。HumanEval 更常见，MBPP 题量更大一些；工业报告常两者一起看。
- BBH：BIG-Bench Hard，挑的是原 BIG-Bench 中模型更容易翻车的任务，常用来看复杂推理。
所以 data recipe 的观察方式不是“总分越高越好”这么粗，而是看能力画像：
网页清洗更严格 → MMLU / C-Eval 是否更稳？
数学数据加权提高 → GSM8K / MATH 是否上涨？
代码数据加权提高 → HumanEval / MBPP 是否上涨？
指令数据质量提高 → IFEval / MT-Bench 是否上涨？
如果一个 recipe 让 HumanEval 涨了 10 分，但 MMLU 掉了 5 分，你就要问：这是你想要的代码专用模型，还是通用能力被代码数据挤掉了？
# === Benchmark 画像：看 data recipe 改动影响了哪些能力 ===
print("=== Benchmark 能力画像：Recipe A vs Recipe B ===")
print()
benchmarks = [
("MMLU", "通识/学科"),
("C-Eval", "中文知识"),
("GSM8K", "数学应用题"),
("MATH", "竞赛数学"),
("HumanEval", "代码函数"),
("MBPP", "代码题"),
("BBH", "复杂推理"),
("IFEval", "指令遵循"),
]
# 教学模拟分数：只用来演示怎么读表，不代表真实模型。
recipe_a = {
"MMLU": 54,
"C-Eval": 50,
"GSM8K": 28,
"MATH": 12,
"HumanEval": 12,
"MBPP": 18,
"BBH": 36,
"IFEval": 52,
}
recipe_b = {
"MMLU": 52,
"C-Eval": 48,
"GSM8K": 39,
"MATH": 20,
"HumanEval": 24,
"MBPP": 31,
"BBH": 39,
"IFEval": 51,
}
print(f"{'Benchmark':<12s} {'能力':<8s} {'A':>5s} {'B':>5s} {'变化':>7s} 观察")
print("-" * 62)
for name, ability in benchmarks:
delta = recipe_b[name] - recipe_a[name]
if delta >= 5:
note = "明显变强"
elif delta <= -3:
note = "需要警惕"
else:
note = "基本持平"
print(
f"{name:<12s} {ability:<8s} "
f"{recipe_a[name]:>5.1f} {recipe_b[name]:>5.1f} {delta:>+7.1f} {note}"
)
print()
print("关键观察：B 像是加了代码/数学数据，代码和数学涨了，但通识略掉。")
print("这不是绝对好坏，要看你想训练通用模型还是代码/数学专用模型。")
4.4 跑 Benchmark 的常见工具
知道 benchmark 之后，下一个问题是：怎么跑？不要自己手写一堆零散脚本。工业和研究里通常会用现成评测框架，因为它们已经处理了 prompt、few-shot、metric、并行、结果汇总这些麻烦事。
| 工具 | 适合谁 | 特点 | 典型场景 |
|---|---|---|---|
lm-evaluation-harness (lm-eval) | 开源模型研究者 | EleutherAI 维护，支持大量经典 NLP/LLM benchmark，论文复现常见 | 跑 MMLU、GSM8K、HellaSwag、BBH 等 |
| EvalScope（ModelScope） | 国内模型和工程团队 | ModelScope 社区的一站式评测框架，支持能力评测、性能压测、结果可视化 | 跑 C-Eval、MMLU、GSM8K，也做推理性能测试 |
| OpenCompass | 系统做模型报告的人 | OpenCompass 是大模型评测平台，支持 100+ 数据集、分布式评测、配置化实验 | 复现模型技术报告、做多维能力榜单 |
| OpenAI Evals | 做自定义产品评测的人 | OpenAI 的 eval 框架和 benchmark registry，适合写业务私有 eval | 评估 API 模型、做回归测试、自定义 judge |
| HELM | 做综合研究评测的人 | Stanford CRFM 的整体评测框架，强调多维度和透明报告 | 研究型横向比较 |
| VLMEvalKit / OpenCompass 多模态 | 做 VLM 的人 | 面向视觉语言模型的多模态评测 | 跑 MMMU、MMBench、MathVista |
三个实用判断：
- 想快速复现实验：优先看
lm-eval或 OpenCompass。 - 模型在 ModelScope / 国内生态里：EvalScope 更顺手。
- 要做产品自己的测试集：OpenAI Evals 或自己写一层 eval harness 更合适。
命令长什么样？先看概念，不要求现在运行：
# lm-evaluation-harness：评测 HuggingFace 模型
lm_eval --model hf --model_args pretrained=Qwen/Qwen2.5-0.5B --tasks mmlu,gsm8k --num_fewshot 5 --batch_size auto
# EvalScope：用 OpenAI-compatible API 跑一个小评测
evalscope eval --model your-model-name --api-url $OPENAI_API_BASE_URL --api-key $OPENAI_API_KEY --eval-type openai_api --datasets gsm8k
# OpenCompass：通常用配置文件组织模型、数据集和评测方式
python run.py --models hf_qwen2_5_7b --datasets mmlu_gen ceval_gen gsm8k_gen
真正写报告时，一定要同时记录：模型版本、benchmark 名称、few-shot 数量、prompt 模板、是否 CoT、temperature、max tokens、评测工具版本。否则同一个模型可能跑出不一样的分数，别人也复现不了。
5. 完整 Pipeline 实战：为一个 1B 模型准备数据
前面四节分别讲了文本提取、质量过滤、去重、数据混合四个环节。但在实际项目中，这些环节并不是独立运行的——它们构成一条流水线，上一步的输出是下一步的输入，任何一个环节出问题都会影响最终训练效果。
下面把整个流程串起来，模拟一次完整的数据工程 pipeline：从原始 Common Crawl WARC 文件出发，经过提取 → 过滤 → 去重 → 混合，最终产出可以直接送入训练的数据。每一步我们都会输出统计数字，让读者对「处理了多少、丢掉了多少」有直观感受。
实际工业 pipeline 的代码量很大（通常数千行），这里用简化版本展示核心逻辑。每条规则的实际效果都 可以通过输出的统计数字来验证。
print("=== 实战: 1B LLM 数据 Pipeline ===")
print()
steps = [
("Step 1: 确定数据预算", [
"Chinchilla 最优: N = 1B → D ≈ 20B tokens",
"过度训练: N = 1B → D ≈ 100B tokens",
"选择: 50B tokens (折中，性价比高)",
]),
("Step 2: 下载 Common Crawl", [
"下载最近的 2-3 个月 dump (约 20TB 压缩 WARC)",
"工具: cc_downloader, HuggingFace datasets",
]),
("Step 3: 文本提取 + 语言过滤", [
"WARC → HTML → 纯文本 (trafilatura / resiliparse)",
"语言检测 (fastText): 只保留英文和中 文",
"产出: ~2TB 纯文本 (约 400B tokens)",
]),
("Step 4: 质量过滤", [
"启发式规则: 长度/词长/特殊字符/行重复",
"KenLM PPL 过滤: 10 < PPL < 1000",
"产出: ~200GB (约 40B tokens) → 只剩下 10%",
]),
("Step 5: 去重", [
"精确去重: SHA256 哈希 → 去掉 ~10%",
"MinHash 近似去重: 相似度 > 80% 的只保留一篇 → 去掉 ~20%",
"产出: ~140GB (约 28B tokens)",
]),
("Step 6: 混合其他来源", [
"Wikipedia (2x epoch): 4B tokens",
"Books (2x epoch): 6B tokens",
"Code GitHub (2x epoch): 10B tokens",
"其他: 2B tokens",
"总计: 28B + 22B = 50B tokens",
]),
("Step 7: Tokenize + 打包", [
"用 BPE tokenizer 把文本转成 token IDs",
"拼接成连续序列，切成 2048/4096 长度的 chunk",
"在文档边界插入 <EOS> token",
"Shuffle + 打包成训练 batch → 开始训练！",
]),
]
for title, details in steps:
print(title)
for d in details:
print(f" {d}")
print()
print("压缩比总结:")
print(" 20TB WARC → 2TB 纯文本 → 200GB 过滤后 → 140GB 去重后")
print(" 最后可用数据只占原始下载的 ~0.7%")
=== 实战: 1B LLM 数据 Pipeline ===
Step 1: 确定数据预算
Chinchilla 最优: N = 1B → D ≈ 20B tokens
过度训练: N = 1B → D ≈ 100B tokens
选择: 50B tokens (折中，性价比高)
Step 2: 下载 Common Crawl
下载最近的 2-3 个月 dump (约 20TB 压缩 WARC)
工具: cc_downloader, HuggingFace datasets
Step 3: 文本提取 + 语言过滤
WARC → HTML → 纯文本 (trafilatura / resiliparse)
语言检测 (fastText): 只保留英文和中文
产出: ~2TB 纯文本 (约 400B tokens)
Step 4: 质量过滤
启发式规则: 长度/词长/特殊字符/行重复
KenLM PPL 过滤: 10 < PPL < 1000
产出: ~200GB (约 40B tokens) → 只剩下 10%
Step 5: 去重
精确去重: SHA256 哈希 → 去掉 ~10%
MinHash 近似去重: 相似度 > 80% 的只保留一篇 → 去掉 ~20%
产出: ~140GB (约 28B tokens)
Step 6: 混合其他来源
Wikipedia (2x epoch): 4B tokens
Books (2x epoch): 6B tokens
Code GitHub (2x epoch): 10B tokens
其他: 2B tokens
总计: 28B + 22B = 50B tokens
Step 7: Tokenize + 打包
用 BPE tokenizer 把文本转成 token IDs
拼接成连续序列，切成 2048/4096 长度的 chunk
在文档边界插入 <EOS> token
Shuffle + 打包成训练 batch → 开始训练！
压缩比总结:
20TB WARC → 2TB 纯文本 → 200GB 过滤后 → 140GB 去重后
最后可用数据只占原始下载的 ~0.7%
6. 数据质量 > 数量
这是 NLP 领域反复验证的结论：
T5 论文 (2019):
清洗后的 750GB 数据训练效果 > 未清洗的 6TB 数据
→ 1/8 的数据量，质量够高反而更好
Textbooks Are All You Need (2023):
用 7B 高质量「教科书」风格数据训练的 1.3B 模型
代码能力超过了用更多数据训练的大模型
→ 数据质量可以部分弥补参数量的不足
LLaMA 2 → LLaMA 3 (2023-2024):
模型架构几乎没变，最大的改进在数据质量和规模
从 2T → 15T tokens，数据量增大的同时质量反而提高了
→ 更好的过滤、更好的去重、更好的混合策略
这个结论对实际工作的指导意义很直接：如果你只有有限的资源，优先投入在提高数据质量上，而不是盲目扩大数据量。100 本好书的训 练价值远高于 10000 本垃圾邮件。
7. 工业案例和工具地图
前面我们手写了一个教学版 pipeline。那工业界是不是也这么做？大方向一样，但工程形态更系统。
先看几个公开案例：
| 案例 | 核心看点 | 你应该学到什么 |
|---|---|---|
| LLaMA 3 | 训练在 15T+ token 的公开可得数据上，强调数据质量和规模 | 架构变化不大时，数据质量仍能带来巨大提升 |
| FineWeb | 从 Common Crawl 构建 18T+ 英文网页 token，公开过滤/去重实验 | 好数据集不是“下载网页”，而是大量 ablation 后的结果 |
| DCLM / DataComp-LM | 固定训练设置，专门比较不同数据过滤和混合策略 | data recipe 可以像模型结构一样被系统评测 |
| Dolma | 公开 3T token 预训练语料和数据处理报告 | 开放模型需要开放数据处理细节，方便复现和分析 |
| RedPajama Data | 提供质量信号、去重信息和多源数据 | 数据集不只是一堆文本，还可以带质量标签 |
再看工具。你不需要现在就会用它们，但要知道它们解决的不 是同一个层次的问题：
| 工具 | 角色 | 典型能力 |
|---|---|---|
| Data-Juicer | 一站式 LLM 数据处理系统 | mapper/filter/dedup/selector/evaluator，组合成 data recipe |
| DataTrove | 大规模文本处理 pipeline | Common Crawl 处理、过滤、去重、分布式执行 |
| NeMo Curator | NVIDIA 的数据清洗与筛选工具 | 文本/图像/视频数据清洗、PII、安全过滤、去重 |
| IBM Data Prep Kit | 通用 AI 数据准备工具集 | 文档转换、去重、PII 检测、质量转换 |
注意一个容易误解的点：这些库不会自动告诉你“什么数据最好”。它们提供的是可组合的积木。真正困难的是设计 recipe、跑实验、看评测，再决定下一轮怎么改。
所以工业 pipeline 可以这样理解：
原始数据
→ 一串 operators（提取、过滤、脱敏、去重、去污染、混合）
→ data recipe
→ 小模型试训 / 评测
→ 根据结果调整 recipe
→ 再处理、再训练、再评测
这比“写一堆清洗规则”更接近真实工作流。
8. 从 Tokenize 到训练流
数据工程最后一步是 tokenize。这之后数据长什么样？
文档 A: "Machine learning is great."
→ tokenize → [42, 567, 18, 891, 15]
文档 B: "Deep learning is also great."
→ tokenize → [123, 567, 18, 456, 891, 15]
然后拼接成一条「token 面条」:
[42, 567, 18, 891, 15, <EOS>, 123, 567, 18, 456, 891, 15, <EOS>, ...]
切成训练 chunk (假设 seq_len=8):
Chunk 1: [42, 567, 18, 891, 15, <EOS>, 123, 567]
Chunk 2: [18, 456, 891, 15, <EOS>, ...]
...
注意：chunk 边界会切断文档！
Chunk 1 的后 3 个 token 来自文档 B，前 5 个来自文档 A
→ 模型有时会在 chunk 里「跨文档」学习
→ 解决方法：加 <EOS> 告诉模型「新文档开始」
8.1 Sequence Packing：把短文档拼成满的 chunk
上一节展示了「拼接 → 切成固定长度」的朴素做法。但它有一个问题：如果文档长度分布不均，有些 chunk 会被大量 padding 填充，浪费算力。
Sequence Packing 用装箱算法解决这个问题：把多条短文档紧凑地塞进一个固定长度的 chunk，尽量不留空位。
朴素做法（有 padding 浪费）:
Chunk 1: [文档A(500 tokens)][padding(3596 tokens)] ← 浪费 88%
Chunk 2: [文档B(800 tokens)][padding(3296 tokens)] ← 浪费 80%
Packing 做法（尽量塞满）:
Chunk 1: [文档A(500)][文档B(800)][文档C(1200)][文档D(1500)][padding(96)] ← 利用率 98%
实际收益：训练吞吐量可以提升 2x~4x（短文档越多收益越大）。
但 Packing 有一个必须解决的技术问题：不同文档之间不能互相 attend，否则会信息泄漏。 解决方法是在 attention 计算时加 block-diagonal mask。
# === Sequence Packing 手算演示 ===
print("=== Sequence Packing：装箱算法手算 ===")
print()
# 模拟 8 条文档的 token 长度
docs = [
("文档A", 120),
("文档B", 80),
("文档C", 200),
("文档D", 50),
("文档E", 150),
("文档F", 90),
("文档G", 60),
("文档H", 40),
]
max_seq_len = 256 # chunk 的固定长度
# First-Fit Decreasing 装箱：先把长文档排前面，再逐个往 chunk 里塞
sorted_docs = sorted(docs, key=lambda x: -x[1]) # 按长度降序
chunks = [] # 每个 chunk 是 (文档列表, 已用长度)
for name, length in sorted_docs:
placed = False
# 尝试塞进已有的 chunk
for chunk in chunks:
if chunk[1] + length + 1 <= max_seq_len: # +1 是 EOS token
chunk[0].append((name, length))
chunk[1] += length + 1
placed = True
break
if not placed:
chunks.append([[(name, length)], length + 1])
print(f"总文档数: {len(docs)}")
print(f"Chunk 容量: {max_seq_len} tokens")
print(f"装箱结果: {len(chunks)} 个 chunk")
print()
total_used = 0
for i, (items, used) in enumerate(chunks):
doc_names = ' + '.join(name for name, _ in items)
padding = max_seq_len - used
util = used / max_seq_len * 100
total_used += used
print(f" Chunk {i+1}: [{doc_names}]")
print(f" 已用 {used} tokens, padding {padding}, 利用率 {util:.0f}%")
overall_util = total_used / (len(chunks) * max_seq_len) * 100
print(f"\n整体利用率: {overall_util:.0f}%")
print()
print("对比朴素做法（每条文档一个 chunk）:")
naive_chunks = len(docs)
print(f" 朴素: {naive_chunks} 个 chunk")
print(f" Packing: {len(chunks)} 个 chunk")
print(f" 节省: {(1 - len(chunks)/naive_chunks)*100:.0f}%")
8.2 Block-Diagonal Attention Mask：防止文档之间互相偷看
Packing 之后，一个 chunk 里可能包含 3~4 条不同的文档。如果不加任何限制，模型做 attention 时会「看到」其他文档的内容——这就是信息泄漏。
解决方法：block-diagonal mask。每个文档只能在内部做 attention，不同文档之间的 attention 权重设为负无穷（softmax 后变 0）。
Chunk: [文档A][文档B][文档C]
Attention Mask:
A A A B B C C
A [ 1 1 1 0 0 0 0 ] ← A 只看 A
A [ 1 1 1 0 0 0 0 ]
A [ 1 1 1 0 0 0 0 ]
B [ 0 0 0 1 1 0 0 ] ← B 只看 B
B [ 0 0 0 1 1 0 0 ]
C [ 0 0 0 0 0 1 1 ] ← C 只看 C
C [ 0 0 0 0 0 1 1 ]
对角线上的块各自独立 = block-diagonal
import numpy as np
# === Block-Diagonal Mask 构造演示 ===
print("=== Block-Diagonal Attention Mask ===")
print()
# 假设一个 packed chunk 包含 3 条文档，长度分别为 3, 2, 2
doc_lengths = [3, 2, 2]
total_len = sum(doc_lengths) # 7
# 构造 block-diagonal mask（纯 numpy）
mask = np.zeros((total_len, total_len))
pos = 0
for length in doc_lengths:
for i in range(length):
for j in range(i + 1):
mask[pos + i, pos + j] = 1.0
pos += length
print(f"文档长度: {doc_lengths}, 总长度: {total_len}")
print()
print("Block-Diagonal Mask (1=可见, 0=屏蔽):")
print(" ", " ".join([f"t{i}" for i in range(total_len)]))
labels = []
pos = 0
for doc_idx, length in enumerate(doc_lengths):
for _ in range(length):
labels.append(f"D{doc_idx}")
print()
for i in range(total_len):
row = " ".join([f" {int(mask[i,j])}" for j in range(total_len)])
print(f" {labels[i]} t{i}: {row}")
print()
print("观察:")
print(" D0(Doc 0) 内部: t0→t0, t1→t0,t1, t2→t0,t1,t2 ← 因果 attention")
print(" D1(Doc 1) 内部: t3→t3, t4→t3,t4 ← 和 D0 完全隔离")
print(" D2(Doc 2) 内部: t5→t5, t6→t5,t6 ← 和其他文档隔离")
print()
print("实际训练中还需要:")
print(" 1. 每个文档的 position id 从 0 重新开始")
print(" 2. Loss 只在真实 token 上计算，padding 不算")
8.3 Fill-in-the-Middle (FIM)：训练代码补全模型的数据格式
前面讨论的全是一般语言模型的数据流：从左到右，自回归预测下一个 token。但对于代码模型，还有一种重要的数据构造方式。
写代码时，光标前后都有代码。左边是已经写好的部分（prefix），右边是还没改动的部分（suffix）。代码补全模型需要同时理解两边，才能准确填好中间。Fill-in-the-Middle（FIM）就是为这个场景设计的训练数据格式。
做法是把每段代码随机拆成三部分，用特殊 token 重新排列：
<PRE> prefix内容 <SUF> suffix内容 <MID> middle内容
训练时，模型看到 <PRE> ... <SUF> ... <MID> 之后，要预测被挖掉的 middle 部分。prefix 和 suffix 部分的 labels 设为 ignore_index——模型能看到这些上下文，但不为它们计算 loss。只有 middle 部分正常计算 loss。
举个例子，一段 5 行的代码：
原始代码:
a = 1
b = 2
c = 3
d = 4
e = 5
拆成前 2 + 中间 2 + 后 1:
prefix = "a = 1\nb = 2\n" (模型看到，不预测)
middle = "c = 3\nd = 4\n" (训练目标)
suffix = "e = 5" (模型看到，不预测)
FIM 格式（模型输入）:
<PRE> a = 1\nb = 2\n <SUF> e = 5 <MID> c = 3\nd = 4\n
└─ 可看到，不预测 ─┘ └ 可看到 ┘ └─── 要预测 ────┘
不是所有样本都做 FIM。通常约 50% 用标准自回归格式（保持一般语言理解和生成能力），50% 用 FIM 格式（学会看前后文做代码补全）。两种格式混合训练，模型既能正常对话，也能在 IDE 里做代码补全。FIM 不影响 tokenize 和 packing 的流程——它是在 tokenize 之前做一次文本级别的重新排列。
代码模型（Code Llama、DeepSeek-Coder、StarCoder 等）都把 FIM 作为标准配置。普通文本模型一般不用，因为写作场景天然是从左到右的。
import numpy as np
# FIM 格式转换的手算演示
def fim_transform(code):
"""
将一段代码转换为 FIM 训练格式
随机选择切点 → 拆成 prefix/middle/suffix → 按 FIM 格式重排
"""
lines = code.split('\n')
n = len(lines)
# 随机选切点
prefix_len = np.random.randint(1, max(2, int(n * 0.4)))
middle_len = np.random.randint(1, max(2, int(n * 0.5)))
prefix_len = min(prefix_len, n - 1)
middle_len = min(middle_len, n - prefix_len)
prefix = '\n'.join(lines[:prefix_len])
middle = '\n'.join(lines[prefix_len:prefix_len + middle_len])
suffix = '\n'.join(lines[prefix_len + middle_len:])
fim_text = f"<PRE> {prefix}\n<SUF> {suffix}\n<MID> {middle}"
return fim_text, prefix, middle, suffix
np.random.seed(42)
sample_code = """def calculate(x, y):
z = x + y
result = z * 2
if result > 100:
return result
else:
return 0
print(calculate(3, 4))"""
fim_text, prefix, middle, suffix = fim_transform(sample_code)
print("=== FIM 格式转换手算 ===")
print()
print("原始代码:")
print(sample_code)
print()
print("--- 拆分后 ---")
print(f"Prefix ({len(prefix)} 字符, 模型看到):")
print(prefix)
print()
print(f"Middle ({len(middle)} 字符, 训练目标):")
print(middle)
print()
print(f"Suffix ({len(suffix)} 字符, 模型看到):")
print(suffix if suffix else '(空)')
print()
print("--- FIM 格式（模型输入）---")
print(fim_text)
print()
print("=== 训练时的 Label Mask ===")
print()
print("规则：")
print(" <PRE> ... <SUF> ... <MID> 这些特殊 token → 知道但不预测")
print(" prefix 和 suffix 的内容 → label = ignore_index")
print(" middle 的内容 → label = 正常 token（参与 loss）")
print()
print("关键观察：")
print("1. FIM 通过重排文本，让模型在训练时就能利用双向上下文")
print("2. <PRE> 和 <SUF> 提供前后文信息，<MID> 之后模型开始预测被挖掉的部分")
print("3. 只有 middle 区域的 loss 被保留——prefix/suffix 给信息但不给惩罚")
print("4. 约 50% 样本用标准格式 + 50% 用 FIM → 兼顾自回归生成和代码补全")
小结
确认你已经搞懂了这些（按顺序检查）：
- ✅ 数据来源：Common Crawl（主体）+ Wikipedia + Books + Code + Papers
- ✅ Pipeline 五步：文本提取 → 语言过滤 → 质量过滤 → 去重 → 数据混合
- ✅ HTML → Text：去 script/style 标签 + 去 HTML 标签 + 清理空白
- ✅ 质量过滤：启发式规则（长度/词长/符号）+ PPL（语言模型打分）
- ✅ PII/安全过滤：质量正常的文本，也可能因为邮箱、手机号、密钥而需要脱敏或丢弃
- ✅ 精确去重：SHA256 哈希，完全相同的只留一份
- ✅ MinHash：把文章变指纹（n-gram → hash → 取最小 K 个），指纹相似 = 文章相似
- ✅ 去污染：训练集和评测集不能高度重叠，否则 benchmark 分数不可信
- ✅ 数据混合 ：从“高质量多 epoch”升级到可评测的 data recipe
- ✅ 常见 benchmark：MMLU/C-Eval 看知识，GSM8K/MATH 看数学，HumanEval/MBPP 看代码，BBH/GPQA 看难题推理
- ✅ 评测工具：lm-eval、EvalScope、OpenCompass、OpenAI Evals 负责把 benchmark 跑得可复现
- ✅ 工业工具：Data-Juicer、DataTrove、NeMo Curator、IBM Data Prep Kit 都是在帮你组合和运行数据 operators
- ✅ Tokenize 后：拼接成「token 面条」→ 切成固定长度 chunk → 开始训练
- ✅ FIM（Fill-in-the-Middle）：用 /
/ 重排文本，训练代码补全能力
一句话总结：数据工程是 LLM 训练中最关键的环节之一。教学版 pipeline 帮你看清主线；工业版 pipeline 的关键是把清洗、脱敏、去重、去污染、混合和评测连成闭环，不断迭代 data recipe。
作业
下面 3 道小作业都是“填空 + assert 检查”。建议你先自己想 3 分钟；可以用 AI 问思路、拆步骤、检查方向，但不建议直接让 AI 做完整答案。
# 作业 1：补全一个邮箱脱敏函数
# 小提示：可以用 re.sub，把命中的邮箱替换成 <EMAIL>。
import re
def redact_email(text):
"""把文本中的邮箱替换成 <EMAIL>"""
email_pattern = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
# TODO: 把下面的 None 改成一行 re.sub(...)
result = None
return result
# 填完后取消下一行注释运行检查
# assert redact_email("mail me at [email protected]") == "mail me at <EMAIL>"
# print("作业 1 通过：你已经会做最基本的 PII 脱敏。")
# 作业 2：补全 Jaccard 相似度
# 小提示：交集用 &，并集用 |。
def jaccard_similarity(a, b):
"""计算两个集合的 Jaccard 相似度"""
# TODO: 把下面的 None 改成 len(交集) / len(并集)
score = None
return score
# 填完后取消下面三行注释运行检查
# s1 = {"the cat", "cat sat", "sat down"}
# s2 = {"the cat", "cat sat", "sat here"}
# assert abs(jaccard_similarity(s1, s2) - 0.5) < 1e-9
# print("作业 2 通过：你已经会计算近似去重的核心相似度。")
# 作业 3：补全 data recipe 的加权采样量
# 小提示：有效采样量 = 原始 token 数 * 采样权重。
sources = [
{"name": "web", "tokens": 1000, "weight": 0.5},
{"name": "wiki", "tokens": 100, "weight": 2.0},
{"name": "code", "tokens": 200, "weight": 1.5},
]
# TODO: 把 None 改成一个列表推导式，算出每个来源的有效采样量
sampled_tokens = None
# 填完后取消下面两行注释运行检查
# assert sampled_tokens == [500, 200, 300]
# print("作业 3 通过：你已经理解 data recipe 里的采样权重。")