Flower XGBoost 联邦学习快速上手用 FedXgbBagging 在 Higgs 数据集上训练分类模型【免费下载链接】flowerFlower: A Friendly Federated AI Framework项目地址: https://gitcode.com/GitHub_Trending/flo/flower本教程以 Flower 官方 Quickstart XGBoost 项目为蓝本讲解如何用 Flower 与 XGBoost 搭建一个基于FedXgbBaggingbootstrap aggregation策略的双节点联邦训练系统在 Higgs 数据集上训练二分类模型。读完本文你将掌握flwr new脚手架生成项目、flwr run本地仿真运行、pyproject.toml参数配置、XGBoost 模型与 FlowerMessage/ArrayRecord之间的二进制序列化转换以及ClientApp/ServerApp的完整实现与底层树聚合原理。背景为什么用 FedXgbBagging 做联邦 XGBoostXGBoost 的梯度提升树模型天然难以像神经网络那样对权重做平均因此 Flower 提供了一种适配树模型的联邦策略 ——FedXgbBaggingbootstrap aggregation装袋聚合。其核心思想是每个客户端对自己的本地数据做自助采样bootstrap subsampling每一轮训练一棵或多棵树服务端把各客户端本轮新增的树**拼接aggregate**进全局模型而不是加权平均参数全局模型随联邦轮次不断“长树”集成来自所有客户端的新树从而提升整体稳定性与精度。从源码看FedXgbBagging继承自FedAvg见 fedxgb_bagging.py但重写了configure_train与aggregate_train前者记录当前正在通信的全局模型字节流current_bst后者调用工具函数aggregate_bagging完成树级拼接。具体拼接逻辑位于 strategy_utils.py将上一轮全局模型与客户端返回的新树按 JSON 结构合并更新num_trees、iteration_indptr并把新树的 id 重编号后追加到全局模型树列表中。这就是“服务端只做树拼接、不做数值平均”的实现基础。环境准备与项目创建教程建议在独立的虚拟环境中运行。Flower 官方提供了虚拟环境搭建指南见 contributor-how-to-set-up-a-virtual-env.rst。环境就绪后先安装 Flower# In a new Python environment $ pip install flwr然后使用flwr new脚手架一键生成完整的 Flower XGBoost 项目。该命令会拉取 Flower Labs 官方模板并生成一个可直接运行的双节点联邦项目默认使用FedXgbBagging策略、本地仿真 profile$ flwr new flwrlabs/quickstart-xgboost执行后当前目录下会出现名为quickstart-xgboost的新目录结构如下quickstart-xgboost ├── quickstart_xgboost │ ├── __init__.py │ ├── client_app.py # Defines your ClientApp │ ├── server_app.py # Defines your ServerApp │ └── task.py # Defines your data loading and utility functions ├── pyproject.toml # Project metadata like dependencies and configs └── README.md仓库中已包含与该模板同源的完整实现可直接对照阅读examples/quickstart-xgboost 目录下的quickstart_xgboost/task.py、quickstart_xgboost/client_app.py、quickstart_xgboost/server_app.py与pyproject.toml。其中pyproject.toml声明了运行时依赖flwr[simulation]1.36.0、flwr-datasets0.6.1、xgboost2.0.0并通过[tool.flwr.app.components]把server_app:app与client_app:app绑定为应用入口见 pyproject.toml。运行联邦训练并查看流式日志进入项目目录用默认参数运行并流式输出日志$ cd quickstart-xgboost # Run with default arguments and stream logs $ flwr run . --stream说明普通flwr run .只提交运行、打印 run ID 后立即返回不会流式输出日志加--stream才会持续滚动显示训练过程。关于本地完整工作流的更多细节可参考 how-to-run-flower-locally.rst。默认参数下你会看到类似下面的流式输出Starting local SuperLink on 127.0.0.1:39091... Successfully started run 1859953118041441032 INFO : Starting FedXgbBagging strategy: INFO : ├── Number of rounds: 3 INFO : [ROUND 1/3] INFO : configure_train: Sampled 2 nodes (out of 2) INFO : aggregate_train: Received 2 results and 0 failures INFO : └── Aggregated MetricRecord: {} INFO : configure_evaluate: Sampled 2 nodes (out of 2) INFO : aggregate_evaluate: Received 2 results and 0 failures INFO : └── Aggregated MetricRecord: {auc: 0.7677505289821278} INFO : [ROUND 2/3] INFO : ... INFO : [ROUND 3/3] INFO : ... INFO : Strategy execution finished in 132.88s INFO : Final results: INFO : ServerApp-side Evaluate Metrics: INFO : {}从日志可以清晰看到运行机制的几个关键点flwr run会先在本机启动一个受管的本地 SuperLink默认监听127.0.0.1:39091再由 Flower Simulation Runtime 执行本次运行FedXgbBagging每轮从 2 个节点中采样 2 个参与训练与评估服务端聚合后输出MetricRecord其中auc即为每轮全局模型在验证集上的 AUC 指标首轮日志中约 0.7678。你还可以通过--run-config覆盖pyproject.toml中[tool.flwr.app.config]定义的任意参数例如# Override some arguments $ flwr run . --run-config num-server-rounds5 params.eta0.2这条命令把联邦轮数改为 5、XGBoost 学习率eta改为 0.2无需改动任何文件即可做参数实验。配置文件pyproject.toml中的超参数所有联邦与 XGBoost 超参数都集中在pyproject.toml的[tool.flwr.app.config]段[tool.flwr.app.config] num-server-rounds 3 fraction-train 0.1 fraction-evaluate 0.1 local-epochs 1 save-model false # XGBoost parameters params.objective binary:logistic params.eta 0.1 # Learning rate params.max-depth 8 params.eval-metric auc params.nthread 16 params.num-parallel-tree 1 params.subsample 1 params.tree-method hist各参数含义如下参数默认值说明num-server-rounds3联邦训练总轮数fraction-train0.1每轮参与训练的节点比例示例模板为 0.1仓库同源示例默认 1.0见 pyproject.tomlfraction-evaluate0.1每轮参与评估的节点比例local-epochs1本地树提升tree boost的迭代轮数即每轮每个客户端新训练几棵树save-modelfalse是否在训练结束后把最终模型保存到磁盘params.objectivebinary:logistic二分类逻辑回归目标函数params.eta0.1学习率learning rateparams.max-depth8树的最大深度params.eval-metricauc评估指标为 AUCparams.nthread16训练使用的 CPU 线程数params.num-parallel-tree1每轮并行训练树的数量params.subsample1训练样本的子采样比例params.tree-methodhist直方图近似算法默认在 CPU 上训练要点默认在 CPU 上训练若想改用 GPU只需把tree-method设为gpu_hist。local-epochs直接决定每个客户端每轮向服务端贡献的树数量它同时被ClientApp的本地提升循环与FedXgbBagging的树拼接逻辑使用下文详解。注意pyproject.toml中键名使用连字符如num-server-rounds而 XGBoost 原生参数名使用下划线如max_depth。因此源码中通过unflatten_dictreplace_keys两步把扁平配置还原成嵌套字典并把键中的-递归替换为_得到可直接传给xgb.train的params字典见 task.py。数据用 Flower Datasets 加载并切分 Higgs本项目使用 Flower Datasets 自动下载并切分Higgs数据集jxie/higgs并用IidPartitioner把训练集划分为num_partitions份分区数等于客户端总数实现 IID 数据分布。Flower Datasets 还提供其他 partitioner如 Dirichlet 分布的非 IID 切分可按需替换。客户端侧的数据加载逻辑如下partitioner IidPartitioner(num_partitionsnum_clients) fds FederatedDataset( datasetjxie/higgs, partitioners{train: partitioner}, ) partition fds.load_partition(partition_id, splittrain) partition.set_format(numpy) # Train/test splitting train_data, valid_data, num_train, num_val train_test_split( partition, test_fraction0.2, seed42 ) # Reformat data to DMatrix for xgboost train_dmatrix transform_dataset_to_dmatrix(train_data) valid_dmatrix transform_dataset_to_dmatrix(valid_data)流程分三步先按partition_id取回该客户端对应的数据分区再按 80/20 比例把本地分区拆成训练集与验证集test_fraction0.2固定随机种子seed42保证可复现最后转换为 XGBoost 所需的DMatrix格式。两个工具函数的完整实现见 task.pydef train_test_split(partition, test_fraction, seed): Split the data into train and validation set given split rate. train_test partition.train_test_split(test_sizetest_fraction, seedseed) partition_train train_test[train] partition_test train_test[test] num_train len(partition_train) num_test len(partition_test) return partition_train, partition_test, num_train, num_test def transform_dataset_to_dmatrix(data): Transform dataset to DMatrix format for xgboost. batch data[:] x np.asarray(batch[inputs], dtypenp.float32) y np.asarray(batch[label], dtypenp.float32) return xgb.DMatrix(x, labely)Higgs 数据集的inputs为 28 维特征、label为二分类标签与params.objective binary:logistic的目标函数相对应。为了让FederatedDataset只初始化一次load_data用模块级变量fds做了缓存见 task.py。ClientApp在Message与 XGBoost 模型之间做转换要把 XGBoost 接入 Flower核心改动在于把Message中收到的ArrayRecord还原成 XGBoost 可加载的二进制对象训练完再把模型序列化回ArrayRecord塞进回复Message。Flower 的ClientApp提供了三个可实现的装饰器方法train用本地数据训练收到的模型、evaluate在本地验证集上评估模型性能、query查询执行ClientApp的节点信息。本教程只用到train和evaluate。train本地提升 回传新树train方法接收两个参数来自ServerApp的Message以及Context。默认情况下该Message携带两部分内容一个ArrayRecord存放待联邦训练的模型数组默认通过 keyarrays访问一个ConfigRecord存放ServerApp下发的配置默认通过 keyconfig访问。Context则用于读取运行配置与节点配置run config 的超参数定义在 Flower App 的pyproject.toml中node config 只在 Deployment Runtime 下可设置仿真运行期间不能直接配置。完整实现见 client_app.py# Flower ClientApp app ClientApp() app.train() def train(msg: Message, context: Context) - Message: # Load model and data partition_id context.node_config[partition-id] num_partitions context.node_config[num-partitions] train_dmatrix, _, num_train, _ load_data(partition_id, num_partitions) # Read from run config num_local_round context.run_config[local-epochs] # Flatted config dict and replace - with _ cfg replace_keys(unflatten_dict(context.run_config)) params cfg[params] global_round msg.content[config][server-round] if global_round 1: # First round local training bst xgb.train( params, train_dmatrix, num_boost_roundnum_local_round, ) else: bst xgb.Booster(paramsparams) global_model bytearray(msg.content[arrays][0].numpy().tobytes()) # Load global model into booster bst.load_model(global_model) # Local training bst _local_boost(bst, num_local_round, train_dmatrix) # Save model local_model bst.save_raw(json) model_np np.frombuffer(local_model, dtypenp.uint8) # Construct reply message # Note: we store the model as the first item in a list into ArrayRecord, # which can be accessed using index [0]. model_record ArrayRecord([model_np]) metrics { num-examples: num_train, } metric_record MetricRecord(metrics) content RecordDict({arrays: model_record, metrics: metric_record}) return Message(contentcontent, reply_tomsg)第一轮server-round 1时服务端尚未有可用模型客户端直接用xgb.train()在本地数据上从零训练前num_local_round棵树。从第二轮起客户端读取服务端下发的全局模型字节流加载进新建的Booster再调用_local_boost在本地数据上继续提升def _local_boost(bst_input, num_local_round, train_dmatrix): # Update trees based on local training data. for i in range(num_local_round): bst_input.update(train_dmatrix, bst_input.num_boosted_rounds()) # Bagging: extract the last Nnum_local_round trees for sever aggregation bst bst_input[ bst_input.num_boosted_rounds() - num_local_round : bst_input.num_boosted_rounds() ] return bst_local_boost做两件事通过bst_input.update(dmatrix, num_boosted_rounds())迭代num_local_round次完成本地树提升随后用切片语法取出最后N num_local_round棵树作为本轮增量返回给服务端。这正是装袋聚合的关键 —— 客户端只上送本轮新增的树服务端把它们拼接到全局模型上见 client_app.py。回复Message的构造要点模型以ArrayRecord列表的第一个元素存放之后用msg.content[arrays][0]即可访问同时通过MetricRecord上报num-examples供服务端按样本量加权聚合指标。整个数据流可概括为Message → ArrayRecord → numpy → bytearray → bst.load_model → 本地提升 → bst.save_raw(json) → np.frombuffer → ArrayRecord → 回复 Message。evaluate本地验证集评估 AUCevaluate与train几乎相同只有两点差异其一不做本地训练直接用收到的全局模型在本地留出的验证集上评估其二回复Message无需再携带模型模型未被本地修改只上报指标。实现见 client_app.pyapp.evaluate() def evaluate(msg: Message, context: Context) - Message: # Load model and data partition_id context.node_config[partition-id] num_partitions context.node_config[num-partitions] _, valid_dmatrix, _, num_val load_data(partition_id, num_partitions) # Load config cfg replace_keys(unflatten_dict(context.run_config)) params cfg[params] # Load global model bst xgb.Booster(paramsparams) global_model bytearray(msg.content[arrays][0].numpy().tobytes()) bst.load_model(global_model) # Run evaluation eval_results bst.eval_set( evals[(valid_dmatrix, valid)], iterationbst.num_boosted_rounds() - 1, ) auc float(eval_results.split(\t)[1].split(:)[1]) # Construct and return reply Message metrics { auc: auc, num-examples: num_val, } metric_record MetricRecord(metrics) content RecordDict({metrics: metric_record}) return Message(contentcontent, reply_tomsg)评估使用bst.eval_set在验证集DMatrix上计算指标并从返回字符串中解析出auc数值连同num-examples一起放入MetricRecord返回。这就是日志中Aggregated MetricRecord: {auc: 0.7677505289821278}的来源。ServerApp用 FedXgbBagging 编排联邦轮次ServerApp的核心是app.main()方法它接收两个参数Grid用于与运行ClientApp的节点交互调度每一轮 train/evaluate/query与Context提供运行配置访问。完整实现见 server_app.py# Create ServerApp app ServerApp() app.main() def main(grid: Grid, context: Context) - None: # Read run config num_rounds context.run_config[num-server-rounds] fraction_train context.run_config[fraction-train] fraction_evaluate context.run_config[fraction-evaluate] # Flatted config dict and replace - with _ cfg replace_keys(unflatten_dict(context.run_config)) params cfg[params] # Init global model # Init with an empty object; the XGBooster will be created # and trained on the client side. global_model b # Note: we store the model as the first item in a list into ArrayRecord, # which can be accessed using index [0]. arrays ArrayRecord([np.frombuffer(global_model, dtypenp.uint8)]) # Initialize FedXgbBagging strategy strategy FedXgbBagging( fraction_trainfraction_train, fraction_evaluatefraction_evaluate, ) # Start strategy, run FedXgbBagging for num_rounds result strategy.start( gridgrid, initial_arraysarrays, num_roundsnum_rounds, ) if context.run_config[save-model]: # Save final model to disk bst xgb.Booster(paramsparams) global_model bytearray(result.arrays[0].numpy().tobytes()) # Load global model into booster bst.load_model(global_model) # Save model print(\nSaving final model to disk...) bst.save_model(final_model.json)实现要点全局模型以空字节串初始化。XGBoost 模型并不在服务端初始化而是在第一轮由客户端创建并训练因此initial_arrays传入的是一个空的ArrayRecord。实例化FedXgbBagging策略传入fraction_train与fraction_evaluate。从 fedxgb_bagging.py 的类文档看该策略还支持min_train_nodes、min_evaluate_nodes、min_available_nodes默认均为 2、weighted_by_key默认num-examples等参数本项目仅使用默认值。调用策略的start方法启动训练传入Grid、初始ArrayRecord与轮数num_rounds。start返回一个Result对象其中包含联邦过程的所有相关信息最终模型权重ArrayRecord以及联邦训练/评估指标MetricRecord。按需保存模型当save-model为 true 时把result.arrays[0]还原为Booster并调用xgb.save_model(final_model.json)落盘。值得一提的边界行为FedXgbBagging在聚合前会通过_ensure_single_array校验每个回复的ArrayRecord恰好只含一个 Array否则抛出InconsistentMessageReplies并跳过聚合见 fedxgb_bagging.py。这提醒我们在自定义客户端回复时务必保持“模型只存一个数组”的约定。服务端树拼接的底层原理服务端聚合是FedXgbBagging的灵魂其树拼接实现在aggregate_bagging中见 strategy_utils.py核心步骤为若上一轮全局模型为空b直接返回当前客户端模型作为起点解析上一轮模型与当前客户端模型的树数量_get_tree_nums统计num_trees与并行树数以 JSON 形式打开两个模型把当前客户端的并行树数量累加到全局模型的num_trees并追加iteration_indptr边界遍历当前客户端的每棵树重新编号id tree_num_prev tree_count后追加进全局模型的trees与tree_info列表序列化回字节流作为新的全局模型继续下一轮。配合客户端_local_boost只回传最后N棵新树的约定服务端与客户端共同构成完整的“本地长树、服务端接树”的 bagging 闭环全局模型就是所有客户端各轮新增树的串联集成。这一机制在FedXgbBagging.aggregate_train中被逐条应用最终把聚合结果放回ArrayRecord并在下一轮下发见 fedxgb_bagging.py。进一步探索仓库内examples/quickstart-xgboost是本文讲解代码的完整可运行版本包含task.py、client_app.py、server_app.py与pyproject.toml可直接对照阅读。若想进一步配置和运行 Flower 仿真可参考 how-to-run-simulations.rst想了解本地完整工作流SuperLink、SuperNode、Simulation Runtime 的角色可阅读 how-to-run-flower-locally.rst。需要更深度的联邦 XGBoost 玩法时可查看仓库中的 xgboost-comprehensive 示例含task.py、client_app.py、server_app.py以及分层联邦的 hfedxgboost baseline后者包含完整的 YAML 配置与实验结果文件results.csv、results_centralized.csv。至此你已经用 Flower XGBoost 跑通了第一个联邦学习系统从脚手架创建、数据切分、ClientApp 训练/评估到 ServerApp 的 FedXgbBagging 编排与树拼接聚合全链路均已打通。【免费下载链接】flowerFlower: A Friendly Federated AI Framework项目地址: https://gitcode.com/GitHub_Trending/flo/flower创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Flower + XGBoost 联邦学习快速上手：用 FedXgbBagging 在 Higgs 数据集上训练分类模型
Flower XGBoost 联邦学习快速上手用 FedXgbBagging 在 Higgs 数据集上训练分类模型【免费下载链接】flowerFlower: A Friendly Federated AI Framework项目地址: https://gitcode.com/GitHub_Trending/flo/flower本教程以 Flower 官方 Quickstart XGBoost 项目为蓝本讲解如何用 Flower 与 XGBoost 搭建一个基于FedXgbBaggingbootstrap aggregation策略的双节点联邦训练系统在 Higgs 数据集上训练二分类模型。读完本文你将掌握flwr new脚手架生成项目、flwr run本地仿真运行、pyproject.toml参数配置、XGBoost 模型与 FlowerMessage/ArrayRecord之间的二进制序列化转换以及ClientApp/ServerApp的完整实现与底层树聚合原理。背景为什么用 FedXgbBagging 做联邦 XGBoostXGBoost 的梯度提升树模型天然难以像神经网络那样对权重做平均因此 Flower 提供了一种适配树模型的联邦策略 ——FedXgbBaggingbootstrap aggregation装袋聚合。其核心思想是每个客户端对自己的本地数据做自助采样bootstrap subsampling每一轮训练一棵或多棵树服务端把各客户端本轮新增的树**拼接aggregate**进全局模型而不是加权平均参数全局模型随联邦轮次不断“长树”集成来自所有客户端的新树从而提升整体稳定性与精度。从源码看FedXgbBagging继承自FedAvg见 fedxgb_bagging.py但重写了configure_train与aggregate_train前者记录当前正在通信的全局模型字节流current_bst后者调用工具函数aggregate_bagging完成树级拼接。具体拼接逻辑位于 strategy_utils.py将上一轮全局模型与客户端返回的新树按 JSON 结构合并更新num_trees、iteration_indptr并把新树的 id 重编号后追加到全局模型树列表中。这就是“服务端只做树拼接、不做数值平均”的实现基础。环境准备与项目创建教程建议在独立的虚拟环境中运行。Flower 官方提供了虚拟环境搭建指南见 contributor-how-to-set-up-a-virtual-env.rst。环境就绪后先安装 Flower# In a new Python environment $ pip install flwr然后使用flwr new脚手架一键生成完整的 Flower XGBoost 项目。该命令会拉取 Flower Labs 官方模板并生成一个可直接运行的双节点联邦项目默认使用FedXgbBagging策略、本地仿真 profile$ flwr new flwrlabs/quickstart-xgboost执行后当前目录下会出现名为quickstart-xgboost的新目录结构如下quickstart-xgboost ├── quickstart_xgboost │ ├── __init__.py │ ├── client_app.py # Defines your ClientApp │ ├── server_app.py # Defines your ServerApp │ └── task.py # Defines your data loading and utility functions ├── pyproject.toml # Project metadata like dependencies and configs └── README.md仓库中已包含与该模板同源的完整实现可直接对照阅读examples/quickstart-xgboost 目录下的quickstart_xgboost/task.py、quickstart_xgboost/client_app.py、quickstart_xgboost/server_app.py与pyproject.toml。其中pyproject.toml声明了运行时依赖flwr[simulation]1.36.0、flwr-datasets0.6.1、xgboost2.0.0并通过[tool.flwr.app.components]把server_app:app与client_app:app绑定为应用入口见 pyproject.toml。运行联邦训练并查看流式日志进入项目目录用默认参数运行并流式输出日志$ cd quickstart-xgboost # Run with default arguments and stream logs $ flwr run . --stream说明普通flwr run .只提交运行、打印 run ID 后立即返回不会流式输出日志加--stream才会持续滚动显示训练过程。关于本地完整工作流的更多细节可参考 how-to-run-flower-locally.rst。默认参数下你会看到类似下面的流式输出Starting local SuperLink on 127.0.0.1:39091... Successfully started run 1859953118041441032 INFO : Starting FedXgbBagging strategy: INFO : ├── Number of rounds: 3 INFO : [ROUND 1/3] INFO : configure_train: Sampled 2 nodes (out of 2) INFO : aggregate_train: Received 2 results and 0 failures INFO : └── Aggregated MetricRecord: {} INFO : configure_evaluate: Sampled 2 nodes (out of 2) INFO : aggregate_evaluate: Received 2 results and 0 failures INFO : └── Aggregated MetricRecord: {auc: 0.7677505289821278} INFO : [ROUND 2/3] INFO : ... INFO : [ROUND 3/3] INFO : ... INFO : Strategy execution finished in 132.88s INFO : Final results: INFO : ServerApp-side Evaluate Metrics: INFO : {}从日志可以清晰看到运行机制的几个关键点flwr run会先在本机启动一个受管的本地 SuperLink默认监听127.0.0.1:39091再由 Flower Simulation Runtime 执行本次运行FedXgbBagging每轮从 2 个节点中采样 2 个参与训练与评估服务端聚合后输出MetricRecord其中auc即为每轮全局模型在验证集上的 AUC 指标首轮日志中约 0.7678。你还可以通过--run-config覆盖pyproject.toml中[tool.flwr.app.config]定义的任意参数例如# Override some arguments $ flwr run . --run-config num-server-rounds5 params.eta0.2这条命令把联邦轮数改为 5、XGBoost 学习率eta改为 0.2无需改动任何文件即可做参数实验。配置文件pyproject.toml中的超参数所有联邦与 XGBoost 超参数都集中在pyproject.toml的[tool.flwr.app.config]段[tool.flwr.app.config] num-server-rounds 3 fraction-train 0.1 fraction-evaluate 0.1 local-epochs 1 save-model false # XGBoost parameters params.objective binary:logistic params.eta 0.1 # Learning rate params.max-depth 8 params.eval-metric auc params.nthread 16 params.num-parallel-tree 1 params.subsample 1 params.tree-method hist各参数含义如下参数默认值说明num-server-rounds3联邦训练总轮数fraction-train0.1每轮参与训练的节点比例示例模板为 0.1仓库同源示例默认 1.0见 pyproject.tomlfraction-evaluate0.1每轮参与评估的节点比例local-epochs1本地树提升tree boost的迭代轮数即每轮每个客户端新训练几棵树save-modelfalse是否在训练结束后把最终模型保存到磁盘params.objectivebinary:logistic二分类逻辑回归目标函数params.eta0.1学习率learning rateparams.max-depth8树的最大深度params.eval-metricauc评估指标为 AUCparams.nthread16训练使用的 CPU 线程数params.num-parallel-tree1每轮并行训练树的数量params.subsample1训练样本的子采样比例params.tree-methodhist直方图近似算法默认在 CPU 上训练要点默认在 CPU 上训练若想改用 GPU只需把tree-method设为gpu_hist。local-epochs直接决定每个客户端每轮向服务端贡献的树数量它同时被ClientApp的本地提升循环与FedXgbBagging的树拼接逻辑使用下文详解。注意pyproject.toml中键名使用连字符如num-server-rounds而 XGBoost 原生参数名使用下划线如max_depth。因此源码中通过unflatten_dictreplace_keys两步把扁平配置还原成嵌套字典并把键中的-递归替换为_得到可直接传给xgb.train的params字典见 task.py。数据用 Flower Datasets 加载并切分 Higgs本项目使用 Flower Datasets 自动下载并切分Higgs数据集jxie/higgs并用IidPartitioner把训练集划分为num_partitions份分区数等于客户端总数实现 IID 数据分布。Flower Datasets 还提供其他 partitioner如 Dirichlet 分布的非 IID 切分可按需替换。客户端侧的数据加载逻辑如下partitioner IidPartitioner(num_partitionsnum_clients) fds FederatedDataset( datasetjxie/higgs, partitioners{train: partitioner}, ) partition fds.load_partition(partition_id, splittrain) partition.set_format(numpy) # Train/test splitting train_data, valid_data, num_train, num_val train_test_split( partition, test_fraction0.2, seed42 ) # Reformat data to DMatrix for xgboost train_dmatrix transform_dataset_to_dmatrix(train_data) valid_dmatrix transform_dataset_to_dmatrix(valid_data)流程分三步先按partition_id取回该客户端对应的数据分区再按 80/20 比例把本地分区拆成训练集与验证集test_fraction0.2固定随机种子seed42保证可复现最后转换为 XGBoost 所需的DMatrix格式。两个工具函数的完整实现见 task.pydef train_test_split(partition, test_fraction, seed): Split the data into train and validation set given split rate. train_test partition.train_test_split(test_sizetest_fraction, seedseed) partition_train train_test[train] partition_test train_test[test] num_train len(partition_train) num_test len(partition_test) return partition_train, partition_test, num_train, num_test def transform_dataset_to_dmatrix(data): Transform dataset to DMatrix format for xgboost. batch data[:] x np.asarray(batch[inputs], dtypenp.float32) y np.asarray(batch[label], dtypenp.float32) return xgb.DMatrix(x, labely)Higgs 数据集的inputs为 28 维特征、label为二分类标签与params.objective binary:logistic的目标函数相对应。为了让FederatedDataset只初始化一次load_data用模块级变量fds做了缓存见 task.py。ClientApp在Message与 XGBoost 模型之间做转换要把 XGBoost 接入 Flower核心改动在于把Message中收到的ArrayRecord还原成 XGBoost 可加载的二进制对象训练完再把模型序列化回ArrayRecord塞进回复Message。Flower 的ClientApp提供了三个可实现的装饰器方法train用本地数据训练收到的模型、evaluate在本地验证集上评估模型性能、query查询执行ClientApp的节点信息。本教程只用到train和evaluate。train本地提升 回传新树train方法接收两个参数来自ServerApp的Message以及Context。默认情况下该Message携带两部分内容一个ArrayRecord存放待联邦训练的模型数组默认通过 keyarrays访问一个ConfigRecord存放ServerApp下发的配置默认通过 keyconfig访问。Context则用于读取运行配置与节点配置run config 的超参数定义在 Flower App 的pyproject.toml中node config 只在 Deployment Runtime 下可设置仿真运行期间不能直接配置。完整实现见 client_app.py# Flower ClientApp app ClientApp() app.train() def train(msg: Message, context: Context) - Message: # Load model and data partition_id context.node_config[partition-id] num_partitions context.node_config[num-partitions] train_dmatrix, _, num_train, _ load_data(partition_id, num_partitions) # Read from run config num_local_round context.run_config[local-epochs] # Flatted config dict and replace - with _ cfg replace_keys(unflatten_dict(context.run_config)) params cfg[params] global_round msg.content[config][server-round] if global_round 1: # First round local training bst xgb.train( params, train_dmatrix, num_boost_roundnum_local_round, ) else: bst xgb.Booster(paramsparams) global_model bytearray(msg.content[arrays][0].numpy().tobytes()) # Load global model into booster bst.load_model(global_model) # Local training bst _local_boost(bst, num_local_round, train_dmatrix) # Save model local_model bst.save_raw(json) model_np np.frombuffer(local_model, dtypenp.uint8) # Construct reply message # Note: we store the model as the first item in a list into ArrayRecord, # which can be accessed using index [0]. model_record ArrayRecord([model_np]) metrics { num-examples: num_train, } metric_record MetricRecord(metrics) content RecordDict({arrays: model_record, metrics: metric_record}) return Message(contentcontent, reply_tomsg)第一轮server-round 1时服务端尚未有可用模型客户端直接用xgb.train()在本地数据上从零训练前num_local_round棵树。从第二轮起客户端读取服务端下发的全局模型字节流加载进新建的Booster再调用_local_boost在本地数据上继续提升def _local_boost(bst_input, num_local_round, train_dmatrix): # Update trees based on local training data. for i in range(num_local_round): bst_input.update(train_dmatrix, bst_input.num_boosted_rounds()) # Bagging: extract the last Nnum_local_round trees for sever aggregation bst bst_input[ bst_input.num_boosted_rounds() - num_local_round : bst_input.num_boosted_rounds() ] return bst_local_boost做两件事通过bst_input.update(dmatrix, num_boosted_rounds())迭代num_local_round次完成本地树提升随后用切片语法取出最后N num_local_round棵树作为本轮增量返回给服务端。这正是装袋聚合的关键 —— 客户端只上送本轮新增的树服务端把它们拼接到全局模型上见 client_app.py。回复Message的构造要点模型以ArrayRecord列表的第一个元素存放之后用msg.content[arrays][0]即可访问同时通过MetricRecord上报num-examples供服务端按样本量加权聚合指标。整个数据流可概括为Message → ArrayRecord → numpy → bytearray → bst.load_model → 本地提升 → bst.save_raw(json) → np.frombuffer → ArrayRecord → 回复 Message。evaluate本地验证集评估 AUCevaluate与train几乎相同只有两点差异其一不做本地训练直接用收到的全局模型在本地留出的验证集上评估其二回复Message无需再携带模型模型未被本地修改只上报指标。实现见 client_app.pyapp.evaluate() def evaluate(msg: Message, context: Context) - Message: # Load model and data partition_id context.node_config[partition-id] num_partitions context.node_config[num-partitions] _, valid_dmatrix, _, num_val load_data(partition_id, num_partitions) # Load config cfg replace_keys(unflatten_dict(context.run_config)) params cfg[params] # Load global model bst xgb.Booster(paramsparams) global_model bytearray(msg.content[arrays][0].numpy().tobytes()) bst.load_model(global_model) # Run evaluation eval_results bst.eval_set( evals[(valid_dmatrix, valid)], iterationbst.num_boosted_rounds() - 1, ) auc float(eval_results.split(\t)[1].split(:)[1]) # Construct and return reply Message metrics { auc: auc, num-examples: num_val, } metric_record MetricRecord(metrics) content RecordDict({metrics: metric_record}) return Message(contentcontent, reply_tomsg)评估使用bst.eval_set在验证集DMatrix上计算指标并从返回字符串中解析出auc数值连同num-examples一起放入MetricRecord返回。这就是日志中Aggregated MetricRecord: {auc: 0.7677505289821278}的来源。ServerApp用 FedXgbBagging 编排联邦轮次ServerApp的核心是app.main()方法它接收两个参数Grid用于与运行ClientApp的节点交互调度每一轮 train/evaluate/query与Context提供运行配置访问。完整实现见 server_app.py# Create ServerApp app ServerApp() app.main() def main(grid: Grid, context: Context) - None: # Read run config num_rounds context.run_config[num-server-rounds] fraction_train context.run_config[fraction-train] fraction_evaluate context.run_config[fraction-evaluate] # Flatted config dict and replace - with _ cfg replace_keys(unflatten_dict(context.run_config)) params cfg[params] # Init global model # Init with an empty object; the XGBooster will be created # and trained on the client side. global_model b # Note: we store the model as the first item in a list into ArrayRecord, # which can be accessed using index [0]. arrays ArrayRecord([np.frombuffer(global_model, dtypenp.uint8)]) # Initialize FedXgbBagging strategy strategy FedXgbBagging( fraction_trainfraction_train, fraction_evaluatefraction_evaluate, ) # Start strategy, run FedXgbBagging for num_rounds result strategy.start( gridgrid, initial_arraysarrays, num_roundsnum_rounds, ) if context.run_config[save-model]: # Save final model to disk bst xgb.Booster(paramsparams) global_model bytearray(result.arrays[0].numpy().tobytes()) # Load global model into booster bst.load_model(global_model) # Save model print(\nSaving final model to disk...) bst.save_model(final_model.json)实现要点全局模型以空字节串初始化。XGBoost 模型并不在服务端初始化而是在第一轮由客户端创建并训练因此initial_arrays传入的是一个空的ArrayRecord。实例化FedXgbBagging策略传入fraction_train与fraction_evaluate。从 fedxgb_bagging.py 的类文档看该策略还支持min_train_nodes、min_evaluate_nodes、min_available_nodes默认均为 2、weighted_by_key默认num-examples等参数本项目仅使用默认值。调用策略的start方法启动训练传入Grid、初始ArrayRecord与轮数num_rounds。start返回一个Result对象其中包含联邦过程的所有相关信息最终模型权重ArrayRecord以及联邦训练/评估指标MetricRecord。按需保存模型当save-model为 true 时把result.arrays[0]还原为Booster并调用xgb.save_model(final_model.json)落盘。值得一提的边界行为FedXgbBagging在聚合前会通过_ensure_single_array校验每个回复的ArrayRecord恰好只含一个 Array否则抛出InconsistentMessageReplies并跳过聚合见 fedxgb_bagging.py。这提醒我们在自定义客户端回复时务必保持“模型只存一个数组”的约定。服务端树拼接的底层原理服务端聚合是FedXgbBagging的灵魂其树拼接实现在aggregate_bagging中见 strategy_utils.py核心步骤为若上一轮全局模型为空b直接返回当前客户端模型作为起点解析上一轮模型与当前客户端模型的树数量_get_tree_nums统计num_trees与并行树数以 JSON 形式打开两个模型把当前客户端的并行树数量累加到全局模型的num_trees并追加iteration_indptr边界遍历当前客户端的每棵树重新编号id tree_num_prev tree_count后追加进全局模型的trees与tree_info列表序列化回字节流作为新的全局模型继续下一轮。配合客户端_local_boost只回传最后N棵新树的约定服务端与客户端共同构成完整的“本地长树、服务端接树”的 bagging 闭环全局模型就是所有客户端各轮新增树的串联集成。这一机制在FedXgbBagging.aggregate_train中被逐条应用最终把聚合结果放回ArrayRecord并在下一轮下发见 fedxgb_bagging.py。进一步探索仓库内examples/quickstart-xgboost是本文讲解代码的完整可运行版本包含task.py、client_app.py、server_app.py与pyproject.toml可直接对照阅读。若想进一步配置和运行 Flower 仿真可参考 how-to-run-simulations.rst想了解本地完整工作流SuperLink、SuperNode、Simulation Runtime 的角色可阅读 how-to-run-flower-locally.rst。需要更深度的联邦 XGBoost 玩法时可查看仓库中的 xgboost-comprehensive 示例含task.py、client_app.py、server_app.py以及分层联邦的 hfedxgboost baseline后者包含完整的 YAML 配置与实验结果文件results.csv、results_centralized.csv。至此你已经用 Flower XGBoost 跑通了第一个联邦学习系统从脚手架创建、数据切分、ClientApp 训练/评估到 ServerApp 的 FedXgbBagging 编排与树拼接聚合全链路均已打通。【免费下载链接】flowerFlower: A Friendly Federated AI Framework项目地址: https://gitcode.com/GitHub_Trending/flo/flower创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考