在本教程中，我将实现联邦学习(FL)的构建块，并从头开始对MNIST机器学习数据集进行训练。

介绍

FL架构的基本形式包括一个位于中心的管理员或服务器，负责协调训练活动。客户端主要是边缘设备，可以达到数百万的数量。这些设备在每次训练迭代中至少与服务器通信两次。首先，它们各自从服务器接收当前全局模型的权重，在各自的本地数据上对其进行训练，以生成更新后的参数，然后将这些参数上传到服务器进行汇总。这种通信循环一直持续到达到预先设定的epochs数或准确度条件为止。在联邦平均算法中，汇总仅仅意味着平均操作。

现在我们已经了解了FL是什么以及它如何工作，让我们继续在Tensorflow中从头开始构建一个FL，并在Kaggle的MNIST数据集（https://www.kaggle.com/scolianni/mnistasjpg）上对其进行训练。

导入所需的Python库

import numpy as np
import random
import cv2
import os
from imutils import paths
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelBinarizer
from sklearn.model_selection import train_test_split
from sklearn.utils import shuffle
from sklearn.metrics import accuracy_score

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv2D
from tensorflow.keras.layers import MaxPooling2D
from tensorflow.keras.layers import Activation
from tensorflow.keras.layers import Flatten
from tensorflow.keras.layers import Dense
from tensorflow.keras.optimizers import SGD
from tensorflow.keras import backend as K

from fl_mnist_implementation_tutorial_utils import *

读取和预处理MNIST数据集

我正在使用MNIST数据集的jpeg版本。它由42000个数字图像组成，每个类保存在单独的文件夹中。我将使用一下Python代码片段将数据加载到内存中，并保留10%的数据，以便稍后测试经过训练的全局模型。

def load(paths, verbose=-1):
    '''expects images for each class in seperate dir, 
    e.g all digits in 0 class in the directory named 0 '''
    data = list()
    labels = list()
    # loop over the input images
    for (i, imgpath) in enumerate(paths):
        # load the image and extract the class labels
        im_gray = cv2.imread(imgpath, cv2.IMREAD_GRAYSCALE)  #9
        image = np.array(im_gray).flatten()
        label = imgpath.split(os.path.sep)[-2]
        # scale the image to [0, 1] and add to list
        data.append(image/255)   #13
        labels.append(label)
        # show an update every `verbose` images
        if verbose > 0 and i > 0 and (i + 1) % verbose == 0:
            print("[INFO] processed {}/{}".format(i + 1, len(paths)))
    # return a tuple of the data and labels
    return data, labels

在第9行，从磁盘读取每个图像作为灰度，然后将其flattened。flattening步骤是导入，因为稍后我们将使用MLP网络架构。然后我们分割图像的路径，在第11行获得它的类ID。我们还在第13行上的将图像缩放到[0,1]，以减弱像素亮度变化的影响。

训练和测试拆分

#declear path to your mnist data folder
img_path = '/path/to/your/training/dataset'

#get the path list using the path object
image_paths = list(paths.list_images(img_path))

#apply our function
image_list, label_list = load(image_paths, verbose=10000)

#binarize the labels
lb = LabelBinarizer()
label_list = lb.fit_transform(label_list)

#split data into training and test set
X_train, X_test, y_train, y_test = train_test_split(image_list, 
                                                    label_list, 
                                                    test_size=0.1, 
                                                    random_state=42)

在这个代码片段中执行了两个步骤。我们应用了在前面的代码块中定义的load函数来获得图像列表(现在是numpy数组)和标签列表。之后，我们使用来自sklearn的LabelBinarizer对象对标签进行one-hot编码。不再把数字1的标签作为数字1，它现在的形式是[0,1,0,0,0,0,0,0,0,0,0]。使用这种标记样式，我将能够使用Tensorflow中的交叉熵损失作为我们模型的损失函数。或者，我可以保持标签不变，而使用稀疏分类熵损失。最后，我使用sklearn的train_test_split将数据拆分成比例为9:1的train/test。

联邦成员(客户端)

在FL的实际实现中，每个联邦成员将独立拥有自己的数据。请记住，FL的目标是将模型传递到数据。我将训练集分成10个碎片，每个客户一个。我写了一个函数create_clients来实现这一目标。

def create_clients(image_list, label_list, num_clients=10, initial='clients'):
    ''' return: a dictionary with keys clients' names and value as 
                data shards - tuple of images and label lists.
        args: 
            image_list: a list of numpy arrays of training images
            label_list:a list of binarized labels for each image
            num_client: number of fedrated members (clients)
            initials: the clients'name prefix, e.g, clients_1 
            
    '''

    #create a list of client names
    client_names = ['{}_{}'.format(initial, i+1) for i in range(num_clients)]  #13

    #randomize the data
    data = list(zip(image_list, label_list))
    random.shuffle(data)

    #shard data and place at each client
    size = len(data)//num_clients
    shards = [data[i:i + size] for i in range(0, size*num_clients, size)]

    #number of clients must equal number of shards
    assert(len(shards) == len(client_names))

    return {client_names[i] : shards[i] for i in range(len(client_names))}   #26

在第13行，我使用前缀字符串创建了一个客户端名称列表。第16–21行将数据和标签压缩，将所得的元组列表随机化并分片为所需数量的客户端（num_clients）。在第26行，返回了一个字典，其中包含作为键的每个客户端名称和作为值的它们的数据共享。现在让我们将这个函数应用到我们的训练数据集。

#create clients
clients = create_clients(X_train, y_train, num_clients=10, initial='client')

批处理客户端和测试数据

接下来是将每个客户端数据处理为tensorflow数据集并进行批处理。为了简化这个步骤并避免重复，我将这个过程封装到一个名为batch_data的小函数中。

def batch_data(data_shard, bs=32):
    '''Takes in a clients data shard and create a tfds object off it
    args:
        shard: a data, label constituting a client's data shard
        bs:batch size
    return:
        tfds object'''
    #seperate shard into data and labels lists
    data, label = zip(*data_shard)    #9
    dataset = tf.data.Dataset.from_tensor_slices((list(data), list(label)))
    return dataset.shuffle(len(label)).batch(bs)

每个客户端数据集都是以create_clients中的数据/标签元组列表的形式出现的。在上面的第9行，为了与TFDS API兼容，我将元组拆分为单独的数据和标签列表。在应用此函数时，我还将处理测试集，并将其保留到以后使用。

#process and batch the training data for each client
clients_batched = dict()
for (client_name, data) in clients.items():
    clients_batched[client_name] = batch_data(data)
    
#process and batch the test set  
test_batched = tf.data.Dataset.from_tensor_slices((X_test, y_test)).batch(len(y_test))

创建模型

在介绍部分我没有提到的一件事是FL最适合参数化学习——所有类型的神经网络。诸如KNN或类似的机器学习技术仅存储训练数据，而学习可能无法从FL中受益。我正在创建一个2层MLP作为分类任务的模型。我将使用Keras API创建此文件。

class SimpleMLP:
    @staticmethod
    def build(shape, classes):
        model = Sequential()
        model.add(Dense(200, input_shape=(shape,)))
        model.add(Activation("relu"))
        model.add(Dense(200))
        model.add(Activation("relu"))
        model.add(Dense(classes))
        model.add(Activation("softmax"))
        return model

要构建新模型，将调用build方法。它需要输入数据的形状和类的数量作为参数。使用MNIST，shape参数将是28*28*1 = 784，而类的数量将是10。此时，我还将为模型编译定义一个优化器、损失函数和度量。

lr = 0.01 
comms_round = 100
loss='categorical_crossentropy'
metrics = ['accuracy']
optimizer = SGD(lr=lr, 
                decay=lr / comms_round, 
                momentum=0.9
               )  

SGD是默认优化器。损失函数为categorical_crossentropy，度量为accuracy。但是，在衰变参数中看起来有些奇怪。comms_round是什么?它只是我想要运行的全局epochs(aggregations)数量。

模型汇总（加权平均）

到目前为止，根据深度学习管道，我所做的一切几乎都是标准的（除了数据分区和客户端创建）。我所使用的数据是水平分区的，因此我将简单地进行组件级参数平均，并根据每个参与客户端贡献的数据点的比例进行加权。这是我用的联邦平均方程

在右侧，我们根据单个客户端持有的每个数据点上记录的损失值来估计权重参数。在左边，我们缩放了客户的参数并对结果求和。

下面我将这个过程封装为三个简单的函数。

def weight_scalling_factor(clients_trn_data, client_name):
    client_names = list(clients_trn_data.keys())
    #get the bs
    bs = list(clients_trn_data[client_name])[0][0].shape[0]
    #first calculate the total training data points across clinets
    global_count = sum([tf.data.experimental.cardinality(clients_trn_data[client_name]).numpy() for client_name in client_names])*bs
    # get the total number of data points held by a client
    local_count = tf.data.experimental.cardinality(clients_trn_data[client_name]).numpy()*bs
    return local_count/global_count


def scale_model_weights(weight, scalar):
    '''function for scaling a models weights'''
    weight_final = []
    steps = len(weight)
    for i in range(steps):
        weight_final.append(scalar * weight[i])
    return weight_final



def sum_scaled_weights(scaled_weight_list):
    '''Return the sum of the listed scaled weights. The is equivalent to scaled avg of the weights'''
    avg_grad = list()
    #get the average grad accross all client gradients
    for grad_list_tuple in zip(*scaled_weight_list):
        layer_mean = tf.math.reduce_sum(grad_list_tuple, axis=0)
        avg_grad.append(layer_mean)
        
    return avg_grad


def test_model(X_test, Y_test,  model, comm_round):
    cce = tf.keras.losses.CategoricalCrossentropy(from_logits=True)
    #logits = model.predict(X_test, batch_size=100)
    logits = model.predict(X_test)
    loss = cce(Y_test, logits)
    acc = accuracy_score(tf.argmax(logits, axis=1), tf.argmax(Y_test, axis=1))
    print('comm_round: {} | global_acc: {:.3%} | global_loss: {}'.format(comm_round, acc, loss))
    return acc, loss

（1）weight_scalling_factor 计算客户的本地训练数据在所有客户持有的总体训练数据中所占的比例。首先，我们估计客户的批次大小，然后使用它来计算自己的数据点数量。然后，我们获得了第6行上的总体全局训练数据大小。最后，我们在第9行以分数的形式计算了比例因子。这当然不可能是实际应用程序中的方法。任何客户都不能访问合并的训练数据。在这种情况下，在每个本地训练步骤之后用新参数更新服务器时，每个客户机都应该指出它们所持有的数据点的数量。

（2）scale_model_weights根据（1）中计算的比例因子的值来缩放每个局部模型的权重

（3）sum_scaled_weights将所有客户的比例权重加在一起。

联邦模型训练

#initialize global model
smlp_global = SimpleMLP()
global_model = smlp_global.build(784, 10)
        
#commence global training loop
for comm_round in range(comms_round):
            
    # get the global model's weights - will serve as the initial weights for all local models
    global_weights = global_model.get_weights()
    
    #initial list to collect local model weights after scalling
    scaled_local_weight_list = list()

    #randomize client data - using keys
    client_names= list(clients_batched.keys())  #15
    random.shuffle(client_names)
    
    #loop through each client and create new local model
    for client in client_names:
        smlp_local = SimpleMLP()
        local_model = smlp_local.build(784, 10)
        local_model.compile(loss=loss, 
                      optimizer=optimizer, 
                      metrics=metrics)
        
        #set local model weight to the weight of the global model
        local_model.set_weights(global_weights)
        
        #fit local model with client's data
        local_model.fit(clients_batched[client], epochs=1, verbose=0)
        
        #scale the model weights and add to list
        scaling_factor = weight_scalling_factor(clients_batched, client)
        scaled_weights = scale_model_weights(local_model.get_weights(), scaling_factor)
        scaled_local_weight_list.append(scaled_weights)
        
        #clear session to free memory after each communication round
        K.clear_session()
        
    #to get the average over all the local model, we simply take the sum of the scaled weights
    average_weights = sum_scaled_weights(scaled_local_weight_list)   #41
    
    #update global model 
    global_model.set_weights(average_weights)

    #test global model and print out metrics after each communications round
    for(X_test, Y_test) in test_batched:
        global_acc, global_loss = test_model(X_test, Y_test, global_model, comm_round)  #48

训练逻辑有两个主循环，外循环用于全局迭代，内循环用于迭代每个客户端的本地训练。

首先构建全局模型，输入形状为(784)，数字类为10。然后我进入了外循环。首先获得全局模型的初始化权值。第15行和第16行随机化了客户端字典顺序。然后开始遍历客户端。

对于每个客户端，我初始化一个新的模型对象，编译它，并将它的初始化权重设置为全局模型的当前参数。然后对局部模型(客户端)进行一个epoch的训练。在训练之后，新的权重将被缩放并附加到scaled_local_weight_list中。

回到第41行的外循环，我获取了所有缩放后的局部训练权重的总和，并将全局模型更新为这个新的汇总。这样就结束了完整的全局训练epoch。按照前面声明的comms_round参数的规定，我运行了100个全局训练循环。

最后在第48行，我使用预留的测试集，在每一轮通信结束后，对训练好的全局模型进行测试，代码如下:

def test_model(X_test, Y_test,  model, comm_round):
    cce = tf.keras.losses.CategoricalCrossentropy(from_logits=True)
    #logits = model.predict(X_test, batch_size=100)
    logits = model.predict(X_test)
    loss = cce(Y_test, logits)
    acc = accuracy_score(tf.argmax(logits, axis=1), tf.argmax(Y_test, axis=1))
    print('comm_round: {} | global_acc: {:.3%} | global_loss: {}'.format(comm_round, acc, loss))
    return acc, loss

结果

测试结果有10个客户端，每个客户端运行1个本地epoch，并进行100次全局通信。

与SGD比较

FL模型测试结果很好，经过100轮通信后，测试准确率达到了96.5%。但它与在相同数据集上训练的标准SGD模型相比如何呢?我将在联邦训练数据上训练一个模型(而不是像在FL中那样训练10个模型)。为此，我将使用分区之前的预处理训练数据来训练完全相同的2层MLP模型。

为了确保一个公平，我将保留用于FL训练的每个超级参数，但batch size除外。不是使用32，我们的SGD模型的batch size将是320。

SGD_dataset = tf.data.Dataset.from_tensor_slices((X_train, y_train)).shuffle(len(y_train)).batch(320)
smlp_SGD = SimpleMLP()
SGD_model = smlp_SGD.build(784, 10) 

SGD_model.compile(loss=loss, 
              optimizer=optimizer, 
              metrics=metrics)

# fit the SGD training data to model
_ = SGD_model.fit(SGD_dataset, epochs=100, verbose=0)

#test the SGD global model and print out metrics
for(X_test, Y_test) in test_batched:
        SGD_acc, SGD_loss = test_model(X_test, Y_test, SGD_model, 1)

在100个epoch之后，SGD模型的测试精度达到了94.5%。在这个数据集上，FL的表现比SGD要好一点，不过这种结果在现实世界中是不可能出现的。客户端持有的真实联邦数据大多是非独立同分布IID）的数据。例如，如果我们根据训练数据集构造客户机碎片，使每个客户机的shad由单个类组成，比如client_1只有数字1的图像，client_2只有数字2的图像，等等，我们就可以模拟这个场景。如果采用这种非IID安排，我们的FL模型的测试准确率可能会下降6%。