Metaphor

Personalized Federated Learning

11分钟阅读

个性化联邦学习

个性化联邦学习旨在为每个客户端训练定制化的模型，在保护隐私的同时适应各客户端独特的数据分布。本章系统介绍个性化联邦学习的理论基础、核心算法和最新进展。

1. 个性化动机

1.1 全局模型的局限

在标准联邦学习中，所有客户端共享一个全局模型。然而：

这个全局最优解 对于某些客户端可能并非最优：

  • 医疗场景：不同医院的病例分布差异显著
  • 金融场景：不同银行的客户特征各异
  • 移动设备：不同用户的打字习惯不同

1.2 个性化-协作权衡

个性化联邦学习需要平衡两个目标：

目标描述权衡
协作收益从其他客户端学习共享知识数据量少、分布单一的客户端更需要
本地适应适应本地数据分布数据丰富、分布独特的客户端更需要

2. 个性化方法分类

2.1 方法分类体系

个性化联邦学习方法
│
├─ 基于全局模型
│   ├─ 全局微调
│   ├─ 知识蒸馏
│   └─ 元学习
│
├─ 基于本地模型
│   ├─ 完全个性化
│   ├─ 部分个性化
│   └─ 聚类方法
│
└─ 混合方法
    ├─ 全局+本地参数分离
    └─ LoRA适配器

2.2 每类方法的优缺点

方法类别优点缺点
全局微调简单、可解释通信开销大
元学习快速适应收敛困难
完全个性化最佳本地适应无协作收益
部分个性化平衡协作与适应参数划分敏感
LoRA适配器高效、灵活秩选择困难

3. 基于元学习的方法

3.1 Per-FedAvg

Per-FedAvg1是首个将元学习应用于联邦学习的个性化方法。

核心思想：学习一个良好的全局初始化，使本地微调能够快速收敛。

数学框架：

  1. 内层优化（客户端本地）：
  1. 外层优化（服务器端）：

其中 是内层学习率。

def per_fedavg_one_round(global_model, client_loaders, 
                         alpha=0.01, beta=0.001):
    """
    Per-FedAvg一轮更新
    """
    client_updates = []
    
    for client_id, data_loader in enumerate(client_loaders):
        # 复制全局模型到客户端
        local_model = copy.deepcopy(global_model)
        
        # 内层优化：本地微调
        optimizer = torch.optim.SGD(local_model.parameters(), lr=alpha)
        for batch in data_loader:
            optimizer.zero_grad()
            loss = compute_loss(local_model, batch)
            loss.backward()
            optimizer.step()
        
        # 保存本地微调后的参数
        theta_k = get_parameters(local_model)
        client_updates.append(theta_k)
        
        # 计算元梯度（在另一个batch上）
        # ...
    
    # 外层优化：聚合元梯度
    meta_grad = aggregate_meta_gradients(client_updates)
    global_model = update(global_model, beta * meta_grad)
    
    return global_model

3.2 Meta-Learning with Memory

class FedMem:
    """
    带有记忆的联邦元学习
    """
    def __init__(self, model, memory_size=100):
        self.global_model = model
        self.client_memory = {}  # 每个客户端的记忆
    
    def local_update(self, client_id, data_loader):
        """使用记忆增强的本地更新"""
        if client_id not in self.client_memory:
            self.client_memory[client_id] = []
        
        local_model = copy.deepcopy(self.global_model)
        optimizer = torch.optim.Adam(local_model.parameters())
        
        # 首先在记忆数据上训练
        for batch in self.client_memory[client_id]:
            optimizer.zero_grad()
            loss = compute_loss(local_model, batch)
            loss.backward()
            optimizer.step()
        
        # 然后在当前数据上训练
        for batch in data_loader:
            optimizer.zero_grad()
            loss = compute_loss(local_model, batch)
            loss.backward()
            optimizer.step()
        
        return local_model.state_dict()

4. 部分个性化方法

4.1 核心思想

部分个性化2将模型参数分为全局部分和本地部分：

其中：

  • ：所有客户端共享的全局参数
  • ：客户端 的本地参数

4.2 PFedHB

class PFedHB:
    """
    Per-FedAvg with Hypernetworks and Balanced weights
    """
    def __init__(self, model, global_layers, local_layers):
        self.global_params = model[global_layers]
        self.local_params = model[local_layers]
        self.hypernet = HyperNetwork(global_params)
    
    def get_personalized_model(self, client_id):
        """生成客户端个性化的模型"""
        # 使用超网络生成客户端特定的本地参数
        local_specific = self.hypernet(
            torch.tensor([client_id])
        )
        return {
            'global': self.global_params,
            'local': local_specific
        }

4.3 Partial Personalized FL公式

过个性化问题：当本地参数主导时，可能导致每个客户端完全独立训练，失去协作收益。

过个性化边界：

其中 是正则化参数，控制本地化的程度。

5. LoRA在联邦学习中的应用

5.1 FedLora

class LoraLinear(nn.Module):
    """LoRA适配器层"""
    def __init__(self, in_features, out_features, rank=4, alpha=1):
        super().__init__()
        self.rank = rank
        self.alpha = alpha
        
        # 冻结原始权重
        self.weight = nn.Parameter(
            torch.randn(out_features, in_features), 
            requires_grad=False
        )
        
        # LoRA低秩分解
        self.lora_A = nn.Parameter(torch.randn(rank, in_features))
        self.lora_B = nn.Parameter(torch.zeros(out_features, rank))
    
    def forward(self, x):
        return x @ self.weight.T + self.lora_B @ self.lora_A * (self.alpha / self.rank)
 
class FedLoraModel:
    """
    联邦学习中的LoRA模型
    """
    def __init__(self, base_model, lora_rank=4, lora_alpha=1):
        self.base_model = base_model
        self.lora_rank = lora_rank
        self.lora_alpha = lora_alpha
        
        # 冻结基座模型
        for param in self.base_model.parameters():
            param.requires_grad = False
        
        # 替换指定层为LoRA层
        self._apply_lora()
    
    def _apply_lora(self):
        """将指定层替换为LoRA层"""
        for name, module in self.base_model.named_modules():
            if 'attention' in name or 'ffn' in name:
                # 替换为LoRA层
                old_layer = module
                new_layer = LoraLinear(
                    old_layer.in_features,
                    old_layer.out_features,
                    rank=self.lora_rank,
                    alpha=self.lora_alpha
                )
                # 复制原始权重
                new_layer.weight.data = old_layer.weight.data
                replace_module(self.base_model, name, new_layer)

5.2 PF2LoRA

PF2LoRA3提出两级LoRA适配：

class TwoLevelLora:
    """
    PF2LoRA: 两级LoRA适配
    - 第一级：所有客户端共享的通用适配器
    - 第二级：客户端特定的个性化适配器
    """
    def __init__(self, base_model, rank_common=8, rank_personal=2):
        self.base_model = base_model
        self.common_lora = {}  # 共享LoRA参数
        self.personal_lora = {}  # 个性化LoRA参数
    
    def local_update(self, client_id, data_loader, 
                     lr=0.01, epochs=5):
        """客户端本地更新"""
        # 初始化客户端个性化LoRA
        if client_id not in self.personal_lora:
            self.personal_lora[client_id] = self._init_personal_lora()
        
        optimizer = torch.optim.Adam([
            {'params': self.common_lora.values()},
            {'params': self.personal_lora[client_id].values()}
        ], lr=lr)
        
        for epoch in range(epochs):
            for batch in data_loader:
                optimizer.zero_grad()
                loss = self.compute_loss(client_id, batch)
                loss.backward()
                optimizer.step()
        
        return {
            'common': self.common_lora,
            'personal': self.personal_lora[client_id]
        }
    
    def aggregate(self, client_updates):
        """聚合LoRA参数"""
        # 聚合通用LoRA
        for key in self.common_lora:
            aggregated = torch.mean(
                torch.stack([u['common'][key] for u in client_updates]),
                dim=0
            )
            self.common_lora[key] = aggregated
        
        # 个性化LoRA保持本地
        return self.common_lora

5.3 pFedGPT

pFedGPT4使用贝叶斯优化进行层级LoRA聚合：

class pFedGPT:
    """
    pFedGPT: 分层贝叶斯优化LoRA聚合
    """
    def __init__(self, model, module_names):
        self.model = model
        self.module_names = module_names
        self.aggregation_weights = {name: None for name in module_names}
    
    def hierarchical_bayesian_optimization(self, client_updates, 
                                           client_data_info):
        """
        层级贝叶斯优化
        """
        for module_name in self.module_names:
            # 收集该模块的所有客户端更新
            module_updates = [u[module_name] for u in client_updates]
            
            # 贝叶斯优化搜索最优权重
            def objective(weights):
                # 重构聚合模型
                aggregated = sum(
                    w * u for w, u in zip(weights, module_updates)
                )
                # 评估在验证集上的性能
                return -evaluate(aggregated, validation_data)
            
            # 多保真度评估
            weights = bayesian_optimize(
                objective,
                n_initial=10,
                n_iterations=20,
                fidelity_levels=[0.1, 0.5, 1.0]  # 课程学习策略
            )
            
            self.aggregation_weights[module_name] = weights
        
        return self.aggregation_weights

6. 聚类方法

6.1 IFCA

class IFCA:
    """
    IFCA: Iterative Federated Cluster Optimization
    """
    def __init__(self, model, num_clusters=3):
        self.num_clusters = num_clusters
        self.cluster_centers = [
            copy.deepcopy(model) for _ in range(num_clusters)
        ]
    
    def local_iteration(self, client_model, data_loader):
        """找到客户端最近的聚类中心"""
        best_cluster = 0
        best_loss = float('inf')
        
        for cluster_id, center in enumerate(self.cluster_centers):
            # 用聚类中心初始化客户端模型
            load_parameters(client_model, center)
            
            # 评估损失
            loss = evaluate(client_model, data_loader)
            
            if loss < best_loss:
                best_loss = loss
                best_cluster = cluster_id
        
        return best_cluster
    
    def update_centers(self, client_assignments, client_updates):
        """更新聚类中心"""
        for cluster_id in range(self.num_clusters):
            cluster_clients = [
                u for a, u in zip(client_assignments, client_updates)
                if a == cluster_id
            ]
            
            if cluster_clients:
                # 更新为该聚类客户端的平均
                self.cluster_centers[cluster_id] = average_models(
                    cluster_clients
                )

7. 知识蒸馏方法

7.1 FedMD

class FedMD:
    """
    FedMD: Federated Model Distillation
    """
    def __init__(self, model, distillation_data):
        self.global_model = model
        self.public_data = distillation_data  # 共享的蒸馏数据
    
    def local_knowledge_distillation(self, client_model, client_data):
        """本地知识蒸馏"""
        # 步骤1：全局模型在公共数据上的"软标签"
        with torch.no_grad():
            global_soft_labels = self.global_model(self.public_data)
        
        # 步骤2：客户端在公共数据上训练，同时学习硬标签和软标签
        optimizer = torch.optim.Adam(client_model.parameters())
        
        for batch in self.public_data:
            optimizer.zero_grad()
            
            # 硬标签损失
            hard_loss = F.cross_entropy(
                client_model(batch), 
                batch.labels
            )
            
            # 软标签损失（知识蒸馏）
            soft_loss = F.kl_div(
                F.log_softmax(client_model(batch) / T),
                F.softmax(global_soft_labels / T),
                reduction='batchmean'
            ) * (T ** 2)
            
            loss = hard_loss + lambda_ * soft_loss
            loss.backward()
            optimizer.step()
        
        return client_model.state_dict()

8. 实验对比

8.1 CIFAR-10非IID设置

方法α=0.1α=0.3α=0.5α=0.7
FedAvg45.2%62.3%71.5%78.2%
Per-FedAvg52.1%67.8%74.3%79.8%
PFedHB55.3%70.2%76.1%81.2%
PF2LoRA58.7%72.5%78.4%82.9%
pFedGPT61.2%74.8%80.1%83.7%

α=0.1表示极端非IID，α=1.0表示IID

8.2 通信效率对比

方法通信量/轮可调参数
全局微调完整模型全部
Per-FedAvg完整模型全部
LoRA低秩矩阵r × (d_in + d_out)
PF2LoRA两级低秩r₁ + r₂ × n
部分个性化全局+本地取决于划分

9. 总结

个性化联邦学习通过多种策略平衡协作收益和本地适应需求：

  1. 元学习方法：学习良好初始化，支持快速本地微调
  2. 部分个性化：分离全局共享和本地特定参数
  3. LoRA适配器：高效的低秩参数化方案
  4. 聚类方法：发现相似客户端组
  5. 知识蒸馏：通过软标签传递知识

选择合适方法的关键考虑因素：

  • 非IID程度
  • 客户端数量和可用性
  • 通信带宽限制
  • 个性化需求强度

参考资料

相关主题：[federated-learning-fundamentals]、[federated-learning-non-iid-strategies]、[lora]

Footnotes

  1. Fallah et al. “Personalized Federated Learning with Theoretical Guarantees: A Model-Agnostic Meta-Learning Approach” (NeurIPS 2020) ↩

  2. Mishchenko et al. “Partially Personalized Federated Learning: Breaking the Curse of Data Heterogeneity” (TMLR 2025) ↩

  3. PF2LoRA: “Personalized Federated Fine-tuning for Heterogeneous Data: An Automatic Rank Learning Approach via Two-Level LoRA” (arXiv:2503.03920) ↩

  4. pFedGPT: “Hierarchically Optimizing LoRA Aggregation Weights for Personalized Federated GPT Models” (EMNLP 2025) ↩

关系图谱