简介：这是一份用Python在本地模拟横向联邦学习的完整项目包，面向想理解联邦学习机制并动手复现的机器学习初学者与工程师。资源以横向联邦为主线，覆盖多客户端本地训练、参数上传与服务器聚合的闭环流程，并配套CIFAR-10图像分类相关代码，可直接用于学习分布式协作训练与隐私保护相关场景。RAR压缩包内共25个文件，以Python脚本为主，包括模型定义、客户端/服务端程序、数据集处理与配置模块，另含编译缓存、PyCharm项目配置和CIFAR-10数据集文件，整体约302.68MB，便于在本地还原实验环境。已有725人学习，适合作为入门横向联邦学习、梳理通信与聚合逻辑的参考资料。通过阅读源码与运行模拟，可以掌握联邦平均的基本写法、数据切分方式以及客户端—服务器通信框架，为进一步使用TensorFlow Federated等工具打下基础。
1. 横向联邦学习本地模拟：用Python把多客户端训练塞进一台机器
横向联邦学习（Horizontal Federated Learning）是一种在不交换原始数据的前提下，通过多方协作训练共享模型的技术。大多数教程都拿真实分布式环境开刀，但如果你只是想理解算法逻辑、调参或者做实验验证，用一台机器本地模拟是最高效的方式。我最初接手这个项目时，以为要搭一套分布式框架，后来发现用Python做本地模拟，把所有客户端循环塞进单机，反而能更清楚地看到每个环节的瓶颈。这份资源就是一套完整的本地模拟实现，覆盖数据划分、客户端训练、参数聚合三个核心步骤，适合想快速上手的开发者。
2. 横向联邦学习核心流程：数据分区、客户端训练与FedAvg聚合
在写完整代码之前，先搞清楚横向联邦学习在干什么。横向联邦学习的特点是每个客户端拥有自己一部分样本，特征维度相同，但样本不同。典型场景是不同机构各自拥有用户数据，它们合作训练同一个模型，但原始数据不能离开本地。
2.1 数据分区：模拟非独立同分布（Non-IID）
真实联邦学习中，不同客户端的本地数据分布往往不一致，称为非独立同分布（Non-IID）。例如，某支付平台的用户数据中，一线城市用户的消费习惯迥异，如果客户端按地域划分，数据分布就会不同。在本地模拟中，我需要用代码生成多个数据片段，并刻意让它们分布不同，这样才能测试联邦学习算法的鲁棒性。
代码示例如下：
import numpy as np def generate_data(num_samples=1000, num_clients=5): # 生成模拟数据：每类样本按正态分布偏移，模拟Non-IID all_x = np.random.rand(num_samples, 10).astype(np.float32) all_y = (np.sum(all_x, axis=1) > 5).astype(np.int32) # 按客户端切片，故意打乱顺序形成Non-IID idx = np.arange(num_samples) np.random.shuffle(idx) split_points = np.linspace(0, num_samples, num_clients+1).astype(np.int32) client_data = [] for i in range(num_clients): start, end = split_points[i], split_points[i+1] client_data.append((all_x[idx[start:end]], all_y[idx[start:end]])) return client_data逻辑说明：这里生成一个简单的二分类问题，特征all_x是10维随机数，标签all_y是特征和的阈值。用np.random.shuffle打乱索引，再按顺序切片，虽然数据是随机生成的，但客户端切片的顺序已经固定，所以每个客户端的数据分布有差异——这就是模拟Non-IID的一种低成本方式。实际项目中可以用狄利克雷分布（Dirichlet）来控制每个客户端的类别比例，但当前代码足够用来测机制。
参数说明：num_samples是总样本数，num_clients是模拟的客户端数量。客户端数量建议4~8个，太少体现不出联邦学习的优势，太多在单机上跑会明显变慢。
2.2 客户端训练与参数回传
每个客户端拿到全局模型参数后，用本地数据训练几轮（epoch），然后返回新的参数。这里的关键是，客户端不能把原始数据上传给服务端，只能上传模型参数或梯度。我在本地模拟时，直接把参数存成一个numpy数组，便于做加权平均。
代码示例如下：
class Client: def __init__(self, client_id, x_data, y_data): self.client_id = client_id self.x = x_data self.y = y_data self.local_epochs = 3 self.learning_rate = 0.01 def train(self, global_params): # 用本地数据跑几轮梯度下降 params = global_params.copy() for _ in range(self.local_epochs): pred = np.dot(self.x, params) errors = self.y - pred grad = -2 * np.dot(self.x.T, errors) / len(self.x) params -= self.learning_rate * grad return params逻辑说明：train方法接收全局参数global_params，复制一份后在本地做梯度下降。我在代码里用线性回归的梯度公式（误差乘以特征），因为逻辑清晰，适合作为演示。如果你想换成逻辑回归或神经网络，只需要改动train内部的模型表达式即可，不会影响联邦学习的整体流程。
参数说明：local_epochs是本地训练轮数，一般设3~5轮；learning_rate是学习率，设0.01~0.1之间。local_epochs太大容易导致模型偏向本地数据，最终聚合时模型泛化能力下降。
2.3 服务端聚合与模型更新
服务端负责聚合所有客户端训练后的参数，最常用的方法是联邦平均（FedAvg）。FedAvg的加权权重等于每个客户端的本地样本数占总数量的比例，权重越大，说明该客户端在全局模型中的话语权越强。
代码示例如下：
class Server: def __init__(self, initial_params): self.global_params = initial_params def aggregate(self, client_params_list, client_sizes): total_size = sum(client_sizes) weighted_params = np.zeros_like(self.global_params) for params, size in zip(client_params_list, client_sizes): weight = size / total_size weighted_params += weight * params self.global_params = weighted_params return self.global_params.copy()逻辑说明：服务端把每个客户端返回的参数按照样本数加权求和。比如客户端A有100个样本，客户端B有300个样本，那么B的权重是75%，A是25%。这样做会避免数据量小的客户端主导模型更新。
参数说明：client_params_list是一个列表，每个元素是客户端训练后的参数（numpy数组）；client_sizes是对应的样本数量列表。务必要保证两个列表的顺序一致，否则聚合结果会错乱。
3. 本地模拟实现：从环境准备到完整训练循环
这部分直接给出一套能跑起来的完整脚本，你可以在自己的Python环境中运行。我的目标是让脚本尽量简单，但把横向联邦学习的每个环节都体现出来。
3.1 环境准备与依赖安装
我用的都是Python标准库加上numpy，安装非常简单。在终端执行以下命令即可：
pip install numpy如果你已经装了Anaconda，可以跳过这一步。我建议用虚拟环境，避免污染系统环境。以下是我测试时的环境版本：Python 3.8.10和numpy 1.21.2，但理论上Python 3.6以上都不影响。
3.2 完整训练循环：5个客户端+1个服务端
下面是一个完整的本地模拟脚本，包含数据生成、客户端和服务端定义，以及主训练循环。
import numpy as np # ---------- 数据生成 ---------- def generate_data(num_samples=1000, num_clients=5): all_x = np.random.rand(num_samples, 10).astype(np.float32) all_y = (np.sum(all_x, axis=1) > 5).astype(np.int32) idx = np.random.permutation(num_samples) split_points = np.linspace(0, num_samples, num_clients+1).astype(np.int32) client_data = [] for i in range(num_clients): start, end = split_points[i], split_points[i+1] client_data.append((all_x[idx[start:end]], all_y[idx[start:end]])) return client_data # ---------- 客户端 ---------- class Client: def __init__(self, client_id, x_data, y_data): self.client_id = client_id self.x = x_data self.y = y_data self.local_epochs = 5 self.learning_rate = 0.05 def train(self, global_params): params = global_params.copy() for _ in range(self.local_epochs): pred = np.dot(self.x, params) errors = self.y - pred grad = -2 * np.dot(self.x.T, errors) / len(self.x) params -= self.learning_rate * grad return params # ---------- 服务端 ---------- class Server: def __init__(self, initial_params): self.global_params = initial_params def aggregate(self, client_params_list, client_sizes): total_size = sum(client_sizes) weighted_params = np.zeros_like(self.global_params) for params, size in zip(client_params_list, client_sizes): weight = size / total_size weighted_params += weight * params self.global_params = weighted_params return self.global_params.copy() # ---------- 主循环 ---------- client_data = generate_data(num_samples=1000, num_clients=5) dims = client_data[0][0].shape[1] initial_params = np.zeros(dims, dtype=np.float32) server = Server(initial_params) clients = [] for i in range(5): x, y = client_data[i] clients.append(Client(i, x, y)) communication_rounds = 20 for round_idx in range(communication_rounds): client_params = [] client_sizes = [] for client in clients: updated_params = client.train(server.global_params) client_params.append(updated_params) client_sizes.append(len(client.y)) server.aggregate(client_params, client_sizes) # 打印每轮损失，观察收敛趋势 pred = np.dot(np.concatenate([c.x for c in clients]), server.global_params) truth = np.concatenate([c.y for c in clients]) loss = np.mean((pred - truth) ** 2) print(f"Round {round_idx+1}, Loss: {loss:.4f}")逻辑说明：主循环执行20轮通信，每轮中所有客户端并行（准确说是逐个）训练，然后将参数和样本数返回给服务端，服务端聚合后更新全局参数。打印的损失值是基于全量数据计算的MSE，方便观察模型是否收敛。实际联邦学习环境里客户端是并行在不同设备上运行的，但本地模拟使用for循环就足够了。
参数说明：communication_rounds是通信轮数，我设置20轮，是因为线性回归问题一般很快收敛；如果你换成神经网络特征，可能需要50~100轮。另外，这里的数据集是随机均匀划分，所以Non-IID效应不强，如果你需要用极端Non-IID场景，建议用Dirichlet分布划分数据，这部分我会在最后一章给出示例。
3.3 如何测试不同参数组合
我一般会写一个自动化脚本来做参数扫描，比如遍历不同的local_epochs和learning_rate，观察损失曲线。以下是一个简单的测试代码：
for lr in [0.01, 0.05, 0.1]: for epochs in [3, 5, 10]: print(f"lr={lr}, epochs={epochs}") # 复制上述主循环，替换对应参数 # 记录最终损失注意：每次扫描时，最好固定随机种子，否则不同参数对比就没有意义。固定方式很简单，在生成数据前加一句np.random.seed(0)，具体见第4章。
4. 避坑与常见问题：本地模拟的四个典型踩坑记录
模拟联邦学习时，我踩过不少坑，尤其是把并行逻辑变成串行逻辑那一步。下面几条是最容易翻车的点，写出来给后来人参考。
4.1 客户端数据分布不同导致聚合发散
现象：第一轮通信后，损失值飙升，甚至比初始值还高。
原因：某个客户端样本极少（比如只有5条数据），它的梯度噪声非常大，服务端加权聚合时把这个噪声放大，整体模型就偏了。
解决：检查客户端样本数，加入最小样本数阈值；或者在聚合时对噪声大的客户端参数做裁剪，限制参数更新的最大值（例如将每次更新的参数范数归一化到1以内）。
4.2 参数回传时误传了global_params本身
现象：每个客户端训练后返回的参数与其他客户端完全一致，损失不下降。
原因：把全局参数当作本地参数直接用，没有先复制。代码里params = global_params.copy()，如果忘了引用.copy()，后续改动会直接污染全局参数。
解决：养成习惯，在任何需要修改全局参数的场景，强制执行一次.copy()，避免共享引用。
4.3local_epochs太大导致模型震荡
现象：通信轮数增加，但损失值忽高忽低，不收敛。
原因：每个客户端在本地训练了太多轮，导致模型过拟合本地数据，服务端聚合出来的模型被反复拉扯。
解决：把local_epochs压缩到3~5之间。我一般先用对比实验扫一次，比如分别测1、3、5、10轮，看哪个值在验证集上表现最好。
4.4 随机种子未固定，实验结果不可复现
现象：两次运行，同样的参数，损失曲线完全不一致。
原因：数据生成本身用了随机函数，客户端训练顺序也可能受随机性影响。
解决：在脚本入口处设置np.random.seed(42)；如果需要多个随机源（比如数据划分和训练拆分），可以用np.random.default_rng(seed)单独生成随机流，避免全局种子冲突。
5. 进阶：加入极端Non-IID数据与调试技巧
如果你想让模拟更贴近真实世界，需要把数据分布做得更不均匀。常见做法是用狄利克雷分布（Dirichlet distribution）来控制每个客户端的类别比例，参数alpha越小，数据分布越倾斜。
def generate_dirichlet_data(num_samples=1000, num_clients=5, alpha=0.5): rng = np.random.default_rng(0) # 先随机生成特征和标签 all_x = rng.rand(num_samples, 10) all_y = (np.sum(all_x, axis=1) > 5).astype(np.int32) # 用Dirichlet分布分配样本索引 proportions = rng.dirichlet([alpha]*num_clients, size=num_samples) client_indices = np.zeros((num_samples, num_clients), dtype=bool) for i in range(num_samples): client_indices[i, np.argmax(proportions[i])] = True client_data = [] for c in range(num_clients): indices = np.where(client_indices[:, c])[0] client_data.append((all_x[indices], all_y[indices])) return client_data逻辑说明：proportions是一个形状为(num_samples, num_clients)的矩阵，每行表示一个样本被分配到每个客户端的概率，np.argmax取概率最大的那个客户端作为归属。alpha控制分布平滑度，alpha越小，每个客户端越可能拿到某类极端样本。
参数说明：alpha通常在0.1（极端Non-IID）到1（接近IID）之间调整。我一般会同时跑alpha=0.1和alpha=1两组实验，对比最终全局模型的精度差异，这样能直观看出Non-IID对联邦学习的影响。
调试技巧：我习惯用一个独立的验证集来衡量每次通信后的全局模型性能。把验证集从客户端数据中抽出来，比如每个客户端留10%样本作为验证集，这样能看到联邦学习在训练过程中是否出现过拟合或崩溃。具体做法是在客户端train方法中单独保存验证集，每次聚合后计算验证损失，我一般用这个值来判断是否该调小local_epochs或learning_rate。
记得有一次我把alpha设成0.1，客户端数据差异极大，结果前5轮聚合后验证损失不降反升，我以为算法有问题，后来排查发现是某个客户端的样本数太少（只有几十条），梯度噪声太大。从那以后，我每次模拟前都会先检查每个客户端的样本分布，如果某个客户端样本量少于总数据量的1%，我会先做数据补全或增加num_samples。希望帮到你。
本文还有配套的精品资源，点击获取
