这是一个使用Delta框架编写的横向联邦学习的任务示例。
数据是分布在多个节点上的MNIST数据集，每个节点上只有其中的一部分样本。任务是训练一个卷积神经网络的模型，进行手写数字的识别。
本示例可以在Deltaboard中直接运行，完整的Jupyter Notebook文件也已经包含在Deltaboard中，进入Deltaboard的Playground，可以在examples文件下看到本示例文件。
在Deltaboard的在线版本中可以直接查看和运行这个示例
https://board.deltampc.com
1. 引入需要的包
我们的计算逻辑是用torch写的。所以首先引入numpy和torch，以及一些辅助的工具，然后从delta-task的包中，引入Delta框架的内容，包括DeltaNode节点，用于调用API发送任务，用于横向联邦学习任务HorizontalLearning，以及用于配置安全聚合策略的FaultTolerantFedAvg等：
from typing import Any, Dict, Iterable, List, Tuple
import logging
import numpy as np
import torch
from PIL.Image import Image
from torch.utils.data import DataLoader, Dataset
from delta.delta_node import DeltaNode
from delta.task.learning import HorizontalLearning, FaultTolerantFedAvg
import delta.dataset
2. 定义神经网络模型
接下来我们来定义神经网络模型，这里和传统的神经网络模型定义完全一样：
class LeNet(torch.nn.Module):
def __init__(self):
super().__init__()
self.conv1 = torch.nn.Conv2d(1, 16, 5, padding=2)
self.pool1 = torch.nn.AvgPool2d(2, stride=2)
self.conv2 = torch.nn.Conv2d(16, 16, 5)
self.pool2 = torch.nn.AvgPool2d(2, stride=2)
self.dense1 = torch.nn.Linear(400, 100)
self.dense2 = torch.nn.Linear(100, 10)
def forward(self, x: torch.Tensor):
x = self.conv1(x)
x = torch.relu(x)
x = self.pool1(x)
x = self.conv2(x)
x = torch.relu(x)
x = self.pool2(x)
x = x.view(-1, 400)
x = self.dense1(x)
x = torch.relu(x)
x = self.dense2(x)
return x
3. 定义隐私计算任务
然后可以开始定义我们的横向联邦任务了，用横向联邦学习的方式，在多节点上训练上面定义的神经网络模型
在定义横向联邦学习任务时，有几部分内容是需要用户自己定义的：
任务配置: 我们需要在 super().__init__() 方法中对任务进行配置。
数据集: 我们需要在dataset方法中定义任务所需要的数据集。
训练集的Dataloader: 我们需要在make_train_dataloader方法中定义训练集的Dataloader。
验证集的Dataloader: 我们需要在make_validate_dataloader方法中定义验证集的Dataloader。
模型训练: 在该方法中，定义整个模型的训练过程，包含整个前向传播和后向传播的。该方法的输入是训练集的Dataloader。
模型验证: 在该方法中，定义整个模型的验证过程。该方法的输入是验证集的Dataloader，输出是一个字典，字典的键是计算出的指标名称，值是对应的指标值。
模型参数: 我们需要在state_dict方法中定义所有需要训练和更新的模型参数。
def transform_data(data: List[Tuple[Image, str]]):
"""
作为dataloader的collate_fn，用于预处理函数。
将输入的mnist图片调整大小、归一化后，变为torch.Tensor返回。
"""
xs, ys = [], []
for x, y in data:
xs.append(np.array(x).reshape((1, 28, 28)))
ys.append(int(y))
imgs = torch.tensor(xs)
label = torch.tensor(ys)
imgs = imgs / 255 - 0.5
return imgs, label
class Example(HorizontalLearning):
def __init__(self) -> None:
super().__init__(
name="example", # 任务名称，用于在Deltaboard中的展示
max_rounds=2, # 任务训练的总轮次，每聚合更新一次权重，代表一轮
validate_interval=1, # 验证的轮次间隔，1表示每完成一轮，进行一次验证
validate_frac=0.1, # 验证集的比例，范围(0,1)
strategy=FaultTolerantFedAvg( # 安全聚合的策略，可选策略目前包含 FedAvg和FaultTolerantFedAvg，都位于delta.task.learning包下
min_clients=2, # 算法所需的最少客户端数，至少为2
max_clients=3, # 算法所支持的最大客户端数，必须大雨等于min_clients
merge_epoch=1, # 聚合更新的间隔，merge_interval_epoch表示每多少个epoch聚合更新一次权重
wait_timeout=30, # 等待超时时间，用来控制一轮计算的超时时间
connection_timeout=10 # 连接超时时间，用来控制流程中每个阶段的超时时间
)
)
self.model = LeNet()
self.loss_func = torch.nn.CrossEntropyLoss()
self.optimizer = torch.optim.SGD(
self.model.parameters(),
lr=0.1,
momentum=0.9,
weight_decay=1e-3,
nesterov=True,
)
def dataset(self) -> delta.dataset.Dataset:
"""
定义任务所需要的数据集。
return: 一个delta.dataset.Dataset
"""
return delta.dataset.Dataset(dataset="mnist")
def make_train_dataloader(self, dataset: Dataset) -> DataLoader:
"""
定义训练集Dataloader，可以对dataset进行各种变换、预处理等操作。
dataset: 训练集的Dataset
return: 训练集的Dataloader
"""
return DataLoader(dataset, batch_size=64, shuffle=True, drop_last=True, collate_fn=transform_data) # type: ignore
def make_validate_dataloader(self, dataset: Dataset) -> DataLoader:
"""
定义验证集Dataloader，可以对dataset进行各种变换、预处理等操作。
dataset: 验证集的Dataset
return: 验证集的Dataloader
"""
return DataLoader(dataset, batch_size=64, shuffle=False, drop_last=False, collate_fn=transform_data) # type: ignore
def train(self, dataloader: Iterable):
"""
训练步骤
dataloader: 训练数据集对应的dataloader
return: None
"""
for batch in dataloader:
x, y = batch
y_pred = self.model(x)
loss = self.loss_func(y_pred, y)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
def validate(self, dataloader: Iterable) -> Dict[str, Any]:
"""
验证步骤，输出验证的指标值
dataloader: 验证集对应的dataloader
return: Dict[str, float]，一个字典，键为指标的名称（str），值为对应的指标值（float）
"""
total_loss = 0
count = 0
ys = []
y_s = []
for batch in dataloader:
x, y = batch
y_pred = self.model(x)
loss = self.loss_func(y_pred, y)
total_loss += loss.item()
count += 1
y_ = torch.argmax(y_pred, dim=1)
y_s.extend(y_.tolist())
ys.extend(y.tolist())
avg_loss = total_loss / count
tp = len([1 for i in range(len(ys)) if ys[i] == y_s[i]])
precision = tp / len(ys)
return {"loss": avg_loss, "precision": precision}
def state_dict(self) -> Dict[str, torch.Tensor]:
"""
需要训练、更新的模型参数
在聚合更新、保存结果时，只会更新、保存get_params返回的参数
return: List[torch.Tensor]， 模型参数列表
"""
return self.model.state_dict()
具体来讲，当定义横向联邦学习任务时，我们需要定义一个继承自HorizontalLearning的类。HorizontalLearning类是一个虚基类，针对横向联邦学习任务的流程，定义了一些虚函数， 需要我们来实现。
首先是构造函数__init__。构造函数在这里，主要是对任务任务进行一些基础的配置。这些配置项包括任务名称（name），任务训练的总轮数（max_rounds），执行验证的频率（每 validate_interval 轮执行一次验证），验证集的比例（validate_frac），以及安全聚合的策略（strategy）。在自己实现构造函数时，必须调用基类的构造函数，即super().__init__()。
之后是dataset方法。dataset方法定义任务所需要的数据集。该方法返回一个delta.dataset.Dataset实例， 其参数dataset代表所需数据集的名称。关于数据集格式的具体细节，请参考这篇文章。目前横向联邦学习任务，只支持实用一个数据集，所以dataset方法只能返回一个delta.dataset.Dataset实例
然后是两个定义DataLoader的方法，分别是定义训练集DataLoader的make_train_dataloader和定义验证集DataLoader的make_validate_dataloader。这两个方法需要实现的逻辑类似，所以放在一起说。这两个方法的输入都是一个torch.utils.data.Dataset实例，分别为训练集和验证集；在方法中，可以按照需要对数据集进行变换、预处理等操作。最后，我们需要返回一个torch.utils.data.Dataloader实例，它会作为模型训练方法的输入。
然后是训练模型的方法train。该方法的输入是在make_train_dataloader中定义的DataLoader；在该方法中，我们需要实现整个模型的训练过程，包括前向传播，后向传播和参数更新操作。
还有验证模型的方法validate。该方法的输入是在make_validate_dataloader中定义的Dataloader；在该方法中，可以根据需要，在验证集上，计算模型的各种性能指标；该方法最后返回一个字典，字典的键是模型的指标的名称，对应的值是指标的值。
最后，我们还需要在state_dict方法中定义所有需要训练和更新的模型参数，方法的返回值就是这些模型参数的列表。
4. 指定执行任务用的Delta Node的API
定义好了任务，我们就可以开始准备在Delta Node上执行任务了。
Delta Task框架可以直接调用Delta Node API发送任务到Delta Node开始执行，只要在任务执行时指定Delta Node的API地址即可。
Deltaboard提供了对于Delta Node的API的封装，为每个用户提供了一个独立的API地址，支持多人同时使用同一个Delta Node，并且能够在Deltaboard中管理自己提交的任务。 在这里，我们使用Deltaboard提供的API来执行任务。如果用户自己搭建了Delta Node，也可以直接使用Delta Node的API。
在Deltaboard导航栏中进入“个人中心”，在Deltaboard API中，复制自己的API地址，并粘贴到下面的代码中：
DELTA_NODE_API = "http://127.0.0.1:6704"
5. 执行隐私计算任务
接下来我们可以开始运行这个模型了：
task = Example().build()
delta_node = DeltaNode(DELTA_NODE_API)
delta_node.create_task(task)
6. 查看执行状态
点击执行后，可以从输出的日志看出，任务已经提交到了Delta Node的节点上。
接下来，可以从左侧的导航栏中，前往“任务列表”，找到刚刚提交的任务，点击进去查看具体的执行日志了。
在系统详细设计的章节中，有隐私计算任务从发送到Delta Node开始，到执行完毕的详细流程说明，在上文中执行的横向联邦学习任务，可以在下面的文章中详细了解其执行过程：
page横向联邦学习