【联邦学习实战】FLGo入门

一、FLGo介绍

二、Quick Start

项目结构

三、运行配置-初始化&配置选项

3.1 联邦算法 algorithm

3.2 训练参数option

3.2.1基本选项：

3.2.2 服务器端选项：

3.2.3 客户端选项：

3.2.4 与实际机器相关的选项：

3.2.5 特定联邦算法的其他超参数：

3.2.6 日志设置：

模型

自定义模型

快速切换模型

一、FLGo介绍

FLGo的开发宗旨是让做联邦学习相关的研究变得更加简单。为FLGo实现了以下功能：

帮助使用者很轻松地自己手写一个联邦学习算法，并和其他算法进行对比；
提供了数个主流的联邦数据集基准（图像、文本、图、表格等），支持各种常见的数据异构性划分，并提供接口辅助大家尝试在自己的数据集上进行联邦学习；
支持模拟常见的系统异构性，并提供接口支持用户自行定制系统异构性，来做相关方面的仿真研究；
提供了基础的实验记录、结果分析功能，方便做实验；支持自动调度GPU进行网格搜索；
支持纵向联邦、异步联邦等算法（对于异步联邦算法，使用了统一的虚拟时钟，使得其可以和同步联邦算法进行公平的比较）

以上来自知乎用户wwwzz

项目地址

二、Quick Start

项目结构

├─ algorithm（算法）
│  ├─ fedavg.py                   // FedAvg算法
│  ├─ ...
│  ├─ fedasync.py                 // 异步分布式算法的基类
│  └─ fedbase.py                  // 分布式算法的基类
├─ benchmark（基准测试）
│  ├─ mnist_classification          // 在MNIST数据集上的分类任务
│  │  ├─ model                   // 相应的模型
│  |  └─ core.py                 // 数据集的核心支持，包含三个必要的类（例如 TaskGen、TaskReader、TaskCalculator）
│  ├─ ...
│  ├─ RAW_DATA                   // 存储下载的原始数据集
│  └─ toolkits                      // 用于生成分布式数据集的基本工具
│     ├─ cv                      // 常见的计算机视觉分布式任务
│     │  ├─ horizontal           // 水平分布任务
│     │  │  └─ image_classification.py   // 图像分类的基类
│     │  └─ ...
│     ├─ ...
│     ├─ base.py                 // 所有分布式任务的基类
│     ├─ partition.py            // 用于数据集分割的分区类
│     └─ visualization.py        // 数据集分割后的可视化工具
├─ experiment（实验）
│  ├─ logger（日志记录器）
│  │  ├─ basic_logger.py         // 基本日志记录器类
│  |  └─ simple_logger.py        // 简单日志记录器类
│  ├─ analyzer.py               // 用于分析和打印实验结果的类
│  ├─ res_config.yml            // 分析器的超参数文件
│  ├─ run_config.yml            // 运行配置的超参数文件
|  └─ runner.py                // 基于超参数组合生成实验命令的类，并处理实验的进程调度
├─ system_simulator（系统模拟器）
│  ├─ base.py                   // 模拟系统异构性的基类
│  ├─ default_simulator.py      // 模拟系统异构性的默认类
|  └─ ...
├─ utils（实用工具）
│  ├─ fflow.py                  // 选项读取、初始化等工具
│  └─ fmodule.py                // 模型级别的操作符
└─ requirements.txt             // 项目的依赖包清单

导入flgo，根据requirements.txt配置环境。

import flgo
import flgo.benchmark.mnist_classification as mnist
import flgo.algorithm.fedavg as fedavg
import os
import flgo.experiment.analyzer as al# 数据集划分，设置一百个客户端
task = './test_mnist'
config = {'benchmark':{'name':'flgo.benchmark.mnist_classification'},'partitioner':{'name': 'IIDPartitioner','para':{'num_clients':100}}}
if not os.path.exists(task): flgo.gen_task(config, task_path = task)# 使用FedAvg训练
# nun_rounds: 通信轮次
# num_epochs: 本地训练epoch数量
# gpu：采用gpu训练，使用卡一fedavg_runner = flgo.init(task=task, algorithm=fedavg, option={'num_rounds':5, 'num_epochs':1, 'gpu':0})
fedavg_runner.run()# 绘制训练过程中的验证集损失和验证集精度
analysis_plan = {'Selector':{'task': task,'header':['fedavg']},'Painter':{'Curve':[{'args':{'x': 'communication_round', 'y':'val_loss'}, 'fig_option':{'title':'valid loss on MNIST'}},{'args':{'x': 'communication_round', 'y':'val_accuracy'},  'fig_option':{'title':'valid accuracy on MNIST'}},]}
}
al.show(analysis_plan)

三、运行配置-初始化&配置选项

def init(task: str, algorithm, option = {}, model=None, Logger: flgo.experiment.logger.BasicLogger = None, Simulator: BasicSimulator=flgo.simulator.DefaultSimulator, scene='horizontal'):

flgo.init函数主要作用是为联邦学习任务提供一个初始化的入口点，根据传入的参数创建并返回一个具有run方法的runner实例，通过调用run方法开启迭代训练。

为了生成该对象，init应当接收的输入包括：

task：联邦任务路径，由flgo.gen_task生成的任务路径；
algorithm：联邦算法，要求algorithm的类型是class或module，横向联邦中需要其具备algorithm.Server和algorithm.Client两个可访问的属性；
option（可选）：运行选项，类型为字典，包含运行时的各类参数；
model（可选）：待优化的模型模块，要求model的类型为类或module，横向联邦中需要其具备model.init_global_module和model.init_local_module两个可访问的方法，为实体初始化module；
Logger（可选）：日志记录器类，要求父类为flgo.experiment.logger.BasicLogger
Simulator（可选）：系统模拟器类，要求父类为flgo.system_simulator.BasicSimulator
scene（可选）：类型为字符串，联邦场景（e.g. 横向 or 纵向）

3.1 联邦算法 algorithm

可选参数有fedavg、fedprox、fedasync、fedbuff等算法，详情见doc

3.2 训练参数option

3.2.1基本选项：

task（任务）：选择拆分数据集的任务，例如mnist_classification_client100_dist0_beta0_noise0等。
algorithm（算法）：选择分散学习算法，例如fedfv、fedavg、fedprox等。
model（模型）：选择与数据集相对应的模型，例如mlp、cnn、resnet18等。

3.2.2 服务器端选项：

sample（采样方式）：决定每轮如何采样客户端，例如uniform（均匀采样）或md（根据概率选择）。
aggregate（聚合方式）：决定如何聚合客户端的模型，例如uniform、weighted_scale、weighted_com。
num_rounds（通信轮次的数量）。
proportion（每轮中选择的客户端比例）。
lr_scheduler（全局学习率调度器）。
learning_rate_decay（学习率的衰减率）。

3.2.3 客户端选项：

num_epochs（本地训练的轮数）。
num_steps（本地更新步数，默认值为-1，如果设置大于0，num_epochs将无效）。
learning_rate（本地训练时的步长）。
batch_size（本地训练期间的批量数据大小）。
optimizer（选择优化器，如SGD、Adam等）。
weight_decay（设置本地训练过程中的权重衰减比例）。
momentum（当使用SGD优化器时，每步采取的动量项的比例）。

3.2.4 与实际机器相关的选项：

seed（初始随机种子）。
gpu（GPU设备的ID，可以选择使用的GPU，也可以使用CPU）。
server_with_cpu（默认为False）。
test_batch_size（在验证集上评估模型时使用的批量大小）。
eval_interval（控制每两次评估之间的间隔）。
num_threads（客户端计算会话中的线程数，以加速训练过程）。
num_workers（torch.utils.data.Dataloader的工作线程数）。

3.2.5 特定联邦算法的其他超参数：

algo_para（用于接收算法相关的超参数，可以在命令行中指定值）。

3.2.6 日志设置：

logger（选择要使用的日志记录器）。
log_level（日志级别）。
log_file（控制是否将运行时信息存储到.log文件中，默认值为false）。
no_log_console（控制是否在控制台上显示运行时间信息，默认值为false）。

模型

切换模型主要是通过指定init函数中的model关键字，来切换模型。例如：

cnn_runner = flgo.init(task, fedavg, option={'num_rounds':5, 'num_epochs':1, 'gpu':0}, model=cnn)
mlp_runner = flgo.init(task, fedavg, option={'num_rounds':5, 'num_epochs':1, 'gpu':0}, model=mlp)

在FLgo中，将模型视作benchmark所包含的一部分。这是因为对于不同的数据集来说，模型的架构往往不一致，且同一个数据集也可以使用不同的模型。因此，每个benchmark都必须具有benchmark_name.model子模块，且子模块中需要包含适用于该benchmark的模型（例如mnist_classification.model.cnn）。

|── benchmark

│ ├── mnist_classification

│ │ ├── model

│ │ ├── cnn.py

│ │ ├── mlp.py

自定义模型

在FLGo中，主要通过重写model.init_local_module接口和model.init_global_module接口，来为每个实体object（例如，横向联邦中Server或Client的实例）分配模型。其中：

init_local_module是为实体分配自身所持有的本地模型（例如personalized models）
init_global_module是为实体分配用于共享的全局模型（例如经典横向联邦中的global model）

由于经典横向联邦中只需要服务器维护最新的全局模型，因此上面代码中init_global_module中只为类别为Server的实例设置一个全局模型，而所有其他实体不持有模型。此外，所有模型需要继承flgo.utils.fmodule.FModule，该类型封装了torch的nn.Module，使得使用者可以直接对模型做一些相加、放缩、平均等常用运算。

class Model(FModule):def __init__(self):super().__init__()self.encoder = nn.Sequential(nn.Conv2d(in_channels=1, out_channels=32, kernel_size=5, padding=2),nn.ReLU(),nn.MaxPool2d(2),nn.Conv2d(in_channels=32, out_channels=64, kernel_size=5, padding=2),nn.ReLU(),nn.MaxPool2d(2),nn.Flatten(1),nn.Linear(3136, 512),nn.ReLU(),)self.head = nn.Linear(512, 10)def forward(self, x):x = self.encoder(x)x = self.head(x)return xdef init_local_module(object):passdef init_global_module(object):if 'Server' in object.__class__.__name__:object.model = Model().to(object.device)

快速切换模型

为了直接可以从模型构筑能被flgo.init加载的模型，我在flgo 0.0.17版本中进一步封装了转换模型的函数flgo.convert_model函数，使用方法如下：

from torch import nn
import torch.nn.functional as F
from flgo.utils.fmodule import FModuleclass Model(nn.Module):def __init__(self):super().__init__()self.conv1 = nn.Conv2d(in_channels=1, out_channels=32, kernel_size=5, padding=2)self.conv2 = nn.Conv2d(in_channels=32, out_channels=64, kernel_size=5, padding=2)self.fc1 = nn.Linear(3136, 512)self.fc2 = nn.Linear(512, 10)def forward(self, x):x = x.view((x.shape[0],28,28))x = x.unsqueeze(1)x = F.max_pool2d(F.relu(self.conv1(x)), 2)x = F.max_pool2d(F.relu(self.conv2(x)), 2)x = x.view(-1, x.shape[1]*x.shape[2]*x.shape[3])x = F.relu(self.fc1(x))x = self.fc2(x)return xmodel = flgo.convert_model(Model)
mycnn_runner2 = flgo.init(task, fedavg, option={'num_rounds':5, 'num_epochs':1, 'gpu':0}, model=model)
mycnn_runner2.run()