当前位置：首页 > news >正文

联邦学习 + 区块链：去中心化 AI 训练的隐私保护与激励设计

news 2026/7/19 15:42:05

联邦学习 + 区块链：去中心化 AI 训练的隐私保护与激励设计

一、医疗数据不能出医院，但 AI 需要跨医院的数据训练

医疗 AI 面临一个典型的"数据孤岛"问题：每家医院的数据都包含隐私信息，不能直接共享。但单家医院的数据量不足以训练出高精度的诊断模型。

联邦学习（Federated Learning）解决了这个问题：各家医院在本地训练模型，只上传模型参数（而非原始数据）到中心服务器做聚合。原始数据不出本地，保护了隐私。

但新的问题出现了：如何激励各家医院参与训练？如何保证上传的参数没有被篡改？区块链可以回答这两个问题——用代币激励参与，用链上记录保证数据完整性。

二、联邦学习 + 区块链的协作架构

在该协作架构中，流程主要分为本地训练、中心聚合与链上记录三个核心环节。首先，各参与医院（如医院 A、B、C）利用本地医疗数据进行模型训练，并对生成的梯度进行加密及差分隐私处理，确保原始数据不出本地。随后，加密后的梯度被发送至聚合服务器，通过联邦平均（FedAvg）算法进行聚合，生成全局模型。与此同时，梯度的哈希值会被上传至区块链层的智能合约中。区块链层负责基于贡献度分发代币激励、记录训练参与情况以供审计追踪，并维护基于历史行为的声誉系统。最终，聚合服务器将更新后的全局模型分发回各医院，完成一轮迭代。

三、Python 实现联邦平均 + 链上记录

联邦学习客户端（各参与方）

import torch import torch.nn as nn import hashlib import json from typing import Dict, List, Optional, Tuple

import numpy as np

class FederatedClient:
"""联邦学习客户端——在医院/数据持有方本地运行"""

def __init__(self, client_id: str, model: nn.Module, local_data_loader): self.client_id = client_id self.model = model self.data_loader = local_data_loader self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.model.to(self.device) # 差分隐私参数 self.dp_noise_multiplier = 0.1 # 噪声系数 self.dp_clip_norm = 1.0 # 梯度裁剪阈值 def local_train( self, global_weights: Dict[str, torch.Tensor], epochs: int = 5, lr: float = 0.001, ) -> Tuple[Dict[str, torch.Tensor], int, str]: """ 本地训练 1. 加载全局模型参数 2. 在本地数据上训练 3. 添加差分隐私噪声 4. 返回更新后的权重 + 样本数量 + 梯度哈希 """ # 加载全局模型 self.model.load_state_dict(global_weights) optimizer = torch.optim.Adam(self.model.parameters(), lr=lr) criterion = nn.CrossEntropyLoss() self.model.train() total_samples = 0 for epoch in range(epochs): for batch_data, batch_labels in self.data_loader: batch_data = batch_data.to(self.device) batch_labels = batch_labels.to(self.device) # 前向传播 outputs = self.model(batch_data) loss = criterion(outputs, batch_labels) # 反向传播 optimizer.zero_grad() loss.backward() # DP 梯度裁剪 + 添加噪声 self._apply_dp_to_gradients() optimizer.step() total_samples += len(batch_data) # 计算梯度更新 = 当前权重 - 全局权重 current_weights = self.model.state_dict() weight_updates = {} for key in current_weights: weight_updates[key] = current_weights[key] - global_weights[key] # 计算梯度哈希（用于链上存证） gradient_hash = self._compute_gradient_hash(weight_updates) return weight_updates, total_samples, gradient_hash def _apply_dp_to_gradients(self): """差分隐私：梯度裁剪 + 高斯噪声""" total_norm = 0.0 for param in self.model.parameters(): if param.grad is not None: total_norm += param.grad.norm(2).item() ** 2 total_norm = total_norm ** 0.5 clip_coef = min(1.0, self.dp_clip_norm / (total_norm + 1e-6)) for param in self.model.parameters(): if param.grad is not None: # 裁剪 param.grad *= clip_coef # 添加高斯噪声 noise = torch.normal( mean=0.0, std=self.dp_noise_multiplier * self.dp_clip_norm, size=param.grad.shape, ).to(self.device) param.grad += noise def _compute_gradient_hash(self, weights: Dict[str, torch.Tensor]) -> str: """计算梯度哈希——链上存证""" hasher = hashlib.sha3_256() for key in sorted(weights.keys()): hasher.update(key.encode()) hasher.update(weights[key].cpu().numpy().tobytes()) return hasher.hexdigest()

### 聚合服务器 + 区块链交互 ```python from collections import OrderedDict class FederatedAggregator: """联邦平均聚合器——集成区块链记录""" def __init__(self, blockchain_client=None): self.blockchain = blockchain_client self.round_history: List[Dict] = [] # 各轮训练的链上记录 def aggregate( self, client_updates: List[Tuple[Dict[str, torch.Tensor], int, str]], round_id: int, ) -> Dict[str, torch.Tensor]: """ FedAvg 聚合：加权平均各客户端的梯度 并将各客户端的梯度哈希记录到区块链 """ # 验证哈希 + 记录到区块链 for i, (weights, samples, grad_hash) in enumerate(client_updates): client_id = f"client_{i}" # 链上记录 if self.blockchain: tx_hash = self.blockchain.record_gradient( client_id=client_id, round_id=round_id, gradient_hash=grad_hash, sample_count=samples, ) self.round_history.append({ "round": round_id, "client": client_id, "gradient_hash": grad_hash, "samples": samples, "tx_hash": tx_hash, }) # 计算总样本数 total_samples = sum(samples for _, samples, _ in client_updates) # FedAvg 加权平均 global_weights = OrderedDict() first_weights = client_updates[0][0] for key in first_weights: # 初始化为零 global_weights[key] = torch.zeros_like(first_weights[key]) for weights, samples, _ in client_updates: global_weights[key] += weights[key] * (samples / total_samples) print(f"轮次 {round_id}: 聚合了 {len(client_updates)} 个客户端, " f"总样本 {total_samples}") return global_weights def get_training_proof(self, round_id: int) -> Dict: """获取指定轮次的训练证明（从链上查询）""" round_records = [r for r in self.round_history if r["round"] == round_id] return { "round": round_id, "participants": len(round_records), "total_samples": sum(r["samples"] for r in round_records), "records": round_records, }

区块链激励合约（概念实现）

class FederatedIncentiveContract: """ 联邦学习激励智能合约 ——根据参与方的贡献度分配代币奖励 """ def __init__(self, web3_client, contract_address: str): self.w3 = web3_client self.contract_address = contract_address # 激励参数 self.base_reward = 100 # 基础奖励（代币） self.data_bonus = 0.01 # 每样本奖励 self.quality_multiplier = 1.0 # 质量系数 def calculate_reward( self, client_id: str, sample_count: int, contribution_score: float, # 模型贡献度（验证集提升） reputation: float, # 历史声誉 ) -> float: """ 计算激励奖励 reward = base_reward + data_bonus * samples * quality * reputation """ data_reward = self.data_bonus * sample_count quality_reward = self.quality_multiplier * contribution_score reputation_bonus = reputation * 0.1 total = (self.base_reward + data_reward) * (1 + quality_reward + reputation_bonus) return max(0, total) def distribute_rewards( self, round_id: int, participants: List[Dict], ) -> List[Dict]: """ 分发奖励——更新链上状态 """ distributions = [] for participant in participants: reward = self.calculate_reward( client_id=participant["client_id"], sample_count=participant["sample_count"], contribution_score=participant.get("contribution_score", 0.5), reputation=participant.get("reputation", 1.0), ) # 链上转账 tx_hash = self._transfer_tokens( to_address=participant["address"], amount=int(reward), ) distributions.append({ "client_id": participant["client_id"], "reward": reward, "tx_hash": tx_hash, "round": round_id, }) return distributions def _transfer_tokens(self, to_address: str, amount: int) -> str: """模拟链上代币转账""" # 实际实现中调用智能合约 return f"0x{hash(f'{to_address}:{amount}')}"