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

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 家社区医院）
  • 联邦数据增强：合成少数类样本

总结：隐私与智能的双赢

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