企业官网建设流程全解析

简介：这份压缩包面向希望入门并实践联邦学习分布式训练的开发者与学习者，围绕 MNIST 手写数字识别任务，提供从模型定义、客户端本地训练、服务端聚合到测试评估的完整代码与数据环境，可用于理解 FedAvg 流程、分布式节点协作以及差分隐私在模型更新中的应用。压缩包共 17 个文件，约 56.29MB，主要包含 4 个 py 脚本、4 个 pth 与 2 个 pt 模型权重、1 个 npz 数据文件，以及 MNIST 原始 ubyte 数据和解压子包等，兼顾训练代码、预训练模型与数据读取验证。已有 669 人学习，适合按目录结构快速复现实验。读者可据此搭建多节点本地训练与服务端聚合循环，观察模型更新与隐私噪声对准确率的影响，并借助测试脚本完成收敛评估，从而形成对联邦学习系统组成、训练流程和差分隐私实现方式的直观认识。

1. 联邦学习跑 MNIST：先拆包再跑的原因

MNIST 几乎人人都跑过，但真正按分布式训练的方式把联邦学习跑一遍的人不多——多数教程在单进程里把数据切一切，就算有了十个客户端。这个包补的正是这一步：一份按客户端切好的 MNIST 数据集，配本地训练与聚合的可运行脚本，把「数据不出端、只传模型参数」这条链路走完。适合两类人：想亲手验证 non-IID 到底掉多少点的算法同学，和要把集中式训练改造成联邦流程、打算先在小数据集上预先处理一遍工程问题的工程师。拆包顺序建议按数据划分、本地训练、聚合、调度四段走，跳过第一段后面全是玄学。

2. 数据先切对：non-IID 划分的两种切法与可复现脚本

2.1 IID 切分会让实验结论整体失真

最常见的错误做法是random_split把 60000 张训练图平均分给 100 个客户端，每个客户端 600 张。这样切出来的每一份都包含全部 10 个类，分布几乎一致，FedAvg 在这种数据上跑出来的精度和集中式训练只差零点几个点，于是很容易得出「联邦学习不掉点」的结论。等到换成真实场景，每个端上只有两三个类的样本，精度直接掉十几个点，回头再改架构已经晚了。

所以数据集的价值不在 MNIST 本身，而在于它把「怎么切」这件事固化成脚本，并且带固定随机种子。业界常用的两种 non-IID 构造方式值得都跑一遍：一种是按标签分片（shard），把样本按标签排序后切成2 × 客户端数份，每个客户端随机领 2 份，结果就是每个客户端只有 2 个类，这是最经典的 pathological non-IID；另一种是狄利克雷分布（Dirichlet）切分，用alpha控制倾斜程度，alpha越小越不均匀，alpha=0.1时已经相当极端，alpha=0.5是比较温和的默认值。这两种切法覆盖了「类别缺失」和「类别比例失衡」两类真实问题，先在这上面把模型跑稳，再谈别的。

2.2 按标签分片与 Dirichlet 切分的实现

先看按标签分片。核心是稳定排序加随机打乱分片编号，排序不稳定会导致同样的种子在不同机器上切出不同的结果，这一点在复现实验时非常关键。

import numpy as np from torchvision import datasets def shard_partition(targets, num_clients, shards_per_client=2, seed=0): """按标签排序后切片：每个客户端只拿到 shards_per_client 个类的样本""" rng = np.random.default_rng(seed) num_shards = num_clients * shards_per_client # kind="stable" 保证同类样本内部顺序固定，换机器换 numpy 版本结果一致 sorted_idx = np.argsort(targets, kind="stable") shards = np.array_split(sorted_idx, num_shards) shard_ids = rng.permutation(num_shards) # 打乱分片归属，避免客户端 0 永远拿到 0/1 类 client_idx = {} for c in range(num_clients): picked = shard_ids[c * shards_per_client:(c + 1) * shards_per_client] client_idx[c] = np.concatenate([shards[s] for s in picked]) return client_idx

这里有两个参数容易设错。shards_per_client设成 2 是 FedAvg 原始论文的设定，每个客户端 2 个类、600 张图；设成 1 会让每个客户端只有 1 个类，本地模型会彻底跑偏，聚合后基本不收敛；设成 10 就退化成了 IID。num_clients决定单客户端样本量，60000 除以 100 是 600，再除以 batch size 64 大约 9 到 10 个 iteration，一轮本地训练的计算量很小，这也是很多人把本地 epoch 调到 5 以上的原因，后面避坑那章会讲这样做为什么反而掉点。

再看 Dirichlet 切分。它的思路是对每个类别单独分配比例，天然保证每个客户端的样本量不完全相等，更贴近真实设备。

def dirichlet_partition(targets, num_clients, alpha=0.5, seed=0): """按类别做 Dirichlet 采样，alpha 越小客户端之间越不均衡""" rng = np.random.default_rng(seed) num_classes = int(targets.max()) + 1 client_idx = {c: [] for c in range(num_clients)} for k in range(num_classes): idx_k = np.where(targets == k)[0] rng.shuffle(idx_k) # 先打乱，否则切出来的永远是前几张图 props = rng.dirichlet(np.repeat(alpha, num_clients)) cuts = (np.cumsum(props)[:-1] * len(idx_k)).astype(int) for c, part in enumerate(np.split(idx_k, cuts)): client_idx[c].extend(part.tolist()) return {c: np.array(v, dtype=np.int64) for c, v in client_idx.items()}

alpha是这个函数唯一需要调的参数。alpha=100基本等价于均匀分配，没有实验价值；alpha=0.1会出现大量只有几十张图的客户端，某些客户端的本地训练几乎不起作用，但聚合时它的权重也小，整体还能收敛，只是每轮精度抖得厉害。如果要做「样本量倾斜」的实验，还可以再乘一个 lognormal 系数去裁剪每个客户端的样本数，这属于进阶玩法，先把alpha扫一遍更划算。

2.3 划分结果的自检与保存方式

切完不检查等于没切。至少要看两件事：每个客户端的样本量和它的标签分布。很多掉点问题的根因就是某个客户端只分到 3 张图，或者某个类的样本全部集中在两三个客户端上。

def check_partition(client_idx, targets, num_classes=10): rows = [] for c, idx in client_idx.items(): counts = np.bincount(targets[idx], minlength=num_classes) rows.append((c, len(idx), counts.max(), (counts > 0).sum())) sizes = np.array([r[1] for r in rows]) print(f"客户端数={len(rows)} 样本量 min/mean/max = " f"{sizes.min()}/{sizes.mean():.1f}/{sizes.max()}") print(f"平均覆盖类别数 = {np.mean([r[3] for r in rows]):.2f}") return rows

shard_partition的理想输出是「平均覆盖类别数 = 2.00」，样本量全部等于 600；dirichlet_partition(alpha=0.5)的覆盖类别数一般在 6 到 8 之间，样本量 min 可能掉到 100 以下。如果分片切完覆盖类别数是 10，说明shards_per_client传错了或者根本没走这个函数。

保存建议用np.savez_compressed，把每个客户端的索引数组存成client_0到client_N，另外单独存一份test_idx。注意测试集绝对不能参与划分，有些人为了图省事把 70000 张图一起切，最后用其中一部分当测试集，那评估结果就是彻底的自我欺骗，这是最典型的翻车点之一。

np.savez_compressed("partition_shard100_s2_seed0.npz", **{f"client_{c}": v for c, v in client_idx.items()}, test_idx=np.arange(len(test_set)))

文件名里带上切法、客户端数、shards_per_client和种子，后面跑不同配置时不会互相覆盖。这个习惯看起来琐碎，但当你同时跑alpha=0.1/0.5和 shard 两组实验、又忘了哪个文件对应哪个配置时，后悔药是买不到的。

3. FedAvg 落地：客户端本地训练与服务端加权聚合

3.1 聚合权重为什么必须取样本数而不是客户端数

FedAvg 的聚合公式看起来简单，w_global = Σ (n_k / n) · w_k，但实际代码里最常见的错误是写成w_global = (1/K) · Σ w_k，也就是每个客户端权重相同。在 IID 切分下这两种写法结果几乎一样，一旦换成 Dirichlet 切分，样本量从几十到几千不等，等权平均就等于让只有 30 张图的客户端和 3000 张图的客户端话语权相同，全局模型会被小客户端拖偏，精度可能差 5 个点以上。

权重必须是n_k / n，其中n_k是客户端 k 的样本数，n是本轮参与聚合的样本总数。注意分母是本轮参与客户端的样本和，不是全局训练集总量，因为每轮采样一部分客户端时，未参与者的样本不该进入分母。

3.2 客户端本地训练函数与优化器配置

模型用两层卷积加两层全连接就够了，MNIST 上不加 BatchNorm 也能到 99% 左右，而且能少一个坑——BN 的 running_mean/running_var 在联邦场景下是每个客户端各自的统计量，聚合时平均这些 buffer 在数学上没有意义，后面会专门讲。

import torch, torch.nn as nn class MnistCNN(nn.Module): def __init__(self, num_classes=10): super().__init__() self.features = nn.Sequential( nn.Conv2d(1, 32, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2), # 28 -> 14 nn.Conv2d(32, 64, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2), # 14 -> 7 ) self.classifier = nn.Sequential( nn.Flatten(), nn.Linear(64 * 7 * 7, 128), nn.ReLU(), nn.Linear(128, num_classes), ) def forward(self, x): return self.classifier(self.features(x))

本地训练的写法要注意两点：一是每次client_update前必须load_state_dict拿到最新全局权重，不能让客户端沿用上一轮的本地模型，那是 FedProx 之类的变体才做的事；二是返回的 state_dict 要先.detach().cpu()，否则持有计算图会把显存撑爆，100 个客户端串行跑必然 OOM。

def client_update(global_sd, loader, local_epochs, lr, device, momentum=0.9, weight_decay=5e-4): model = MnistCNN().to(device) model.load_state_dict(global_sd) # 每轮都从全局模型出发，这是 FedAvg 的前提 model.train() opt = torch.optim.SGD(model.parameters(), lr=lr, momentum=momentum, weight_decay=weight_decay) loss_fn = nn.CrossEntropyLoss() for _ in range(local_epochs): for x, y in loader: x, y = x.to(device), y.to(device) opt.zero_grad() loss_fn(model(x), y).backward() opt.step() return {k: v.detach().cpu() for k, v in model.state_dict().items()}

超参方面，lr=0.01、momentum=0.9、batch_size=64、local_epochs=1是 FedAvg 原文附近比较稳的一组，先按这组跑通再调。local_epochs是最容易被误调的参数，很多人觉得「本地多训几轮能少通信几次」，从 1 调到 5、10，结果 non-IID 下精度反而下降，原因在避坑章展开。weight_decay保留 5e-4 有助于缓解客户端漂移，不建议一开始就设 0。

3.3 服务端聚合与整轮闭环

聚合函数要处理两类张量：浮点的权重做加权平均，整型的 buffer 直接取第一个客户端的值。后者最典型的就是 BN 层的num_batches_tracked，如果对它做加权平均，类型会变成 float，load_state_dict时直接报类型不匹配。

def fedavg(global_sd, client_sds, client_sizes): total = float(sum(client_sizes)) new_sd = {} for k, v in global_sd.items(): if v.dtype.is_floating_point: acc = torch.zeros_like(v, dtype=torch.float32) for sd, n in zip(client_sds, client_sizes): acc += sd[k].float() * (n / total) new_sd[k] = acc.to(v.dtype) else: new_sd[k] = client_sds[0][k] # 整型 buffer 不参与平均 return new_sd

整轮循环的结构就是「采样客户端 → 并行或串行本地训练 → 加权聚合 → 全局评估」，frac控制每轮参与的客户端比例，100 个客户端取frac=0.1就是每轮 10 个，这是通信效率和收敛速度之间最直接的旋钮。评估用完整测试集，每轮记一次，别只在最后一轮看结果，因为联邦训练的精度曲线经常是先升后降，最后那轮未必是最好的一轮。

for r in range(num_rounds): picked = rng.choice(num_clients, size=max(1, int(frac * num_clients)), replace=False) sds, sizes = [], [] for c in picked: loader = make_loader(client_idx[c], batch_size=64) sd = client_update(global_sd, loader, local_epochs, lr, device) sds.append(sd); sizes.append(len(client_idx[c])) global_sd = fedavg(global_sd, sds, sizes) acc = evaluate(global_sd, test_loader, device) print(f"round {r:03d} acc={acc:.4f}")

make_loader里记得给每个客户端的 DataLoader 传不同的generator种子，但同一轮内不同客户端之间不要用同一个种子，否则在样本量相同的情况下它们的 batch 顺序完全一致，虽然影响不大，但会让实验的随机性看起来比实际更小。

4. 分布式调度：多进程模拟集群与通信轮次控制

4.1 单机多进程与真实多机的差别

数据集自带的调度脚本一般是两种形态：串行 for 循环，和多进程进程池。串行版本好调试，但 100 个客户端每轮跑一遍，CPU 上大约要几十秒，跑 100 轮就是小一个小时；进程池能把这个时间压到核数量分之一。

需要提前说清楚单机多进程和真实分布式的差距：进程池的「通信」本质是 pickle 序列化，进程间共享同一块内存和磁盘，没有网络延迟、没有带宽瓶颈、没有客户端掉线。所以进程池跑出来的精度曲线是可信的，但通信开销的结论不可信。真实场景里每轮上传下载的参数量、客户端的可用率、网络抖动，都是进程池模拟不出来的。做算法实验用进程池足够，做工程评估必须换成真实的 HTTP/gRPC 传输，而且要加超时和重试。

另一个坑是启动方式。Linux 默认fork，spawn会重新导入主模块。如果集成了 GPU，fork在 CUDA 初始化之后调用会直接报错，所以统一用spawn更省心，代价是每个 worker 要重新 import 一遍依赖。

4.2 用进程池跑一轮联邦

数据通过 initializer 一次性传给各 worker，用uint8传图和标签，比传 float 省四倍带宽。worker 里按索引取子集，避免每个客户端都完整读一遍 MNIST。

import multiprocessing as mp from concurrent.futures import ProcessPoolExecutor _CTX = {} def _init_worker(images_u8, targets, cfg): # 每个 worker 进程只执行一次，图像用 uint8 传入，用时再转 float _CTX["x"] = images_u8 _CTX["y"] = targets _CTX["cfg"] = cfg def _run_client(args): cid, global_sd, indices = args cfg = _CTX["cfg"] torch.manual_seed(cfg["seed"] + cid) # 每个客户端固定但不同的种子 ds = torch.utils.data.TensorDataset( torch.from_numpy(_CTX["x"][indices]).float().div_(255.0).unsqueeze(1), torch.from_numpy(_CTX["y"][indices]).long(), ) loader = torch.utils.data.DataLoader(ds, batch_size=cfg["bs"], shuffle=True) sd = client_update(global_sd, loader, cfg["local_epochs"], cfg["lr"], "cpu") return cid, sd, len(indices)

主循环里进程池只创建一次，放在轮次循环外面，否则每轮重建 100 个进程，光启动开销就吃掉大部分收益。

ctx = mp.get_context("spawn") with ProcessPoolExecutor(max_workers=8, mp_context=ctx, initializer=_init_worker, initargs=(train_images_u8, train_targets, cfg)) as pool: for r in range(num_rounds): picked = rng.choice(num_clients, size=frac_n, replace=False) jobs = [(c, global_sd, client_idx[c]) for c in picked] results = list(pool.map(_run_client, jobs)) results.sort(key=lambda t: t[0]) # map 保序，但显式排序更保险 global_sd = fedavg(global_sd, [r[1] for r in results], [r[2] for r in results])

参数上，max_workers不要超过物理核数，超了只会互相抢 CPU；global_sd每轮都要作为参数传给 worker，模型只有几百 KB 时开销可忽略，如果换成 ResNet 级别，就要考虑用共享内存或者只传增量的方式。pool.map在某个 worker 抛异常时会整体失败，建议在_run_client里包一层 try，返回原始全局权重并把n_k设成 0，避免一个客户端把整轮训练打断，这也是真实联邦系统里必须做的容错。

4.3 断点续跑与轮次日志

联邦训练动辄上百轮，中途中断是常态。每轮结束存一次 checkpoint，包含轮次、模型、随机数状态和最优精度，恢复时把这三样都读回来，否则续跑出来的曲线和连续跑对不上。

ckpt = {"round": r + 1, "sd": global_sd, "rng": rng.bit_generator.state, "best_acc": best_acc, "cfg": cfg} torch.save(ckpt, f"ckpt/last.pt") if acc > best_acc: best_acc = acc torch.save(ckpt, "ckpt/best.pt")

注意rng.bit_generator.state存的是 numpy Generator 的完整内部状态，只存种子是不够的，因为种子只能让序列从头开始，不能从第 40 轮恢复。日志建议每轮一行 CSV，字段固定成round, n_clients, n_samples, loss, acc, lr, local_epochs，后面画对比曲线时直接读表，不用再从 stdout 里 grep。

5. 避坑与排查：non-IID 掉点、BN 与聚合错位

5.1 全局精度比集中式低 8 个点

现象：同样的 CNN，集中式训练 5 个 epoch 到 99.2%，联邦跑 100 轮只有 91%。

原因：这不是 bug，是 non-IID 下的客户端漂移。每个客户端只见过 2 个类，本地 SGD 会把模型往自己那几个类上拽，聚合时平均出来的方向偏离全局最优。local_epochs越大漂移越严重。

解决：先把local_epochs降回 1，确认精度回升；再加服务端动量（对聚合后的更新量做指数滑动平均），或者改成 FedProx，在客户端 loss 里加mu/2 * ||w - w_global||^2的近端项，mu从 0.01 试起。如果目标只是复现基准，接受 3 到 5 个点的差距是正常的，先别怀疑代码。

5.2 加了 BatchNorm 之后每轮精度剧烈震荡

现象：模型里加 BN 后，全局精度在 85% 到 93% 之间来回跳，去掉 BN 就稳定。

原因：running_mean和running_var是每个客户端本地数据的统计量，non-IID 下各客户端的统计量差异极大，直接加权平均得到的 buffer 不属于任何一个分布，推理时 BN 就是在用错误的均值和方差做归一化。

解决：联邦场景下要么不用 BN，改用 GroupNorm 或者 LayerNorm，它们不依赖 batch 统计；要么把 BN 的 buffer 排除在聚合之外，只聚合weight和bias，让每个客户端保留自己的统计量，但这要求推理发生在客户端侧。做 MNIST 这种小任务，最省事的做法就是不加 BN。

5.3 聚合后模型反而变差，甚至不如单个客户端

现象：某一轮聚合完成后，全局精度比上一轮还低，比单个客户端本地精度低得更多。

原因：大概率是权重和样本数没有对齐。client_sds和client_sizes的顺序必须一一对应，用了pool.map又在中途过滤失败任务，很容易出现错位；另一个常见原因是把验证集样本也算进了n_k。

解决：在fedavg开头加一句断言assert len(client_sds) == len(client_sizes)，并且让 worker 返回(cid, sd, n)三元组后在主进程统一排序，用 cid 对齐而不是依赖返回顺序。加完之后这类问题基本绝迹。

5.4 每轮结果抖动大，两次同样的配置跑出不同曲线

现象：完全相同的参数跑两遍，最终精度差 2 个点以上。

原因：随机性来源太多——客户端采样、DataLoader 的 shuffle、参数初始化、进程池的调度顺序。只要有一处没固定种子，结果就不可复现。

解决：在主进程固定numpy和torch的种子，worker 里用base_seed + cid派生独立种子，DataLoader 显式传generator。另外注意torch.use_deterministic_algorithms(True)在 CPU 上开销很小，可以打开。做完这些，两次运行的曲线应该逐轮一致。

5.5 通信轮数翻倍但精度不涨

现象：从 100 轮加到 300 轮，精度卡在同一个值附近，后 200 轮基本是平的。

原因：要么是学习率没有衰减，模型早就在最优点附近震荡；要么是frac太小，每轮只有 5 个客户端参与，梯度噪声盖过了有效信号；还有一种可能是测试集精度已经饱和，需要看训练损失才能判断。

解决：先加学习率衰减，比如每 30 轮乘 0.5；再把frac从 0.05 提到 0.1 到 0.2 看曲线是否继续上升。如果两条路都不动，那就是模型容量到顶了，换更大的网络比加轮数有意义。

6. 收在验证上：通信预算表与影子集

联邦实验最容易失控的地方不是代码，是「到底该跑多少轮」。我的做法是先固定一张通信预算表，把轮数、每轮客户端数、本地 epoch 三者的组合列出来，用小规模试跑填精度，再决定主实验怎么配。

看 D 和 A 的对比就明白了：同样 500 次上传，本地多训 4 轮反而掉了 1.7 个点，这就是客户端漂移的代价。所以「省通信」不能靠加本地 epoch，得靠加客户端并行度或者用梯度压缩。

另一件事是影子集。从训练集里切 2000 张出来当作服务端专属验证集，全程不参与任何客户端的本地训练，每轮聚合后在它上面评估一次。测试集只在最后用一次，拿来调轮数就是自欺欺人。这两个集子的曲线一旦分叉——影子集还在涨、测试集开始平——就说明开始过拟合了，该停就停。

从那以后我每次跑联邦实验，都强制先把划分脚本的输出打印一遍、把影子集单独落盘、把种子写进 checkpoint 文件名，三步走完才允许启动训练。这三件事花不了五分钟，能省掉后面几天的返工。这套 MNIST 分片数据、划分脚本和聚合代码在同一个压缩包里，解压后data/和src/的目录结构直接对应，路径不用改就能跑。希望帮到你。

本文还有配套的精品资源，点击获取