【Bug已解决】How to take the average of the weights of two networks? 解决方案

【Bug已解决】How to take the average of the weights of two networks? 解决方案问题描述在深度学习实践中模型平均Model Averaging是一种广泛使用的正则化和集成技术。其核心思想是将两个或多个独立训练的神经网络的权重进行平均得到一个新的模型。这种技术在以下场景中特别有用SGD 权重平均随机权重平均Stochastic Weight Averaging, SWA通过平均多个检查点的权重来提升泛化性能联邦学习多个客户端的模型权重需要聚合模型集成将多个独立训练的模型合并为一个更稳健的模型EMA指数移动平均维护模型权重的移动平均版本然而许多开发者在尝试对两个网络的权重取平均时会遇到各种错误和意外行为。最常见的问题包括权重形状不匹配、BatchNorm 统计量未正确处理、平均后模型性能下降等。本文将深入分析这些问题并提供完整的解决方案。错误复现错误一直接使用 state_dict 平均导致形状不匹配import torch import torch.nn as nn import copy # 创建两个结构相同但训练状态不同的模型 model_a nn.Sequential( nn.Linear(784, 256), nn.ReLU(), nn.Linear(256, 10) ) model_b nn.Sequential( nn.Linear(784, 256), nn.ReLU(), nn.Linear(256, 10) ) # 错误尝试直接遍历 state_dict 取平均 averaged_state_dict {} for key in model_a.state_dict(): averaged_state_dict[key] (model_a.state_dict()[key] model_b.state_dict()[key]) / 2 # 尝试加载到新模型 model_avg copy.deepcopy(model_a) try: model_avg.load_state_dict(averaged_state_dict) print(成功加载) except Exception as e: print(fError: {e})虽然这段代码在结构相同的模型上可能不会报错但存在隐患state_dict()每次调用都返回新的字典效率低下且容易在更复杂的模型中出错。错误二模型结构不同导致维度不匹配# 两个结构不同的模型 model_a nn.Sequential( nn.Linear(784, 256), nn.ReLU(), nn.Linear(256, 10) ) model_b nn.Sequential( nn.Linear(784, 128), # 不同的隐藏层大小 nn.ReLU(), nn.Linear(128, 10) ) # 错误尝试直接平均不同结构的权重 try: for (name_a, param_a), (name_b, param_b) in zip( model_a.named_parameters(), model_b.named_parameters() ): avg_param (param_a.data param_b.data) / 2 param_a.data avg_param except RuntimeError as e: print(fError: {e}) # 输出: The size of tensor a (256) must match the size of tensor b (128)错误三忽略 BatchNorm 的 running statisticsimport torch import torch.nn as nn # 包含 BatchNorm 的模型 model_a nn.Sequential( nn.Conv2d(3, 64, 3, padding1), nn.BatchNorm2d(64), nn.ReLU(), nn.AdaptiveAvgPool2d(1), nn.Flatten(), nn.Linear(64, 10) ) model_b copy.deepcopy(model_a) # 随机初始化不同的权重 for param in model_b.parameters(): param.data torch.randn_like(param.data) # 错误尝试只平均 parameters忽略 buffersBatchNorm 的 running_mean/var for (name_a, param_a), (name_b, param_b) in zip( model_a.named_parameters(), model_b.named_parameters() ): param_a.data (param_a.data param_b.data) / 2 # 此时 model_a 的 BatchNorm running_mean/var 仍然是 model_a 原来的值 # 没有被平均导致推理时统计量与权重不匹配 print(BatchNorm running_mean:, model_a[1].running_mean[:5]) # 这个值没有被平均错误四平均后模型性能急剧下降# 两个在不同数据分布上训练的模型 model_a nn.Linear(100, 10) # 在分布A上训练 model_b nn.Linear(100, 10) # 在分布B上训练 # 简单平均 for (name_a, param_a), (name_b, param_b) in zip( model_a.named_parameters(), model_b.named_parameters() ): param_a.data (param_a.data param_b.data) / 2 # 测试发现性能远低于任一原始模型 # 原因两个模型可能处于损失曲面的不同区域简单平均可能落在不好的位置根因分析1. state_dict 与 named_parameters 的区别PyTorch 模型的权重分为两类Parameters可训练的权重通过model.parameters()获取Buffers不可训练但需要保存的状态如 BatchNorm 的running_mean、running_var通过model.buffers()获取state_dict()包含两者而named_parameters()只包含前者。只平均 parameters 而忽略 buffers 是最常见的错误。2. 权重空间的非凸性神经网络的损失函数是非凸的两个独立训练的模型可能处于损失曲面的不同碗中。简单线性平均两个模型的权重可能落在一个高损失区域导致性能下降。这就是为什么 SWA 要求模型来自同一个训练轨迹的不同检查点。3. BatchNorm 统计量的特殊性BatchNorm 层的running_mean和running_var是在训练过程中通过指数移动平均更新的它们与权重紧密耦合。如果只平均权重而不更新这些统计量推理时会出现分布不匹配的问题。4. 模型结构验证缺失在平均权重之前必须验证两个模型的结构完全相同层名、参数形状、buffer 形状。缺少这个验证步骤会导致运行时错误。解决方案方案一使用 state_dict 进行完整平均推荐这是最安全、最完整的方法同时处理 parameters 和 buffersimport torch import torch.nn as nn import copy from typing import List, Dict, Optional def average_state_dicts(state_dicts: List[Dict[str, torch.Tensor]]) - Dict[str, torch.Tensor]: 平均多个 state_dict Args: state_dicts: state_dict 列表 Returns: 平均后的 state_dict if not state_dicts: raise ValueError(state_dicts 列表不能为空) # 验证所有 state_dict 的键一致 base_keys set(state_dicts[0].keys()) for i, sd in enumerate(state_dicts[1:], 1): if set(sd.keys()) ! base_keys: diff_keys base_keys.symmetric_difference(set(sd.keys())) raise ValueError( fstate_dict {i} 的键与 state_dict 0 不匹配。 f差异键: {diff_keys} ) # 验证所有张量形状一致 for key in base_keys: base_shape state_dicts[0][key].shape for i, sd in enumerate(state_dicts[1:], 1): if sd[key].shape ! base_shape: raise ValueError( f键 {key} 在 state_dict {i} 中的形状 {sd[key].shape} f与 state_dict 0 的形状 {base_shape} 不匹配 ) # 计算平均 averaged {} for key in base_keys: # 使用浮点数进行平均 tensors [sd[key].float() for sd in state_dicts] averaged[key] torch.stack(tensors).mean(dim0) # 恢复原始 dtype original_dtype state_dicts[0][key].dtype averaged[key] averaged[key].to(original_dtype) return averaged def average_models(models: List[nn.Module], weights: Optional[List[float]] None) - nn.Module: 平均多个模型的权重 Args: models: 模型列表 weights: 各模型的权重加权平均None 表示等权平均 Returns: 权重平均后的新模型基于第一个模型的结构 if not models: raise ValueError(模型列表不能为空) if weights is None: weights [1.0 / len(models)] * len(models) else: assert len(weights) len(models), 权重数量必须与模型数量一致 total sum(weights) weights [w / total for w in weights] # 验证模型结构一致 base_model models[0] base_state_dict base_model.state_dict() base_keys set(base_state_dict.keys()) for i, model in enumerate(models[1:], 1): model_keys set(model.state_dict().keys()) if model_keys ! base_keys: raise ValueError( f模型 {i} 的结构与模型 0 不匹配。 f差异: {base_keys.symmetric_difference(model_keys)} ) # 创建新模型 averaged_model copy.deepcopy(base_model) # 加权平均 state_dict new_state_dict {} for key in base_state_dict: # 加权平均 weighted_sum sum( w * model.state_dict()[key].float() for w, model in zip(weights, models) ) original_dtype base_state_dict[key].dtype new_state_dict[key] weighted_sum.to(original_dtype) averaged_model.load_state_dict(new_state_dict) return averaged_model方案二使用 SWAStochastic Weight AveragingPyTorch 内置了 SWA 支持这是官方推荐的权重平均方法import torch import torch.nn as nn from torch.optim.swa_utils import AveragedModel, SWALR from torch.utils.data import DataLoader def train_with_swa(): 使用 PyTorch 内置 SWA 进行训练 # 创建模型 model nn.Sequential( nn.Conv2d(3, 64, 3, padding1), nn.BatchNorm2d(64), nn.ReLU(), nn.MaxPool2d(2), nn.Conv2d(64, 128, 3, padding1), nn.BatchNorm2d(128), nn.ReLU(), nn.AdaptiveAvgPool2d(1), nn.Flatten(), nn.Linear(128, 10) ) # 创建 SWA 模型自动维护权重的移动平均 swa_model AveragedModel(model) # 基础优化器 optimizer torch.optim.SGD(model.parameters(), lr0.1, momentum0.9, weight_decay1e-4) # SWA 学习率调度器 swa_scheduler SWALR(optimizer, swa_lr0.05) # 模拟训练数据 num_epochs 100 swa_start 75 # 从第75个epoch开始SWA for epoch in range(num_epochs): # 模拟一个 batch inputs torch.randn(32, 3, 32, 32) targets torch.randint(0, 10, (32,)) # 前向传播 outputs model(inputs) loss nn.CrossEntropyLoss()(outputs, targets) # 反向传播 optimizer.zero_grad() loss.backward() optimizer.step() # SWA 阶段 if epoch swa_start: swa_model.update_parameters(model) # 更新SWA平均 swa_scheduler.step() else: # 正常学习率调度 pass # 更新 BatchNorm 统计量 torch.optim.swa_utils.update_bn(DataLoader( [(torch.randn(32, 3, 32, 32), torch.randint(0, 10, (32,)))], ), swa_model, devicecpu) print(SWA 训练完成) print(fSWA 模型参数数量: {sum(p.numel() for p in swa_model.parameters())}) return swa_model完整修复代码下面是一个完整的、可运行的权重平均实现包含模型验证、加权平均、BatchNorm 更新和性能评估import torch import torch.nn as nn import torch.nn.functional as F import copy from typing import List, Optional, Tuple import warnings class ModelWeightAverager: 神经网络权重平均器 支持功能 - 等权平均 - 加权平均 - 指数移动平均EMA - BatchNorm 统计量更新 - 模型结构验证 def __init__(self, base_model: nn.Module): Args: base_model: 基础模型用于确定结构 self.base_model base_model self.averaged_model copy.deepcopy(base_model) self.base_state_dict_keys set(base_model.state_dict().keys()) self.num_averaged 0 self.ema_decay None def validate_model(self, model: nn.Module) - bool: 验证模型结构是否与基础模型一致 model_keys set(model.state_dict().keys()) if model_keys ! self.base_state_dict_keys: diff self.base_state_dict_keys.symmetric_difference(model_keys) raise ValueError( f模型结构不匹配。差异键: {diff} ) # 验证形状 for key in self.base_state_dict_keys: base_shape self.base_model.state_dict()[key].shape model_shape model.state_dict()[key].shape if base_shape ! model_shape: raise ValueError( f键 {key} 形状不匹配: {base_shape} vs {model_shape} ) return True def average_two_models( self, model_a: nn.Module, model_b: nn.Module, alpha: float 0.5 ) - nn.Module: 平均两个模型的权重 Args: model_a: 模型A model_b: 模型B alpha: 模型A的权重0-11-alpha 为模型B的权重 Returns: 平均后的模型 assert 0 alpha 1, alpha 必须在 [0, 1] 范围内 self.validate_model(model_a) self.validate_model(model_b) result_model copy.deepcopy(model_a) result_state_dict {} for key in self.base_state_dict_keys: tensor_a model_a.state_dict()[key].float() tensor_b model_b.state_dict()[key].float() averaged alpha * tensor_a (1 - alpha) * tensor_b result_state_dict[key] averaged.to(model_a.state_dict()[key].dtype) result_model.load_state_dict(result_state_dict) return result_model def average_multiple_models( self, models: List[nn.Module], weights: Optional[List[float]] None ) - nn.Module: 平均多个模型的权重 Args: models: 模型列表 weights: 各模型权重None 表示等权 Returns: 平均后的模型 if not models: raise ValueError(模型列表不能为空) for model in models: self.validate_model(model) if weights is None: weights [1.0 / len(models)] * len(models) else: assert len(weights) len(models) total sum(weights) weights [w / total for w in weights] result_model copy.deepcopy(models[0]) result_state_dict {} for key in self.base_state_dict_keys: weighted_sum sum( w * model.state_dict()[key].float() for w, model in zip(weights, models) ) result_state_dict[key] weighted_sum.to(models[0].state_dict()[key].dtype) result_model.load_state_dict(result_state_dict) self.num_averaged len(models) return result_model def update_ema(self, model: nn.Module, decay: float 0.999): 更新指数移动平均EMA Args: model: 当前模型 decay: EMA 衰减率 self.validate_model(model) self.ema_decay decay ema_state_dict self.averaged_model.state_dict() model_state_dict model.state_dict() for key in self.base_state_dict_keys: ema_tensor ema_state_dict[key].float() model_tensor model_state_dict[key].float() ema_tensor decay * ema_tensor (1 - decay) * model_tensor ema_state_dict[key] ema_tensor.to(ema_state_dict[key].dtype) self.averaged_model.load_state_dict(ema_state_dict) self.num_averaged 1 def update_batchnorm_stats(self, model: nn.Module, dataloader, devicecpu): 更新 BatchNorm 统计量 在权重平均后BatchNorm 的 running_mean/running_var 可能不准确。 需要用新数据重新计算。 Args: model: 需要更新BN的模型 dataloader: 数据加载器 device: 计算设备 model.train() model.to(device) with torch.no_grad(): for inputs, _ in dataloader: inputs inputs.to(device) model(inputs) # 前向传播会更新BN统计量 model.eval() def get_averaged_model(self) - nn.Module: 获取平均后的模型 return self.averaged_model # 完整使用示例 class SimpleCNN(nn.Module): 简单的 CNN 模型用于演示 def __init__(self, num_classes10): super().__init__() self.features nn.Sequential( nn.Conv2d(3, 32, 3, padding1), nn.BatchNorm2d(32), nn.ReLU(inplaceTrue), nn.MaxPool2d(2), nn.Conv2d(32, 64, 3, padding1), nn.BatchNorm2d(64), nn.ReLU(inplaceTrue), nn.MaxPool2d(2), nn.Conv2d(64, 128, 3, padding1), nn.BatchNorm2d(128), nn.ReLU(inplaceTrue), nn.AdaptiveAvgPool2d(1), ) self.classifier nn.Linear(128, num_classes) def forward(self, x): x self.features(x) x x.view(x.size(0), -1) x self.classifier(x) return x def train_model(model, train_loader, num_epochs5, lr0.001, devicecpu): 训练一个模型 model model.to(device) optimizer torch.optim.Adam(model.parameters(), lrlr) criterion nn.CrossEntropyLoss() model.train() for epoch in range(num_epochs): total_loss 0 correct 0 total 0 for inputs, targets in train_loader: inputs, targets inputs.to(device), targets.to(device) optimizer.zero_grad() outputs model(inputs) loss criterion(outputs, targets) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.max(1) correct predicted.eq(targets).sum().item() total targets.size(0) print(f Epoch {epoch1}/{num_epochs} - Loss: {total_loss/len(train_loader):.4f} f- Acc: {100.*correct/total:.2f}%) return model def evaluate_model(model, test_loader, devicecpu): 评估模型 model model.to(device) model.eval() correct 0 total 0 with torch.no_grad(): for inputs, targets in test_loader: inputs, targets inputs.to(device), targets.to(device) outputs model(inputs) _, predicted outputs.max(1) correct predicted.eq(targets).sum().item() total targets.size(0) return 100. * correct / total def main(): 完整演示 print( * 60) print(神经网络权重平均完整示例) print( * 60) device cpu # 创建模拟数据 print(\n[1] 创建模拟数据集...) from torch.utils.data import TensorDataset, DataLoader # 训练数据 train_x torch.randn(1000, 3, 32, 32) train_y torch.randint(0, 10, (1000,)) train_dataset TensorDataset(train_x, train_y) train_loader DataLoader(train_dataset, batch_size32, shuffleTrue) # 测试数据 test_x torch.randn(200, 3, 32, 32) test_y torch.randint(0, 10, (200,)) test_dataset TensorDataset(test_x, test_y) test_loader DataLoader(test_dataset, batch_size32, shuffleFalse) # 训练两个独立模型 print(\n[2] 训练模型 A...) model_a SimpleCNN(num_classes10) model_a train_model(model_a, train_loader, num_epochs3, devicedevice) print(\n[3] 训练模型 B...) model_b SimpleCNN(num_classes10) model_b train_model(model_b, train_loader, num_epochs3, devicedevice) # 评估原始模型 print(\n[4] 评估原始模型:) acc_a evaluate_model(model_a, test_loader, device) acc_b evaluate_model(model_b, test_loader, device) print(f 模型 A 准确率: {acc_a:.2f}%) print(f 模型 B 准确率: {acc_b:.2f}%) # 使用权重平均器 print(\n[5] 权重平均...) averager ModelWeightAverager(model_a) # 等权平均 avg_model averager.average_two_models(model_a, model_b, alpha0.5) acc_avg evaluate_model(avg_model, test_loader, device) print(f 等权平均 (0.5/0.5) 准确率: {acc_avg:.2f}%) # 加权平均 weighted_avg averager.average_two_models(model_a, model_b, alpha0.7) acc_weighted evaluate_model(weighted_avg, test_loader, device) print(f 加权平均 (0.7/0.3) 准确率: {acc_weighted:.2f}%) # 多模型平均 print(\n[6] 训练第三个模型并三模型平均...) model_c SimpleCNN(num_classes10) model_c train_model(model_c, train_loader, num_epochs3, devicedevice) acc_c evaluate_model(model_c, test_loader, device) print(f 模型 C 准确率: {acc_c:.2f}%) multi_avg averager.average_multiple_models([model_a, model_b, model_c]) acc_multi evaluate_model(multi_avg, test_loader, device) print(f 三模型平均准确率: {acc_multi:.2f}%) # EMA 演示 print(\n[7] EMA (指数移动平均) 演示...) ema_averager ModelWeightAverager(model_a) for i in range(5): # 模拟训练过程中的模型更新 temp_model copy.deepcopy(model_a) for param in temp_model.parameters(): param.data torch.randn_like(param.data) * 0.01 ema_averager.update_ema(temp_model, decay0.9) ema_model ema_averager.get_averaged_model() acc_ema evaluate_model(ema_model, test_loader, device) print(f EMA 模型准确率: {acc_ema:.2f}%) # 更新 BatchNorm 统计量 print(\n[8] 更新 BatchNorm 统计量...) averager.update_batchnorm_stats(avg_model, train_loader, device) acc_bn_updated evaluate_model(avg_model, test_loader, device) print(f BN更新后准确率: {acc_bn_updated:.2f}%) # 汇总 print(\n * 60) print(结果汇总:) print(f 模型 A: {acc_a:.2f}%) print(f 模型 B: {acc_b:.2f}%) print(f 模型 C: {acc_c:.2f}%) print(f 等权平均(A,B): {acc_avg:.2f}%) print(f 加权平均(A,B): {acc_weighted:.2f}%) print(f 三模型平均: {acc_multi:.2f}%) print(f EMA: {acc_ema:.2f}%) print(f BN更新后: {acc_bn_updated:.2f}%) print( * 60) if __name__ __main__: main()常见陷阱与注意事项1. 必须同时平均 parameters 和 buffers最常见的错误是只平均named_parameters()而忘记named_buffers()。BatchNorm 的running_mean、running_var和num_batches_tracked都是 buffers必须一起处理。# 错误只处理 parameters for (name, param) in model.named_parameters(): param.data (param_a.data param_b.data) / 2 # 正确使用 state_dict 同时处理 parameters 和 buffers for key in model.state_dict(): model.state_dict()[key] (sd_a[key] sd_b[key]) / 22. 权重平均的前提条件权重平均只在以下条件下有效两个模型来自同一个训练轨迹的不同检查点如 SWA两个模型在相同数据上用不同超参数训练两个模型使用相同架构和相同初始化种子如果两个模型完全独立训练不同初始化、不同数据简单平均通常不会带来性能提升。3. BatchNorm 更新是必须的权重平均后BatchNorm 的统计量与新的权重不匹配。必须用训练数据重新前向传播来更新running_mean和running_var。# 必须在 train 模式下前向传播来更新BN model.train() with torch.no_grad(): for inputs, _ in dataloader: model(inputs) model.eval()4. 数据类型一致性某些 buffer如num_batches_tracked是整数类型直接平均会丢失精度。需要先转为浮点数平均后再转回。5. 加权平均的权重选择在 SWA 中通常使用等权平均。但在联邦学习中不同客户端的数据量不同需要按数据量加权。权重选择应根据具体场景调整。6. 模型处于 eval 模式平均权重后确保将模型切换到eval()模式否则 Dropout 和 BatchNorm 的行为不正确。7. 深拷贝的重要性在创建平均模型时必须使用copy.deepcopy()否则修改平均模型会影响原始模型。8. GPU 内存管理在 GPU 上平均大型模型时注意内存使用。可以逐层平均而不是一次性加载所有 state_dict。总结神经网络权重平均是一种简单但强大的技术本文详细介绍了在 PyTorch 中正确实现权重平均的完整方案使用 state_dict 而非 named_parameters确保同时处理可训练参数和不可训练缓冲区如 BatchNorm 统计量。模型结构验证在平均前验证两个模型的键和形状完全一致。加权平均支持支持等权和加权平均适用于不同场景。EMA 支持实现指数移动平均适用于训练过程中的在线权重平均。BatchNorm 更新权重平均后必须重新计算 BatchNorm 统计量。PyTorch SWA 集成介绍了使用 PyTorch 内置 SWA 工具的方法。关键要点权重平均最适用于同一训练轨迹的不同检查点BatchNorm 统计量的更新是不可省略的步骤数据类型处理需要特别注意整数 buffer深拷贝防止原始模型被意外修改SWA 是 PyTorch 官方推荐的权重平均方案通过本文的实现开发者可以在各种场景下安全地进行模型权重平均包括 SWA 训练、联邦学习、模型集成和 EMA 维护。正确使用权重平均技术可以显著提升模型的泛化性能和训练稳定性。