【Bug已解决】Pytorch Change the learning rate based on number of epochs 解决方案问题描述在训练深度学习模型时学习率Learning Rate是最关键的超参数之一。许多开发者希望根据训练的 epoch 数动态调整学习率以获得更好的收敛效果。然而在 PyTorch 中实现这一功能时经常会遇到学习率不更新、调度器不生效、学习率调整不符合预期等问题。典型的问题场景包括调用了scheduler.step()但学习率没有变化scheduler.step()和optimizer.step()的调用顺序错误导致警告或异常自定义学习率调整逻辑不生效使用LambdaLR时 lambda 函数编写错误多参数组的学习率调整不一致断点续训时调度器状态恢复不正确这些问题的核心在于理解 PyTorch 中学习率调度器的工作机制以及optimizer和scheduler之间的交互关系。错误复现场景一调度器不生效import torch import torch.nn as nn from torch.optim.lr_scheduler import StepLR model nn.Linear(10, 2) optimizer torch.optim.SGD(model.parameters(), lr0.1) scheduler StepLR(optimizer, step_size2, gamma0.5) for epoch in range(10): optimizer.step() # 忘记调用 scheduler.step() print(fEpoch {epoch}, LR: {optimizer.param_groups[0][lr]}) # 输出学习率始终为 0.1没有变化场景二调用顺序错误# PyTorch 1.1 的正确顺序 for epoch in range(10): for batch in dataloader: optimizer.step() # 先更新参数 scheduler.step() # 再更新学习率 # 错误顺序PyTorch 1.1 会警告 for epoch in range(10): scheduler.step() # 错误在 optimizer.step() 之前 for batch in dataloader: optimizer.step() # 警告信息 # UserWarning: Detected call of lr_scheduler.step() before optimizer.step().场景三自定义 Lambda 函数错误from torch.optim.lr_scheduler import LambdaLR optimizer torch.optim.SGD(model.parameters(), lr0.1) # 错误lambda 函数参数理解错误 # 期望epoch 0-4 用 lr0.1, epoch 5-9 用 lr0.01 def wrong_lambda(epoch): if epoch 5: return 0.1 elif epoch 10: return 0.01 else: return 0.001 scheduler LambdaLR(optimizer, lr_lambdawrong_lambda) # 问题LambdaLR 的 lr_lambda 返回的是乘法因子不是最终学习率 # 初始 lr0.1, epoch 0: lr 0.1 * 0.1 0.01不是期望的 0.1场景四断点续训状态丢失# 保存检查点 checkpoint { model_state: model.state_dict(), optimizer_state: optimizer.state_dict(), # 忘记保存 scheduler 状态 } torch.save(checkpoint, checkpoint.pt) # 加载后继续训练 checkpoint torch.load(checkpoint.pt) model.load_state_dict(checkpoint[model_state]) optimizer.load_state_dict(checkpoint[optimizer_state]) # scheduler 从头开始学习率调整不连续根因分析1. PyTorch 调度器的工作原理PyTorch 的学习率调度器本质上是一个在每次调用step()时修改 optimizer 中各参数组学习率的对象。其核心机制如下# 调度器内部简化逻辑 class LRScheduler: def __init__(self, optimizer, last_epoch-1): self.optimizer optimizer self.last_epoch last_epoch self.base_lrs [group[lr] for group in optimizer.param_groups] def step(self): self.last_epoch 1 new_lrs self.get_lr() for group, new_lr in zip(self.optimizer.param_groups, new_lrs): group[lr] new_lr2.step()调用顺序的重要性在 PyTorch 1.1.0 之后调度器的step()应该在每个 epoch 结束时调用而不是在 epoch 开始时。这是因为optimizer.step()在每个 batch 更新参数scheduler.step()在每个 epoch 更新学习率。如果scheduler.step()在optimizer.step()之前调用第一个 epoch 的学习率会被跳过。3. LambdaLR 的乘法因子机制LambdaLR的lr_lambda函数返回的是一个乘法因子最终学习率 初始学习率 × 乘法因子。这是许多开发者容易误解的地方。4. 多参数组的独立性当 optimizer 有多个参数组时如对 bias 和 weight 使用不同的学习率调度器会独立调整每个参数组的学习率。解决方案方案一使用内置调度器推荐import torch import torch.nn as nn from torch.optim.lr_scheduler import ( StepLR, MultiStepLR, ExponentialLR, CosineAnnealingLR, CosineAnnealingWarmRestarts, ReduceLROnPlateau, LambdaLR, OneCycleLR ) model nn.Linear(10, 2) optimizer torch.optim.SGD(model.parameters(), lr0.1) # 1. StepLR: 每 step_size 个 epoch学习率乘以 gamma scheduler_step StepLR(optimizer, step_size3, gamma0.5) # 2. MultiStepLR: 在指定的 epoch 调整学习率 scheduler_multi MultiStepLR(optimizer, milestones[10, 20, 30], gamma0.5) # 3. ExponentialLR: 每个 epoch 学习率乘以 gamma scheduler_exp ExponentialLR(optimizer, gamma0.95) # 4. CosineAnnealingLR: 余弦退火 scheduler_cos CosineAnnealingLR(optimizer, T_max50, eta_min0.001) # 5. ReduceLROnPlateau: 当指标不再改善时降低学习率 scheduler_plateau ReduceLROnPlateau( optimizer, modemin, factor0.5, patience5, min_lr1e-6 )方案二使用 LambdaLR 实现自定义调度from torch.optim.lr_scheduler import LambdaLR import math optimizer torch.optim.SGD(model.parameters(), lr0.1) # 1. 线性衰减 def linear_decay(epoch): return 1.0 - epoch / 100.0 # 2. Warmup 余弦退火 def warmup_cosine(epoch, warmup_epochs5, total_epochs100): if epoch warmup_epochs: return (epoch 1) / warmup_epochs else: progress (epoch - warmup_epochs) / (total_epochs - warmup_epochs) return 0.5 * (1 math.cos(math.pi * progress)) scheduler_warmup LambdaLR(optimizer, lr_lambdawarmup_cosine) # 3. 分段常数 def piecewise_constant(epoch): if epoch 10: return 1.0 # lr 0.1 * 1.0 0.1 elif epoch 20: return 0.5 # lr 0.1 * 0.5 0.05 elif epoch 30: return 0.25 # lr 0.1 * 0.25 0.025 else: return 0.1 # lr 0.1 * 0.1 0.01 scheduler_piece LambdaLR(optimizer, lr_lambdapiecewise_constant) # 4. 多参数组的不同调度 optimizer_multi torch.optim.SGD([ {params: model.weight, lr: 0.1}, {params: model.bias, lr: 0.01}, ]) def weight_lambda(epoch): return 0.95 ** epoch def bias_lambda(epoch): return 1.0 # bias 学习率不变 scheduler_multi LambdaLR(optimizer_multi, lr_lambda[weight_lambda, bias_lambda])方案三手动调整学习率def manual_lr_schedule(optimizer, epoch): 手动调整学习率 if epoch 10: lr 0.1 elif epoch 20: lr 0.01 elif epoch 30: lr 0.001 else: lr 0.0001 for param_group in optimizer.param_groups: param_group[lr] lr return lr for epoch in range(50): current_lr manual_lr_schedule(optimizer, epoch) print(fEpoch {epoch}, LR: {current_lr})方案四使用 OneCycleLR超级收敛from torch.optim.lr_scheduler import OneCycleLR optimizer torch.optim.SGD(model.parameters(), lr0.01) scheduler OneCycleLR( optimizer, max_lr0.1, total_steps1000, pct_start0.3, anneal_strategycos, ) # OneCycleLR 在每个 batch 调用 step() for epoch in range(10): for batch in dataloader: loss model(batch) loss.backward() optimizer.step() optimizer.zero_grad() scheduler.step() # 每个 batch 调用完整修复代码 完整的 PyTorch 学习率调度方案 涵盖多种调度策略、断点续训、多参数组、自定义调度 import torch import torch.nn as nn  import torch.optim as optim from torch.optim.lr_scheduler import ( StepLR, MultiStepLR, ExponentialLR, CosineAnnealingLR, CosineAnnealingWarmRestarts, ReduceLROnPlateau, LambdaLR, OneCycleLR ) from torch.utils.data import DataLoader, TensorDataset import math import os import copy class LRSchedulerManager: 学习率调度管理器支持多种调度策略和断点续训 def __init__(self, optimizer, strategycosine, **kwargs): self.optimizer optimizer self.strategy strategy self.kwargs kwargs self.scheduler self._create_scheduler() def _create_scheduler(self): opt self.optimizer kw self.kwargs if self.strategy step: return StepLR(opt, step_sizekw.get(step_size, 10), gammakw.get(gamma, 0.5)) elif self.strategy multistep: return MultiStepLR(opt, milestoneskw.get(milestones, [30, 60, 90]), gammakw.get(gamma, 0.1)) elif self.strategy exp: return ExponentialLR(opt, gammakw.get(gamma, 0.95)) elif self.strategy cosine: return CosineAnnealingLR(opt, T_maxkw.get(T_max, 50), eta_minkw.get(eta_min, 1e-6)) elif self.strategy cosine_warmup: warmup_epochs kw.get(warmup_epochs, 5) total_epochs kw.get(total_epochs, 100) eta_min kw.get(eta_min, 1e-6) base_lr opt.param_groups[0][lr] def warmup_cosine_fn(epoch): if epoch warmup_epochs: return (epoch 1) / warmup_epochs else: progress (epoch - warmup_epochs) / (total_epochs - warmup_epochs) cosine_factor 0.5 * (1 math.cos(math.pi * progress)) return cosine_factor * (1 - eta_min / base_lr) eta_min / base_lr return LambdaLR(opt, lr_lambdawarmup_cosine_fn) elif self.strategy plateau: return ReduceLROnPlateau(opt, modekw.get(mode, min), factorkw.get(factor, 0.5), patiencekw.get(patience, 5), min_lrkw.get(min_lr, 1e-6)) elif self.strategy onecycle: return OneCycleLR(opt, max_lrkw.get(max_lr, 0.1), total_stepskw.get(total_steps, 1000), pct_startkw.get(pct_start, 0.3), anneal_strategykw.get(anneal_strategy, cos)) else: raise ValueError(f未知策略: {self.strategy}) def step(self, metricNone): if self.strategy plateau: if metric is not None: self.scheduler.step(metric) else: self.scheduler.step() def get_last_lr(self): return [group[lr] for group in self.optimizer.param_groups] def state_dict(self): return self.scheduler.state_dict() def load_state_dict(self, state_dict): self.scheduler.load_state_dict(state_dict) def plot_lr_schedule(self, num_epochs): original_state copy.deepcopy(self.scheduler.state_dict()) original_lrs [group[lr] for group in self.optimizer.param_groups] lrs [] for epoch in range(num_epochs): lrs.append(self.get_last_lr()[0]) if self.strategy plateau: self.step(metric0.5) else: self.step() self.scheduler.load_state_dict(original_state) for group, lr in zip(self.optimizer.param_groups, original_lrs): group[lr] lr return lrs class Trainer: 集成学习率调度的训练器 def __init__(self, model, optimizer, criterion, scheduler_manager, devicecpu): self.model model self.optimizer optimizer self.criterion criterion self.scheduler scheduler_manager self.device torch.device(device) self.model.to(self.device) self.train_losses [] self.val_losses [] self.learning_rates [] def train_epoch(self, dataloader): self.model.train() total_loss 0 num_batches 0 for batch_idx, (data, target) in enumerate(dataloader): data, target data.to(self.device), target.to(self.device) self.optimizer.zero_grad() output self.model(data) loss self.criterion(output, target) loss.backward() self.optimizer.step() if self.scheduler.strategy onecycle: self.scheduler.step() total_loss loss.item() num_batches 1 return total_loss / num_batches def validate(self, dataloader): self.model.eval() total_loss 0 num_batches 0 with torch.no_grad(): for data, target in dataloader: data, target data.to(self.device), target.to(self.device) output self.model(data) loss self.criterion(output, target) total_loss loss.item() num_batches 1 return total_loss / num_batches def fit(self, train_loader, val_loader, num_epochs, save_dirNone): print(f{Epoch:6} | {Train Loss:12} | {Val Loss:12} | {LR:12}) print(- * 55) for epoch in range(num_epochs): current_lr self.scheduler.get_last_lr()[0] self.learning_rates.append(current_lr) train_loss self.train_epoch(train_loader) self.train_losses.append(train_loss) val_loss self.validate(val_loader) self.val_losses.append(val_loss) print(f{epoch:6d} | {train_loss:12.6f} | {val_loss:12.6f} | {current_lr:12.8f}) if self.scheduler.strategy plateau: self.scheduler.step(metricval_loss) elif self.scheduler.strategy ! onecycle: self.scheduler.step() if save_dir and (epoch 1) % 10 0: self.save_checkpoint(save_dir, epoch) print(- * 55) def save_checkpoint(self, save_dir, epoch): os.makedirs(save_dir, exist_okTrue) checkpoint { epoch: epoch, model_state_dict: self.model.state_dict(), optimizer_state_dict: self.optimizer.state_dict(), scheduler_state_dict: self.scheduler.state_dict(), train_losses: self.train_losses, val_losses: self.val_losses, learning_rates: self.learning_rates, } path os.path.join(save_dir, fcheckpoint_epoch_{epoch}.pt) torch.save(checkpoint, path) print(f 检查点已保存: {path}) def load_checkpoint(self, path): checkpoint torch.load(path) self.model.load_state_dict(checkpoint[model_state_dict]) self.optimizer.load_state_dict(checkpoint[optimizer_state_dict]) self.scheduler.load_state_dict(checkpoint[scheduler_state_dict]) self.train_losses checkpoint.get(train_losses, []) self.val_losses checkpoint.get(val_losses, []) self.learning_rates checkpoint.get(learning_rates, []) print(f检查点已加载: {path}, 从 epoch {checkpoint[epoch]} 继续) def create_synthetic_data(n_samples1000, n_features10, n_classes3): torch.manual_seed(42) X torch.randn(n_samples, n_features) W torch.randn(n_features, n_classes) y (X W).argmax(dim1) dataset TensorDataset(X, y) train_size int(0.8 * n_samples) val_size n_samples - train_size train_dataset, val_dataset torch.utils.data.random_split(dataset, [train_size, val_size]) return train_dataset, val_dataset def demo_step_lr(): print(\n * 60) print(示例 1: StepLR (固定步长衰减)) print( * 60) model nn.Linear(10, 3) optimizer optim.SGD(model.parameters(), lr0.1, momentum0.9) scheduler LRSchedulerManager(optimizer, strategystep, step_size5, gamma0.5) lrs scheduler.plot_lr_schedule(num_epochs20) for epoch, lr in enumerate(lrs): print(f Epoch {epoch:3d}: LR {lr:.8f}) def demo_cosine_warmup(): print(\n * 60) print(示例 2: Warmup Cosine Annealing) print( * 60) model nn.Linear(10, 3) optimizer optim.SGD(model.parameters(), lr0.1) scheduler LRSchedulerManager(optimizer, strategycosine_warmup, warmup_epochs5, total_epochs50, eta_min1e-5) lrs scheduler.plot_lr_schedule(num_epochs50) for epoch in [0, 1, 4, 5, 10, 25, 40, 49]: print(f Epoch {epoch:3d}: LR {lrs[epoch]:.8f}) def demo_plateau(): print(\n * 60) print(示例 3: ReduceLROnPlateau (自适应衰减)) print( * 60) model nn.Linear(10, 3) optimizer optim.SGD(model.parameters(), lr0.1) scheduler LRSchedulerManager(optimizer, strategyplateau, factor0.5, patience3, min_lr1e-5) val_losses [0.5, 0.4, 0.35, 0.35, 0.35, 0.35, 0.3, 0.3, 0.3, 0.3, 0.25, 0.25, 0.25, 0.25, 0.2] for epoch, val_loss in enumerate(val_losses): current_lr scheduler.get_last_lr()[0] print(f Epoch {epoch:3d}: Val Loss {val_loss:.4f}, LR {current_lr:.8f}) scheduler.step(metricval_loss) def demo_full_training(): print(\n * 60) print(示例 4: 完整训练流程) print( * 60) train_dataset, val_dataset create_synthetic_data() train_loader DataLoader(train_dataset, batch_size32, shuffleTrue) val_loader DataLoader(val_dataset, batch_size32) model nn.Sequential(nn.Linear(10, 64), nn.ReLU(), nn.Linear(64, 32), nn.ReLU(), nn.Linear(32, 3)) optimizer optim.Adam(model.parameters(), lr0.01) scheduler LRSchedulerManager(optimizer, strategycosine_warmup, warmup_epochs3, total_epochs20, eta_min1e-5) trainer Trainer(modelmodel, optimizeroptimizer, criterionnn.CrossEntropyLoss(), scheduler_managerscheduler) trainer.fit(train_loadertrain_loader, val_loaderval_loader, num_epochs20, save_dir./checkpoints) def demo_multi_param_groups(): print(\n * 60) print(示例 5: 多参数组的不同学习率) print( * 60) model nn.Sequential(nn.Linear(10, 64), nn.ReLU(), nn.Linear(64, 3)) optimizer optim.SGD([ {params: [p for n, p in model.named_parameters() if weight in n], lr: 0.1}, {params: [p for n, p in model.named_parameters() if bias in n], lr: 0.01}, ], momentum0.9) def weight_lr_lambda(epoch): return 0.95 ** epoch def bias_lr_lambda(epoch): return 1.0 scheduler LambdaLR(optimizer, lr_lambda[weight_lr_lambda, bias_lr_lambda]) for epoch in range(10): lrs [group[lr] for group in optimizer.param_groups] print(f Epoch {epoch:3d}: Weight LR {lrs[0]:.8f}, Bias LR {lrs[1]:.8f}) scheduler.step() def demo_custom_lambda(): print(\n * 60) print(示例 6: 自定义 Lambda 调度) print( * 60) model nn.Linear(10, 3) optimizer optim.SGD(model.parameters(), lr0.1) def custom_schedule(epoch, warmup5, cosine_start5, cosine_end45, total50): if epoch warmup: return (epoch 1) / warmup elif epoch cosine_end: progress (epoch - cosine_start) / (cosine_end - cosine_start) return 0.5 * (1 math.cos(math.pi * progress)) else: return 0.01 scheduler LambdaLR(optimizer, lr_lambdacustom_schedule) for epoch in range(50): lr optimizer.param_groups[0][lr] if epoch % 5 0: print(f Epoch {epoch:3d}: LR {lr:.8f}) scheduler.step() if __name__ __main__: demo_step_lr() demo_cosine_warmup() demo_plateau() demo_full_training() demo_multi_param_groups() demo_custom_lambda() print(\n所有示例执行完毕)常见陷阱与注意事项1.step()调用顺序# 正确PyTorch 1.1 for epoch in range(epochs): for batch in dataloader: optimizer.step() scheduler.step() # epoch 结束后调用 # OneCycleLR 例外每个 batch 调用 for epoch in range(epochs): for batch in dataloader: optimizer.step() scheduler.step() # batch 级别2. ReduceLROnPlateau 的特殊用法# ReduceLROnPlateau 需要传入指标 scheduler ReduceLROnPlateau(optimizer, modemin) for epoch in range(epochs): train_loss train() val_loss validate() scheduler.step(val_loss) # 传入验证损失3. 断点续训必须保存调度器状态# 保存 checkpoint { model: model.state_dict(), optimizer: optimizer.state_dict(), scheduler: scheduler.state_dict(), # 必须保存 epoch: epoch, } torch.save(checkpoint, checkpoint.pt) # 加载 ckpt torch.load(checkpoint.pt) model.load_state_dict(ckpt[model]) optimizer.load_state_dict(ckpt[optimizer]) scheduler.load_state_dict(ckpt[scheduler]) # 恢复调度器状态4. LambdaLR 返回的是乘法因子# LambdaLR 的 lambda 返回值是乘法因子不是最终学习率 # 最终 lr base_lr * lambda(epoch) # 正确返回 1.0 表示保持原始学习率 def correct_lambda(epoch): if epoch 5: return 1.0 # lr 0.1 * 1.0 0.1 else: return 0.5 # lr 0.1 * 0.5 0.055. OneCycleLR 的 total_steps 计算# total_steps num_epochs * batches_per_epoch num_epochs 10 batches_per_epoch len(dataloader) total_steps num_epochs * batches_per_epoch scheduler OneCycleLR(optimizer, max_lr0.1, total_stepstotal_steps) # 如果 total_steps 计算错误调度器会提前结束或报错6. 多调度器组合from torch.optim.lr_scheduler import SequentialLR, ConstantLR # 先 warmup 5 个 epoch再余弦退火 scheduler1 ConstantLR(optimizer, factor0.1, total_iters5) scheduler2 CosineAnnealingLR(optimizer, T_max45) scheduler SequentialLR(optimizer, [scheduler1, scheduler2], milestones[5])总结在 PyTorch 中根据 epoch 动态调整学习率关键要点如下理解调度器本质调度器在每次step()时修改 optimizer 的param_groups中的lr值。正确的调用顺序optimizer.step()在每个 batch 调用scheduler.step()在每个 epoch 结束时调用OneCycleLR 例外。LambdaLR 返回乘法因子最终学习率 初始学习率 × lambda 返回值不是直接返回学习率。断点续训保存调度器状态scheduler.state_dict()和scheduler.load_state_dict()确保学习率调整连续。选择合适的策略StepLR/MultiStepLR 适合简单场景CosineAnnealingLR 适合大多数场景ReduceLROnPlateau 适合不确定收敛情况的场景OneCycleLR 适合追求快速收敛的场景。多参数组独立调度可以为不同参数组设置不同的 lambda 函数实现差异化学习率调整。Warmup 提升稳定性在训练初期使用较小的学习率逐步增加可以避免梯度爆炸和训练不稳定。通过合理选择和配置学习率调度策略可以显著提升模型的训练效果和收敛速度。
