Flwr框架中fit_round接收全失败问题排查求助

问题概述 #

使用Flower框架实现客户端向服务器发送随机数组，服务器合并后返回给所有客户端的功能。预期2个客户端分别持[1,2]和[3,4]时，服务器合并为[1,2,3,4]并返回，但实际运行时服务端日志显示fit_round接收0个有效结果、3个失败。

故障原因分析 #

1. 状态码错误：客户端get_parameters和fit方法使用Code.EVALUATE_NOT_IMPLEMENTED状态码，该状态码会被服务端标记为请求失败，需替换为Code.SUCCESS。
2. 参数序列化/反序列化格式错误：
   • ndarrays_to_parameters要求输入为列表形式的ndarray，客户端直接传入单个ndarray会导致序列化失败。
   • 服务端aggregate_fit中直接对Parameters对象执行np.concatenate，未先转换为ndarray，且未处理列表格式。
3. 策略配置不匹配：服务端策略设置min_available_clients=3、min_fit_clients=3，与预期的2个客户端不符，导致服务端等待超时后判定请求失败。
4. 客户端fit方法逻辑错误：parameters_to_ndarrays返回列表，直接赋值给self.array会导致后续长度计算错误，需取出列表中的ndarray元素。
5. evaluate方法未合规实现：NumPyClient要求evaluate方法返回特定格式的结果，直接pass会触发服务端判定失败。

修复后的完整代码 #

服务端代码 #

import numpy as np
import flwr as fl
from typing import List

class MergeArraysStrategy(fl.server.strategy.FedAvg): 
    def aggregate_fit(self, rnd, results, failures):
        # 处理客户端返回的参数
        client_arrays = []
        for result in results:
            # 将Parameters转换为ndarray列表，取第一个元素（目标数组）
            ndarray_list = fl.common.parameters_to_ndarrays(result.parameters)
            client_arrays.append(ndarray_list[0])

        if client_arrays:
            merged_array = np.concatenate(client_arrays)
            # 将合并后的数组转为Parameters格式返回
            merged_parameters = fl.common.ndarrays_to_parameters([merged_array])
            return merged_parameters, {} 
        else:
            # 返回空的Parameters对象
            return fl.common.ndarrays_to_parameters([np.array([])]), {} 
        
    def configure_evaluate(self, server_round, parameters, client_manager):
        # 禁用评估环节
        return []

    def evaluate(self, server_round, parameters):
        # 禁用评估逻辑
        return None, {}

# 创建服务端，适配2个客户端的场景
strategy = MergeArraysStrategy(min_available_clients=2, min_fit_clients=2)
fl.server.start_server(
    server_address="127.0.0.1:8080",
    config=fl.server.ServerConfig(num_rounds=2),
    strategy=strategy
)

客户端代码 #

import flwr as fl
import numpy as np
from flwr.common import (
    Code,
    FitIns,
    FitRes,
    GetParametersIns,
    GetParametersRes,
    Status,
    ndarrays_to_parameters,
    parameters_to_ndarrays,
)

class Client(fl.client.NumPyClient):
    def __init__(self, array):
        self.array = array 
        print(f"初始数组: {self.array}")
    
    def get_parameters(self, ins: GetParametersIns) -> GetParametersRes:
        # 传入列表形式的ndarray进行序列化
        parameters = ndarrays_to_parameters([self.array])
        # 使用SUCCESS状态码
        status = Status(code=Code.SUCCESS, message="Success")
        return GetParametersRes(
            status=status,
            parameters=parameters,
        )
    
    def fit(self, ins: FitIns) -> FitRes:
        # 反序列化参数，取出第一个元素（合并后的数组）
        ndarray_list = parameters_to_ndarrays(ins.parameters)
        self.array = ndarray_list[0]

        # 序列化更新后的数组（用列表包裹）
        parameters_updated = ndarrays_to_parameters([self.array])
        
        status = Status(code=Code.SUCCESS, message="Success")
        return FitRes(
            status=status,
            parameters=parameters_updated,
            num_examples=len(self.array),
            metrics={},
        )
    
    def evaluate(self, ins):
        # 合规实现evaluate方法，返回标准格式结果
        return fl.common.EvaluateRes(
            status=Status(code=Code.SUCCESS, message="Success"),
            loss=0.0,
            num_examples=len(self.array),
            metrics={},
        )

# 创建客户端，生成随机数组
client = Client(array=np.random.randint(1, 10, size=2))
# 启动客户端连接服务端
fl.client.start_client(server_address="127.0.0.1:8080", client=client.to_client())
# 打印最终合并后的数组
print(f"最终合并数组: {client.array}")

验证方法 #

1. 启动服务端代码。
2. 启动2个客户端实例（可在不同终端运行客户端代码）。
3. 查看服务端日志，确认fit_round接收2个有效结果，无失败。
4. 查看客户端终端输出，确认初始数组不同，最终输出合并后的完整数组。

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