如何在Flower与TensorFlow联邦学习中向服务器传递额外参数？

核心问题分析 #

直接在get_parameters方法里返回额外参数会触发服务器端FedAvg策略的解析错误——因为FedAvg默认仅期望接收模型参数张量列表，非张量类型的额外数据会导致类型不匹配。

正确实现方式 #

方法1：利用客户端fit方法返回值传递（推荐） #

Flower客户端的fit方法支持返回四个值：(weights, num_examples, metrics, other_info)，可通过metrics或other_info字段传递额外参数：

• 客户端代码修改：

  def fit(self, parameters, config):
      # 原有模型训练逻辑
      self.set_parameters(parameters)
      self.model.fit(self.x_train, self.y_train, epochs=1)
      
      # 定义要传递的额外参数
      extra_params = {"client_id": self.client_id, "local_train_samples": len(self.x_train)}
      # 返回模型参数、样本数、空metrics、额外参数
      return self.get_parameters(config), len(self.x_train), {}, extra_params
  

• 服务器端自定义策略（继承FedAvg）接收并处理：

  from flwr.server.strategy import FedAvg
  from flwr.common import FitRes, ClientProxy
  
  class CustomFedAvg(FedAvg):
      def aggregate_fit(
          self,
          rnd: int,
          results: list[tuple[ClientProxy, FitRes]],
          failures: list,
      ):
          # 先执行父类的模型参数聚合逻辑
          aggregated_weights, metrics = super().aggregate_fit(rnd, results, failures)
          
          # 提取并处理每个客户端的额外参数
          for client, fit_res in results:
              client_extra = fit_res.metrics  # 或fit_res.other_info，取决于客户端存在哪个字段
              print(f"客户端{client.cid}传递的额外参数：{client_extra}")
              # 这里可添加存储、计算等自定义逻辑
          
          return aggregated_weights, metrics
  

  服务器启动时使用自定义策略：

  strategy = CustomFedAvg()
  flwr.server.start_server(
      server_address="0.0.0.0:8080",
      config=flwr.server.ServerConfig(num_rounds=5),
      strategy=strategy
  )
  

方法2：修改get_parameters返回格式（需同步服务器解析逻辑） #

如果必须通过get_parameters传递，需确保服务器端能正确拆分模型参数与额外数据：

• 客户端返回包含模型参数和额外数据的元组：

  def get_parameters(self, config):
      model_weights = self.model.get_weights()
      extra_params = {"custom_data": "your_value"}
      # 将额外数据序列化（避免非张量类型序列化失败）
      import json
      serialized_extra = json.dumps(extra_params).encode("utf-8")
      return (model_weights, serialized_extra)
  

• 服务器端自定义策略拆分数据：

  class CustomFedAvg(FedAvg):
      def aggregate_fit(self, rnd, results, failures):
          processed_results = []
          for client, fit_res in results:
              # 拆分客户端返回的(模型参数, 序列化的额外数据)
              model_weights, serialized_extra = fit_res.parameters.tensors
              # 反序列化额外数据
              import json
              extra_params = json.loads(serialized_extra.decode("utf-8"))
              
              # 重新包装模型参数为标准Parameters对象
              new_params = flwr.common.Parameters(
                  tensors=model_weights,
                  tensor_type=fit_res.parameters.tensor_type
              )
              # 将额外参数存入metrics方便后续处理
              fit_res.metrics["extra_params"] = extra_params
              processed_results.append((client, FitRes(fit_res.status, new_params, fit_res.num_examples, fit_res.metrics)))
          
          # 调用父类聚合逻辑
          return super().aggregate_fit(rnd, processed_results, failures)
  

关键注意事项 #

• 不要破坏Flower默认的参数传递协议，优先使用官方预留的fit返回字段传递额外数据。
• 非张量类型的额外数据需先序列化（如JSON、pickle），避免跨节点传输时出现序列化错误。
• 自定义策略时需确保父类的核心聚合逻辑不受影响，仅在其基础上扩展额外参数处理。

内容的提问来源于stack exchange，提问作者Saim