章节导航 第 10 章 / 共 20 章
第 10 章 第二部分：预训练数据工程
评估、归因与闭环
数据卡片(Datasheet)、数据探针(PPL分布/毒性分布/领域分布)、消融实验(1B小模型/Scaling Law外推)、训练数据归因TDA(Influence Function/TracIn/Datamodels/EK-FAC)、Scaling Law拟合(Loss(N,D)=E+A/N^α+B/D^β)、闭环设计
数据卡片 数据探针 消融实验 TDA Scaling Law 闭环设计
没有评估的数据工程是盲修。
数据-训练-评估的闭环，是工业级团队的分水岭。
10.1 数据卡片（Datasheet）
10.1.1 什么是数据卡片
数据卡片（Datasheet for Datasets, Gebru et al., 2021）是描述数据集的标准化文档，回答以下问题：
- 数据是什么：来源、规模、构成
- 为什么创建：动机、用途
- 如何创建：采集、清洗、过滤流程
- 如何评估：质量指标、已知局限
- 如何使用：推荐用途、禁忌场景
10.1.2 数据卡片模板
# 数据卡片示例
datasheet:
# 基础信息
name: 'MyPretrain-V1'
version: '1.0.0'
created_at: '2026-07-06'
maintainer: 'data-team@company.com'
# 动机
motivation: |
用于训练公司 7B 通用大模型
重点关注中文能力与代码能力
# 构成
composition:
total_tokens: 5_000_000_000_000 # 5T
num_documents: 2_500_000_000
languages:
en: 0.40
zh: 0.45
other: 0.15
sources:
cc: 0.50
github: 0.15
arxiv: 0.10
wiki: 0.10
books: 0.10
other: 0.05
# 创建流程
collection_process:
sources_detail: '...'
time_period: '2020-2024'
collection_method: 'Common Crawl S3 download'
preprocessing:
- 'HTML extraction (trafilatura)'
- 'Language identification (fastText lid.176)'
- 'Quality filtering (Gopher rules + FineWeb-Edu ≥3)'
- 'Safety filtering (Detoxify threshold=0.7)'
- 'Deduplication (MinHash LSH, threshold=0.8)'
- 'PII redaction (Presidio)'
- 'Decontamination (13-gram with MMLU/GSM8K/...)'
# 质量指标
metrics:
avg_ppl: 2.45
avg_educational_value: 3.2
toxicity_rate: 0.003
duplicate_rate: 0.02 # 去重后残留
pii_detected_rate: 0.001
# 已知局限
known_limitations:
- 'CC 数据存在英文偏置（60%英文）'
- 'GitHub 代码可能含测试数据残留'
- 'arXiv 论文仅到 2024-06'
- '未对低资源语言（如斯瓦希里语）做专门处理'
# 推荐用途
recommended_uses:
- '通用预训练'
- '中文能力训练'
- '代码能力训练'
prohibited_uses:
- '医疗诊断（未做医疗领域精筛）'
- '法律咨询（未做法律领域精筛）'
- '高风险决策（如信贷、招聘）'10.1.3 数据卡片的价值
- 可追溯：知道数据从哪来、怎么处理
- 可复现：别人能复现你的数据集
- 合规证明：满足法规对数据透明度的要求
- 风险沟通：明确告诉下游使用者数据的局限
10.2 数据探针（Data Probe）
数据探针是对数据集做多维度统计分析，发现数据问题。
10.2.1 核心探针维度
class DataProbe:
"""数据探针"""
def __init__(self, dataset):
self.dataset = dataset
def run_all_probes(self):
"""运行所有探针"""
return {
'ppl_distribution': self.ppl_distribution(),
'toxicity_distribution': self.toxicity_distribution(),
'domain_distribution': self.domain_distribution(),
'length_distribution': self.length_distribution(),
'token_frequency': self.token_frequency(),
'language_distribution': self.language_distribution(),
'quality_distribution': self.quality_distribution(),
'dedup_check': self.dedup_check(),
'contamination_check': self.contamination_check()
}
def ppl_distribution(self):
"""
PPL 分布探针
正常：钟形分布，峰值在 2-4
异常：双峰（数据有"两层"）、长尾（有大量高 PPL 异常）
"""
ppls = []
for doc in self.dataset.sample(10000):
ppl = compute_perplexity(self.ref_model, doc['text'])
ppls.append(ppl)
return {
'mean': np.mean(ppls),
'median': np.median(ppls),
'p25': np.percentile(ppls, 25),
'p75': np.percentile(ppls, 75),
'p95': np.percentile(ppls, 95),
'histogram': np.histogram(ppls, bins=50)
}
def toxicity_distribution(self):
"""毒性分布探针"""
from detoxify import Detoxify
detector = Detoxify('multilingual')
toxicities = []
for doc in self.dataset.sample(10000):
result = detector.predict(doc['text'][:2000])
toxicities.append(result['toxicity'])
return {
'mean': np.mean(toxicities),
'toxic_rate': np.mean(np.array(toxicities) > 0.7),
'severe_toxic_rate': np.mean(np.array(toxicities) > 0.3),
'histogram': np.histogram(toxicities, bins=20)
}
def domain_distribution(self):
"""领域分布探针"""
from collections import Counter
domains = Counter()
for doc in self.dataset:
for d in doc.get('domain', ['unknown']):
domains[d] += 1
return dict(domains.most_common(20))
def length_distribution(self):
"""长度分布探针"""
lengths = [len(doc['text']) for doc in self.dataset.sample(10000)]
token_lengths = [doc.get('token_count', 0) for doc in self.dataset.sample(10000)]
return {
'char_length': {
'mean': np.mean(lengths),
'median': np.median(lengths),
'p95': np.percentile(lengths, 95)
},
'token_length': {
'mean': np.mean(token_lengths),
'median': np.median(token_lengths)
}
}
def token_frequency(self):
"""Token 频率分布（Zipf 分布）"""
from collections import Counter
token_freq = Counter()
for doc in self.dataset.sample(10000):
for token in doc['tokens']:
token_freq[token] += 1
# 检查是否符合 Zipf 分布
sorted_freqs = sorted(token_freq.values(), reverse=True)
ranks = np.arange(1, len(sorted_freqs) + 1)
# 拟合 Zipf: freq = c / rank^s
from scipy.optimize import curve_fit
def zipf(r, c, s): return c / r**s
popt, _ = curve_fit(zipf, ranks, sorted_freqs)
return {
'vocab_size': len(token_freq),
'top_10_tokens': token_freq.most_common(10),
'rare_token_count': sum(1 for f in token_freq.values() if f < 5),
'zipf_exponent': popt[1] # 通常在 1.0-1.5
}
def contamination_check(self):
"""污染检测：与评测集重叠"""
eval_sets = load_eval_sets(['mmlu', 'gsm8k', 'human_eval', 'math'])
contamination = {}
for name, eval_data in eval_sets.items():
ngram_set = build_ngram_set(eval_data, n=13)
overlap_count = 0
for doc in self.dataset.sample(10000):
doc_ngrams = build_ngrams(doc['text'], n=13)
if doc_ngrams & ngram_set:
overlap_count += 1
contamination[name] = overlap_count / 10000
return contamination10.2.2 探针结果的解读
| 指标 | 正常范围 | 异常 | 原因 |
|---|---|---|---|
| 平均 PPL | 2-3 | >5 | 数据未清洗 |
| 毒性率 | <0.5% | >2% | 安全过滤失效 |
| 领域分布 | 多样 | 单一领域占 80%+ | 数据偏置 |
| 长度中位数 | 500-2000 字符 | <200 | 短文本过多 |
| 稀有 token | <10% | >30% | 词表设计问题 |
| 评测污染率 | <0.5% | >5% | 去污染失败 |
10.3 消融实验
10.3.1 为什么用 1B 小模型
消融实验用 1B 小模型的原因：
- 算力成本低：1B 模型训练成本是 70B 的 1/100
- 趋势外推有效：1B 上的相对趋势能外推到 70B
- 迭代速度快：1B 训练 1 天能跑完，70B 要 1 个月
10.3.2 消融实验设计
class AblationExperiment:
"""消融实验框架"""
def __init__(self, base_config):
self.base_config = base_config
self.results = []
def run_ablation(self, ablation_name, modified_config, tokens=10_000_000_000):
"""
运行单个消融实验
"""
print(f'Running ablation: {ablation_name}')
# 训练 1B 模型
model = train_1b_model(
config=modified_config,
tokens=tokens
)
# 评测
scores = {
'mmlu': eval_mmlu(model),
'arc': eval_arc(model),
'gsm8k': eval_gsm8k(model),
'human_eval': eval_human_eval(model)
}
self.results.append({
'name': ablation_name,
'config': modified_config,
'scores': scores
})
return scores
def compare_to_baseline(self):
"""对比基线"""
baseline = self.results[0]
print(f'{"Experiment":<30} {"MMLU":>8} {"ARC":>8} {"GSM8K":>8} {"HumE":>8}')
print(f'{"-"*70}')
print(f'{"baseline":<30} {baseline["scores"]["mmlu"]:>8.1f} {baseline["scores"]["arc"]:>8.1f} ...')
for r in self.results[1:]:
delta = {k: v - baseline['scores'][k] for k, v in r['scores'].items()}
print(f'{r["name"]:<30} {r["scores"]["mmlu"]:>8.1f} {r["scores"]["arc"]:>8.1f} ... (Δ MMLU: {delta["mmlu"]:+.1f})')
# 消融实验设计示例
ablation = AblationExperiment(base_config=DEFAULT_CONFIG)
# 实验1: 基线
ablation.run_ablation('baseline', DEFAULT_CONFIG)
# 实验2: 去掉 Gopher 过滤
config_no_gopher = {**DEFAULT_CONFIG, 'use_gopher': False}
ablation.run_ablation('no_gopher', config_no_gopher)
# 实验3: 去掉 MinHash 去重
config_no_dedup = {**DEFAULT_CONFIG, 'use_minhash': False}
ablation.run_ablation('no_dedup', config_no_dedup)
# 实验4: 用 FineWeb-Edu ≥4 替代 ≥3
config_edu4 = {**DEFAULT_CONFIG, 'edu_threshold': 4}
ablation.run_ablation('edu_threshold_4', config_edu4)
# 实验5: 调整配比（增加代码比例）
config_more_code = {**DEFAULT_CONFIG, 'mix': {'cc': 0.4, 'code': 0.3, ...}}
ablation.run_ablation('more_code', config_more_code)
# 对比结果
ablation.compare_to_baseline()10.3.3 典型消融结果
某次真实消融实验结果（1B 模型，100B tokens）：
| 实验 | MMLU | GSM8K | HumanEval | 相对基线 |
|---|---|---|---|---|
| Baseline | 24.5 | 5.2 | 18.3 | - |
| -Gopher | 22.1 | 4.8 | 16.5 | -2.4 MMLU |
| -MinHash | 23.0 | 5.0 | 17.8 | -1.5 MMLU |
| -FineWeb-Edu | 22.8 | 4.9 | 17.0 | -1.7 MMLU |
| +Edu≥4 | 26.1 | 5.8 | 19.5 | +1.6 MMLU |
| +More code | 25.2 | 5.5 | 22.1 | +0.7 MMLU, +3.8 HE |
关键发现：
- Gopher 规则最重要（去掉 -2.4）
- FineWeb-Edu 提升明显（+1.6）
- 配比调整对不同任务影响不同（更多代码提升 HumanEval）
10.4 训练数据归因（TDA）
训练数据归因（Training Data Attribution, TDA）回答一个问题：训练集中的哪些样本对模型在测试样本上的输出有最大贡献？
10.4.1 主流方法对比
| 方法族 | 方法 | 思路 | 可扩展性 | 准确性 |
|---|---|---|---|---|
| 实例级 | Influence Function | 用 Hessian 逆近似 | 差 | 中-高 |
| 实例级 | TracIn | 检查点梯度内积 | 中 | 中 |
| 实例级 | Datamodels | 学习回归模型 | 差 | 高 |
| 实例级 | EK-FAC | Kronecker 分解近似 Hessian | 中-高 | 高 |
| 投影式 | TRAK/LogRA | 梯度投影到低秩空间 | 一般 | 一般 |
| 投影式 | LoRIF | 低秩梯度存储+截断SVD逆Hessian | 好(70B可用) | 一般 |
| 机制级 | MDA | 定位可解释单元→EK-FAC归因→因果验证 | 中 | 高 |
| 任务无关 | Task-Agnostic TDA | 参数空间距离定义影响，无需任务目标 | 中 | 待验证 |
| 审计视角 | Beyond Shapley | 单样本上下文似然移位 | 待验证 | 待验证 |
LoRIF（Low-Rank Influence Functions, arXiv:2601.21929） 首次让梯度 TDA 在 70B 参数模型上实用：利用梯度低秩结构将存储从 O(D) 降至 O(c√D)（存储减少 20×），查询加速 20×，使归因从”3天”缩短到”数小时”。
MDA（Mechanistic Data Attribution, arXiv:2601.21996, ICML 2026 Oral） 将 TDA 从实例级推向机制级：回答”哪条数据塑造了归纳头”而非”哪条数据影响 loss”。反直觉发现——塑造归纳头的核心数据是 XML 标签、Base64 字符串等重复结构数据，而非人类认为的”高质量文本”。
Task-Agnostic TDA（COLM 2026, arXiv:2608.13515） 打破了上表”全部任务相关（task-specific）“的隐性前提：传统方法必须绑定一个具体的下游任务目标（如”对这条 MMLU 题目”），而任务无关影响估计用参数空间距离定义数据影响，无需指定任务。它通过文学→STEM 的时序交叉实验验证有效性——当训练数据从文学切到 STEM 时，参数空间的迁移轨迹能反映各数据点的影响，摆脱了”换一个任务就要重算一次归因”的限制。
Beyond Shapley（arXiv:2607.22766） 提供了跳出 Shapley 分配框架的审计视角：用单样本上下文似然移位审计训练数据影响，即观察”移除/替换单个样本后，模型在该样本上下文上的似然如何移动”。它提醒我们——归因不必绑定 Shapley 值的公理化分配，细粒度的上下文似然审计是另一条互补路径。
10.4.2 Influence Function
import torch
class InfluenceFunction:
"""Influence Function 简化实现"""
def __init__(self, model, train_loader):
self.model = model
self.train_loader = train_loader
def compute_influence(self, test_sample, train_sample):
"""
计算训练样本对测试样本的影响
"""
# Step 1: 计算测试样本的梯度
self.model.zero_grad()
test_loss = self.model.loss(test_sample)
test_grad = torch.autograd.grad(test_loss, self.model.parameters())
test_grad = flatten_grads(test_grad)
# Step 2: 计算训练样本的梯度
self.model.zero_grad()
train_loss = self.model.loss(train_sample)
train_grad = torch.autograd.grad(train_loss, self.model.parameters())
train_grad = flatten_grads(train_grad)
# Step 3: 计算逆 Hessian-向量积
# 简化版：用共轭梯度近似
hvp = self._compute_hvp(test_grad)
inv_hvp = self._conjugate_gradient(hvp, test_grad)
# Step 4: Influence = -train_grad^T * H^-1 * test_grad
influence = -torch.dot(train_grad, inv_hvp)
return influence.item()
def _compute_hvp(self, vector):
"""计算 Hessian-向量积"""
# 用 Pearlmutter 的方法
...
def _conjugate_gradient(self, hvp_fn, b, max_iter=10):
"""共轭梯度法近似 H^-1 * b"""
x = torch.zeros_like(b)
r = b.clone()
p = b.clone()
for _ in range(max_iter):
Ap = hvp_fn(p)
alpha = torch.dot(r, r) / torch.dot(p, Ap)
x += alpha * p
r_new = r - alpha * Ap
beta = torch.dot(r_new, r_new) / torch.dot(r, r)
p = r_new + beta * p
r = r_new
return x10.4.3 TracIn
TracIn（Pruthi et al., 2020）更简单——用训练过程中保存的检查点：
class TracIn:
"""TracIn 简化实现"""
def __init__(self, checkpoints):
"""
checkpoints: 训练过程中的检查点列表
"""
self.checkpoints = checkpoints
def compute_influence(self, test_sample, train_sample):
"""计算影响分数"""
total_influence = 0
for ckpt in self.checkpoints:
model = load_checkpoint(ckpt)
# 计算测试样本梯度
test_loss = model.loss(test_sample)
test_grad = torch.autograd.grad(test_loss, model.parameters())
test_grad = flatten_grads(test_grad)
# 计算训练样本梯度
train_loss = model.loss(train_sample)
train_grad = torch.autograd.grad(train_loss, model.parameters())
train_grad = flatten_grads(train_grad)
# Influence = test_grad · train_grad
influence = torch.dot(test_grad, train_grad)
total_influence += influence.item()
# 用学习率加权
return total_influence / len(self.checkpoints)10.4.4 Datamodels
Datamodels（Ilyas et al., 2022）用回归模型学习数据影响：
import lightgbm as lgb
class Datamodels:
"""Datamodels 实现"""
def __init__(self, num_subsets=10000, subset_ratio=0.5):
"""
num_subsets: 训练多少个子集
subset_ratio: 每个子集保留多少训练数据
"""
self.num_subsets = num_subsets
self.subset_ratio = subset_ratio
self.regressor = None
def collect_data(self, train_samples, eval_samples):
"""
Step 1: 训练多个子集模型，记录每个子集在 eval 上的表现
"""
results = []
for i in range(self.num_subsets):
# 随机采样训练数据子集
indices = np.random.choice(
len(train_samples),
int(len(train_samples) * self.subset_ratio),
replace=False
)
subset = [train_samples[j] for j in indices]
# 训练小模型
model = train_small_model(subset)
# 在 eval 上测试
scores = evaluate(model, eval_samples)
results.append({
'subset_mask': self._to_mask(indices, len(train_samples)),
'scores': scores
})
return results
def train_datamodel(self, collected_data):
"""
Step 2: 训练回归模型
输入: 训练数据子集的 mask (0/1)
输出: eval 分数
"""
X = np.array([d['subset_mask'] for d in collected_data])
y = np.array([d['scores'] for d in collected_data])
self.regressor = lgb.LGBMRegressor(
n_estimators=200,
learning_rate=0.05
)
self.regressor.fit(X, y)
def attribute(self, train_sample_idx, eval_sample_idx):
"""
Step 3: 归因 - 训练样本对 eval 的影响
"""
# 用 datamodel 的特征重要性
importance = self.regressor.feature_importances_[train_sample_idx]
return importance10.4.5 TDA 的应用
- 找有害数据：影响为负的训练样本应该删除
- 找关键数据：影响最大的训练样本应该加倍
- 解释模型行为：模型为什么给出这个答案？因为它被这些训练数据影响了
10.4.6 MDA：机制级训练数据归因
MDA（Mechanistic Data Attribution, ICML 2026 Oral & Spotlight）将 TDA 从”哪条数据影响模型 loss”推向”哪条数据塑造了模型的内部机制”。三阶段框架：
- 定位可解释单元：如定位 Transformer 中的归纳头（Induction Head）
- 高效归因：用 EK-FAC 计算海量数据对特定参数子空间的影响
- 因果验证：通过数据删除/增强验证归因的因果性
反直觉发现：塑造归纳头的核心数据不是人类认为的”高质量文本”，而是 XML 标签、Base64 字符串、LaTeX 源码等重复结构数据。这对数据工程有重要启示：
反思：什么数据真正塑造了模型能力？
- 不要仅用”质量分数”筛选数据——结构重复的数据也有独特价值
- 低质量数据在特定训练阶段可能是必要组件（如 Base64 编码训练 tokenization 鲁棒性）
- 归因结果应反哺数据配比决策——用 MDA 发现的”机制级关键数据”指导混合比例
10.5 Scaling Law 拟合
10.5.1 Chinchilla Scaling Law
经典的 Chinchilla Scaling Law：
Loss(N, D) = E + A/N^α + B/D^β其中：
- N: 参数量
- D: 数据量
- E: 不可约损失（Irreducible Loss）
- A, B: 常数
- α, β: 衰减指数（Chinchilla 论文中 α=β=0.34）
10.5.2 拟合实现
import numpy as np
from scipy.optimize import curve_fit
def chinchilla_law(N_D, E, A, alpha, B, beta):
"""
Chinchilla Scaling Law
N_D: (N, D) tuple
"""
N, D = N_D
return E + A / N**alpha + B / D**beta
def fit_scaling_law(experiments):
"""
用实验数据拟合 Scaling Law
experiments: list of {
'params': N, 'tokens': D, 'loss': L
}
"""
N_data = np.array([e['params'] for e in experiments])
D_data = np.array([e['tokens'] for e in experiments])
L_data = np.array([e['loss'] for e in experiments])
# 拟合
popt, pcov = curve_fit(
chinchilla_law,
(N_data, D_data),
L_data,
p0=[1.5, 100, 0.3, 100, 0.3], # 初始猜测
maxfev=10000
)
E, A, alpha, B, beta = popt
print(f'Fitted parameters:')
print(f' E (irreducible loss): {E:.4f}')
print(f' A: {A:.4f}, α: {alpha:.4f}')
print(f' B: {B:.4f}, β: {beta:.4f}')
return popt
# 示例：用 1B/3B/7B 模型实验拟合
experiments = [
{'params': 1e9, 'tokens': 20e9, 'loss': 2.85},
{'params': 1e9, 'tokens': 100e9, 'loss': 2.65},
{'params': 3e9, 'tokens': 60e9, 'loss': 2.55},
{'params': 3e9, 'tokens': 300e9, 'loss': 2.40},
{'params': 7e9, 'tokens': 140e9, 'loss': 2.30},
{'params': 7e9, 'tokens': 700e9, 'loss': 2.18},
]
params = fit_scaling_law(experiments)10.5.3 最优算力分配
def optimal_allocation(compute_budget, params):
"""
给定算力预算，计算最优 (N, D)
"""
E, A, alpha, B, beta = params
# Chinchilla 推导：最优 N/D 比例
# N* / D* = alpha * B / (beta * A)
optimal_ratio = alpha * B / (beta * A)
# 算力 C = 6 * N * D（近似）
C = compute_budget
N_star = (C * optimal_ratio / 6) ** 0.5
D_star = N_star / optimal_ratio
return N_star, D_star
# 示例：算力 = 1e21 FLOPs
C = 1e21
N_opt, D_opt = optimal_allocation(C, params)
print(f'Optimal: N={N_opt/1e9:.1f}B, D={D_opt/1e9:.1f}B')
# 输出类似：Optimal: N=10.5B, D=210B10.5.4 Scaling Law 外推的踩坑
坑1：数据范围不够
如果实验只在 1B-7B 范围内做，外推到 70B 可能严重偏差。实验范围至少覆盖 2 个数量级。
坑2：忽略了数据质量
经典 Chinchilla 假设数据质量均匀，实际不同质量的数据 Scaling Law 不同。用 Quality-aware Scaling Law 拟合。
坑3：过拟合 Scaling Law 参数
参数过多（5 个）而数据点少（6 个），容易过拟合。增加实验点，至少 10+ 个。
10.6 闭环设计
10.6.1 数据-训练-评估反馈环
┌──────────────────────────────────┐
│ │
▼ │
数据采集/清洗 ──→ 训练 ──→ 评估 ──→ 归因
│ │
│ │
└────────── 反馈调整 ──────────────┘10.6.2 闭环实现
class DataTrainingLoop:
"""数据-训练-评估闭环"""
def __init__(self):
self.iteration = 0
self.history = []
def run_iteration(self):
"""运行一次闭环迭代"""
self.iteration += 1
print(f'\n=== Iteration {self.iteration} ===')
# Step 1: 数据准备
print('Step 1: Preparing data...')
data = self.prepare_data()
# Step 2: 训练小模型
print('Step 2: Training 1B model...')
model = self.train_small_model(data, tokens=100e9)
# Step 3: 评估
print('Step 3: Evaluating...')
scores = self.evaluate(model)
# Step 4: 归因分析
print('Step 4: Attribution analysis...')
attribution = self.attribute(model, data)
# Step 5: 调整数据策略
print('Step 5: Adjusting data strategy...')
adjustments = self.suggest_adjustments(scores, attribution)
# Step 6: 应用调整
print('Step 6: Applying adjustments...')
self.apply_adjustments(adjustments)
# 记录历史
self.history.append({
'iteration': self.iteration,
'scores': scores,
'adjustments': adjustments
})
return scores
def suggest_adjustments(self, scores, attribution):
"""基于评估和归因，建议调整"""
adjustments = {}
# 如果数学能力低
if scores['gsm8k'] < 10:
# 找影响数学的关键数据，加倍权重
math_influential = find_influential_data(attribution, domain='math')
adjustments['increase_math_weight'] = math_influential
# 如果代码能力低
if scores['human_eval'] < 20:
code_influential = find_influential_data(attribution, domain='code')
adjustments['increase_code_weight'] = code_influential
# 如果有有害数据
harmful_data = find_harmful_data(attribution)
if harmful_data:
adjustments['remove_harmful'] = harmful_data
return adjustments10.6.3 闭环的工业实践
某团队的真实闭环迭代：
| 迭代 | MMLU | GSM8K | 调整 |
|---|---|---|---|
| 1 | 24.5 | 5.2 | 基线 |
| 2 | 26.1 | 8.3 | 增加 5% 数学数据 |
| 3 | 27.8 | 12.5 | 移除 0.5% 有害数据 |
| 4 | 28.5 | 15.2 | 退火数据加 10% 合成推理 |
| 5 | 29.1 | 18.7 | 调整配比，代码 +5% |
每次迭代 1-2 周，5 次迭代后 MMLU 提升 4.6，GSM8K 提升 13.5。
闭环的前瞻延伸：从”事后归因”到”训练前引导”（Interpretability-Guided Training）：上述闭环的归因环节是事后的——先训练、再评估、最后归因。Bergen et al.（2026, learnmechinterp.com）系统梳理的”可解释性引导训练”则把顺序反过来——先理解模型机制，再反哺训练数据决策：先定位模型需要什么样的内部机制（如需要更强的归纳头），再用可解释性工具筛选/合成能塑造该机制的数据，最后才投入训练。这把 TDA 从”诊断工具”前移为”处方工具”，是数据工程闭环的下一阶段演进方向。
10.7 评估的踩坑经验
坑1：评测集污染未察觉
某团队每次迭代 MMLU 都在涨，但上真实环境后崩盘。后发现训练数据里混入了 MMLU 题目的解析文章。每次迭代都要重新做去污染检查。
坑2：用同一份 eval 调参
某团队用 MMLU 调参，结果在 MMLU 上”过拟合”——但 ARC、HellaSwag 等其他 benchmark 不涨。至少用 5 个不同维度的 benchmark。
坑3：消融实验不够
某团队只跑了基线实验，没做消融，结果不知道哪个改进有效。每次迭代至少做 5 个消融实验。
坑4：归因太慢
某团队用 Influence Function 做归因，单次迭代要 3 天。用 TracIn 或 Datamodels，能在几小时内完成。
坑5：闭环太慢
某团队每次迭代要 1 个月，结果一年只迭代 12 次，跟不上数据变化。用 1B 小模型做快速迭代，每月能跑 4-8 次。
10.8 工业级评估流水线
class EvaluationPipeline:
"""工业级评估流水线"""
def __init__(self):
self.probes = DataProbe()
self.ablation = AblationExperiment()
self.tda = Datamodels()
self.scaling_law = None
def evaluate_dataset(self, dataset):
"""评估数据集质量"""
# 1. 数据探针
probe_results = self.probes.run_all_probes(dataset)
# 2. 生成数据卡片
datasheet = self.generate_datasheet(dataset, probe_results)
return {
'probes': probe_results,
'datasheet': datasheet
}
def evaluate_model(self, model, train_data):
"""评估模型 + 归因"""
# 1. Benchmark 评估
scores = self.run_benchmarks(model)
# 2. 训练数据归因
attribution = self.tda.attribute(model, train_data)
# 3. Scaling Law 拟合
if len(self.history) >= 5:
self.scaling_law = fit_scaling_law(self.history)
return {
'scores': scores,
'attribution': attribution,
'scaling_law': self.scaling_law
}
def run_benchmarks(self, model):
"""运行所有 benchmark"""
return {
'mmlu': eval_mmlu(model),
'arc': eval_arc(model),
'gsm8k': eval_gsm8k(model),
'human_eval': eval_human_eval(model),
'hellaswag': eval_hellaswag(model),
'winogrande': eval_winogrande(model),
'piqa': eval_piqa(model),
'openbookqa': eval_openbookqa(model),
'triviaqa': eval_triviaqa(model),
'natural_questions': eval_nq(model),
'mt_bench': eval_mt_bench(model),
'arena': eval_arena(model)
}10.9 本课总结
- 数据卡片：标准化描述数据集，回答 What/Why/How/Evaluate/Use
- 数据探针：8 大维度统计分析（PPL/毒性/领域/长度/token/语言/质量/污染）
- 消融实验：用 1B 小模型做单变量对照，找最有价值的改进
- 训练数据归因：Influence Function/TracIn/Datamodels，找出对模型影响最大的训练样本
- Scaling Law 拟合：Loss(N,D)=E+A/N^α+B/D^β，外推最优算力分配
- 闭环设计：数据→训练→评估→归因→调整，每月迭代 4-8 次
课后练习
- 为你的数据集生成一份完整的数据卡片
- 实现 8 大维度的数据探针，分析你的数据集
- 设计 5 个消融实验，找出最有效的改进
- 用 1B 模型做 6 个不同 (N, D) 的实验，拟合 Scaling Law
第二部分（预训练数据工程）到此结束。接下来第三部分将进入后训练数据工程——SFT、偏好数据、RL/RLVR、合成数据等更”精炼”的数据工程。