第一章：为什么需要联邦学习？

1.1 数据孤岛与隐私困境

行业	数据价值	隐私约束

• 医疗 | 多中心数据提升诊断准确率 | 患者病历严禁外传
• 金融 | 跨机构行为识别欺诈 | 客户交易记录高度敏感
• IoT | 海量设备数据优化体验 | 用户语音/图像本地存储

1.2 联邦学习 vs 传统方案

方案	隐私性	模型性能	合规性

• 数据集中训练 | ❌ 高风险 | ✅ 最优 | ❌ 违反 GDPR
• 本地独立训练 | ✅ 安全 | ❌ 差（小样本） | ✅ 合规
• 联邦学习 | ✅ 仅交换加密参数 | ✅ 接近集中式 | ✅ 合规 |

关键突破：打破“隐私-效用”权衡。

---

第二章：联邦学习架构设计

2.1 整体流程（横向联邦）

[协调服务器 (Flask + Flower)]
        ↑↓ 加密模型参数
[客户端 1: 医院 A] ←→ [客户端 2: 医院 B] ←→ ... ←→ [客户端 N: 医院 E]
（本地数据 never leave）

2.2 技术栈选型

组件	技术	说明

• 联邦框架 | Flower（Python） | 轻量、灵活、支持 PyTorch/TensorFlow
• 加密协议 | Secure Aggregation（SecAgg） | 客户端间 Diffie-Hellman 密钥交换
• 差分隐私 | Opacus（PyTorch） | 在梯度中添加噪声
• 前端 | Vue 3 + Chart.js | 参与方监控仪表盘

为何不用 TensorFlow Federated？：Flower 更易集成到现有 Flask 应用，且支持异构客户端。

---

第三章：协调服务器实现（Flask + Flower）

3.1 Flower 服务封装

# services/federated_server.py
import flwr as fl
from flwr.server import ServerConfig
from flwr.server.strategy import FedAvg

class FederatedCoordinator:
    def __init__(self, model_path: str, rounds: int = 5):
        self.strategy = FedAvg(
            fraction_fit=1.0,  # 所有在线客户端参与
            min_fit_clients=3, # 最少3家医院
            evaluate_fn=self.get_evaluate_fn()
        )
        self.config = ServerConfig(num_rounds=rounds)

    def start(self):
        fl.server.start_server(
            server_address="0.0.0.0:8080",
            config=self.config,
            strategy=self.strategy
        )

    def get_evaluate_fn(self):
        # 使用公共验证集（如公开医学数据集）
        def evaluate(server_round, parameters, config):
            model = load_model(parameters)
            loss, accuracy = evaluate_on_public_data(model)
            return loss, {"accuracy": accuracy}
        return evaluate

3.2 Flask 管理 API

# routes/federated_api.py
@app.post('/federated/start')
def start_federated_training():
    task_id = str(uuid.uuid4())
    # 异步启动 Flower 服务器
    threading.Thread(
        target=lambda: federated_coordinator.start(),
        daemon=True
    ).start()
    
    # 注册任务
    db.tasks.insert_one({"task_id": task_id, "status": "running"})
    return jsonify({"task_id": task_id})

@app.get('/federated/status/<task_id>')
def get_task_status(task_id: str):
    task = db.tasks.find_one({"task_id": task_id})
    return jsonify(task)

注意：生产环境应使用 Celery 替代 threading。

---

第四章：客户端实现（医院/银行 SDK）

4.1 客户端注册与认证

# clients/hospital_client.py
import flwr as fl
import torch
from opacus import PrivacyEngine

class HospitalClient(fl.client.NumPyClient):
    def __init__(self, hospital_id: str, data_loader):
        self.hospital_id = hospital_id
        self.data_loader = data_loader
        self.model = get_local_model()

    def get_parameters(self, config):
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]

    def set_parameters(self, parameters):
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = {k: torch.tensor(v) for k, v in params_dict}
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters, config):
        self.set_parameters(parameters)
        
        # 启用差分隐私
        privacy_engine = PrivacyEngine()
        model, optimizer, data_loader = privacy_engine.make_private(
            module=self.model,
            optimizer=torch.optim.SGD(self.model.parameters(), lr=0.01),
            data_loader=self.data_loader,
            noise_multiplier=1.2,
            max_grad_norm=1.0,
        )
        
        # 本地训练
        train_model(model, optimizer, data_loader, epochs=1)
        
        return self.get_parameters(config), len(self.data_loader.dataset), {}

    def evaluate(self, parameters, config):
        self.set_parameters(parameters)
        loss, accuracy = test_model(self.model, self.test_loader)
        return float(loss), len(self.test_loader.dataset), {"accuracy": float(accuracy)}

4.2 安全聚合（SecAgg）集成

• 原理：客户端两两建立密钥，上传掩码后的参数，服务器解掩码后得和
• Flower 支持：通过 Strategy 插件实现（需自定义）

# strategies/secagg_fedavg.py
class SecAggFedAvg(FedAvg):
    def configure_fit(self, server_round, parameters, client_manager):
        # 分发公钥给客户端
        instructions = super().configure_fit(server_round, parameters, client_manager)
        for ins in instructions:
            ins.config["public_keys"] = self.get_client_public_keys()
        return instructions

    def aggregate_fit(self, server_round, results, failures):
        # 解密聚合
        decrypted_params = secagg_decrypt([res.parameters for res in results])
        return ndarrays_to_parameters(decrypted_params), {}

简化方案：初期可先用 TLS 保证传输安全，SecAgg 作为二期优化。

---

第五章：隐私增强技术

5.1 差分隐私（DP）

• 核心思想：在梯度中添加噪声，使单个样本无法被推断
• 关键参数：

  • ε（epsilon）：隐私预算（越小越安全，通常 1–10）
  • δ（delta）：失败概率（通常 1e-5）

# DP 训练后，可计算实际隐私预算
from opacus.accountants import RDPAccountant

accountant = RDPAccountant()
# ... 训练过程中 accountant.step(...)
epsilon, best_alpha = accountant.get_privacy_spent(delta=1e-5)
print(f"(ε={epsilon:.2f}, δ=1e-5)")

5.2 同态加密（可选）

• 适用场景：对 SecAgg 不信任时
• 库推荐：Microsoft SEAL（C++）或 TenSEAL（Python 封装）
• 代价：计算开销大，仅适合小模型

---

第六章：场景实战

6.1 医疗联合诊断（横向联邦）

• 数据：5 家医院各 1,000 例肺部 CT（标签：良性/恶性）
• 模型：ResNet-18
• 结果：

  • 独立训练平均准确率：78%
  • 联邦学习准确率：89%（接近集中式 91%）
  • 隐私保障：原始影像 never leave 医院

6.2 金融反欺诈（纵向联邦）

• 参与方：

  • 银行 A：交易金额、频率
  • 电商 B：商品类别、收货地址

• 技术：

  • 使用 Private Set Intersection (PSI) 对齐用户 ID
  • 联邦逻辑回归训练

• 效果：AUC 提升 12%，且无用户数据交换

6.3 智能家居（跨设备联邦）

• 挑战：设备异构（手机/音箱）、网络不稳定
• 优化：

  • FedProx：处理非 IID 数据
  • 模型压缩：MobileNet 替代 ResNet

• 规模：10,000+ 设备参与，每日一轮

---

第七章：前端管理平台（Vue）

7.1 参与方监控面板

<template>
  <div class="federated-dashboard">
    <h2>联邦训练任务：{{ taskId }}</h2>
    
    <!-- 参与方状态 -->
    <div class="participants">
      <div v-for="p in participants" :key="p.id" 
           :class="['participant', { online: p.status === 'online' }]">
        {{ p.name }} ({{ p.samples }} 样本)
      </div>
    </div>

    <!-- 模型性能 -->
    <LineChart :data="accuracyHistory" title="全局模型准确率" />

    <!-- 贡献度排名 -->
    <table>
      <tr v-for="c in contributions" :key="c.hospital">
        <td>{{ c.hospital }}</td>
        <td>{{ (c.shapley * 100).toFixed(2) }}%</td>
      </tr>
    </table>
  </div>
</template>

<script setup>
const props = defineProps({
  taskId: String
})

// 从 Flask API 获取实时数据
const { data: status } = await useFetch(`/api/federated/status/${props.taskId}`)
const participants = computed(() => status.value?.participants || [])
const accuracyHistory = computed(() => status.value?.metrics?.accuracy || [])
const contributions = computed(() => status.value?.contributions || [])
</script>

7.2 贡献度评估（Shapley Value）

• 原理：衡量每个参与方对模型性能的边际贡献
• 近似算法：蒙特卡洛采样（避免指数复杂度）

# services/contribution.py
def compute_shapley_value(global_acc, participant_accuracies):
    """近似计算 Shapley Value"""
    n = len(participant_accuracies)
    shapley = [0.0] * n
    
    for _ in range(1000):  # 蒙特卡洛采样
        perm = np.random.permutation(n)
        marginal = 0.0
        for i in range(n):
            coalition = perm[:i]
            acc_without = evaluate_without(coalition)
            acc_with = evaluate_without(coalition + [perm[i]])
            marginal = acc_with - acc_without
            shapley[perm[i]] += marginal
    
    return [s / 1000 for s in shapley]

激励应用：贡献度高的医院可获得更多模型使用权或经济补偿。

---

第八章：安全与合规

8.1 攻击防御

攻击类型	防御措施

• 模型反演 | 差分隐私 + 梯度裁剪
• 成员推断 | 限制模型过拟合（早停）
• 后门攻击 | 异常检测（如 Krum 聚合）

8.2 合规审计

• 日志记录：所有参数交换写入区块链（可选）
• 隐私报告：自动生成 ε-δ 隐私证明
• 数据最小化：仅上传必要梯度，不传原始数据

---

第九章：性能优化

9.1 通信压缩

• 量化：32-bit → 8-bit 浮点
• 稀疏化：仅上传 Top-k 梯度

# utils/compression.py
def quantize(params, bits=8):
    scale = (2 ** bits - 1) / (np.max(params) - np.min(params))
    return np.round(params * scale) / scale

9.2 异步联邦

• 适用场景：客户端上线时间不一致
• 策略：FedAsync —— 服务器随时聚合可用客户端

---

第十章：伦理与公平

10.1 数据偏见放大

• 问题：若某医院数据质量差，拉低全局模型
• 对策：

  • 个性化联邦（Personalized FL）：为每方微调模型
  • 公平聚合：加权平均时考虑数据质量

10.2 参与门槛

• 小机构困境：样本少 → 贡献低 → 被边缘化
• 解决方案：

  • 最小参与保障（如强制包含至少 1 家社区医院）
  • 联邦数据增强：合成少数类样本

---

总结：隐私与智能的双赢

联邦学习不是技术的妥协，而是数据文明时代的必然选择。