
简介这是一份面向计算机、人工智能及相关专业学生与开发者的联邦学习实验资源围绕 FedAvg、FedPer、FedRep 与 FedOur 等算法展开对比研究适合课程设计、毕业设计及算法入门进阶使用。资源包共 43 个文件包含 14 个 Python 源码文件、18 张 png 与 2 张 jpg 实验曲线图、5 个 xml 配置及说明文档等压缩包约 631KB结构清晰涵盖模型定义、数据采样、训练与聚合等核心模块。实验一在 Cifar-10 上对比多种算法的准确率与目标损失实验二基于 MedMNIST 测试 10、50、100 等不同客户端数量下的表现实验三使用 Chest X-Ray Images 数据集验证全局模型与本地模型经 Meta-Transfer 后的效果。目前已有 227 人学习配套源码、模型与图片演示齐全代码均经测试运行成功可帮助读者快速复现实验、理解联邦学习流程并在此基础上修改扩展。1. 联邦学习实验从零复现三个实验到底在验证什么你可能已经在 GitHub 上翻过不少联邦学习仓库打开一看动辄依赖七八个自研库、配置文件几十个字段、跑起来先报三个环境错误。而“基于 Python 实现的联邦学习实验三个实验 源代码 模型 图片演示”这类项目核心价值恰恰在于它把联邦学习最关键的三个验证点——非独立同分布数据下的收敛性、通信轮次对精度的影响、以及灾难性遗忘在联邦场景中的表现——用尽量少的依赖、尽量短的代码路径跑通并且给出可视化结果。换句话说它不是教你什么是联邦学习而是让你亲手看到当数据被切分到多个客户端后模型到底发生了什么变化。这篇文章面向两类人一是刚入门联邦学习、想找一个能跑通的最小实验框架的 Python 开发者二是已经了解 FedAvg 基本流程、但需要一套可复现实验来验证自己改进思路的算法工程师。我会按“环境搭建 → 三个实验逐个拆解 → 参数怎么调 → 踩过的坑 → 进阶验证技巧”的顺序把整个实验链路讲清楚。你不需要 GPU 集群一台普通笔记本就能跑完所有实验。2. 环境准备与实验框架搭建从 Python 安装到第一个联邦训练循环2.1 Python 环境与核心依赖的版本选择联邦学习实验对环境的敏感度比普通深度学习项目更高因为涉及多客户端模拟、进程间通信和随机种子控制。我一般会锁定以下版本组合避免因为 NumPy 或 PyTorch 的小版本差异导致聚合结果对不上# 创建独立环境避免污染系统 Python python -m venv fed_env source fed_env/bin/activate # Windows 用 fed_env\Scripts\activate # 核心依赖版本号是血泪经验别随意升级 pip install torch2.1.0 torchvision0.16.0 pip install numpy1.24.3 pip install matplotlib3.7.2 pip install tqdm4.66.1这里重点说三个参数torch2.1.0是因为该版本对torch.utils.data.DataLoader的多 worker 随机种子控制最稳定numpy1.24.3是最后一个不强制要求distutils的版本在部分 Linux 发行版上少一个报错matplotlib3.7.2保证savefig的 dpi 参数行为一致图片演示不会糊。如果你用的是 macOS尤其是 M 系列芯片把torch换成对应支持 MPS 的版本即可但要注意联邦学习实验里客户端数量多、单次计算量小MPS 的启动开销反而可能拖慢整体速度建议先用 CPU 跑通逻辑再考虑加速。2.2 联邦学习实验的最小代码骨架不管三个实验具体做什么联邦学习的训练循环骨架是固定的。下面这段代码是我从多个项目中提炼出的最小可运行版本你可以直接抄import torch import torch.nn as nn import copy import numpy as np class FederatedServer: def __init__(self, model, clients, fraction0.5): self.global_model model self.clients clients self.fraction fraction # 每轮参与训练的客户端比例 def aggregate(self, client_models, client_sizes): FedAvg 聚合按样本量加权平均 total sum(client_sizes) new_state copy.deepcopy(self.global_model.state_dict()) for key in new_state: new_state[key] torch.zeros_like(new_state[key], dtypetorch.float32) for cm, size in zip(client_models, client_sizes): for key in new_state: new_state[key] cm.state_dict()[key] * (size / total) self.global_model.load_state_dict(new_state) def train_round(self): # 按比例随机选客户端 selected np.random.choice( self.clients, sizemax(1, int(len(self.clients) * self.fraction)), replaceFalse ) models, sizes [], [] for client in selected: model copy.deepcopy(self.global_model) model client.local_train(model) models.append(model) sizes.append(len(client.dataset)) self.aggregate(models, sizes) return self.global_model逻辑说明aggregate方法里先把全局模型的每个参数置零再按客户端样本量加权累加。这里有个容易翻车的点——torch.zeros_like必须指定dtypetorch.float32否则如果全局模型里有int类型的 buffer比如 BatchNorm 的num_batches_tracked会直接报类型错误。fraction0.5是参与率控制每轮有多少客户端上传模型这个参数直接决定通信开销和收敛速度的平衡。参数怎么改客户端数量少比如 5 个以下时fraction可以设 1.0让所有客户端都参与客户端多20 个以上时设 0.1 到 0.3 更接近真实场景。local_train里的本地 epoch 数一般设 1 到 5设太大容易导致客户端模型差异过大聚合时反而震荡。3. 三个实验逐个拆解非独立同分布、通信轮次与灾难性遗忘3.1 实验一非独立同分布数据切分与收敛对比第一个实验要验证的是当客户端数据不再独立同分布时FedAvg 的收敛曲线会发生什么变化。常见做法是用 Dirichlet 分布来模拟非独立同分布参数alpha越小数据分布越偏。import numpy as np def dirichlet_split(labels, num_clients, alpha0.5): 按 Dirichlet 分布把数据集切分给多个客户端 num_classes len(np.unique(labels)) client_indices [[] for _ in range(num_clients)] for c in range(num_classes): idx_c np.where(labels c)[0] np.random.shuffle(idx_c) # 为每个客户端生成该类别下的比例 proportions np.random.dirichlet([alpha] * num_clients) # 按比例切分 split_points (np.cumsum(proportions) * len(idx_c)).astype(int)[:-1] for i, idx in enumerate(np.split(idx_c, split_points)): client_indices[i].extend(idx.tolist()) return client_indices逻辑说明对每个类别先用 Dirichlet 分布生成num_clients个比例值再按累计和切分该类别的样本索引。alpha0.5时客户端之间数据分布差异明显但不极端alpha0.1时某些客户端可能只有一两个类别的数据接近极端非独立同分布。参数说明alpha是核心调节参数。我一般会跑三组对比alpha10.0接近独立同分布、alpha0.5中等偏斜、alpha0.1极端偏斜。每组跑 50 轮通信记录全局模型在测试集上的准确率。图片演示里通常会画出三条收敛曲线你会看到alpha0.1的曲线前期上升慢、后期震荡大这就是非独立同分布带来的典型现象。3.2 实验二通信轮次与本地 epoch 的权衡第二个实验要回答一个工程上很实际的问题给定总计算预算是增加通信轮次、减少本地 epoch 好还是减少通信轮次、增加本地 epoch 好def run_experiment(server, rounds50, local_epochs1): acc_history [] for r in range(rounds): server.train_round() acc evaluate(server.global_model, test_loader) acc_history.append(acc) print(fRound {r1}/{rounds}, Acc: {acc:.4f}) return acc_history # 对比两组配置 configs [ {rounds: 100, local_epochs: 1}, {rounds: 20, local_epochs: 5}, ]逻辑说明第一组配置通信 100 轮、每轮本地训练 1 个 epoch第二组通信 20 轮、每轮本地训练 5 个 epoch。总本地计算量相同但通信开销差 5 倍。参数说明local_epochs超过 5 之后客户端模型会严重偏向本地数据聚合时全局模型反而被拖偏这就是文献里常说的“客户端漂移”。我实测下来local_epochs3左右是个比较稳的甜点值。如果你要写论文做对比实验建议至少跑local_epochs为 1、3、5、10 四组画出的曲线能清楚展示拐点。3.3 实验三灾难性遗忘在联邦场景下的表现第三个实验是最容易被忽略但最有价值的当客户端数据分布随时间变化时全局模型会不会出现灾难性遗忘具体做法是让每个客户端的数据按任务顺序到达每完成一个任务就评估模型在所有已见任务上的表现。def continual_federated_experiment(server, task_loaders, rounds_per_task10): task_loaders: 每个任务对应的测试集列表 acc_matrix np.zeros((len(task_loaders), len(task_loaders))) for task_id, train_loader in enumerate(task_loaders): # 当前任务训练 for client in server.clients: client.dataset train_loader # 切换到当前任务数据 for r in range(rounds_per_task): server.train_round() # 评估所有已见任务 for eval_id in range(task_id 1): acc evaluate(server.global_model, task_loaders[eval_id]) acc_matrix[task_id][eval_id] acc return acc_matrix逻辑说明acc_matrix[i][j]表示在第i个任务训练完成后模型在第j个任务上的准确率。如果acc_matrix[2][0]明显低于acc_matrix[0][0]说明发生了灾难性遗忘。参数说明rounds_per_task控制每个任务训练多久设太小模型还没学好当前任务设太大遗忘更严重。我一般设 10 到 20 轮。这个实验的输出矩阵可以直接画成热力图图片演示里通常用颜色深浅表示遗忘程度一眼就能看出哪些任务被“忘得最狠”。4. 避坑与排查联邦学习实验里最容易翻车的五个地方4.1 现象聚合后模型精度突然掉到随机水平原因客户端返回的模型 state_dict 里包含了 BatchNorm 的 running_mean 和 running_var但聚合时没有正确处理这些 buffer导致全局模型的统计量被平均后偏离真实分布。解决要么在聚合时跳过所有非浮点参数要么在客户端本地训练时冻结 BatchNorm 层。我一般直接在模型定义里把 BatchNorm 换成 GroupNorm省去这个麻烦。4.2 现象每轮训练结果波动极大无法复现原因客户端采样、数据切分、DataLoader 的 shuffle 都用了独立的随机源没有统一种子。解决在实验开始处固定所有随机种子包括torch.manual_seed、np.random.seed、random.seed并且给每个客户端的 DataLoader 设置generatortorch.Generator().manual_seed(seed client_id)。4.3 现象非独立同分布实验里某些客户端 loss 始终不下降原因alpha设得太小某个客户端只分到了极少数类别的样本本地模型直接过拟合到这些类别聚合时把全局模型带偏。解决设置客户端最小样本数阈值低于阈值的客户端不参与本轮聚合或者改用更温和的切分方式比如先按类别排序再轮流分配。4.4 现象灾难性遗忘实验里 acc_matrix 全为零原因task_loaders里的测试集没有正确对应到训练时的类别空间模型输出维度对不上评估时直接报错被吞掉。解决确保所有任务的分类头维度一致如果任务类别数不同要么统一到最大类别数要么在评估时只取对应类别的 logits。4.5 现象通信轮次实验跑得特别慢内存占用越来越高原因每轮都把客户端模型深拷贝一份存到列表里没有及时释放Python 的 GC 没跟上。解决聚合完成后手动del client_models并调用torch.cuda.empty_cache()如果用了 GPU或者改用生成器逐轮聚合不保存历史模型。5. 进阶验证用滑动窗口滤波看联邦训练的稳定性跑完三个基础实验后如果你想进一步判断训练过程是否稳定我建议加一个滑动窗口滤波。具体做法是对准确率序列做窗口大小为 5 的移动平均然后观察滤波后的曲线是否单调上升。如果滤波后曲线出现明显下降段说明当前配置下联邦训练不稳定需要调小学习率或降低本地 epoch 数。def moving_average(seq, window5): 滑动窗口滤波用于平滑准确率曲线 if len(seq) window: return seq return np.convolve(seq, np.ones(window)/window, modevalid) # 使用示例 raw_acc run_experiment(server, rounds50, local_epochs3) smooth_acc moving_average(raw_acc, window5)这个技巧在写论文时特别有用原始曲线震荡大审稿人会觉得实验不稳定加一条滤波后的曲线趋势一目了然。窗口大小我一般设 5太小起不到平滑作用太大又会掩盖真实波动。另外验证联邦学习实验是否真正复现成功不要只看最终准确率。我习惯同时检查三个指标全局模型在独立同分布测试集上的准确率、在非独立同分布测试集上的准确率、以及客户端之间模型参数的方差。如果参数方差在训练后期仍然很大说明聚合没有真正把客户端知识融合起来这时候回去检查fraction和local_epochs的搭配。我自己踩过最深的坑是早期为了省时间把local_epochs设成 20结果全局模型在 10 轮后就完全不涨了查了两天才发现是客户端漂移太严重。后来养成习惯任何联邦学习实验先跑local_epochs1的基线确认聚合逻辑没问题再逐步往上加。希望帮到你。本文还有配套的精品资源点击获取