个性化联邦学习方法
全局联邦模型可能不适合所有参与方。个性化联邦学习为每个参与方生成定制化模型，平衡全局知识与本地适配。
个性化方法分类
| 类别 | 代表方法 | 原理 | 个性化程度 |
|---|---|---|---|
| 元学习 | Per-FedAvg | 学习好的初始化 | 中 |
| 多任务学习 | FedMTL | 每方一个任务 | 高 |
| 基底+头部 | FedPer | 共享基底，私有头部 | 中高 |
| 聚类 | IFCA | 相似方共享模型 | 中 |
| 正则化 | FedRep | 表示共享，分类器私有 | 中 |
FedPer：基底+头部
python
class FedPerModel(nn.Module):
"""FedPer: 共享特征提取器 + 私有分类头"""
def __init__(self, backbone, num_classes_per_client):
super().__init__()
self.backbone = backbone # 全局共享
def get_backbone(self):
return self.backbone
def create_head(self, num_classes):
"""为每个客户端创建私有分类头"""
return nn.Linear(self.backbone.hidden_dim, num_classes)
class FedPerClient:
def __init__(self, model, num_classes):
self.backbone = model.get_backbone()
self.head = model.create_head(num_classes)
def local_train(self, dataloader, lr=0.01):
"""本地训练：更新 backbone 和 head"""
optimizer = torch.optim.SGD(
list(self.backbone.parameters()) + list(self.head.parameters()),
lr=lr
)
for batch in dataloader:
features = self.backbone(batch["x"])
logits = self.head(features)
loss = F.cross_entropy(logits, batch["y"])
loss.backward()
optimizer.step()
optimizer.zero_grad()
# 只上传 backbone 参数
return self.backbone.state_dict()元学习：Per-FedAvg
python
class PerFedAvg:
"""Per-FedAvg: 元学习联邦"""
def __init__(self, model, alpha=0.01, beta=0.01):
self.model = model
self.alpha = alpha # 内循环学习率
self.beta = beta # 外循环学习率
def meta_update(self, support_data, query_data):
"""MAML 风格的元更新"""
# 内循环：在 support set 上适应
adapted_params = self.adapt(support_data, self.alpha)
# 外循环：在 query set 上评估
query_loss = self.evaluate(query_data, adapted_params)
# 计算元梯度
meta_grads = torch.autograd.grad(query_loss, self.model.parameters())
# 更新全局模型
for param, grad in zip(self.model.parameters(), meta_grads):
param.data -= self.beta * grad聚类联邦
python
class IFCA:
"""IFCA: 联邦聚类"""
def __init__(self, num_clusters=3):
self.num_clusters = num_clusters
self.cluster_models = [] # 每个聚类一个模型
def assign_cluster(self, client_data, client_model):
"""将客户端分配到最近的聚类"""
min_loss = float('inf')
best_cluster = 0
for c, model in enumerate(self.cluster_models):
loss = self.evaluate(client_data, model)
if loss < min_loss:
min_loss = loss
best_cluster = c
return best_cluster
def aggregate(self, cluster_id, client_updates):
"""在聚类内聚合"""
self.cluster_models[cluster_id] = FedAvg.aggregate(client_updates)个性化方法选择
- 参与方数据分布差异大：FedPer 或聚类
- 快速适应新任务：Per-FedAvg（元学习）
- 参与方数量多且可分组：IFCA
- 追求简单实现：FedRep（只共享表示层）
个性化 vs 全局
个性化程度越高，越偏离全局知识，可能导致：1) 过拟合本地数据；2) 丢失跨方知识；3) 需要更多本地数据。需要在个性化和泛化间权衡。