联邦学习开源平台Flower：在移动装备上运行联邦学习 - ToB企服应用市场:ToB评测及商务社交产业平台

conda create -n flwr python=3.8

复制代码

conda activate flwr

复制代码

pip install flwr

复制代码

python -c "import flwr;print(flwr.__version__)"

复制代码

# 传统的机器学习流程代码
from collections import OrderedDict
import flwr as fl
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision.datasets import CIFAR10
from torchvision.transforms import Compose, Normalize, ToTensor
from tqdm import tqdm
DEVICE = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
# 定义神经网络模型
class Net(nn.Module):
def __init__(self) -> None:
super(Net, self).__init__()
self.conv1 = nn.Conv2d(3, 6, 5)
self.pool = nn.MaxPool2d(2, 2)
self.conv2 = nn.Conv2d(6, 16, 5)
self.fc1 = nn.Linear(16 * 5 * 5, 120)
self.fc2 = nn.Linear(120, 84)
self.fc3 = nn.Linear(84, 10)
def forward(self, x: torch.Tensor) -> torch.Tensor:
x = self.pool(F.relu(self.conv1(x)))
x = self.pool(F.relu(self.conv2(x)))
x = x.view(-1, 16 * 5 * 5)
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
return self.fc3(x)
# 定义模型训练流程
def train(net, trainloader, epochs):
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(net.parameters(), lr=0.001, momentum=0.9)
for _ in range(epochs):
for images, labels in tqdm(trainloader):
optimizer.zero_grad()
criterion(net(images.to(DEVICE)), labels.to(DEVICE)).backward()
optimizer.step()
# 定义模型推理流程
def test(net, testloader):
criterion = torch.nn.CrossEntropyLoss()
correct, loss = 0, 0.0
with torch.no_grad():
for images, labels in tqdm(testloader):
outputs = net(images.to(DEVICE))
labels = labels.to(DEVICE)
loss += criterion(outputs, labels).item()
correct += (torch.max(outputs.data, 1)[1] == labels).sum().item()
accuracy = correct / len(testloader.dataset)
return loss, accuracy
# 定义数据集的获取
def load_data():
"""Load CIFAR-10 (training and test set)."""
trf = Compose([ToTensor(), Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])
trainset = CIFAR10("./data", train=True, download=True, transform=trf)
testset = CIFAR10("./data", train=False, download=True, transform=trf)
return DataLoader(trainset, batch_size=32, shuffle=True), DataLoader(testset)
# 生成模型对象，实际获取训练与测试数据集
net = Net().to(DEVICE)
trainloader, testloader = load_data()

复制代码

# 客户端类实现
class FlowerClient(fl.client.NumPyClient):
# 获取本地模型对应的参数
def get_parameters(self, config):
return [val.cpu().numpy() for _, val in net.state_dict().items()]
# 接收模型参数，并更新本地模型
def set_parameters(self, parameters):
params_dict = zip(net.state_dict().keys(), parameters)
state_dict = OrderedDict({k: torch.tensor(v) for k, v in params_dict})
net.load_state_dict(state_dict, strict=True)
# 本地模型训练，会先调用 set_parameters() 基于收到的全局模型参数更新本地模型
def fit(self, parameters, config):
self.set_parameters(parameters)
train(net, trainloader, epochs=1)
return self.get_parameters(config={}), len(trainloader.dataset), {}
# 基于测试数据集进行测试
def evaluate(self, parameters, config):
self.set_parameters(parameters)
loss, accuracy = test(net, testloader)
return loss, len(testloader.dataset), {"accuracy": accuracy}
# 启动 Flower 客户端
fl.client.start_numpy_client(
server_address="127.0.0.1:8080",
client=FlowerClient(),
)

复制代码

from typing import List, Tuple
import flwr as fl
from flwr.common import Metrics
# 定义指标聚合方法
def weighted_average(metrics: List[Tuple[int, Metrics]]) -> Metrics:
accuracies = [num_examples * m["accuracy"] for num_examples, m in metrics]
examples = [num_examples for num_examples, _ in metrics]
return {"accuracy": sum(accuracies) / sum(examples)}
# 定义模型聚合策略
strategy = fl.server.strategy.FedAvg(evaluate_metrics_aggregation_fn=weighted_average)
# 启动 Flower 服务端
fl.server.start_server(
server_address="0.0.0.0:8080", # 服务器地址
config=fl.server.ServerConfig(num_rounds=3),
strategy=strategy,
)

复制代码

python server.py

复制代码

python client.py

复制代码

import flwr as fl
import tensorflow as tf
# 加载数据
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()
# 加载模型，10分类模型MobilNetV2
model = tf.keras.applications.MobileNetV2((32, 32, 3), classes=10, weights=None)
model.compile("adam", "sparse_categorical_crossentropy", metrics=["accuracy"])
# 定义client类，也就是flower客户端
class CifarClient(fl.client.NumPyClient):
def get_parameters(self, config):
return model.get_weights()
def fit(self, parameters, config):
model.set_weights(parameters)
model.fit(x_train, y_train, epochs=1, batch_size=32, steps_per_epoch=3)
return model.get_weights(), len(x_train), {}
def evaluate(self, parameters, config):
model.set_weights(parameters)
loss, accuracy = model.evaluate(x_test, y_test)
return loss, len(x_test), {"accuracy": float(accuracy)}
fl.client.start_client(server_address="127.0.0.1:8080", client=CifarClient().to_client()) # 启动flower客户端

复制代码

import flwr as fl
if __name__ == '__main__':
fl.server.start_server(server_address="0.0.0.0:8080", config=fl.server.ServerConfig(num_rounds=3)) # 启动flower服务器

复制代码

min_fit_clients=2, # 根据自己需要更改最少客户端
min_evaluate_clients=2,
min_available_clients=2,

复制代码

pip install -r requirements.txt

复制代码

python server.py

复制代码