预训练评估与可归因分析

没有评估的数据工程是盲修。

数据-训练-评估的闭环，是工业级团队的分水岭。

---

11.1 数据卡片（Datasheet）

11.1.1 什么是数据卡片

数据卡片（Datasheet for Datasets, Gebru et al., 2021）是描述数据集的标准化文档，回答以下问题：

• 数据是什么：来源、规模、构成
• 为什么创建：动机、用途
• 如何创建：采集、清洗、过滤流程
• 如何评估：质量指标、已知局限
• 如何使用：推荐用途、禁忌场景

11.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:
    - '医疗诊断（未做医疗领域精筛）'
    - '法律咨询（未做法律领域精筛）'
    - '高风险决策（如信贷、招聘）'

11.1.3 数据卡片的价值

1. 可追溯：知道数据从哪来、怎么处理
2. 可复现：别人能复现你的数据集
3. 合规证明：满足法规对数据透明度的要求
4. 风险沟通：明确告诉下游使用者数据的局限

11.2 数据探针（Data Probe）

数据探针是对数据集做多维度统计分析，发现数据问题。

11.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 contamination

11.2.2 探针结果的解读

指标	正常范围	异常	原因
平均 PPL	2-3	>5	数据未清洗
毒性率	<0.5%	>2%	安全过滤失效
领域分布	多样	单一领域占 80%+	数据偏置
长度中位数	500-2000 字符	<200	短文本过多
稀有 token	<10%	>30%	词表设计问题
评测污染率	<0.5%	>5%	去污染失败

11.3 消融实验

11.3.1 为什么用 1B 小模型

消融实验用 1B 小模型的原因：

1. 算力成本低：1B 模型训练成本是 70B 的 1/100
2. 趋势外推有效：1B 上的相对趋势能外推到 70B
3. 迭代速度快：1B 训练 1 天能跑完，70B 要 1 个月

11.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()

11.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

关键发现：

1. Gopher 规则最重要（去掉 -2.4）
2. FineWeb-Edu 提升明显（+1.6）
3. 配比调整对不同任务影响不同（更多代码提升 HumanEval）

11.4 训练数据归因（TDA）

训练数据归因（Training Data Attribution, TDA）回答一个问题：训练集中的哪些样本对模型在测试样本上的输出有最大贡献？

11.4.1 主流方法对比

方法族	方法	思路	存储开销	可扩展性	准确性
实例级	Influence Function	用 Hessian 逆近似	高（需存全量训练梯度）	差	中-高
实例级	TracIn	检查点梯度内积	中（存检查点梯度）	中	中
实例级	Datamodels	学习回归模型	极高（需存数千子集模型）	差	高
实例级	EK-FAC	Kronecker 分解近似 Hessian	中-高（Kronecker 因子，比全梯度小但仍随模型增长）	中-高	高
投影式	TRAK/LogRA	梯度投影到低秩空间	中（投影后梯度）	一般	一般
投影式	LoRIF	低秩梯度存储+截断SVD逆Hessian	低（20× 压缩）	好(70B可用)	一般
投影式	EOGP	EK-FAC 降维 → 子空间 PCA 学压缩方向 → 1-bit 量化	极低（1/16）	好（OLMo 2 SFT 1B–32B 实测）	高（GPT-2 上更准地预测重训结果）
机制级	MDA	定位可解释单元→EK-FAC归因→因果验证	中	中	高
元微分	MAGIC	对整段训练过程求导，得到”精确影响”	不依赖梯度存储（成本在训练次数）	差（每条预测一次完整训练）	高（精确影响）
元微分	MAGE / SPELL	预算化归因＝用少量测量恢复大影响矩阵	低（跑在既有元梯度设施上，零额外成本）	中-高	高
任务无关	Task-Agnostic TDA	参数空间距离定义影响，无需任务目标	低（可从中间 checkpoint 估计，无需重训）	中	中（COLM 2026 已发表）
审计视角	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 时，参数空间的迁移轨迹能反映各数据点的影响，摆脱了”换一个任务就要重算一次归因”的限制。更具体地：该方法以一个样本的梯度更新在多大程度上缩短”与最终参数的平方距离”来定义其影响力，因此可从中间 checkpoint 估计而无需重训；在 Pythia / PolyPythia 的 18 个配置上，它发现影响力数据存在系统性的时序变化——训练早期文学类数据与”通往最终参数的轨迹”更对齐，后期转为 STEM 类。

Beyond Shapley（arXiv:2607.22766，预印本） 提供了跳出 Shapley 分配框架的审计视角：用单样本上下文似然移位审计训练数据影响，即观察”移除/替换单个样本后，模型在该样本上下文上的似然如何移动”。它提醒我们——归因不必绑定 Shapley 值的公理化分配，细粒度的上下文似然审计是另一条互补路径。

EOGP（arXiv:2609.37842，2026-09-29）：把存储这根轴从 20× 推到 1/16。 影响函数要反复用，重用已存的训练梯度可以省算力，但 LLM 尺度上存全量梯度贵到不可行。EOGP 研究的是”在查询未知的情况下如何压缩梯度而保住影响估计”：先做最坏情况分析刻画最优的固定维度线性表示，再给出 EK-FAC 降维 → 在保留子空间内做 PCA 学压缩方向 → 1-bit 量化（量化省下的预算可以多保留坐标）。实测：GPT-2 上以基线 1/16 的每样本存储更准地预测重训结果；OLMo 2 SFT 1B → 32B 上，与”每样本存储多 100 倍以上”的基线竞争力相当。

工程结论：“能不能在 LLM 上做影响函数”这个问题的瓶颈，已经从”算不算得动”变成”存不存得下”——而存储这一根轴近两年的压缩幅度（LoRIF 20× → EOGP 1/16）远快于算法本身的改进。选型时先看瓶颈在哪一侧。

元微分（metagradient）一族：本章此前缺的那一格。 上表”实例级”四法（Influence Function / TracIn / Datamodels / EK-FAC）都用 Hessian 或梯度做近似；而 MAGIC（Ilyas & Engstrom, 2025）这一族直接对整段训练过程求导，得到的是”精确影响”而非近似——代价是每条预测都要单独跑一次完整训练。2026-09 的 MAGE / SPELL（arXiv:2609.15044，作者含 Andrew Ilyas）把它重构成 budgeted attribution：用少量测量恢复一个大影响矩阵（矩阵补全视角），并指出一个反直觉点——最适合”重建矩阵”的测量，与最适合”归因”的测量不是同一批；据此给出两个算法：MAGE（面向重建）与 SPELL（面向归因），两者都跑在既有元梯度设施上、零额外成本，跨训练规模与测量预算均强于现有基线。

顺便给本章补一根新的评估轴：“给定预算能归因多少” 比”可扩展性 好/中/差”更贴近工程决策。§11.8 的流水线里 TDA 是每轮都跑的环节，读者真正需要知道的是”我这轮有多少预算、能覆盖多少 query”，而不是一个三档定性词。

⚠️ 归因选出样本 ≠ 干预就能生效：注意”重加权”的局限

上面整节（以及 11.6 的闭环）都隐含一条链路：归因 → 找出高影响样本 → 通过重加权/删除改变模型行为。2026-09 的一项工作给这条链路加了一道边界条件（arXiv:2609.02771，2026-09-02，四个开源 LLM 上验证）：影响函数选出的高影响样本，在常规重加权/删除干预下往往不比随机采样强。

这不意味着归因失效，而是干预方式的问题：

• 重加权（reweighting）：对同一批 IF 选出的样本做升/降权，行为改变弱且不一致；
• 重写（rewriting）：保持 instruction 不变，把 response 替换为”行为对齐”或”行为反向”的监督信号，改变更强、更持久、且双向可控；在 safety refusal 上同样复现。

实践含义：如果你的目标是改掉某个行为，优先改写高影响样本的监督信号，而不是给它们调权重。这条信息是保住归因链路的关键——否则实践中一旦复现出”删了没用”，很容易误判为 TDA 整体不可用而放弃。它也提示 TDA 的评测应当是 intervention-aware 的：只报”选得准不准”不足以判断一个归因方法有没有实用价值。（该工作目前为预印本，按预印本证据对待。）

11.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 x

11.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)

11.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 importance

11.4.5 TDA 的应用

1. 找有害数据：影响为负的训练样本应该删除
2. 找关键数据：影响最大的训练样本应该加倍
3. 解释模型行为：模型为什么给出这个答案？因为它被这些训练数据影响了

11.4.6 MDA：机制级训练数据归因

MDA（Mechanistic Data Attribution, ICML 2026 Oral & Spotlight）将 TDA 从”哪条数据影响模型 loss”推向”哪条数据塑造了模型的内部机制”。三阶段框架：

1. 定位可解释单元：如定位 Transformer 中的归纳头（Induction Head）
2. 高效归因：用 EK-FAC 计算海量数据对特定参数子空间的影响
3. 因果验证：通过数据删除/增强验证归因的因果性

反直觉发现：塑造归纳头的核心数据不是人类认为的”高质量文本”，而是 XML 标签、Base64 字符串、LaTeX 源码等重复结构数据。这对数据工程有重要启示：

反思：什么数据真正塑造了模型能力？

• 不要仅用”质量分数”筛选数据——结构重复的数据也有独特价值
• 低质量数据在特定训练阶段可能是必要组件（如 Base64 编码训练 tokenization 鲁棒性）
• 归因结果应反哺数据配比决策——用 MDA 发现的”机制级关键数据”指导混合比例

11.5 Scaling Law 拟合

11.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、β = 0.28）
• 原文拟合常数：E = 1.69、A = 406.4、B = 410.7

⚠️ α ≠ β 不是小数点级别的差异，它决定了后面所有的分配结论。 请务必读完这一条再往下走：

1. 原文明写 α = 0.34、β = 0.28（不是两者都等于 0.34）。α > β 的物理含义是：参数项的损失衰减快于数据项，因此固定算力下应该把更多预算投到数据侧。
2. 由此推出的最优分配指数是 a = β/(α+β) ≈ 0.45（参数）与 b = α/(α+β) ≈ 0.55（数据）——两者并不相等。（原文 Table 2 中 Approach 3 报告的经验拟合值为 a = 0.46、b = 0.54，与上式推导值同量级但不完全相等，引用时以原文为准。）
3. 广为流传的”每参数 20 个 token”是 a ≈ b ≈ 0.5 取整之后的结论。这个取整是 load-bearing 的：只有 a = b 时 D*/N* 才是常数；按原始指数，最优 token/参数比会随算力缓慢爬升。
4. 对本章的直接后果：最优分配必须走原文式 (4) 的闭式（见 §11.5.3），它同时依赖 α 与 β 以及两者之比。若误按 α = β 代入，拉格朗日条件 αA/N^α = βB/D^β 退化为 A/N^α = B/D^α，指数项相互抵消、D*/N* 变成与算力无关的常数 (B/A)^{1/α}，算出的 N*/D* 是系统性偏离的。此前版本的 N=10.5B, D=210B（正好 20×）看起来像”算出来的”，实际复现的是取整结论，且当时那行代码用的 optimal_ratio = α·B/(β·A) 与它所声称的拉格朗日条件并不一致（A/B 写反、且漏了 1/(α+β) 次幂）。

11.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,
        # 初始猜测：用 Chinchilla 原文拟合值作锚点（E, A, alpha, B, beta）
        # 章节原为 [1.5, 100, 0.3, 100, 0.3]，把两个指数都初始化成 0.3；
        # 官方值见 §11.5.1，用自己的数据重拟合时可作为初值与合理性检查的锚点
        p0=[1.69, 406.4, 0.34, 410.7, 0.28],
        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)

11.5.3 最优算力分配

def optimal_allocation(compute_budget, params):
    """
    给定算力预算，计算最优 (N, D)
    """
    E, A, alpha, B, beta = params

    # Chinchilla 原文式 (4) 的闭式最优前沿（Hoffmann et al., arXiv:2203.15556）
    #   在约束 FLOPs(N, D) ≈ 6·N·D 下最小化 L̂(N, D) = E + A/N^α + B/D^β，
    #   等价的拉格朗日条件是  alpha*A / N^alpha == beta*B / D^beta，
    #   解出的闭式为：
    #     G     = (alpha * A / (beta * B)) ** (1 / (alpha + beta))
    #     N*(C) = G       * (C/6) ** a,   a = beta  / (alpha + beta)
    #     D*(C) = G ** -1 * (C/6) ** b,   b = alpha / (alpha + beta)
    # ⚠️ 注意三处容易写错的地方：① 分子分母是 alpha*A / beta*B（不是 alpha*B / beta*A）；
    #    ② 外面必须套 1/(alpha+beta) 次幂，不能直接用比值；
    #    ③ D*/N* 不是常数，它等于 G**-2 * (C/6)**(b-a)，随算力缓慢爬升。
    G = (alpha * A / (beta * B)) ** (1.0 / (alpha + beta))
    a = beta / (alpha + beta)
    b = alpha / (alpha + beta)

    # 算力 C = 6 * N * D（近似）
    u = compute_budget / 6
    N_star = G * u ** a
    D_star = (G ** -1) * u ** b

    return N_star, D_star

# 示例：算力 = 1e21 FLOPs（参数用 Chinchilla 原文拟合值）
params = (1.69, 406.4, 0.34, 410.7, 0.28)
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=1.8B, D=91.4B（D/N ≈ 50，且 6·N·D ≈ C 成立）
# ⚠️ 广为流传的"每参数 20 个 token"并不等于本节闭式解在这个算力上的输出：
#    按原文 α=0.34 / β=0.28，D*/N* 从 C=1e19 的约 32 爬到 C=1e25 的约 122
#    （1e21 处约 50）。"20×"是 a ≈ b ≈ 0.5 取整之后才成立的常数比，
#    取整本身是 load-bearing 的（详见 §11.5.1 第 3、4 条）。

11.5.4 Scaling Law 外推的踩坑

坑1：数据范围不够

如果实验只在 1B-7B 范围内做，外推到 70B 可能严重偏差。实验范围至少覆盖 2 个数量级。

坑2：忽略了数据质量

经典 Chinchilla 假设数据质量均匀，实际不同质量的数据 Scaling Law 不同。用 Quality-aware Scaling Law 拟合。

坑3：过拟合 Scaling Law 参数

参数过多（5 个）而数据点少（6 个），容易过拟合。增加实验点，至少 10+ 个。

11.6 闭环设计

11.6.1 数据-训练-评估反馈环

       ┌──────────────────────────────────┐
       │                                  │
       ▼                                  │
  数据采集/清洗 ──→ 训练 ──→ 评估 ──→ 归因
       │                                  │
       │                                  │
       └────────── 反馈调整 ──────────────┘

11.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 adjustments

11.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。

上表的”调整”一列默认采用配比调整 / 上采样 / 剔除这类权重与集合层面的动作。若某一轮的目标是改变特定行为（而非提升某个 benchmark 分数），参考 §11.4 的边界条件：改写高影响样本的监督信号通常比调整权重更有效——重加权对同一批高影响样本产生的行为改变弱且不一致。

闭环的前瞻延伸：从”事后归因”到”训练前引导”（Interpretability-Guided Training）：上述闭环的归因环节是事后的——先训练、再评估、最后归因。Bergen et al.（2026, learnmechinterp.com）系统梳理的”可解释性引导训练”则把顺序反过来——先理解模型机制，再反哺训练数据决策：先定位模型需要什么样的内部机制（如需要更强的归纳头），再用可解释性工具筛选/合成能塑造该机制的数据，最后才投入训练。这把 TDA 从”诊断工具”前移为”处方工具”，是数据工程闭环的下一阶段演进方向。

11.7 评估的踩坑经验

坑1：评测集污染未察觉

某团队每次迭代 MMLU 都在涨，但上真实环境后崩盘。后发现训练数据里混入了 MMLU 题目的解析文章。每次迭代都要重新做去污染检查。

坑2：用同一份 eval 调参

某团队用 MMLU 调参，结果在 MMLU 上”过拟合”——但 ARC、HellaSwag 等其他 benchmark 不涨。至少用 5 个不同维度的 benchmark。

坑3：消融实验不够

某团队只跑了基线实验，没做消融，结果不知道哪个改进有效。每次迭代至少做 5 个消融实验。

坑4：归因太慢

某团队用 Influence Function 做归因，单次迭代要 3 天。用 TracIn 或 Datamodels，能在几小时内完成。

但换方法不是唯一、也不是最好的一条路。 先判断瓶颈在哪一侧：

• 瓶颈在计算（Hessian 逆算不动）→ 换 TracIn / Datamodels 这类近似，或走 TRAK / LogRA 投影，是合理的。
• 瓶颈在梯度存储（存不下全量梯度）→ 应走压缩路线，而不是换方法：LoRIF 存储减少 20×、查询加速 20×，EOGP 进一步到 1/16 每样本存储且精度不降反升（§11.4.1）。换方法会牺牲”影响函数”这一族的理论性质（如反事实可解释性），压缩不会。

判据一句话：“算不动”换方法，“存不下”做压缩。

坑5：闭环太慢

某团队每次迭代要 1 个月，结果一年只迭代 12 次，跟不上数据变化。用 1B 小模型做快速迭代，每月能跑 4-8 次。

多模态特化：VLM 的归因实验怎么做

本章的归因框架（数据探针、消融实验、TDA）在 VLM 上依然成立，但归因的对象变了：文本侧问”哪批语料贡献了这个能力”，多模态侧还要再问一句”这个能力来自哪个模态的哪批数据”。

第一层：模态消融。只看整体指标无法定位问题，最低限度做三组对照——去掉图文交错数据（只留图文对）、补入 OCR/文档数据、只剩纯文本。Cambrian-1 做的就是这类系统性消融（在 20 多个视觉编码器上跑同一套指令微调流程当”评估协议”），它给出两条彼此独立的发现：一是组合多个视觉编码器协同优于任何单一编码器，尤其在以视觉为中心的任务上；二是数据整编比数据量重要——精心配平的 7M 样本胜过更大的 10M 未整编池。两者都是从消融里拿到的，盯 loss 曲线永远拿不到。

第二层：分能力建 probe 集。VLM 的能力极不均衡：OCR 与文档理解、图表推理、GUI 定位、多图对比、空间关系，各自的数据来源几乎不重叠，混在一个总分里会互相掩盖。做法是给每个能力配一个小规模 probe（几十到几百条即可），再用 MMMU、POPE、HallBench 一类公开集做交叉验证。这里直接复用 11.2 的探针思路，只是维度从”语言 / 领域 / 难度”换成了”能力 × 模态”。

第三层：小规模 ablations，而非全量训练。多模态预训练的成本比文本高一个量级（见第 9 章视觉 token 的开销），全量重训做对照不现实。可行路径是固定总 token 预算、只换数据配比的短程消融，用 1B-3B 规模或截断训练步数做代理。11.3”为什么用 1B 小模型”的论证在这里更强——多模态下大模型消融的代价高到不可接受。

必须加一条监控线：视觉能力退化往往不伴随文本能力下降。这是模态消融最容易骗人的地方：文本基准稳如泰山，你以为配方没问题，视觉侧其实已经悄悄崩了；反过来，大量灌入多模态数据也可能侵蚀纯文本能力。因此每次配方变更都要同时报文本基准与视觉基准，并观察两者的差值而非绝对值。缓解手段上，往配比里混入纯文本回放数据（replay）是目前最便宜的一招，其原理见第 15 章灾难性遗忘。

（详见第 26 章多模态数据工程）

11.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)
        }

11.9 本课总结

• 数据卡片：标准化描述数据集，回答 What/Why/How/Evaluate/Use
• 数据探针：8 大维度统计分析（PPL/毒性/领域/长度/token/语言/质量/污染）
• 消融实验：用 1B 小模型做单变量对照，找最有价值的改进
• 训练数据归因：Influence Function/TracIn/Datamodels，找出对模型影响最大的训练样本（ 归因选出样本 ≠ 干预就能生效：改行为时优先改写高影响样本的监督信号，而非调权重；arXiv:2609.02771，预印本）
• Scaling Law 拟合：Loss(N,D)=E+A/N^α+B/D^β，外推最优算力分配（ Chinchilla 原文 α = 0.34、β = 0.28，E = 1.69、A = 406.4、B = 410.7；两者不相等，最优分配依赖这个差异，详见 §11.5.1）
• 闭环设计：数据→训练→评估→归因→调整，每月迭代 4-8 次

课后练习

1. 为你的数据集生成一份完整的数据卡片
2. 实现 8 大维度的数据探针，分析你的数据集
3. 设计 5 个消融实验，找出最有效的改进
4. 用 1B 模型做 6 个不同 (N, D) 的实验，拟合 Scaling Law

---

第二部分（预训练数据工程）到此结束。接下来第三部分将进入后训练数据工程——SFT、偏好数据、RL/RLVR、合成数据等更”精炼”的数据工程。