Clustered Federated Learning

Flower Help - Intermediate
1

I am trying to implement a Strategy utilizing Clustered Federated Lerning based on the paper Efficient Distribution Similarity Identification in Clustered Federated Learning via Principal Angles Between Client Data Subspaces. I figured out to send the principal components for the clustering within the fit method in the client:

    def fit(self, parameters, config: dict[str, bool | bytes | float | int | str]):
        # TODO: FOR PACFL, calculate here the svd and other things within the first epoch and then do normal training, this is part of client fit method
        lr = config['lr']
        momentum = config['momentum']
        epochs = config['local_epochs']
        current_server_round = config['current_server_round']
        print(current_server_round)   
        if current_server_round == 1:
            
            data_by_label = gather_data_by_label(self.trainloader)
            svd_results = compute_svd_for_each_label_top_k(data_by_label, 3) 
            res = np.hstack(svd_results)
            return self.get_parameters({}), len(self.trainloader), {"svd": json.dumps(res.tolist())}
            
        else:
            optim = torch.optim.SGD(self.model.parameters(), lr=lr, momentum=momentum)
            train(self.model,self.trainloader,optimizer=optim,epochs=epochs, device=self.device)

            return self.get_parameters({}), len(self.trainloader), {}

Now for the server strategy: In the aggregate_fit the clustering was also implemented with the same intention of only doing it if server_round is equal to 1 and save the clusters to the Strategy Object. Here is where my problem starts, i do not really figure out how to pass/send the individual model parameters per cluster to the clients belonging to a specific cluster. I assume the clusterwise aggregation would happen within the aggregate_fit based on the clients ids matched with the ids in the saved clusters. But how do i send the individual model parameters to the clients belonging to a cluster?
Greetings

2

Hi,
One solution to create different models is to use the configure_fit, but instead of using the parameters that are being passed via the function argument, you can keep different models as the class attributes and specify them based on the client.
Here’s a code that does this in FedAvg flower/src/py/flwr/server/strategy/fedavg.py at a8740389ef5066132c9bfb1875ebbb3be9bf727f · adap/flower · GitHub. You’d need to create FitIns based on the client ids.

3

Thanks to your reply @adam-narozniak! I also came up with two possible solutions, not yet implemented, where one is the same thought process as yours like


def aggregate_fit(self,
        server_round: int,
        results: List[Tuple[ClientProxy, FitRes]],
        failures: List[Union[Tuple[ClientProxy, FitRes], BaseException]],
    ) -> Tuple[Optional[Parameters], Dict[str, Scalar]]:

....
for cluster_id in self.clusters:
       fit_res_per_cluster = get_res_per_cluster(results, cluster_id)
       self.model_params[cluster_id] = aggregate_params(fit_res_per_cluster )
....

def configure_fit(
        self, server_round: int, parameters: Parameters, client_manager: ClientManager
    ) -> List[Tuple[ClientProxy, FitIns]]:
.....
for cluster_id in self.clusters:
       fit_ins[cluster_id] = FitIns(self.model_params[cluster_id])
....

Second option coming to my mind was to stack the params within the aggregate_fit of multiple models like:

params_aggregated = ndarrays_to_parameters([model.params for model in clustererd_models])
return parameters_aggregated, {}

and split them up in the configure_fit again.

This code right here is just pseudo code but i think everyone gets the idea. I will stick to the first one but if i have time will also try to implement the second approach and keep the community updated with the results.