简介本资源是一套基于Python实现的联邦学习实验项目面向人工智能、计算机及相关专业的在校学生、教师与企业研发人员适合作为毕业设计、课程设计或算法入门进阶的参考范例。项目围绕FedAvg、FedPer、FedRep与自研FedOur等方法展开三组对比实验在Cifar-10上比较各算法准确率与目标损失在MedMNIST上测试10、50、100个客户端数量对性能的影响并在Chest X-Ray Images数据集上验证全局模型与本地模型经Meta-Transfer微调后的效果。压缩包共43个文件包含14个Python源码、18张png与2张jpg实验曲线图、5个xml配置及md说明文档整体约631KB代码结构清晰、模块划分明确。目前已有227人学习下载。读者可获得完整可运行的实验代码、模型定义、训练与聚合脚本以及准确率与损失可视化结果便于快速复现实验、理解联邦学习流程并在此基础上进行二次开发。1. 联邦学习实验到底在做什么从一次「模型越训越差」的翻车说起很多人第一次接触联邦学习是被「数据不出本地也能联合建模」这句话吸引的。但真正动手跑一个基于 Python 的联邦学习实验时最常见的翻车不是代码报错而是模型精度越训越低甚至比单机训练差一大截。这背后往往不是算法写错了而是数据分布、聚合策略、通信轮次这几个环节没对齐。这个标题里的「三个实验源代码模型图片演示」本质上就是一套能让你在本地把联邦学习从概念跑到可视化的最小闭环用 Python 搭起客户端-服务端结构模拟多个数据持有方各自训练再通过参数聚合得到一个全局模型。它适合两类人一是想入门联邦学习但不想一上来就啃框架源码的开发者二是需要快速验证某个聚合策略或数据划分方式是否有效的研究型工程师。下面我按自己复现这类实验的路径把三个实验拆开讲清楚包括每一步的代码、参数和那些只有跑过才知道的坑。2. 三个实验的骨架数据划分、本地训练与全局聚合怎么串起来联邦学习实验的核心不是某个高深算法而是把「数据留在本地」这件事用代码表达出来。三个实验通常对应三种典型场景IID 数据下的基准实验、Non-IID 数据下的挑战实验、以及带攻击或异常客户端的鲁棒性实验。要跑通它们先得把骨架搭对。2.1 用 Python 模拟多个客户端的数据划分最直接的做法是用 PyTorch 的Subset把一份数据集切给多个客户端。IID 划分就是随机打乱后均分Non-IID 则按标签或 Dirichlet 分布切。下面这段代码是我常用的划分方式支持两种模式。import numpy as np import torch from torch.utils.data import Subset, DataLoader from torchvision import datasets, transforms def split_data(dataset, num_clients, modeiid, alpha0.5): dataset: 完整训练集 num_clients: 客户端数量 mode: iid 或 noniid alpha: Dirichlet 参数越小越不均衡 if mode iid: indices np.random.permutation(len(dataset)) splits np.array_split(indices, num_clients) else: labels np.array([y for _, y in dataset]) num_classes len(np.unique(labels)) # 为每个客户端生成类别分布 splits [[] 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, chunk in enumerate(np.split(idx_c, split_points)): splits[i].extend(chunk.tolist()) return [Subset(dataset, s) for s in splits] # 使用示例 transform transforms.Compose([transforms.ToTensor()]) train_set datasets.MNIST(root./data, trainTrue, downloadTrue, transformtransform) clients split_data(train_set, num_clients10, modenoniid, alpha0.3)这段代码的关键在alpha参数当alpha0.5时每个客户端拿到的类别分布还比较均匀当alpha0.1时会出现某些客户端只有一两类样本的极端情况这正是 Non-IID 实验要复现的场景。num_clients一般设 10 到 100太少体现不出联邦特性太多则单机模拟会慢。划分完记得检查每个客户端的样本数和类别分布否则后面精度上不去你都不知道是算法问题还是数据问题。2.2 本地训练循环与模型定义每个客户端在本地跑若干轮 SGD只上传模型参数不上传数据。模型可以用简单的 CNN 或 MLP重点是训练循环要能独立运行。import torch.nn as nn import torch.optim as optim class SimpleCNN(nn.Module): def __init__(self, num_classes10): super().__init__() self.conv nn.Sequential( nn.Conv2d(1, 32, 3, padding1), nn.ReLU(), nn.MaxPool2d(2), nn.Conv2d(32, 64, 3, padding1), nn.ReLU(), nn.MaxPool2d(2) ) self.fc nn.Sequential( nn.Linear(64 * 7 * 7, 128), nn.ReLU(), nn.Linear(128, num_classes) ) def forward(self, x): x self.conv(x) x x.view(x.size(0), -1) return self.fc(x) def local_train(model, dataloader, epochs1, lr0.01): model.train() optimizer optim.SGD(model.parameters(), lrlr, momentum0.9) criterion nn.CrossEntropyLoss() for _ in range(epochs): for x, y in dataloader: optimizer.zero_grad() loss criterion(model(x), y) loss.backward() optimizer.step() return model.state_dict()epochs通常设 1 到 5设太大客户端会过拟合本地数据反而拖累全局模型。lr在联邦场景下一般比单机训练小0.01 是常见起点。返回的state_dict就是待聚合的参数注意不要返回整个模型对象否则通信开销会失控。2.3 服务端聚合FedAvg 的实现与参数更新服务端收到各客户端参数后按样本量加权平均这就是 FedAvg 的核心。def fed_avg(global_model, client_states, client_sizes): global_model: 全局模型 client_states: 各客户端 state_dict 列表 client_sizes: 各客户端样本数列表 total sum(client_sizes) new_state {} for key in global_model.state_dict().keys(): new_state[key] sum( client_states[i][key] * (client_sizes[i] / total) for i in range(len(client_states)) ) global_model.load_state_dict(new_state) return global_model加权平均比简单平均更合理因为样本多的客户端对全局贡献应该更大。如果做鲁棒性实验这里可以换成中位数聚合或剔除异常客户端这也是第三个实验的切入点。聚合轮次一般设 50 到 200轮次太少模型没收敛太多则收益递减且通信成本高。每轮结束后在测试集上评估全局模型把准确率曲线画出来就是标题里说的「图片演示」部分。3. 把实验跑起来环境、命令与结果可视化骨架搭好后剩下的是让三个实验真正跑出结果。这一章讲环境配置、运行方式和可视化都是能直接抄的步骤。3.1 Python 环境与依赖安装联邦学习实验对环境的依赖不算重但版本要对齐。我一般用 Python 3.8 到 3.10太新的版本有时和 PyTorch 的某些算子不兼容。# 创建虚拟环境 python -m venv fl_env source fl_env/bin/activate # Windows 用 fl_env\Scripts\activate # 安装核心依赖 pip install torch torchvision numpy matplotlibtorch和torchvision版本要匹配比如 torch 2.0 配 torchvision 0.15。matplotlib用来画准确率曲线和客户端数据分布图。如果要用 LightGBM 做对比实验再装pip install lightgbm但联邦场景下树模型聚合比较麻烦一般还是用神经网络。3.2 三个实验的运行入口与参数配置三个实验可以写在一个主脚本里用命令行参数区分。下面是一个典型的入口。import argparse def main(): parser argparse.ArgumentParser() parser.add_argument(--exp, typestr, defaultiid, choices[iid, noniid, robust]) parser.add_argument(--num_clients, typeint, default10) parser.add_argument(--rounds, typeint, default100) parser.add_argument(--local_epochs, typeint, default1) parser.add_argument(--alpha, typefloat, default0.5) args parser.parse_args() if args.exp iid: run_experiment(modeiid, **vars(args)) elif args.exp noniid: run_experiment(modenoniid, **vars(args)) else: run_experiment(modenoniid, robustTrue, **vars(args)) if __name__ __main__: main()运行命令就是python main.py --exp noniid --alpha 0.1 --rounds 150。rounds在 Non-IID 下要比 IID 多因为数据异构需要更多轮次才能收敛。local_epochs在鲁棒性实验里可以适当加大让恶意客户端的影响更明显。3.3 结果可视化准确率曲线与数据分布图图片演示是这类项目的加分项也是判断实验是否正常的依据。我通常画两张图一张是全局模型准确率随轮次的变化另一张是各客户端的数据类别分布。import matplotlib.pyplot as plt def plot_accuracy(acc_list, titleGlobal Model Accuracy): plt.figure(figsize(8, 5)) plt.plot(range(1, len(acc_list) 1), acc_list, markero, markersize3) plt.xlabel(Communication Round) plt.ylabel(Accuracy (%)) plt.title(title) plt.grid(True, alpha0.3) plt.savefig(accuracy_curve.png, dpi150) plt.close() def plot_client_distribution(client_labels, num_classes10): plt.figure(figsize(10, 5)) for i, labels in enumerate(client_labels): counts np.bincount(labels, minlengthnum_classes) plt.bar(np.arange(num_classes) i * 0.1, counts, width0.1, labelfClient {i}) plt.xlabel(Class) plt.ylabel(Sample Count) plt.title(Client Data Distribution) plt.legend() plt.savefig(client_distribution.png, dpi150) plt.close()准确率曲线如果出现剧烈震荡通常是学习率太大或聚合权重有问题如果一直不上升先检查数据划分是不是把某类样本全分给了同一个客户端。数据分布图能直观看出 Non-IID 程度alpha越小柱子越集中。4. 避坑与排查联邦学习实验里最容易翻车的五件事这一章是我自己踩过的坑按「现象 → 原因 → 解决」写你遇到问题时可以对照排查。4.1 全局模型精度始终低于单机训练现象同样模型和数据联邦训练 100 轮后准确率比单机低 5 到 10 个百分点。原因Non-IID 下各客户端模型漂移太大简单加权平均无法对齐。解决增加通信轮次到 200 以上或改用 FedProx在本地损失里加一项近端项约束模型不要偏离全局太远。4.2 某些客户端准确率极低甚至为 0现象全局模型在测试集上还行但个别客户端本地评估惨不忍睹。原因这些客户端数据量太少或类别太偏聚合时被边缘化。解决检查数据划分保证每个客户端至少有几百条样本或者在聚合时对样本少的客户端做上采样但要注意这会引入偏差。4.3 训练过程中 loss 变成 NaN现象前几轮正常突然 loss 爆炸。原因学习率过大或者某个客户端的梯度异常。解决把本地学习率降到 0.001 到 0.005加梯度裁剪torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)。如果做鲁棒性实验恶意客户端故意发大梯度中位数聚合能缓解。4.4 通信轮次增加但准确率不升反降现象50 轮时准确率最高继续训练反而下降。原因过拟合全局测试集或客户端本地过拟合。解决减少本地 epoch 到 1加早停策略每轮评估后保存最佳模型而不是最后一轮模型。4.5 图片演示里曲线和预期完全相反现象Non-IID 实验的准确率曲线比 IID 还高。原因数据划分时随机种子没固定或者测试集泄漏到了训练集。解决固定np.random.seed(42)和torch.manual_seed(42)确保测试集只在全局评估时使用绝不参与任何客户端训练。5. 进阶技巧用鲁棒性实验验证聚合策略的真实边界三个实验里最有价值的是第三个——鲁棒性实验。它不只是跑通代码而是让你看到联邦学习在真实威胁下的边界。我一般会模拟两类异常客户端一类是标签翻转攻击把本地标签随机打乱另一类是梯度放大攻击上传时把参数乘以一个大系数。然后对比 FedAvg、中位数聚合、Krum 三种策略的表现。具体做法是在聚合前对客户端参数做筛选。中位数聚合的实现如下def median_aggregation(global_model, client_states): new_state {} for key in global_model.state_dict().keys(): stacked torch.stack([state[key] for state in client_states]) new_state[key] torch.median(stacked, dim0).values global_model.load_state_dict(new_state) return global_model中位数聚合对少量恶意客户端有天然抵抗但当恶意比例超过 50% 时会失效。Krum 则是选一个与其他客户端距离最小的参数作为聚合结果适合恶意客户端比例较低的场景。我通常把恶意比例从 10% 逐步加到 40%观察三种策略的准确率拐点。这个拐点就是你在实际部署时能容忍的异常客户端上限。还有一个容易被忽略的技巧在每轮聚合前记录各客户端参数与全局参数的余弦相似度。如果某个客户端持续低于阈值可以直接剔除。这个阈值不用拍脑袋用前 10 轮正常客户端的相似度均值减两倍标准差就能算出来。我试过在 MNIST 和 CIFAR-10 上这个方法能把 20% 的标签翻转攻击影响压到 2% 以内的准确率损失。最后说个习惯每次跑实验前先跑一轮单机训练作为基线把准确率记下来。联邦实验的结果如果比这个基线低太多先别怀疑算法去查数据划分和聚合权重。这个基线就是你的后悔药能省下大量无效调参时间。希望帮到你。本文还有配套的精品资源点击获取