案例 8：批量归一化、超参数搜索与训练故障诊断#
学习目标与记号#
当模型表现不理想时，原因可能来自数据准备、程序实现、参数更新过程，也可能是模型无法很好地应用到新样本。本案例使用 UCI Dry Bean 数据集，建立能够保存训练记录、在异常时停止并可重复运行的训练流程，重点考察批量归一化（Batch Normalization，BN）、对数尺度随机搜索、逐步缩减与训练故障诊断。
本案例主要讨论以下内容：
区分 BN 在训练阶段使用当前批次统计量、在推断阶段使用训练过程中保存的运行统计量；
区分模型参数与超参数，并理解在线性尺度和对数尺度上搜索的差异；
实现不读取测试集的随机搜索与简化的逐步缩减；
通过小数据过拟合测试、有效数值检查、梯度范数、检查点和训练记录查找错误；
同时报告任务表现、训练时间、参数量和模型大小，并说明快速模式的使用范围。
记一个小批量的样本数为 \(B\)，隐藏层特征数为 \(d\)，隐藏层的线性运算结果为 \(\boldsymbol{Z}\in\mathbb{R}^{B\times d}\)。模型与训练方案确定后，测试集只用于一次最终评价。
数据来源与运行范围#
数据集：UCI Dry Bean；
说明页：https://archive.ics.uci.edu/dataset/602/dry；
匿名 HTTPS 下载地址：https://archive.ics.uci.edu/static/public/602/dry+bean+dataset.zip；
许可：CC BY 4.0；
数据规模：13,611 个样本、16 个数值特征和 7 个类别。
本案例需要比较多组超参数，因此完整运行时间较长。快速模式默认使用 6,000 个训练样本、4 组随机候选和两阶段的较短训练轮数；设置环境变量 AI_COURSE_FAST_MODE=0 可以进行较完整的搜索。所有数据均写入 AI_COURSE_DATA_DIR 指定的目录或用户缓存目录，不写入课程源码目录。
from pathlib import Path
import hashlib
import io
import json
import os
import shutil
import tempfile
import time
import urllib.request
import zipfile
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import torch
from torch import nn
from torch.utils.data import DataLoader, TensorDataset
SEED = 42
FAST_MODE = os.environ.get('AI_COURSE_FAST_MODE', '1') != '0'
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(SEED)
DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
CACHE_ROOT = Path(os.environ.get('AI_COURSE_DATA_DIR', Path.home() / '.cache' / 'ai-course-cases')).expanduser()
CACHE_ROOT.mkdir(parents=True, exist_ok=True)
def download(url, filename):
destination = CACHE_ROOT / filename
if not destination.exists():
request = urllib.request.Request(url, headers={'User-Agent': 'ai-course-case/1.0'})
with urllib.request.urlopen(request, timeout=180) as response:
with tempfile.NamedTemporaryFile(dir=CACHE_ROOT, delete=False) as tmp:
shutil.copyfileobj(response, tmp)
temporary = Path(tmp.name)
temporary.replace(destination)
digest = hashlib.sha256(destination.read_bytes()).hexdigest()
print(f'{destination.name}：{destination.stat().st_size / 1024**2:.2f} MiB；SHA-256：{digest}')
return destination
print({'计算设备': str(DEVICE), '快速模式': FAST_MODE, '缓存目录': str(CACHE_ROOT)})
{'计算设备': 'cpu', '快速模式': True, '缓存目录': '/private/tmp/ai-course-case-data'}
第 1 步：读取并检查原始数据#
ARFF 文件同时保存属性说明和数据。读取后，程序将类别标签转换为字符串，将数值特征转换为统一的数据类型，并检查是否存在无穷大或无效数值。查找训练问题应从检查数据开始：如果类别集合、数据维度或数值范围不符合预期，应在建立模型之前停止，而不要等到损失函数出现无效数值后再查找原因。
from scipy.io import arff
URL = 'https://archive.ics.uci.edu/static/public/602/dry+bean+dataset.zip'
archive = download(URL, 'uci_dry_bean.zip')
with zipfile.ZipFile(archive) as zf:
members = [name for name in zf.namelist() if name.lower().endswith('.arff')]
if len(members) != 1:
raise RuntimeError(f'未唯一找到 ARFF：{members}')
text = zf.read(members[0]).decode('utf-8', errors='replace')
records, metadata = arff.loadarff(io.StringIO(text))
raw = pd.DataFrame(records)
raw['Class'] = raw['Class'].str.decode('utf-8')
feature_names = [column for column in raw.columns if column != 'Class']
raw[feature_names] = raw[feature_names].astype(np.float64)
assert raw.shape == (13611, 17)
assert raw['Class'].nunique() == 7
assert np.isfinite(raw[feature_names].to_numpy()).all()
class_display_names = {'BARBUNYA': '巴尔布尼亚豆', 'BOMBAY': '孟买豆', 'CALI': '卡利豆', 'DERMASON': '德尔马松豆', 'HOROZ': '霍罗兹豆', 'SEKER': '塞克尔豆', 'SIRA': '西拉豆'}
class_counts_display = raw['Class'].value_counts().rename(index=class_display_names).to_dict()
print({'原始数据维度': raw.shape, '各类别样本数': class_counts_display})
uci_dry_bean.zip：4.52 MiB；SHA-256：0a64eff5be87f48c3dbbfc0a12a56c5d5b5167ef8e61cd45d69b3e7c7130c06f
{'原始数据维度': (13611, 17), '各类别样本数': {'德尔马松豆': 3546, '西拉豆': 2636, '塞克尔豆': 2027, '霍罗兹豆': 1928, '卡利豆': 1630, '巴尔布尼亚豆': 1322, '孟买豆': 522}}
第 2 步：训练前检查数据#
训练前至少应确认：输入数值范围是否合理、类别编号是否连续、各类别样本数是否相差过大、数据中是否存在重复或缺失，以及准备使用什么评价指标。本数据集中各类别样本数存在差异，但不属于极端不平衡，因此主要使用宏平均 F1（macro-averaged F1）评价各类别的综合表现，并用交叉熵观察概率预测的训练过程。
不同特征的数值范围相差较大。即使网络包含 BN，对输入进行标准化仍有助于改善第一层的计算尺度；输入标准化和 BN 作用在不同位置，不能把 BN 当作修复所有数据问题的方法。
feature_display_names = {'Area': '面积', 'Perimeter': '周长', 'MajorAxisLength': '长轴长度', 'MinorAxisLength': '短轴长度', 'AspectRation': '纵横比', 'Eccentricity': '离心率', 'ConvexArea': '凸包面积', 'EquivDiameter': '等效直径', 'Extent': '范围占比', 'Solidity': '实心度', 'roundness': '圆度', 'Compactness': '紧致度', 'ShapeFactor1': '形状因子 1', 'ShapeFactor2': '形状因子 2', 'ShapeFactor3': '形状因子 3', 'ShapeFactor4': '形状因子 4'}
summary_display = raw[feature_names].describe().T[['mean', 'std', 'min', 'max']].rename(index=feature_display_names, columns={'mean': '均值', 'std': '标准差', 'min': '最小值', 'max': '最大值'}).rename_axis('特征')
display(summary_display)
counts = raw['Class'].value_counts().sort_index()
counts.rename(index=class_display_names).plot.bar(title='类别数量'); plt.xlabel('类别'); plt.ylabel('样本数'); plt.show()
print({'缺失值数': int(raw.isna().sum().sum()), '重复行数': int(raw.duplicated().sum()), '最大/最小类别样本数之比': float(counts.max() / counts.min())})
| 均值 | 标准差 | 最小值 | 最大值 | |
|---|---|---|---|---|
| 特征 | ||||
| 面积 | 53048.284549 | 29324.095717 | 20420.000000 | 254616.000000 |
| 周长 | 855.283459 | 214.289696 | 524.736000 | 1985.370000 |
| 长轴长度 | 320.141867 | 85.694186 | 183.601165 | 738.860153 |
| 短轴长度 | 202.270714 | 44.970091 | 122.512653 | 460.198497 |
| 纵横比 | 1.583242 | 0.246678 | 1.024868 | 2.430306 |
| 离心率 | 0.750895 | 0.092002 | 0.218951 | 0.911423 |
| 凸包面积 | 53768.200206 | 29774.915817 | 20684.000000 | 263261.000000 |
| 等效直径 | 253.064220 | 59.177120 | 161.243764 | 569.374358 |
| 范围占比 | 0.749733 | 0.049086 | 0.555315 | 0.866195 |
| 实心度 | 0.987143 | 0.004660 | 0.919246 | 0.994677 |
| 圆度 | 0.873282 | 0.059520 | 0.489618 | 0.990685 |
| 紧致度 | 0.799864 | 0.061713 | 0.640577 | 0.987303 |
| 形状因子 1 | 0.006564 | 0.001128 | 0.002778 | 0.010451 |
| 形状因子 2 | 0.001716 | 0.000596 | 0.000564 | 0.003665 |
| 形状因子 3 | 0.643590 | 0.098996 | 0.410339 | 0.974767 |
| 形状因子 4 | 0.995063 | 0.004366 | 0.947687 | 0.999733 |
{'缺失值数': 0, '重复行数': 68, '最大/最小类别样本数之比': 6.793103448275862}
第 3 步：划分数据、标准化训练集并检查张量维度#
先按类别比例划分训练集、验证集和测试集，再只根据训练集计算 StandardScaler 所需的均值和标准差。快速模式还会从训练集中按类别比例保留 6,000 个样本；验证集和测试集保持独立。输入矩阵和观测标签向量的维度分别为 \(n\times16\) 和 \(n\)。CrossEntropyLoss 要求观测标签使用整数类别编号，而不是独热编码（one-hot encoding）的浮点矩阵。
测试集对象创建后，不会作为参数传给超参数搜索函数，从程序接口上减少误用测试集的可能。
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
classes = sorted(raw['Class'].unique())
class_to_id = {name: index for index, name in enumerate(classes)}
X = raw[feature_names].to_numpy(np.float32)
y = raw['Class'].map(class_to_id).to_numpy(np.int64)
X_train, X_hold, y_train, y_hold = train_test_split(X, y, test_size=0.30, stratify=y, random_state=SEED)
X_validation, X_test, y_validation, y_test = train_test_split(X_hold, y_hold, test_size=0.50, stratify=y_hold, random_state=SEED)
if FAST_MODE and len(X_train) > 6000:
X_train, _, y_train, _ = train_test_split(X_train, y_train, train_size=6000, stratify=y_train, random_state=SEED)
scaler = StandardScaler().fit(X_train)
X_train_z = scaler.transform(X_train).astype(np.float32)
X_validation_z = scaler.transform(X_validation).astype(np.float32)
X_test_z = scaler.transform(X_test).astype(np.float32)
train_data = TensorDataset(torch.from_numpy(X_train_z), torch.from_numpy(y_train))
validation_data = TensorDataset(torch.from_numpy(X_validation_z), torch.from_numpy(y_validation))
test_data = TensorDataset(torch.from_numpy(X_test_z), torch.from_numpy(y_test))
assert train_data.tensors[0].shape == (len(y_train), 16)
assert train_data.tensors[1].shape == (len(y_train),)
print({'训练集样本数': len(train_data), '验证集样本数': len(validation_data), '测试集样本数': len(test_data)})
{'训练集样本数': 6000, '验证集样本数': 2042, '测试集样本数': 2042}
第 4 步：建立非神经网络基线#
标准化后的多项逻辑回归提供快速、稳定的线性基线。若后续复杂训练流程没有超过基线，首先检查损失、学习率、初始化和数据加载，而不是把搜索范围继续扩大。
这里只在验证集报告；测试集仍封存。
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, f1_score, log_loss
baseline = LogisticRegression(max_iter=2000, random_state=SEED).fit(X_train_z, y_train)
baseline_validation_probability = baseline.predict_proba(X_validation_z)
baseline_validation_prediction = baseline_validation_probability.argmax(axis=1)
print({'验证集损失': log_loss(y_validation, baseline_validation_probability), '验证集准确率': accuracy_score(y_validation, baseline_validation_prediction), '验证集宏平均 F1': f1_score(y_validation, baseline_validation_prediction, average='macro')})
{'验证集损失': 0.2122594714164734, '验证集准确率': 0.9245837414299706, '验证集宏平均 F1': 0.9377785928784051}
第 5 步：批量归一化在训练与推断阶段的差异#
对于一批隐藏层的线性运算结果 \(\boldsymbol{Z}\in\mathbb{R}^{B\times d}\)，训练阶段沿批量方向分别计算每个特征的均值和方差，再使用可学习的尺度参数 \(\boldsymbol{\gamma}\) 与平移参数 \(\boldsymbol{\beta}\) 进行调整。推断阶段不能依赖当前单个样本或临时组成的批次，而应使用训练过程中逐步得到并保存的运行均值与运行方差。
下面让同一批输入分别在 train() 和 eval() 模式下进行前向传播。训练模式会更新运行统计量，评估模式不会；两种模式的结果可能不同，这是 BN 的预期行为，并不表示模型在评估时进行了随机计算。
class DiagnosticMLP(nn.Module):
def __init__(self, width=64, use_batchnorm=True, dropout=0.0):
super().__init__()
layers = [nn.Linear(16, width)]
if use_batchnorm:
layers.append(nn.BatchNorm1d(width))
layers.extend([nn.ReLU(), nn.Dropout(dropout), nn.Linear(width, width)])
if use_batchnorm:
layers.append(nn.BatchNorm1d(width))
layers.extend([nn.ReLU(), nn.Dropout(dropout), nn.Linear(width, 7)])
self.network = nn.Sequential(*layers)
def forward(self, X_batch):
return self.network(X_batch)
probe_model = DiagnosticMLP(width=32, use_batchnorm=True)
probe_batch = train_data.tensors[0][:64]
bn_layers = [module for module in probe_model.modules() if isinstance(module, nn.BatchNorm1d)]
before = bn_layers[0].running_mean.clone()
probe_model.train()
train_output = probe_model(probe_batch)
after_train = bn_layers[0].running_mean.clone()
probe_model.eval()
with torch.no_grad():
eval_output = probe_model(probe_batch)
after_eval = bn_layers[0].running_mean.clone()
assert train_output.shape == eval_output.shape == (64, 7)
assert not torch.equal(before, after_train)
assert torch.equal(after_train, after_eval)
print({'运行均值变化量': float((after_train - before).norm()), '训练/评估模式输出平均差异': float((train_output - eval_output).abs().mean())})
{'运行均值变化量': 0.08452160656452179, '训练/评估模式输出平均差异': 0.2803036868572235}
第 6 步：参数、超参数和搜索尺度#
权重、偏置以及 BN 中的 \(\boldsymbol{\gamma}\) 和 \(\boldsymbol{\beta}\) 都由训练过程根据梯度更新，属于模型参数。隐藏层宽度、是否使用 BN、学习率、权重衰减和批量大小需要在训练前指定，属于超参数。学习率和权重衰减可能相差多个数量级，适合在对数尺度上均匀生成候选值：若 \(u\sim\operatorname{Uniform}(a,b)\)，则取 \(10^u\)。
候选数量相同时，随机搜索通常能够为每个连续超参数尝试更多不同取值。程序使用固定随机种子 42 生成候选配置，因此可以重复得到相同的候选集合。
search_rng = np.random.default_rng(SEED)
N_CANDIDATES = 4 if FAST_MODE else 12
configs = []
for config_id in range(N_CANDIDATES):
configs.append({
'config_id': config_id,
'width': int(search_rng.choice([32, 64, 128])),
'use_batchnorm': bool(search_rng.choice([False, True])),
'dropout': float(search_rng.choice([0.0, 0.2])),
'batch_size': int(search_rng.choice([64, 128, 256])),
'learning_rate': float(10 ** search_rng.uniform(-4.0, -2.0)),
'weight_decay': float(10 ** search_rng.uniform(-6.0, -2.0)),
})
config_table = pd.DataFrame(configs)
config_display = config_table.replace({'use_batchnorm': {True: '是', False: '否'}}).rename(columns={'config_id': '配置编号', 'width': '隐藏层宽度', 'use_batchnorm': '使用批归一化', 'dropout': '丢弃率', 'batch_size': '批大小', 'learning_rate': '学习率', 'weight_decay': '权重衰减'})
display(config_display)
assert config_table['learning_rate'].between(1e-4, 1e-2).all()
assert config_table['weight_decay'].between(1e-6, 1e-2).all()
| 配置编号 | 隐藏层宽度 | 使用批归一化 | 丢弃率 | 批大小 | 学习率 | 权重衰减 | |
|---|---|---|---|---|---|---|---|
| 0 | 0 | 32 | 是 | 0.2 | 128 | 0.005214 | 0.000616 |
| 1 | 1 | 32 | 否 | 0.2 | 256 | 0.003329 | 0.001394 |
| 2 | 2 | 64 | 否 | 0.2 | 128 | 0.000552 | 0.005094 |
| 3 | 3 | 128 | 是 | 0.0 | 256 | 0.000771 | 0.000008 |
第 7 步：统一训练程序、检查点与有效数值检查#
训练程序执行以下检查：
每个小批量的输入、输出层线性运算结果和损失都必须是有限数值；
输出层线性运算结果的维度必须为 \(B\times7\)；
梯度范数必须是有限数值。若超过预先设置的上限，应立即报告，而不能直接用梯度裁剪掩盖原因；
每轮记录训练集与验证集损失、宏平均 F1、梯度范数和训练时间；
只根据验证集损失保存最佳检查点，也就是当时的一组模型参数。
函数只接收训练集和验证集，不接收测试集，因此超参数搜索阶段不能读取测试指标。
def evaluate(model, dataset):
model.eval()
logits_parts, target_parts = [], []
with torch.no_grad():
for X_batch, y_batch in DataLoader(dataset, batch_size=512, shuffle=False):
logits = model(X_batch.to(DEVICE))
if not torch.isfinite(logits).all():
raise FloatingPointError('评估时输出层的线性运算结果出现 NaN 或无穷值')
logits_parts.append(logits.cpu())
target_parts.append(y_batch)
logits = torch.cat(logits_parts)
target = torch.cat(target_parts).numpy()
probability = torch.softmax(logits, dim=1).numpy()
prediction = probability.argmax(axis=1)
return {'loss': log_loss(target, probability), 'accuracy': accuracy_score(target, prediction), 'macro_f1': f1_score(target, prediction, average='macro')}
def train_config(config, epochs, seed=SEED):
torch.manual_seed(seed)
model = DiagnosticMLP(config['width'], config['use_batchnorm'], config['dropout']).to(DEVICE)
optimizer = torch.optim.AdamW(model.parameters(), lr=config['learning_rate'], weight_decay=config['weight_decay'])
generator = torch.Generator().manual_seed(seed)
loader = DataLoader(train_data, batch_size=config['batch_size'], shuffle=True, generator=generator, drop_last=config['use_batchnorm'])
best_loss, best_state, history = np.inf, None, []
started = time.perf_counter()
for epoch in range(1, epochs + 1):
model.train()
loss_sum, count, last_gradient_norm = 0.0, 0, np.nan
for batch_number, (X_batch, y_batch) in enumerate(loader):
if not torch.isfinite(X_batch).all():
raise FloatingPointError(f'输入异常：epoch={epoch}, batch={batch_number}')
X_batch, y_batch = X_batch.to(DEVICE), y_batch.to(DEVICE)
optimizer.zero_grad(set_to_none=True)
logits = model(X_batch)
assert logits.shape == (y_batch.shape[0], 7)
loss = nn.functional.cross_entropy(logits, y_batch)
if not torch.isfinite(loss):
raise FloatingPointError(f'损失异常：epoch={epoch}, batch={batch_number}')
loss.backward()
gradient_norm = torch.sqrt(sum((parameter.grad.detach() ** 2).sum() for parameter in model.parameters() if parameter.grad is not None))
if not torch.isfinite(gradient_norm) or gradient_norm.item() > 1e4:
raise FloatingPointError(f'梯度异常：{gradient_norm.item()}')
last_gradient_norm = gradient_norm.item()
optimizer.step()
loss_sum += loss.item() * y_batch.shape[0]
count += y_batch.shape[0]
validation_metrics = evaluate(model, validation_data)
history.append({'epoch': epoch, 'train_loss': loss_sum / count, 'validation_loss': validation_metrics['loss'], 'validation_accuracy': validation_metrics['accuracy'], 'validation_macro_f1': validation_metrics['macro_f1'], 'last_gradient_norm': last_gradient_norm})
if validation_metrics['loss'] < best_loss:
best_loss = validation_metrics['loss']
best_state = {key: value.detach().cpu().clone() for key, value in model.state_dict().items()}
elapsed = time.perf_counter() - started
model.load_state_dict(best_state)
return model, pd.DataFrame(history), {'best_validation_loss': best_loss, 'seconds': elapsed, 'parameters': sum(parameter.numel() for parameter in model.parameters())}
probe_config = {'width': 32, 'use_batchnorm': True, 'dropout': 0.0, 'batch_size': 128, 'learning_rate': 1e-3, 'weight_decay': 1e-4}
probe_trained, probe_history, probe_summary = train_config(probe_config, epochs=2)
print({'最佳验证集损失': probe_summary['best_validation_loss'], '训练耗时（秒）': probe_summary['seconds'], '参数量': probe_summary['parameters']})
display(probe_history.rename(columns={'epoch': '训练轮次', 'train_loss': '训练集损失', 'validation_loss': '验证集损失', 'validation_accuracy': '验证集准确率', 'validation_macro_f1': '验证集宏平均 F1', 'last_gradient_norm': '最后一批梯度范数'}))
{'最佳验证集损失': 0.6029327511787415, '训练耗时（秒）': 0.1416257500241045, '参数量': 1959}
| 训练轮次 | 训练集损失 | 验证集损失 | 验证集准确率 | 验证集宏平均 F1 | 最后一批梯度范数 | |
|---|---|---|---|---|---|---|
| 0 | 1 | 1.350787 | 1.004863 | 0.826641 | 0.832183 | 0.781621 |
| 1 | 2 | 0.776447 | 0.602933 | 0.908423 | 0.918552 | 0.702356 |
第 8 步：主动加入错误并检查训练程序#
可靠的训练程序不仅要能完成正常实验，还应在错误首次出现时停止。这里先复制一个小批量，并主动加入 NaN（not a number，无效数值），用于确认有效数值检查会在模型参数更新之前发现问题。随后再设置一个非常大的学习率：它可能立即使损失或梯度出现异常，也可能在短时间内仍得到有限数值，但验证损失明显变差。因此，查找问题时不能只看程序是否抛出异常。
观测标签与样本错位时，损失通常仍是有限数值，需要通过人工核对少量样本和小数据过拟合测试来发现。
faulty_X = train_data.tensors[0][:32].clone()
faulty_X[0, 0] = float('nan')
try:
if not torch.isfinite(faulty_X).all():
raise FloatingPointError('故障注入成功：输入含 NaN，训练前被有限性检查拦截')
except FloatingPointError as error:
print(error)
extreme_config = {**probe_config, 'learning_rate': 10.0, 'use_batchnorm': False}
try:
_, extreme_history, _ = train_config(extreme_config, epochs=2)
print('极大学习率仍保持有限；其验证损失为：', extreme_history['validation_loss'].tolist())
except FloatingPointError as error:
print('极大学习率触发诊断：', error)
故障注入成功：输入含 NaN，训练前被有限性检查拦截
极大学习率触发诊断： 梯度异常：13892.2275390625
第 9 步：小数据过拟合测试#
从每个类别选取少量训练样本，让一个具有足够表达能力的网络反复学习这些样本。若模型连这几十个训练样本都无法达到很高的准确率，应优先检查观测标签、输出层、损失函数、梯度计算和参数更新，而不是继续增加正则化。
通过这项测试，只能说明模型至少能够学习一小组训练样本，不能证明验证过程不存在数据泄漏，也不能说明模型在新样本上具有良好表现。
tiny_indices = np.concatenate([np.flatnonzero(y_train == class_id)[:8] for class_id in range(7)])
tiny_X = torch.from_numpy(X_train_z[tiny_indices])
tiny_y = torch.from_numpy(y_train[tiny_indices])
torch.manual_seed(SEED)
tiny_model = DiagnosticMLP(width=128, use_batchnorm=False, dropout=0.0).to(DEVICE)
tiny_optimizer = torch.optim.Adam(tiny_model.parameters(), lr=0.01)
for step in range(300 if FAST_MODE else 800):
tiny_optimizer.zero_grad(set_to_none=True)
tiny_logits = tiny_model(tiny_X.to(DEVICE))
tiny_loss = nn.functional.cross_entropy(tiny_logits, tiny_y.to(DEVICE))
tiny_loss.backward()
tiny_optimizer.step()
with torch.no_grad():
tiny_accuracy = (tiny_model(tiny_X.to(DEVICE)).argmax(dim=1).cpu() == tiny_y).float().mean().item()
print({'小样本数量': len(tiny_y), '训练集准确率': tiny_accuracy, '最终损失': float(tiny_loss)})
assert tiny_accuracy > 0.90, '小样本无法拟合，应先排查训练流程'
{'小样本数量': 56, '训练集准确率': 1.0, '最终损失': 3.6273288515076274e-06}
第 10 步：简化的逐步缩减#
逐步缩减先让所有候选方案使用相同的较短训练轮数，再根据验证集表现保留其中较好的一部分，并为保留下来的方案增加训练轮数。本案例在第一阶段根据验证集损失保留前一半候选；第二阶段使用相同随机种子重新训练这些候选，并给予更长的总训练轮数。与让所有候选都进行长时间训练相比，这种方法可以节省计算资源，但训练初期的排名不一定等于最终排名，因此可能过早淘汰收敛较慢的方案。
程序会记录每个候选的运行状态、训练时间、参数量和学习曲线。运行失败的候选也会保留失败记录，而不会在汇总结果前直接删除。快速模式的两个阶段分别训练 3 轮和 8 轮，完整模式分别训练 5 轮和 20 轮。
stage1_epochs, stage2_epochs = ((3, 8) if FAST_MODE else (5, 20))
stage1_rows, run_histories = [], {}
for config in configs:
try:
model_i, history_i, summary_i = train_config(config, stage1_epochs)
run_histories[(config['config_id'], 1)] = history_i
stage1_rows.append({**config, **summary_i, 'status': 'ok'})
except Exception as error:
stage1_rows.append({**config, 'best_validation_loss': np.inf, 'seconds': np.nan, 'parameters': np.nan, 'status': f'failed: {type(error).__name__}: {error}'})
stage1_table = pd.DataFrame(stage1_rows).sort_values('best_validation_loss')
stage1_display = stage1_table.replace({'use_batchnorm': {True: '是', False: '否'}, 'status': {'ok': '成功'}})
stage1_display['status'] = stage1_display['status'].str.replace(r'^failed:', '失败：', regex=True)
stage1_display = stage1_display.rename(columns={'config_id': '配置编号', 'width': '隐藏层宽度', 'use_batchnorm': '使用批归一化', 'dropout': '丢弃率', 'batch_size': '批大小', 'learning_rate': '学习率', 'weight_decay': '权重衰减', 'best_validation_loss': '最佳验证集损失', 'seconds': '训练耗时（秒）', 'parameters': '参数量', 'status': '状态'})
display(stage1_display)
successful = stage1_table.query("status == 'ok'")
if successful.empty:
raise RuntimeError('所有第一阶段候选均失败')
promoted_ids = successful.head(max(1, int(np.ceil(len(successful) / 2))))['config_id'].astype(int).tolist()
print('晋级配置：', promoted_ids)
| 配置编号 | 隐藏层宽度 | 使用批归一化 | 丢弃率 | 批大小 | 学习率 | 权重衰减 | 最佳验证集损失 | 训练耗时（秒） | 参数量 | 状态 | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 0 | 32 | 是 | 0.2 | 128 | 0.005214 | 0.000616 | 0.222285 | 0.183918 | 1959 | 成功 |
| 3 | 3 | 128 | 是 | 0.0 | 256 | 0.000771 | 0.000008 | 0.277784 | 0.133219 | 20103 | 成功 |
| 1 | 1 | 32 | 否 | 0.2 | 256 | 0.003329 | 0.001394 | 0.348416 | 0.074909 | 1831 | 成功 |
| 2 | 2 | 64 | 否 | 0.2 | 128 | 0.000552 | 0.005094 | 0.525237 | 0.112734 | 5703 | 成功 |
晋级配置： [0, 3]
第 11 步：第二阶段、最佳检查点与训练日志#
入围配置使用更长预算重新训练。这里“重新训练”而不是从第一阶段继续，是为了让 train_config 保持简单、每次预算含义明确；代价是重复了前几轮计算。真正的逐步缩减通常会从检查点继续，并精确记录累计资源。
选择规则只看验证损失。并列时优先参数更少，再优先运行时间更短。测试集依然没有传入搜索函数。
stage2_rows, stage2_models = [], {}
for config_id in promoted_ids:
config = next(item for item in configs if item['config_id'] == config_id)
try:
model_i, history_i, summary_i = train_config(config, stage2_epochs)
stage2_models[config_id] = model_i
run_histories[(config_id, 2)] = history_i
stage2_rows.append({**config, **summary_i, 'status': 'ok'})
except Exception as error:
stage2_rows.append({**config, 'best_validation_loss': np.inf, 'seconds': np.nan, 'parameters': np.nan, 'status': f'failed: {type(error).__name__}: {error}'})
stage2_table = pd.DataFrame(stage2_rows).sort_values(['best_validation_loss', 'parameters', 'seconds'])
stage2_display = stage2_table.replace({'use_batchnorm': {True: '是', False: '否'}, 'status': {'ok': '成功'}})
stage2_display['status'] = stage2_display['status'].str.replace(r'^failed:', '失败：', regex=True)
stage2_display = stage2_display.rename(columns={'config_id': '配置编号', 'width': '隐藏层宽度', 'use_batchnorm': '使用批归一化', 'dropout': '丢弃率', 'batch_size': '批大小', 'learning_rate': '学习率', 'weight_decay': '权重衰减', 'best_validation_loss': '最佳验证集损失', 'seconds': '训练耗时（秒）', 'parameters': '参数量', 'status': '状态'})
display(stage2_display)
best_row = stage2_table.query("status == 'ok'").iloc[0]
best_config_id = int(best_row['config_id'])
best_model = stage2_models[best_config_id]
best_history = run_histories[(best_config_id, 2)]
best_config_display = best_row.copy()
best_config_display['use_batchnorm'] = {True: '是', False: '否'}[bool(best_config_display['use_batchnorm'])]
best_config_display['status'] = {'ok': '成功'}.get(best_config_display['status'], best_config_display['status'])
best_config_display = best_config_display.rename(index={'config_id': '配置编号', 'width': '隐藏层宽度', 'use_batchnorm': '使用批归一化', 'dropout': '丢弃率', 'batch_size': '批大小', 'learning_rate': '学习率', 'weight_decay': '权重衰减', 'best_validation_loss': '最佳验证集损失', 'seconds': '训练耗时（秒）', 'parameters': '参数量', 'status': '状态'})
print('最佳配置：', best_config_display.to_dict())
best_history_display = best_history.rename(columns={'epoch': '训练轮次', 'train_loss': '训练集损失', 'validation_loss': '验证集损失'})
best_history_display.plot(x='训练轮次', y=['训练集损失', '验证集损失'], marker='o', title='最佳配置训练日志')
plt.xlabel('训练轮次'); plt.ylabel('交叉熵'); plt.show()
| 配置编号 | 隐藏层宽度 | 使用批归一化 | 丢弃率 | 批大小 | 学习率 | 权重衰减 | 最佳验证集损失 | 训练耗时（秒） | 参数量 | 状态 | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 0 | 32 | 是 | 0.2 | 128 | 0.005214 | 0.000616 | 0.205504 | 0.491726 | 1959 | 成功 |
| 1 | 3 | 128 | 是 | 0.0 | 256 | 0.000771 | 0.000008 | 0.209949 | 0.354913 | 20103 | 成功 |
最佳配置： {'配置编号': 0, '隐藏层宽度': 32, '使用批归一化': '是', '丢弃率': 0.2, '批大小': 128, '学习率': 0.005214297905300546, '权重衰减': 0.0006158459876169436, '最佳验证集损失': 0.20550382137298584, '训练耗时（秒）': 0.4917257910128683, '参数量': 1959, '状态': '成功'}
第 12 步：确定配置后评价测试表现与计算代价#
超参数搜索结束并确定训练配置后，才使用测试集。结果报告测试集交叉熵、准确率、宏平均 F1、参数量、训练时间以及模型参数文件的大小。程序只在内存中计算模型参数文件的大小，不会在课程源码目录生成权重文件。
测量固定批量的预测时间时，先进行若干次预运行，再重复计时并取平均。若使用图形处理器，还要等待设备完成计算，否则计时程序可能在图形处理器真正完成任务之前就停止计时。
import io as binary_io
test_metrics = evaluate(best_model, test_data)
buffer = binary_io.BytesIO()
torch.save(best_model.state_dict(), buffer)
model_size_mb = buffer.tell() / 1024**2
fixed_batch = test_data.tensors[0][:512].to(DEVICE)
best_model.eval()
with torch.no_grad():
_ = best_model(fixed_batch)
if DEVICE.type == 'cuda':
torch.cuda.synchronize()
started = time.perf_counter()
repeats = 50
for _ in range(repeats):
_ = best_model(fixed_batch)
if DEVICE.type == 'cuda':
torch.cuda.synchronize()
prediction_seconds = (time.perf_counter() - started) / repeats
final_report = {
**test_metrics,
'parameters': int(sum(parameter.numel() for parameter in best_model.parameters())),
'state_dict_mb': model_size_mb,
'training_seconds': float(best_row['seconds']),
'fixed_batch_512_seconds': prediction_seconds,
}
final_report_display = pd.Series(final_report).rename(index={'loss': '测试集损失', 'accuracy': '测试集准确率', 'macro_f1': '测试集宏平均 F1', 'parameters': '参数量', 'state_dict_mb': '模型状态大小（MiB）', 'training_seconds': '训练耗时（秒）', 'fixed_batch_512_seconds': '固定 512 样本批次推理耗时（秒）'}).rename_axis('指标')
display(final_report_display.to_frame('数值'))
| 数值 | |
|---|---|
| 指标 | |
| 测试集损失 | 0.214114 |
| 测试集准确率 | 0.920176 |
| 测试集宏平均 F1 | 0.932632 |
| 参数量 | 1959.000000 |
| 模型状态大小（MiB） | 0.013049 |
| 训练耗时（秒） | 0.491726 |
| 固定 512 样本批次推理耗时（秒） | 0.000110 |
第 13 步：比较使用与不使用批量归一化的结果#
为了观察 BN 可能带来的影响，保留最佳配置中的其他条件，只把 use_batchnorm 从“是”改为“否”或从“否”改为“是”，然后使用相同的较长训练轮数重新训练。由于只改变了一个因素，结果差异更容易与 BN 联系起来。不过，这里仍只使用一个随机种子。批量很小时，批次均值和方差的波动可能较大；输入已经标准化且网络较浅时，BN 也不一定改善验证结果。
最后查看测试集中各类别的召回率，但不根据这些结果返回修改训练配置。
best_config = next(item for item in configs if item['config_id'] == best_config_id)
ablated_config = {**best_config, 'use_batchnorm': not best_config['use_batchnorm']}
ablated_model, ablated_history, ablated_summary = train_config(ablated_config, stage2_epochs)
ablated_validation = evaluate(ablated_model, validation_data)
ablation_display = pd.DataFrame([
{'配置': '已选配置', '使用批归一化': best_config['use_batchnorm'], '验证集损失': float(best_row['best_validation_loss']), '训练耗时（秒）': float(best_row['seconds'])},
{'配置': '切换批归一化开关', '使用批归一化': ablated_config['use_batchnorm'], '验证集损失': ablated_validation['loss'], '训练耗时（秒）': ablated_summary['seconds']},
])
ablation_display['使用批归一化'] = ablation_display['使用批归一化'].map({True: '是', False: '否'})
display(ablation_display)
from sklearn.metrics import classification_report
best_model.eval()
with torch.no_grad():
test_prediction = best_model(test_data.tensors[0].to(DEVICE)).argmax(dim=1).cpu().numpy()
class_report = pd.DataFrame(classification_report(y_test, test_prediction, target_names=classes, output_dict=True)).T
class_report_display = class_report.loc[classes].sort_values('recall').rename(index=class_display_names, columns={'precision': '精确率', 'recall': '召回率', 'f1-score': 'F1 分数', 'support': '样本数'}).rename_axis('类别')
display(class_report_display)
| 配置 | 使用批归一化 | 验证集损失 | 训练耗时（秒） | |
|---|---|---|---|---|
| 0 | 已选配置 | 是 | 0.205504 | 0.491726 |
| 1 | 切换批归一化开关 | 否 | 0.214126 | 0.272096 |
| 精确率 | 召回率 | F1 分数 | 样本数 | |
|---|---|---|---|---|
| 类别 | ||||
| 德尔马松豆 | 0.945122 | 0.874060 | 0.908203 | 532.0 |
| 巴尔布尼亚豆 | 0.941799 | 0.894472 | 0.917526 | 199.0 |
| 西拉豆 | 0.828375 | 0.916456 | 0.870192 | 395.0 |
| 卡利豆 | 0.926230 | 0.922449 | 0.924335 | 245.0 |
| 塞克尔豆 | 0.944625 | 0.953947 | 0.949264 | 304.0 |
| 霍罗兹豆 | 0.949153 | 0.968858 | 0.958904 | 289.0 |
| 孟买豆 | 1.000000 | 1.000000 | 1.000000 | 78.0 |
本案例小结与局限#
BN 在训练阶段更新运行统计量，在评估阶段使用已经保存的运行统计量。若忘记调用
eval()，预测结果会依赖当前批次的样本组成，模型中的运行统计量也可能继续变化；超参数搜索只能使用训练集与验证集结果。若测试集参与候选排序，最终测试结果就不再是独立评价；
跨越多个数量级的正超参数适合在对数尺度上搜索。比较不同候选时，还应使用相同的数据划分、随机种子集合、评价方法和计算资源标准；
逐步缩减可以节省计算，却可能淘汰收敛较慢的配置。快速模式中的短训练尤其容易偏向训练初期收敛较快的方法；
梯度裁剪不能用来掩盖无效数值、错误观测标签或过大的学习率。出现异常时，应先保存并检查第一个异常批次及其训练配置；
单次随机搜索具有偶然性。正式结论需要增加候选数量，使用多个随机种子重复实验，并报告失败运行、峰值内存和运行环境；
当前随机划分只能评价来自相似分布的测试样本，不能证明模型跨设备、产地或时间仍然稳定。
综合练习#
手算一个 \(4\times3\) 小批量的 BN 均值、方差、标准化结果以及使用 \(\boldsymbol{\gamma}\) 和 \(\boldsymbol{\beta}\) 调整后的输出。
让同一个样本分别处于不同训练批次，比较
train()模式下的输出；再切换到eval()模式，验证同一个样本的输出不会随同批其他样本改变。实现可以从第一阶段检查点继续训练的逐步缩减，准确记录累计训练轮数和已经处理的样本数。
比较在线性尺度和对数尺度上均匀生成学习率候选的方法，在候选数量相同时统计能够正常完成训练的比例。
分别加入观测标签错位、输出维度错误、NaN 输入、过大学习率以及忘记调用
eval()五类问题，并为每类问题设置最先触发的检查或运行记录。对表现最好的两个配置分别使用 5 个随机种子重复训练，报告宏平均 F1、训练时间、模型大小与失败次数，而不是只保留最好的一次运行。