# 训练调试：Loss 不下降、震荡、NaN 的排查清单 **TL;DR：** 训练不收敛是最令人沮丧的问题之一。本文提供一个系统化的排查清单，覆盖 Loss 不下降、Loss 震荡、Loss 变 NaN 三大类问题的诊断和修复方法。 ## 为什么这很重要 你花了两周准备数据、设计模型架构、搭建训练流水线。终于开始训练了，然后： - **Loss不下降**：训练了100个epoch，loss和随机初始化一样 - **Loss震荡**：loss忽上忽下，看起来像是心电图 - **Loss变NaN**：训练了5000步后，loss突然变成NaN，之前的努力全部白费 这些问题在深度学习中极其常见。经验丰富的工程师都知道：**训练一个模型，80%的时间在调试，20%的时间在等结果。** 但调试不应该靠猜。一个系统化的排查方法可以将调试时间从数天缩短到数小时。核心原则是：**从简单到复杂，从已知到未知。** ## 核心概念 ### Loss曲线的五种常见模式 ``` 模式1: Loss不下降（平坦线） Loss │ │───────────────────────── └────────────────────────── Step 可能原因：学习率太小、梯度消失、数据问题 模式2: Loss缓慢下降后停止（过早饱和） Loss │\ │ \ │ \_______________________ └────────────────────────── Step 可能原因：学习率太小、模型容量不足、数据质量差 模式3: Loss震荡 Loss │ /\ /\ /\ /\ │ / \/ \/ \/ \ │ └────────────────────────── Step 可能原因：学习率太大、batch size太小、数据shuffle不充分 模式4: Loss爆炸后变NaN Loss │\ │ \ ╱╲ │ ╲╱ ╲╱╲╱ │ \→ NaN └────────────────────────── Step 可能原因：学习率太大、数值溢出、除零、FP16精度问题 模式5: 过拟合 Loss │\ 训练Loss │ \_________ │ \ 验证Loss │ \───/\──/── └────────────────────────── Step 可能原因：模型太大、数据太少、正则化不足 ``` ### 调试的黄金法则 Andrej Karpathy在他的"A Recipe for Training Neural Networks"中提出了一个核心原则： > **先把你的模型在单个batch上训练到过拟合。** 如果你连一个batch都过拟合不了，说明你的代码或数据有根本性的问题。在解决这些基本问题之前，调整超参数或换模型架构都是浪费时间。 调试的正确顺序： ``` 1. 单batch过拟合测试 → 验证代码正确性 2. 数据质量检查 → 验证输入正确 3. 小数据集训练 → 验证模型能力 4. 完整训练 → 验证超参数和训练策略 5. 长时间训练 → 验证收敛性和泛化性 ``` ## 工作原理（简化的心智模型） ### 用学开车来理解训练调试 训练模型就像教一个人开车： **Loss不下降** = 教练说了半天，学员完全没听懂。可能是：教练的话太小声了（学习率太小）、学员在睡觉（梯度消失）、或者教的教材是错的（数据有问题）。先确认教练能把一句话说清楚（单batch过拟合），再考虑教学策略。 **Loss震荡** = 学员时好时坏，有时候开得不错，有时候突然失控。可能是：教练给的指令太激进（学习率太大）、练习场景太混乱（batch size太小，梯度方差大）。降低指令强度（减小学习率），多给一些稳定的练习环境（增大batch size）。 **Loss变NaN** = 学员把车开出了道路，撞墙了。一旦撞墙，后面的练习全部废了。通常是某个操作导致了不可恢复的错误（数值溢出、除零）。需要找出是哪个操作导致了问题，然后加上保护措施。 ## 工作原理（详细机制） ### 一、问题1：Loss不下降 #### 排查清单 ```python """ 训练调试工具集 提供系统化的问题诊断方法 """ import torch import torch.nn as nn import numpy as np from typing import Dict, List, Optional, Tuple from collections import defaultdict class TrainingDebugger: """训练调试器""" def __init__(self, model: nn.Module, device: str = 'auto'): if device == 'auto': self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') else: self.device = torch.device(device) self.model = model.to(self.device) # === 诊断1：单batch过拟合测试 === def test_single_batch_overfit( self, input_data: torch.Tensor, target: torch.Tensor, loss_fn, n_steps: int = 500, lr: float = 1e-3, ) -> Dict: """ 黄金法则：先在单个batch上过拟合 如果模型不能在单个batch上把loss降到接近0， 说明代码或数据有根本性问题 参数: input_data: 单个batch的输入 target: 单个batch的目标 loss_fn: 损失函数 n_steps: 训练步数 lr: 学习率（通常比正常训练大一些） 返回: 诊断结果字典 """ self.model.train() optimizer = torch.optim.Adam(self.model.parameters(), lr=lr) input_data = input_data.to(self.device) target = target.to(self.device) losses = [] for step in range(n_steps): optimizer.zero_grad() output = self.model(input_data) loss = loss_fn(output, target) loss.backward() optimizer.step() losses.append(loss.item()) if step % 100 == 0: print(f" Step {step}: loss = {loss.item():.6f}") initial_loss = losses[0] final_loss = losses[-1] min_loss = min(losses) # 诊断 diagnosis = { 'initial_loss': initial_loss, 'final_loss': final_loss, 'min_loss': min_loss, 'loss_reduction': initial_loss - final_loss, 'loss_reduction_pct': (initial_loss - final_loss) / initial_loss * 100, } if final_loss < initial_loss * 0.01: diagnosis['status'] = 'PASS' diagnosis['message'] = '模型成功在单batch上过拟合，代码基本正确' elif final_loss < initial_loss * 0.5: diagnosis['status'] = 'PARTIAL' diagnosis['message'] = 'Loss下降了但没有收敛，可能有以下问题：学习率不够大、模型容量不足、或数据有问题' else: diagnosis['status'] = 'FAIL' diagnosis['message'] = 'Loss几乎没下降！存在根本性问题，检查：1)loss计算 2)梯度是否为None 3)参数是否在更新' print(f"\n诊断结果: {diagnosis['status']}") print(f" 初始Loss: {initial_loss:.6f}") print(f" 最终Loss: {final_loss:.6f}") print(f" 最低Loss: {min_loss:.6f}") print(f" Loss降低: {diagnosis['loss_reduction_pct']:.1f}%") print(f" 说明: {diagnosis['message']}") return diagnosis # === 诊断2：梯度检查 === def check_gradients(self, input_data: torch.Tensor, target: torch.Tensor, loss_fn) -> Dict: """ 检查模型梯度的健康状况 检测： - 梯度为None（参数没有参与计算） - 梯度为零（梯度消失或计算错误） - 梯度为NaN/Inf（数值溢出） - 梯度过大（梯度爆炸） - 梯度过小（梯度消失） """ self.model.train() self.model.zero_grad() input_data = input_data.to(self.device) target = target.to(self.device) output = self.model(input_data) loss = loss_fn(output, target) loss.backward() gradient_stats = {} issues = [] for name, param in self.model.named_parameters(): if param.grad is None: gradient_stats[name] = { 'status': 'NO_GRADIENT', 'message': '参数没有被计算图包含' } issues.append(f" ⚠ {name}: 梯度为None") continue grad = param.grad.data stats = { 'shape': list(param.shape), 'mean': grad.mean().item(), 'std': grad.std().item(), 'min': grad.min().item(), 'max': grad.max().item(), 'has_nan': torch.isnan(grad).any().item(), 'has_inf': torch.isinf(grad).any().item(), 'zero_pct': (grad == 0).float().mean().item(), } # 判断健康状况 if stats['has_nan']: stats['status'] = 'NAN' issues.append(f" 🚨 {name}: 梯度包含NaN!") elif stats['has_inf']: stats['status'] = 'INF' issues.append(f" 🚨 {name}: 梯度包含Inf!") elif stats['std'] < 1e-10: stats['status'] = 'VANISHING' issues.append(f" ⚠ {name}: 梯度几乎为零（std={stats['std']:.2e}）") elif stats['std'] > 1e3: stats['status'] = 'EXPLODING' issues.append(f" ⚠ {name}: 梯度爆炸（std={stats['std']:.2e}）") else: stats['status'] = 'OK' gradient_stats[name] = stats # 打印报告 print("梯度检查报告:") if not issues: print(" ✓ 所有梯度正常") else: for issue in issues: print(issue) # 打印梯度统计（按层） print(f"\n {'层名':<40} {'状态':>10} {'均值':>10} {'标准差':>10} {'零%':>6}") print(" " + "-" * 80) for name, stats in gradient_stats.items(): if isinstance(stats.get('std'), float): print(f" {name:<40} {stats['status']:>10} {stats['mean']:>10.2e} {stats['std']:>10.2e} {stats.get('zero_pct', 0):>5.1%}") return {'gradient_stats': gradient_stats, 'issues': issues} # === 诊断3：数据检查 === @staticmethod def check_data(data: torch.Tensor, name: str = "data") -> Dict: """ 检查数据质量 检测：NaN、Inf、全零、异常值、值范围 """ issues = [] stats = { 'shape': list(data.shape), 'dtype': str(data.dtype), } # 基本统计 if data.is_floating_point(): stats['mean'] = data.mean().item() stats['std'] = data.std().item() stats['min'] = data.min().item() stats['max'] = data.max().item() # NaN检查 nan_count = torch.isnan(data).sum().item() stats['nan_count'] = nan_count if nan_count > 0: issues.append(f"🚨 发现 {nan_count} 个NaN值") # Inf检查 inf_count = torch.isinf(data).sum().item() stats['inf_count'] = inf_count if inf_count > 0: issues.append(f"🚨 发现 {inf_count} 个Inf值") # 全零检查 zero_pct = (data == 0).float().mean().item() stats['zero_pct'] = zero_pct if zero_pct > 0.99: issues.append(f"⚠ 数据几乎全是零 ({zero_pct:.1%})") # 异常值检查（使用z-score） if data.is_floating_point() and stats.get('std', 0) > 0: z_scores = (data - stats['mean']).abs() / stats['std'] outlier_pct = (z_scores > 10).float().mean().item() stats['extreme_outlier_pct'] = outlier_pct if outlier_pct > 0.01: issues.append(f"⚠ 发现 {outlier_pct:.2%} 的极端异常值 (|z|>10)") print(f"数据检查 [{name}]:") if not issues: print(f" ✓ 数据正常 (shape={stats['shape']}, dtype={stats['dtype']})") else: for issue in issues: print(f" {issue}") if data.is_floating_point(): print(f" 统计: mean={stats['mean']:.4f}, std={stats['std']:.4f}, " f"range=[{stats['min']:.4f}, {stats['max']:.4f}]") return {'stats': stats, 'issues': issues} # === 诊断4：学习率诊断 === def diagnose_learning_rate( self, dataloader, loss_fn, lr_range: List[float] = None, n_steps_per_lr: int = 20, ) -> Dict: """ 学习率范围测试（Learning Rate Range Test） 方法：从一个很小的学习率开始，逐步增大， 观察loss的变化。当loss开始上升或变NaN时， 说明学习率太大了。 最优学习率通常在loss下降最快的地方附近。 """ if lr_range is None: lr_range = [1e-7, 1e-6, 1e-5, 1e-4, 1e-3, 1e-2, 1e-1, 1.0] results = {} for lr in lr_range: # 重新初始化模型（重要！每次测试从相同的初始化开始） for layer in self.model.modules(): if hasattr(layer, 'reset_parameters'): layer.reset_parameters() optimizer = torch.optim.Adam(self.model.parameters(), lr=lr) losses = [] for step, batch in enumerate(dataloader): if step >= n_steps_per_lr: break input_data, target = batch input_data = input_data.to(self.device) target = target.to(self.device) optimizer.zero_grad() output = self.model(input_data) loss = loss_fn(output, target) if torch.isnan(loss) or torch.isinf(loss): losses.append(float('nan')) break loss.backward() # 检查梯度是否爆炸 max_grad = 0 for p in self.model.parameters(): if p.grad is not None: max_grad = max(max_grad, p.grad.abs().max().item()) optimizer.step() losses.append(loss.item()) avg_loss = np.nanmean(losses) if losses else float('nan') results[lr] = { 'losses': losses, 'avg_loss': avg_loss, 'final_loss': losses[-1] if losses else float('nan'), 'has_nan': any(np.isnan(l) for l in losses), 'max_grad': max_grad, } status = "NaN!" if results[lr]['has_nan'] else f"loss={avg_loss:.4f}" print(f" LR={lr:.0e}: {status}") # 找到最优学习率 valid_lrs = [(lr, r['avg_loss']) for lr, r in results.items() if not np.isnan(r['avg_loss'])] if valid_lrs: best_lr, best_loss = min(valid_lrs, key=lambda x: x[1]) # 找到loss开始发散的学习率 diverge_lr = None for lr in sorted(results.keys()): if results[lr]['has_nan'] or (results[lr]['avg_loss'] > best_loss * 10): diverge_lr = lr break recommendation = best_lr * 0.5 # 推荐使用最优lr的一半 else: best_lr = None diverge_lr = None recommendation = None return { 'results': results, 'best_lr': best_lr, 'diverge_lr': diverge_lr, 'recommended_lr': recommendation, } # === 诊断5：权重检查 === def check_weights(self) -> Dict: """ 检查模型权重的健康状况 在训练过程中定期调用，监控权重的变化 """ weight_stats = {} issues = [] for name, param in self.model.named_parameters(): data = param.data stats = { 'shape': list(param.shape), 'mean': data.mean().item(), 'std': data.std().item(), 'min': data.min().item(), 'max': data.max().item(), 'has_nan': torch.isnan(data).any().item(), 'has_inf': torch.isinf(data).any().item(), } if stats['has_nan']: issues.append(f"🚨 {name}: 权重包含NaN!") elif stats['has_inf']: issues.append(f"🚨 {name}: 权重包含Inf!") elif stats['std'] < 1e-10: issues.append(f"⚠ {name}: 权重几乎全相同 (std={stats['std']:.2e})") elif stats['std'] > 1e3: issues.append(f"⚠ {name}: 权重值过大 (std={stats['std']:.2e})") weight_stats[name] = stats if issues: print("权重检查发现问题:") for issue in issues: print(f" {issue}") else: print("权重检查: ✓ 所有权重正常") return {'weight_stats': weight_stats, 'issues': issues} # ============================================================ # 使用示例 # ============================================================ # 创建一个简单的模型和数据进行诊断 class SimpleClassifier(nn.Module): def __init__(self, input_dim=100, hidden_dim=256, output_dim=10): super().__init__() self.net = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim), ) def forward(self, x): return self.net(x) # 创建调试器 model = SimpleClassifier() debugger = TrainingDebugger(model) # 生成模拟数据 batch_size = 32 input_data = torch.randn(batch_size, 100) target = torch.randint(0, 10, (batch_size,)) loss_fn = nn.CrossEntropyLoss() print("="*60) print("诊断1：单batch过拟合测试") print("="*60) result = debugger.test_single_batch_overfit(input_data, target, loss_fn, n_steps=300) print(f"\n{'='*60}") print("诊断2：梯度检查") print("="*60) grad_result = debugger.check_gradients(input_data, target, loss_fn) print(f"\n{'='*60}") print("诊断3：数据检查") print("="*60) data_result = TrainingDebugger.check_data(input_data, "input_data") print(f"\n{'='*60}") print("诊断4：权重检查") print("="*60) weight_result = debugger.check_weights() ``` ### 二、问题2：Loss震荡 Loss震荡通常由以下原因引起： ```python """ Loss震荡的诊断和修复 """ import numpy as np import matplotlib matplotlib.use('Agg') # 非交互式后端 import matplotlib.pyplot as plt def diagnose_oscillation(losses: list, window: int = 100) -> dict: """ 分析loss曲线的震荡特征 参数: losses: loss值列表 window: 滑动窗口大小 """ losses = np.array(losses) # 计算滑动平均 if len(losses) >= window: moving_avg = np.convolve(losses, np.ones(window)/window, mode='valid') else: moving_avg = losses # 计算震荡幅度 if len(losses) >= window: local_std = np.array([ losses[max(0, i-window):i+1].std() for i in range(len(losses)) ]) else: local_std = np.full(len(losses), losses.std()) # 震荡指标 overall_trend = np.polyfit(np.arange(len(losses)), losses, 1)[0] oscillation_amplitude = np.mean(local_std[-window:]) if len(local_std) >= window else np.mean(local_std) avg_loss = np.mean(losses[-window:]) if len(losses) >= window else np.mean(losses) oscillation_ratio = oscillation_amplitude / max(avg_loss, 1e-8) diagnosis = { 'overall_trend': overall_trend, 'oscillation_amplitude': oscillation_amplitude, 'oscillation_ratio': oscillation_ratio, 'avg_recent_loss': avg_loss, } # 诊断 if oscillation_ratio > 0.5: diagnosis['severity'] = 'SEVERE' diagnosis['recommendation'] = '严重震荡。建议：1)降低学习率10x 2)增大batch size 3)增加梯度裁剪' elif oscillation_ratio > 0.1: diagnosis['severity'] = 'MODERATE' diagnosis['recommendation'] = '中等震荡。建议：1)适当降低学习率 2)使用warmup策略 3)检查数据shuffle' else: diagnosis['severity'] = 'NORMAL' diagnosis['recommendation'] = '震荡在正常范围内' if overall_trend > 0: diagnosis['trend'] = 'INCREASING' diagnosis['recommendation'] += '\n ⚠ Loss整体趋势在上升！可能需要降低学习率或检查数据质量' elif overall_trend < -1e-5: diagnosis['trend'] = 'DECREASING' else: diagnosis['trend'] = 'FLAT' print(f"Loss震荡分析:") print(f" 震荡幅度: {oscillation_amplitude:.6f}") print(f" 震荡比率: {oscillation_ratio:.2%}") print(f" 整体趋势: {diagnosis['trend']}") print(f" 严重程度: {diagnosis['severity']}") print(f" 建议: {diagnosis['recommendation']}") return diagnosis # 模拟不同类型的loss曲线 np.random.seed(42) # 正常下降 normal_losses = [2.0 * np.exp(-0.01 * i) + np.random.normal(0, 0.02) for i in range(1000)] # 震荡 oscillating_losses = [2.0 * np.exp(-0.005 * i) + 0.3 * np.sin(i * 0.1) + np.random.normal(0, 0.05) for i in range(1000)] # 严重震荡 severe_losses = [2.0 * np.exp(-0.002 * i) + np.sin(i * 0.3) * 0.5 + np.random.normal(0, 0.1) for i in range(1000)] print("=== 正常Loss ===") diagnose_oscillation(normal_losses) print(f"\n=== 轻度震荡 ===") diagnose_oscillation(oscillating_losses) print(f"\n=== 严重震荡 ===") diagnose_oscillation(severe_losses) ``` ### 三、问题3：Loss变NaN Loss变NaN是最严重的问题——训练无法继续。必须找出NaN第一次出现的位置。 ```python """ NaN问题的系统化诊断 """ import torch import torch.nn as nn import numpy as np from typing import Optional class NaNDetector: """ NaN检测器 核心策略：在训练循环中的每个关键点插入检查点， 定位NaN第一次出现的位置 """ def __init__(self): self.checkpoints = [] self.found_nan = False def check(self, name: str, tensor: torch.Tensor, step: int): """检查张量是否包含NaN""" if self.found_nan: return # 已经找到第一个NaN，跳过后续检查 has_nan = torch.isnan(tensor).any().item() if isinstance(tensor, torch.Tensor) else np.isnan(tensor).any() has_inf = torch.isinf(tensor).any().item() if isinstance(tensor, torch.Tensor) else np.isinf(tensor).any() self.checkpoints.append({ 'step': step, 'name': name, 'has_nan': has_nan, 'has_inf': has_inf, 'shape': list(tensor.shape) if isinstance(tensor, torch.Tensor) else 'scalar', }) if has_nan or has_inf: self.found_nan = True print(f"\n🚨 NaN/Inf 首次出现在:") print(f" Step: {step}") print(f" 检查点: {name}") print(f" 形状: {self.checkpoints[-1]['shape']}") if isinstance(tensor, torch.Tensor): print(f" 统计: mean={tensor[~torch.isnan(tensor)].mean().item() if (~torch.isnan(tensor)).any() else 'all NaN':.6f}") # 打印之前的检查点（帮助理解NaN出现前的状态） print(f"\n 前5个检查点:") for cp in self.checkpoints[-6:-1]: status = "OK" if not cp['has_nan'] and not cp['has_inf'] else "PROBLEM" print(f" Step {cp['step']}: {cp['name']} - {status}") def instrumented_train_step( self, model: nn.Module, optimizer: torch.optim.Optimizer, input_data: torch.Tensor, target: torch.Tensor, loss_fn, step: int, ) -> Optional[float]: """ 带检查点的训练步骤 在以下位置插入NaN检测： 1. 输入数据 2. 前向传播输出 3. Loss值 4. 反向传播后的梯度 5. 参数更新后的权重 """ self.check(f"input_data", input_data, step) self.check(f"target", target, step) # 检查权重（前向传播前） for name, param in model.named_parameters(): self.check(f"weight/{name}", param.data, step) optimizer.zero_grad() # 前向传播 output = model(input_data) self.check(f"forward_output", output, step) # Loss计算 loss = loss_fn(output, target) self.check(f"loss", loss, step) if torch.isnan(loss): print(f"\n NaN出现在Loss计算中!") print(f" 输出范围: [{output.min().item():.4f}, {output.max().item():.4f}]") if hasattr(target, 'unique'): print(f" 目标值: {target.unique().tolist()}") return None # 反向传播 loss.backward() # 检查梯度 for name, param in model.named_parameters(): if param.grad is not None: self.check(f"gradient/{name}", param.grad, step) # 梯度裁剪（防止爆炸） torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 参数更新 optimizer.step() # 检查更新后的权重 for name, param in model.named_parameters(): self.check(f"updated_weight/{name}", param.data, step) return loss.item() # 常见NaN原因和修复方法 NAN_TROUBLESHOOTING_GUIDE = """ NaN Loss 排查清单 ================ 1. 检查输入数据 □ 输入中是否有NaN/Inf值？ □ 输入值范围是否合理？（归一化了吗？） □ 标签是否正确？（负数索引？超出范围？） 修复：添加数据验证，使用torch.isnan()和torch.isinf()检查 2. 检查学习率 □ 学习率是否太大？（最常见的NaN原因） 修复：降低学习率10倍，观察是否还出现NaN 3. 检查Loss函数 □ 是否有除法操作？（除以零→NaN） □ 是否有log操作？（log(0)→-Inf，log(负数)→NaN） □ 是否有sqrt操作？（sqrt(负数)→NaN） 修复：添加epsilon保护，如 log(x + 1e-8)，x / (y + 1e-8) 4. 检查混合精度 □ 是否使用FP16？（FP16范围太小，容易溢出） □ 是否使用了loss scaling？ 修复：切换到BF16，或调整loss scaling策略 5. 检查模型架构 □ 是否有未初始化的参数？ □ 是否有指数运算导致数值溢出？（softmax中的exp） □ 是否有循环中的梯度累积导致爆炸？ 修复：使用xavier/kaiming初始化，softmax前减去最大值 6. 检查优化器 □ 是否使用了正确的优化器配置？ □ Adam的epsilon是否太小？ 修复：确保eps=1e-8（默认值），不要设太小 """ print(NAN_TROUBLESHOOTING_GUIDE) # 演示：制造和修复NaN print("\n=== 制造NaN的场景和修复 ===\n") # 场景1：除零 print("场景1：Log中的零值") x = torch.tensor([1.0, 0.0, 2.0]) # 错误：log(0) = -inf bad_result = torch.log(x) print(f" torch.log([1, 0, 2]) = {bad_result}") # 修复：添加epsilon good_result = torch.log(x + 1e-8) print(f" torch.log(x + 1e-8) = {good_result}") # 场景2：FP16溢出 print("\n场景2：FP16溢出") large_val = torch.tensor([100000.0]) fp16_val = large_val.half() print(f" FP32: {large_val.item()}") print(f" FP16: {fp16_val.item()}") # 可能是inf # 修复：使用BF16 bf16_val = large_val.bfloat16() print(f" BF16: {bf16_val.item()}") # 不会溢出 # 场景3：Softmax溢出 print("\n场景3：Softmax数值溢出") logits = torch.tensor([1000.0, 1001.0, 1002.0]) # 朴素softmax会溢出 bad_softmax = torch.exp(logits) / torch.exp(logits).sum() print(f" 朴素softmax: {bad_softmax}") # [nan, nan, nan] # 修复：减去最大值 logits_stable = logits - logits.max() good_softmax = torch.exp(logits_stable) / torch.exp(logits_stable).sum() print(f" 稳定softmax: {good_softmax}") # 正确结果 # 或者直接使用PyTorch的softmax pytorch_softmax = torch.softmax(logits, dim=0) print(f" torch.softmax: {pytorch_softmax}") ``` ### 四、系统化调试流程 ```python """ 系统化调试流程 将所有诊断工具组合成完整的调试流程 """ import torch import torch.nn as nn from typing import Dict, List def systematic_debug( model: nn.Module, train_dataloader, val_dataloader, loss_fn, optimizer_class=torch.optim.AdamW, device: str = 'auto', ) -> Dict: """ 系统化调试流程 按以下顺序逐步排查： 1. 数据检查 2. 单batch过拟合 3. 梯度检查 4. 小数据集训练 5. 学习率范围测试 """ if device == 'auto': device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') else: device = torch.device(device) model = model.to(device) debugger = TrainingDebugger(model, device=str(device)) results = {} # === Step 1: 数据检查 === print("=" * 60) print("Step 1/5: 数据质量检查") print("=" * 60) batch = next(iter(train_dataloader)) input_data, target = batch input_check = TrainingDebugger.check_data(input_data, "训练输入") target_check = TrainingDebugger.check_data(target, "训练目标") results['data_check'] = { 'input_issues': input_check['issues'], 'target_issues': target_check['issues'], } if input_check['issues'] or target_check['issues']: print("\n⚠ 数据有问题！先修复数据问题再继续调试。") return results # === Step 2: 单batch过拟合 === print(f"\n{'=' * 60}") print("Step 2/5: 单batch过拟合测试") print("=" * 60) input_data = input_data.to(device) target = target.to(device) overfit_result = debugger.test_single_batch_overfit( input_data, target, loss_fn, n_steps=300, lr=1e-3 ) results['single_batch_overfit'] = overfit_result if overfit_result['status'] == 'FAIL': print("\n⚠ 单batch无法过拟合！进行更深入的诊断...") # 梯度检查 grad_result = debugger.check_gradients(input_data, target, loss_fn) results['gradient_check'] = grad_result if grad_result['issues']: print("\n梯度有问题。常见原因：") print(" 1. 模型中有些层没有参与计算图（检查forward函数）") print(" 2. Loss函数返回了常数（检查loss计算）") print(" 3. 学习率为0（检查optimizer配置）") return results # === Step 3: 梯度健康检查 === print(f"\n{'=' * 60}") print("Step 3/5: 梯度健康检查") print("=" * 60) grad_result = debugger.check_gradients(input_data, target, loss_fn) results['gradient_check'] = grad_result # === Step 4: 权重检查 === print(f"\n{'=' * 60}") print("Step 4/5: 权重初始化检查") print("=" * 60) weight_result = debugger.check_weights() results['weight_check'] = weight_result # === Step 5: 综合建议 === print(f"\n{'=' * 60}") print("Step 5/5: 综合诊断和建议") print("=" * 60) recommendations = [] if overfit_result['status'] == 'PASS': recommendations.append("✓ 单batch过拟合成功，模型和代码基本正确") if not grad_result['issues']: recommendations.append("✓ 梯度检查通过") if not weight_result['issues']: recommendations.append("✓ 权重初始化正常") if len(recommendations) == 3: recommendations.append("\n下一步建议：") recommendations.append(" 1. 使用完整数据集训练，从较小的学习率开始（如1e-4）") recommendations.append(" 2. 监控训练loss和验证loss曲线") recommendations.append(" 3. 如果loss不下降，尝试学习率范围测试") recommendations.append(" 4. 如果loss震荡，降低学习率或增大batch size") recommendations.append(" 5. 如果出现过拟合，增加正则化或减少模型大小") for rec in recommendations: print(f" {rec}") results['recommendations'] = recommendations return results # 运行完整调试流程 model = SimpleClassifier(input_dim=100, hidden_dim=256, output_dim=10) # 创建模拟数据 from torch.utils.data import TensorDataset, DataLoader train_data = TensorDataset(torch.randn(1000, 100), torch.randint(0, 10, (1000,))) val_data = TensorDataset(torch.randn(200, 100), torch.randint(0, 10, (200,))) train_loader = DataLoader(train_data, batch_size=32, shuffle=True) val_loader = DataLoader(val_data, batch_size=32) results = systematic_debug( model=model, train_dataloader=train_loader, val_dataloader=val_loader, loss_fn=nn.CrossEntropyLoss(), ) ``` ## 代码示例（完整可运行的 Python） ```python """ 完整的训练监控和预警系统 在训练过程中实时检测异常并报警 """ import torch import torch.nn as nn import numpy as np from collections import deque from typing import Dict, Optional, List import time class TrainingMonitor: """ 训练监控器 功能： - 实时监控loss曲线 - 自动检测异常（NaN, 震荡, 不下降） - 提供修复建议 - 记录训练历史 """ def __init__( self, model: nn.Module, patience_nan: int = 1, # 连续NaN多少步后报警 patience_stall: int = 100, # loss多少步不下降后报警 oscillation_threshold: float = 0.3, # 震荡比率阈值 grad_clip_norm: float = 1.0, # 梯度裁剪阈值 ): self.model = model self.patience_nan = patience_nan self.patience_stall = patience_stall self.oscillation_threshold = oscillation_threshold self.grad_clip_norm = grad_clip_norm self.loss_history = deque(maxlen=1000) self.lr_history = deque(maxlen=1000) self.grad_norm_history = deque(maxlen=1000) self.step = 0 self.nan_count = 0 self.best_loss = float('inf') self.steps_since_improvement = 0 self.alerts = [] def pre_step(self, optimizer: torch.optim.Optimizer): """训练步骤前调用""" self.current_lr = optimizer.param_groups[0]['lr'] def post_step(self, loss: float, extra_metrics: Dict = None) -> Dict: """ 训练步骤后调用 返回：状态和警告信息 """ self.step += 1 # 检查NaN if np.isnan(loss) or np.isinf(loss): self.nan_count += 1 alert = { 'step': self.step, 'type': 'NaN_DETECTED', 'message': f'Loss = {loss}, 连续NaN计数: {self.nan_count}', 'action': '检查输入数据、学习率、loss函数' } self.alerts.append(alert) if self.nan_count >= self.patience_nan: alert['severity'] = 'CRITICAL' return {'should_stop': True, 'alert': alert} return {'should_stop': False, 'alert': alert} self.nan_count = 0 # 重置NaN计数 self.loss_history.append(loss) self.lr_history.append(self.current_lr) # 检查loss是否在改善 if loss < self.best_loss: self.best_loss = loss self.steps_since_improvement = 0 else: self.steps_since_improvement += 1 status = {'should_stop': False, 'loss': loss, 'step': self.step} # 检查停滞 if self.steps_since_improvement >= self.patience_stall: alert = { 'step': self.step, 'type': 'LOSS_STALLED', 'message': f'Loss已{self.steps_since_improvement}步未改善', 'action': '尝试: 降低学习率、增加学习率调度、检查数据质量' } self.alerts.append(alert) status['alert'] = alert # 检查震荡 if len(self.loss_history) >= 50: recent_losses = list(self.loss_history)[-50:] avg = np.mean(recent_losses) std = np.std(recent_losses) oscillation_ratio = std / max(abs(avg), 1e-8) if oscillation_ratio > self.oscillation_threshold: alert = { 'step': self.step, 'type': 'OSCILLATION', 'message': f'Loss震荡比率: {oscillation_ratio:.2%} (阈值: {self.oscillation_threshold:.0%})', 'action': '尝试: 降低学习率、增大batch size、增加梯度裁剪' } self.alerts.append(alert) status['alert'] = alert # 梯度范数 total_norm = 0 for p in self.model.parameters(): if p.grad is not None: total_norm += p.grad.data.norm(2).item() ** 2 total_norm = total_norm ** 0.5 self.grad_norm_history.append(total_norm) status['grad_norm'] = total_norm if total_norm > 100: alert = { 'step': self.step, 'type': 'GRADIENT_EXPLOSION', 'message': f'梯度范数: {total_norm:.2f} (过大)', 'action': '增加梯度裁剪强度，或降低学习率' } self.alerts.append(alert) status['alert'] = alert return status def get_summary(self) -> Dict: """获取训练摘要""" return { 'total_steps': self.step, 'best_loss': self.best_loss, 'current_loss': self.loss_history[-1] if self.loss_history else None, 'nan_count_total': sum(1 for a in self.alerts if a['type'] == 'NaN_DETECTED'), 'total_alerts': len(self.alerts), 'recent_grad_norm': list(self.grad_norm_history)[-10:], } # 使用示例 model = nn.Sequential( nn.Linear(100, 256), nn.ReLU(), nn.Linear(256, 10), ) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3) loss_fn = nn.CrossEntropyLoss() monitor = TrainingMonitor(model, patience_stall=50) print("训练监控演示:") for step in range(100): monitor.pre_step(optimizer) x = torch.randn(32, 100) y = torch.randint(0, 10, (32,)) optimizer.zero_grad() output = model(x) loss = loss_fn(output, y) loss.backward() # 梯度裁剪 torch.nn.utils.clip_grad_norm_(model.parameters(), monitor.grad_clip_norm) optimizer.step() status = monitor.post_step(loss.item()) if status.get('alert'): print(f" Step {step}: [{status['alert']['type']}] {status['alert']['message']}") elif step % 20 == 0: print(f" Step {step}: loss={loss.item():.4f}, grad_norm={status.get('grad_norm', 0):.4f}") if status.get('should_stop'): print(" 🛑 训练被监控器终止!") break summary = monitor.get_summary() print(f"\n训练摘要: {summary['total_steps']}步, 最佳loss={summary['best_loss']:.4f}, " f"告警={summary['total_alerts']}次") ``` ## 真实案例 ### 案例1：GPT训练中的NaN OpenAI在GPT-2的训练中遇到过NaN问题。原因是： - 某些训练样本包含异常大的数值（日志中的极大值） - FP16混合精度训练中，softmax计算溢出 - **修复**：使用log-softmax代替直接softmax，添加数值稳定性保护 ### 案例2：DeepMind训练Chinchilla DeepMind在训练Chinchilla时遇到loss plateau（loss停滞不下降）。原因是： - 学习率调度与训练数据量不匹配 - **修复**：重新设计学习率调度，使学习率衰减与数据消耗速度匹配 ### 案例3：Stable Diffusion训练中的梯度爆炸 Stability AI在训练Stable Diffusion时遇到梯度爆炸。原因是： - UNet中某些层的梯度在反向传播中被指数放大 - **修复**：对特定层使用梯度裁剪，将梯度范数限制在合理范围内 ## 权衡取舍以及何时不该使用 ### 调试花多少时间 调试本身也有ROI。花一天时间排查一个loss不下降的问题可能比花一天时间换一个模型架构更有价值——因为代码错误是确定性的，修复后一定会解决；而换架构可能只是掩盖了真正的问题。 ### 梯度裁剪的权衡 梯度裁剪是防止梯度爆炸的标准方法，但它也会限制模型的探索能力。如果你的任务确实需要大的梯度更新（例如某些强化学习场景），过度裁剪可能导致训练缓慢。 ### 何时不该花时间调试 1. **Loss在缓慢下降**：如果loss在下降，只是速度慢，可能只是需要更多的训练时间。 2. **验证loss略高于训练loss**：轻微的过拟合是正常的，不需要调试。 3. **最终的精度差几个百分点**：这时候应该考虑换数据、换模型、换任务定义，而不是调试训练过程。 ## 关键要点 1. **先在单个batch上过拟合**，这是最重要的调试原则。如果模型连一个batch都学不了，后续的调试都是浪费时间。 2. **Loss不下降时，按以下顺序排查**：数据质量 → 代码正确性（loss计算、梯度流）→ 学习率 → 模型容量。90%的问题出在前两个。 3. **Loss变NaN时，找到NaN第一次出现的位置**。用NaNDetector在训练循环中插入检查点，定位是哪个操作产生了NaN。最常见的原因是：学习率太大、FP16溢出、除零/log(零)。 4. **Loss震荡时，降低学习率是最直接的方法**。然后考虑增大batch size、使用学习率warmup、添加梯度裁剪。 5. **梯度检查是诊断的核心工具**。检查每层梯度的均值、标准差、是否有NaN/Inf、是否有过多零值。梯度消失（std接近0）和梯度爆炸（std极大）都有明确的修复方案。 6. **系统化地调试，不要靠猜**。使用本文提供的TrainingDebugger和TrainingMonitor工具，逐步排除可能的原因，而不是随机尝试不同的超参数。 7. **预防胜于治疗**。在训练开始前就做好数据检查、在训练过程中使用监控器实时检测异常，比事后调试高效得多。 ## 延伸阅读 **文章**： - Andrej Karpathy "A Recipe for Training Neural Networks" — 系统化的训练方法论 - "Debugging the Dreaded NaN" (Chaim Rand) — NaN问题的深度分析 - Google ML Crash Course "Interpreting Loss Curves" — Loss曲线解读 **工具**： - PyTorch `torch.autograd.detect_anomaly()` — 自动检测梯度计算中的异常 - PyTorch `torch.autograd.set_detect_anomaly(True)` — 在反向传播中定位NaN源头 - Weights & Biases — 训练可视化，实时监控loss曲线和梯度 - TensorBoard — 训练日志可视化 **社区讨论**： - PyTorch Forums — 大量关于训练问题的讨论和解决方案 - nanoGPT Issues — 小型GPT实现中的常见问题和修复 - r/MachineLearning — 社区经验分享