无需重构开销的联邦学习

最有价值的数据往往最难以移动。监管边界、数据主权规则以及组织风险承受能力通常阻碍集中聚合。与此同时，仅仅是数据引力就使得即使是被允许的传输也变得缓慢、昂贵且在大规模时脆弱。

最新版本的 NVIDIA FLARE 通过一个联邦学习计算运行时来应对这一现实，它将训练逻辑移动到数据所在之处，而原始数据则保持原地。查看如何在联邦学习系统中利用 PyTorch 的示例。

阅读全文：

使用 NVIDIA FLARE 实现无需重构开销的联邦学习

来源：https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/ 联邦学习不再是研究上的新奇事物——它是应对一个硬约束的实际响应：最有价值的数据往往最难以移动。监管边界、数据主权规则以及组织风险承受能力通常阻碍集中聚合。与此同时，仅仅是数据引力就使得即使是被允许的传输也变得缓慢、昂贵且在大规模时脆弱。

最新版本的 NVIDIA FLARE 通过一个联邦计算运行时来应对这一现实，它将训练逻辑移动到数据所在之处，而原始数据则保持原地。在高风险环境中，集中聚合数据通常不可能或不切实际，因此一个现代的联邦平台必须将数据隔离、合规性和隐私增强技术作为首要需求。

历史上阻碍采用的并不是联邦学习的概念本身，而是开发者体验。如果从“我的本地脚本可以训练”到“我的任务在联邦站点上运行”需要深入的重构、新的类层次结构或脆弱的配置，许多项目在试点阶段就会停滞。

FLARE API 的演进正是针对这一点：通过将工作拆分为两个具体步骤来消除重构开销，这些步骤清晰地映射到团队实际构建和交付机器学习系统的方式上：

• 步骤 1（客户端 API）： 用大约 5-6 行代码将现有的本地训练脚本转换为联邦客户端，无需更改训练循环结构。
• 步骤 2（任务配方）： 选择联邦工作流并将其绑定到客户端训练脚本，然后通过仅交换执行环境，在模拟、概念验证和生产中运行同一任务。

‘无数据拷贝’ 作为系统需求

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#%E2%80%98no_data_copy%E2%80%99_as_a_system_requirement

在受监管或高敏感度环境中，“集中数据集”越来越被排除在外。一个实用的联邦计算平台需要支持：

• 无数据拷贝： 数据保留在本地，仅移动模型更新（或等效信号）。
• 合规姿态： 支持主权和审计要求的部署和治理控制。
• 隐私增强技术： 多层防御（示例包括同态加密、差分隐私和机密计算）。

图示展示了集中式与联邦计算的前后对比。左侧（“之前”），三个独立数据孤岛将数据发送到单一集中数据库，在此训练模型。右侧（“之后”），数据保留在多个站点的独立锁定数据库中，而共享模型跨站点进行协调，箭头表示仅交换模型更新，而非复制原始数据。中间显示了不同行业（如金融、医疗和公共部门）的数据孤岛。
图 1. 联邦计算将数据保留在本地，通过模型更新实现协作，同时支持合规和隐私增强保护。

重构悬崖：联邦学习项目为何停滞

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#the_refactoring_cliff_why_fl_projects_stall

团队通常在试点后遇到以下两个悬崖之一：

• 代码悬崖： 将现有的 PyTorch/TensorFlow/Lightning 训练转换为联邦学习可能需要侵入式重构——新的抽象层、消息粘合代码以及框架特定的脚手架。
• 生命周期悬崖： 即使模拟成功，迁移到概念验证和生产阶段也会通过任务重定义、重新配置和环境特定分支引发重写。

FLARE 通过将工作流标准化为两个步骤来平坦化这两个悬崖：

1. 让你的脚本联邦化（客户端 API）
2. 将其作为可移植任务执行（任务配方）

预期的体验明确是组合这两者，以便你能够快速从零开始到运行一个联邦任务。

步骤 1：将本地训练脚本转换为联邦客户端（客户端 API）

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#step_1_convert_your_local_training_script_into_a_federated_client_client_api

适用对象：拥有现有训练代码并希望最小化差异的实践者和机器学习工程师。

心智模型有意保持简单：

1. 初始化客户端运行时
2. 在任务运行时循环
3. 接收当前全局模型
4. 本地训练（你的代码）
5. 发送更新后的权重和指标

FLARE 的客户端 API 旨在实现最小代码更改，并避免强制你使用繁重的“Executor/Learner”继承——使用 FLModel 结构或简单的数据交换与运行时通信。

示例 1a：将 PyTorch 转换为 FLARE

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#example_1a_convert_pytorch_to_flare_

以下是你可应用于许多脚本的具体模式。关键接触点是：flare.init()、flare.receive()、加载模型权重，以及使用更新后的权重和指标调用 flare.send()。

我们在左侧显示本地训练代码，右侧显示联邦版本，并突出显示：导入、flare.init()、receive()、send()。

train.py

# train.py

import torch
import torchvision
import torchvision.transforms as transforms

from model import Net

batch_size = 4
epochs = 1
lr = 0.01
model = Net()
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
loss = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr=lr, momentum=0.9)
transform = transforms.Compose(
   [
       transforms.ToTensor(),
       transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),
   ]
)

train_dataset = torchvision.datasets.CIFAR10(
   root="/tmp/data/cifar10", transform=transform, download=True, train=True
)

trainloader = torch.utils.data.DataLoader(
   train_dataset, batch_size=batch_size, shuffle=True
)

model.to(device)

for epoch in range(epochs):
   running_loss = 0.0

   for i, batch in enumerate(trainloader):
       images, labels = batch[0].to(device), batch[1].to(device)

       optimizer.zero_grad()

       predictions = model(images)
       cost = loss(predictions, labels)
       cost.backward()
       optimizer.step()

       running_loss += cost.cpu().detach().numpy() / batch_size

       if i % 3000 == 2999:
           print(
               f"Epoch: {epoch + 1}/{epochs}, batch: {i + 1}, Loss: {running_loss / 3000}"
           )
           running_loss = 0.0

   print(
       f"Epoch: {epoch + 1}/{epochs}, batch: {i + 1}, Loss: {running_loss / (i + 1)}"
   )

print("Finished Training")

torch.save(model.state_dict(), "./cifar_net.pth")

client.py

# client.py

# 1. Import client API
import nvflare.client as flare
import torch
import torchvision
import torchvision.transforms as transforms

from model import Net

batch_size = 4
epochs = 1
lr = 0.01
model = Net()
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
loss = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr=lr, momentum=0.9)
transform = transforms.Compose(
   [
       transforms.ToTensor(),
       transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),
   ]
)

train_dataset = torchvision.datasets.CIFAR10(
   root="/tmp/data/cifar10", transform=transform, download=True, train=True
)

trainloader = torch.utils.data.DataLoader(
   train_dataset, batch_size=batch_size, shuffle=True
)

# 2. Initialize FLARE
flare.init()

# At each round while FLARE is running
while flare.is_running():
   # 3. Receive the global model
   input_model = flare.receive()

   # 4. Load global model
   model.load_state_dict(input_model.params)
   model.to(device)

   for epoch in range(epochs):
       running_loss = 0.0

       for i, batch in enumerate(trainloader):
           images, labels = batch[0].to(device), batch[1].to(device)

           optimizer.zero_grad()

           predictions = model(images)
           cost = loss(predictions, labels)
           cost.backward()
           optimizer.step()

           running_loss += cost.cpu().detach().numpy() / batch_size

           if i % 3000 == 2999:
               print(
                   f"Epoch: {epoch + 1}/{epochs}, batch: {i + 1}, Loss: {running_loss / 3000}"
               )
               running_loss = 0.0

       print(
           f"Epoch: {epoch + 1}/{epochs}, batch: {i + 1}, Loss: {running_loss / (i + 1)}"
       )

   print("Finished Training")

   torch.save(model.state_dict(), "./cifar_net.pth")

   # 5. Send back the updated model
   output_model = flare.FLModel(
       params=model.cpu().state_dict(),
       meta={"NUM_STEPS_CURRENT_ROUND": len(trainloader) * epochs},
   )
   flare.send(output_model)

示例 1b：PyTorch Lightning 客户端

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#example_1b_pytorch_lightning_client_the_lightning_integration_keeps_the_same

Lightning 集成保持了相同的意图——接收全局模型、训练、发送更新——但以一种 Lightning 友好的方式暴露出来：导入 Lightning 客户端适配器并修补 Trainer。

典型流程是：导入、修补、（可选）验证、照常训练。

# lightning_client.py
import pytorch_lightning as pl
from pytorch_lightning import Trainer

import nvflare.client.lightning as flare  # Lightning Client API  

from model import LitNet
from data import CIFAR10DataModule
def main():
   model = LitNet()
   dm = CIFAR10DataModule()

   trainer = Trainer(max_epochs=1, accelerator="gpu", devices=1)

   # Patch trainer to participate in FL
   flare.patch(trainer)

   while flare.is_running():
       # Optional: validate current global model (useful for server-side selection flows)
       trainer.validate(model, datamodule=dm)

       # Train starting from received global model (handled internally after patch)
       trainer.fit(model, datamodule=dm)

if __name__ == "__main__":
   main()

重点：Lightning 用户无需降级到自定义联邦消息传递——他们保持 Trainer 抽象，仍然正确参与联邦轮次。

步骤 2：在任何地方打包并执行联邦任务（任务配方）

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#step_2_package_and_execute_the_federated_job_anywhere_job_recipes

适用对象： 希望使用代码优先的任务定义，且该定义在不同环境中保持稳定的数据科学家和应用团队。

在步骤 1 之后，你拥有了一个联邦客户端脚本。步骤 2 使其成为一个联邦任务，你可以重复运行并干净地跨生命周期移动。

任务配方旨在用基于 Python 的任务定义替代基于 JSON 的任务配置：

• 代码优先： 用 Python 定义完整的联邦任务，而不是复杂的配置文件
• 一次编写，随处运行： 相同的配方可在模拟器、概念验证或生产中运行
• 加速部署： 无需更改代码结构，即可从实验阶段进入部署阶段

示例 2a：在模拟中执行 FedAvg 配方

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#example_2a_execute_a_fedavg_recipe_in_simulation

关键的连接点是你的配方引用了你在步骤 1 中创建的客户端训练脚本（例如 train_script="client.py"），然后你在一个环境中执行它。

# job.py
from nvflare.app_common.workflows.job import FedAvgRecipe
from nvflare.job_config import SimEnv  # exact import path can vary by NVFlare version

from model import SimpleNetwork

def main():
   n_clients = 3
   num_rounds = 5
   batch_size = 32

   recipe = FedAvgRecipe(
       name="hello-pt",
       min_clients=n_clients,
       num_rounds=num_rounds,
       model=SimpleNetwork(),
       train_script="client.py",  # <-- Step A script
       train_args=f"--batch_size {batch_size} --epochs 1",
   )

   env = SimEnv(num_clients=n_clients, num_threads=n_clients)
   recipe.execute(env=env)

if __name__ == "__main__":
   main()

这就是“一次编写”理念的实践：一旦配方正确引用了你的客户端脚本，其余部分就变成了执行问题。

示例 2b：通过环境交换从模拟迁移到现实世界

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#example_2b_move_from_simulation_to_real-world_with_an_environment_swap%C2%A0

任务配方通过交换执行环境来形式化一个渐进式工作流：

1. SimEnv（模拟）： 轻松开发，快速调试
2. PocEnv（概念验证）： 本地运行时，多进程，实际测试
3. ProdEnv（生产）： 在安全的可扩展基础设施上进行分布式部署

替代文本：图示显示了一个三阶段 JobRecipe 管道流向三个执行环境。一个标有“JobRecipe”的盒子顶部引出三个箭头，指向并排的面板：SimEnv（模拟）用于轻松开发和快速调试，PocEnv（概念验证）用于本地运行时中真实的多进程测试，ProdEnv（生产）用于安全的分布式部署。
图 2. 一个任务配方，多种执行环境：在 SimEnv 中调试，在 PocEnv 中验证，在 ProdEnv 中部署，无需重写任务定义。

开始使用

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#getting_started

• 从你已信任的脚本开始。
• 步骤 1： 添加客户端 API 握手（或修补你的 Lightning Trainer）。
• 步骤 2： 将其封装在任务配方中，首先在模拟中执行，然后概念验证，最后通过交换环境进入生产。

FLARE 在新闻中

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#flare_in_the_news

FLARE 正在真实部署中亮相——从 Eli Lilly TuneLab 的联邦学习平台（由 Rhino Federated Computing 使用 NVFlare 构建）到 台湾卫生福利部的国家医疗联邦学习计划，以及 三实验室（Sandia/LANL/LLNL） 跨敏感数据集的联邦 AI 试点。

更进一步

https://developer.nvidia.com/blog/federated-learning-without-the-refactoring-overhead-using-nvidia-flare/#going_further

从你已信任的脚本开始。添加最小的 FLARE 客户端握手（接收 → 训练 → 发送）。然后，当你准备好时，从单节点模拟扩展到多站点部署。

• 从这里开始： 你好世界示例（实现第一次联邦运行的最快路径）——NVFlare Hello World
• 观看演示： 查看简化的 API 栈在行动