企业官网建设流程全解析

简介：本资源是一份面向计算机及相关专业本科生的联邦学习安全方向毕业设计实践包，聚焦于论文级攻击防御方案的代码复现与工程落地，适用于毕设、课程设计、科研入门及AI安全方向自学。压缩包共184个文件，含109个Python源码（涵盖FL训练框架、后门攻击注入、鲁棒聚合算法等核心模块）、14个YAML配置文件（定义数据集划分、模型结构与攻击参数）、12个Shell脚本（支持一键环境部署与实验启动），以及5份Markdown文档（含详细运行说明、答辩要点与扩展建议），整体体积仅391KB，轻量易部署。已有171人下载学习，资源经作者实测全部可运行，答辩平均分96分，附带清晰目录结构与README引导，支持小白快速上手，也便于进阶者基于现有代码修改适配新攻击场景或防御策略。

1. 联邦学习攻击预防不是加个“安全层”就完事：毕业设计里真正要复现的，是攻击者怎么绕过聚合、客户端如何被毒化、以及防御代码跑起来后指标为什么反而掉——这三件事必须闭环验证

你手里的毕业设计标题写着“联邦学习攻击预防与论文代码复现”，但实际打开仓库发现：README里只有pip install -r requirements.txt和一句“运行main.py即可”，训练日志里acc从92%掉到63%却没报错，测试集准确率波动像心电图，而导师问“你复现的是哪篇论文的哪个攻击？防御模块插在Aggregator还是Client侧？消融实验对比了baseline吗？”——你卡住了。这不是Python环境配不配得上的问题，而是联邦学习的攻击与防御天然嵌套在系统级交互中：模型更新被篡改、梯度被投毒、客户端被伪造、聚合规则被利用……这些动作不发生在单机训练循环里，而藏在client上传→server校验→aggregation→下发的四步链路上。本篇不讲“联邦学习是什么”，只聚焦你正在跑、跑不通、跑出来结果不对的那个复现任务：用Python复现经典攻击（如label-flipping、model poisoning）与对应防御（如Krum、RFA、Norm Clipping），并确保每一步都能观测、可调试、能解释下降原因。适合已跑通FedAvg baseline、正卡在“防御后精度崩塌”或“攻击没生效”的本科生与硕士生。

2. 复现前必须厘清的三道生死线：攻击类型、防御位置、评估协议——选错任意一项，代码再全也是无效劳动

联邦学习攻击预防的复现，本质是在特定威胁模型下，验证某防御机制对某类攻击的有效性。跳过威胁建模直接写代码，等于在没画靶心的情况下开枪。下面三条线，是你启动复现前必须亲手划清的边界。

2.1 攻击类型决定代码结构：Label-Flipping和Model Poisoning根本不是同一层的事

Label-Flipping（标签翻转）发生在数据层：客户端本地训练时，把猫的图片标成狗，再上传被污染的梯度。它不修改模型参数，只污染训练信号。复现时需在client端train()函数中插入标签扰动逻辑：

# client.py 中 train() 函数片段 def train(self, model, dataloader, epochs=1): model.train() optimizer = torch.optim.SGD(model.parameters(), lr=0.01) for epoch in range(epochs): for x, y in dataloader: # 【关键插入点】仅对恶意客户端执行标签翻转 if self.is_malicious: # 需提前设置 malicious_client_ids y = (y + 1) % 10 # 假设10分类，将标签循环+1 x, y = x.to(self.device), y.to(self.device) optimizer.zero_grad() loss = F.cross_entropy(model(x), y) loss.backward() optimizer.step() return model.state_dict() # 返回污染后的state_dict

注意：这段代码必须绑定到具体客户端ID，不能全局生效；self.is_malicious需在初始化client时传入，而非运行时随机判定——否则无法复现论文中“20%客户端被攻陷”的设定。

Model Poisoning（模型投毒）则发生在参数层：客户端上传前，直接篡改state_dict()，例如注入后门触发器或放大特定层梯度。复现时需在upload()环节拦截：

# client.py 中 upload() 函数 def upload(self, model_state_dict): if self.is_malicious: # 【典型投毒操作】放大最后一层bias，制造类别偏移 if 'fc2.bias' in model_state_dict: model_state_dict['fc2.bias'] *= 5.0 # 放大5倍 # 或注入高斯噪声破坏收敛性 for k, v in model_state_dict.items(): if 'weight' in k: noise = torch.randn_like(v) * 0.1 model_state_dict[k] = v + noise return model_state_dict

逻辑说明：Label-Flipping影响的是梯度方向，Model Poisoning直接影响聚合输入值。前者需配合数据集（如CIFAR-10）的label映射表操作，后者直接操作tensor——二者调试方式、观测指标（loss曲线 vs 参数范数分布）完全不同。你复现的论文若未明确攻击类型，立刻查原文Methodology章节的Threat Model小节，别猜。

2.2 防御位置决定代码挂载点：Server端聚合防御和Client端鲁棒训练不可混用

几乎所有毕业设计代码库都把防御写在server.py里，这是对的——因为Krum、RFA、Bulyan等主流防御都在server端对收到的N个client update做筛选或加权。但必须确认你复现的论文是否真在server端防御。例如：

• Krum：计算每个client update与其他所有update的欧氏距离平方和，选距离和最小的那个update参与聚合；
• RFA（Robust Federated Averaging）：对每个参数维度，取N个client该维度值的几何中位数（geometric median）；
• Norm Clipping：对每个client上传的梯度向量做L2范数裁剪，超阈值则缩放。

它们的共同点是：输入是N个state_dict，输出是1个聚合后state_dict。代码必须放在server端aggregate()函数内：

# server.py 中 aggregate() 函数（以Krum为例） def aggregate(self, client_updates): # client_updates: List[Dict[str, torch.Tensor]], 长度为N n = len(client_updates) scores = torch.zeros(n) # 计算每个client update的score（距离和） for i in range(n): dist_sum = 0 for j in range(n): if i != j: # 对每个参数key计算欧氏距离平方 dist_sq = 0 for k in client_updates[i].keys(): diff = client_updates[i][k] - client_updates[j][k] dist_sq += torch.sum(diff ** 2).item() dist_sum += dist_sq scores[i] = dist_sum # 选score最小的client update作为聚合结果 best_idx = torch.argmin(scores).item() return copy.deepcopy(client_updates[best_idx])

参数说明：scores[i]越小代表第i个client的update越“中心”，Krum假设恶意client的update会偏离正常分布。阈值不需设——它天然选1个，不是过滤。若你看到代码里有if score < threshold:，那大概率是作者自己魔改的非标准Krum，与原论文不符。

而Client端防御（如FedProx、SCAFFOLD）需修改client本地训练目标函数，与server端防御完全隔离。若你的论文同时用了两者（如“Server用RFA + Client用FedProx”），必须拆成两个独立分支复现，禁止在一个main.py里硬塞两种逻辑——否则无法归因精度变化是哪部分起效。

2.3 评估协议是精度数字的唯一判据：没有global test set重测，一切acc都是幻觉

最致命的复现错误：用client本地test set算accuracy当global性能。联邦学习的global accuracy必须在server持有的、独立于所有client的global test set上计算。这个set不能是任何client的test data切分出来的——它必须是全新采集/划分的数据。

常见做法是：

• 使用CIFAR-10时，将原始10000张test image按类别均匀切分为global_test（5000张）+ reserved_for_poisoning（5000张）；
• reserved部分用于构造后门攻击的trigger样本（如贴小方块），global_test纯用于最终评估；
• 所有client的train/test split仅用于本地训练，绝不参与global指标计算。

验证代码必须显式加载global test set：

# evaluate.py def global_evaluate(model, global_test_loader): model.eval() correct, total = 0, 0 with torch.no_grad(): for x, y in global_test_loader: # 注意：这里用的是global_test_loader，不是client的loader x, y = x.to(device), y.to(device) pred = model(x).argmax(dim=1) correct += pred.eq(y).sum().item() total += len(y) return 100. * correct / total # 在main.py训练循环末尾调用 global_acc = global_evaluate(server_model, global_test_loader) print(f"Global Test Accuracy: {global_acc:.2f}%")

血泪经验：曾见三个毕业设计代码库，global_test_loader实际加载的是某个client的test data，导致防御后acc虚高15%——因为那个client恰好数据干净。务必检查global_test_loader.dataset的len()和class distribution，确认它独立于所有client dataset。

3. 用PyTorch+Flower复现FedAvg baseline：50行核心代码跑通，但必须亲手改这3处才能进攻击防御阶段

很多同学卡在第一步：连基础FedAvg都跑不起来，更别说加攻击和防御。问题不在代码量，而在三个隐藏依赖必须手动补全。以下用PyTorch + Flower框架（轻量、易调试、社区活跃）给出最小可行复现路径，所有代码均可直接粘贴运行。

3.1 安装与环境：Flower 1.3+ PyTorch 2.0+ Python 3.8——版本错一个，client注册就失败

# 创建干净虚拟环境（强烈推荐） python3.8 -m venv fed_env source fed_env/bin/activate # Linux/Mac；Windows用 fed_env\Scripts\activate.bat # 安装指定版本（Flower 1.4+对PyTorch 2.1支持有bug，锁定1.3.1） pip install torch==2.0.1 torchvision==0.15.2 pip install flwr==1.3.1 # 关键！不要pip install flwr最新版 pip install numpy scikit-learn tqdm

提示：Flower 1.3.1是最后一个稳定支持PyTorch 2.0的版本。若用Python 3.9+，需降级到PyTorch 1.13，否则flwr.client.NumPyClient序列化失败。环境变量PYTHONPATH无需设置，Flower自动处理。

3.2 Server端：50行代码启动，但必须重写fit_config和evaluate_fn

# server.py import flwr as fl import torch import numpy as np from collections import OrderedDict # 1. 定义全局模型（此处用简单CNN，与client一致） class Net(torch.nn.Module): def __init__(self): super().__init__() self.conv1 = torch.nn.Conv2d(3, 6, 5) self.pool = torch.nn.MaxPool2d(2, 2) self.conv2 = torch.nn.Conv2d(6, 16, 5) self.fc1 = torch.nn.Linear(16 * 5 * 5, 120) self.fc2 = torch.nn.Linear(120, 84) self.fc3 = torch.nn.Linear(84, 10) def forward(self, x): x = self.pool(torch.nn.functional.relu(self.conv1(x))) x = self.pool(torch.nn.functional.relu(self.conv2(x))) x = torch.flatten(x, 1) x = torch.nn.functional.relu(self.fc1(x)) x = torch.nn.functional.relu(self.fc2(x)) x = self.fc3(x) return x # 2. 初始化全局模型 net = Net().to("cpu") # server通常不用GPU params = [val.cpu().numpy() for _, val in net.state_dict().items()] # 3. 定义聚合策略（FedAvg） strategy = fl.server.strategy.FedAvg( fraction_fit=1.0, # 所有client参与训练 fraction_evaluate=0.0, # 不在server端评估，我们自己做 min_available_clients=2, # 【关键修改1】必须重写fit_config，让client知道训练轮次和batch_size on_fit_config_fn=lambda server_round: { "server_round": server_round, "local_epochs": 1, # 每轮client只训1 epoch，避免过拟合 "batch_size": 32, }, # 【关键修改2】重写evaluate_fn，在server端用global test set评估 evaluate_fn=global_evaluate_fn, # 下方定义 ) # 4. global_evaluate_fn：必须加载global test set def global_evaluate_fn(server_round, parameters, config): # 将NumPy参数转回PyTorch state_dict params_dict = zip(net.state_dict().keys(), parameters) state_dict = OrderedDict({k: torch.tensor(v) for k, v in params_dict}) net.load_state_dict(state_dict, strict=True) # 加载global test set（此处简化，实际需从文件读取） from torchvision import datasets, transforms transform = transforms.Compose([transforms.ToTensor()]) global_testset = datasets.CIFAR10( root="./data", train=False, download=True, transform=transform ) global_testloader = torch.utils.data.DataLoader( global_testset, batch_size=100, shuffle=False ) # 计算global accuracy net.eval() correct, total = 0, 0 with torch.no_grad(): for x, y in global_testloader: pred = net(x).argmax(dim=1) correct += pred.eq(y).sum().item() total += len(y) accuracy = correct / total print(f"Round {server_round} Global Accuracy: {accuracy:.4f}") return float(accuracy), {"accuracy": accuracy} # 5. 启动server fl.server.start_server( server_address="0.0.0.0:8080", config=fl.server.ServerConfig(num_rounds=10), strategy=strategy, )

逻辑说明：on_fit_config_fn返回的字典会通过gRPC传给每个client，client据此设置local_epochs和batch_size；evaluate_fn在每轮结束后被调用，它必须重新加载global test set（不能复用client的loader），且返回(loss, metrics)元组——loss可设为0，metrics必须含accuracy键供Flower记录。

3.3 Client端：继承NumPyClient，但必须重写get_parameters/set_parameters

# client.py import flwr as fl import torch import numpy as np from torch import nn, optim from torchvision import datasets, transforms class CIFARClient(fl.client.NumPyClient): def __init__(self, cid, is_malicious=False): self.cid = cid self.is_malicious = is_malicious # 【关键修改3】client必须自己加载数据，且train/test split独立 transform = transforms.Compose([transforms.ToTensor()]) trainset = datasets.CIFAR10( root="./data", train=True, download=True, transform=transform ) # 按cid划分数据：假设10个client，每个拿10%数据 indices = list(range(len(trainset))) np.random.seed(int(cid)) # 确保划分可复现 np.random.shuffle(indices) client_size = len(trainset) // 10 client_indices = indices[int(cid)*client_size:(int(cid)+1)*client_size] self.trainloader = torch.utils.data.DataLoader( torch.utils.data.Subset(trainset, client_indices), batch_size=32, shuffle=True ) # test set也独立划分（不用于global评估！） testset = datasets.CIFAR10( root="./data", train=False, download=True, transform=transform ) self.testloader = torch.utils.data.DataLoader( testset, batch_size=100, shuffle=False ) def get_parameters(self, config): # 返回当前模型参数（NumPy格式） return [val.cpu().numpy() for _, val in net.state_dict().items()] def fit(self, parameters, config): # 加载server下发的参数 params_dict = zip(net.state_dict().keys(), parameters) state_dict = OrderedDict({k: torch.tensor(v) for k, v in params_dict}) net.load_state_dict(state_dict, strict=True) # 本地训练（此处插入攻击逻辑） if self.is_malicious: self.poison_labels() # 或 self.poison_model() # 标准训练 net.train() optimizer = optim.SGD(net.parameters(), lr=0.01) for _ in range(config["local_epochs"]): for x, y in self.trainloader: optimizer.zero_grad() loss = nn.CrossEntropyLoss()(net(x), y) loss.backward() optimizer.step() # 返回更新后的参数 return self.get_parameters({}), len(self.trainloader.dataset), {} def poison_labels(self): # 示例：对trainloader中的batch做label翻转 pass # 实现见2.1节 def poison_model(self): # 示例：对state_dict做投毒 pass # 实现见2.1节 def evaluate(self, parameters, config): # client本地评估（仅用于debug，不计入global指标） params_dict = zip(net.state_dict().keys(), parameters) state_dict = OrderedDict({k: torch.tensor(v) for k, v in params_dict}) net.load_state_dict(state_dict, strict=True) net.eval() correct, total = 0, 0 with torch.no_grad(): for x, y in self.testloader: pred = net(x).argmax(dim=1) correct += pred.eq(y).sum().item() total += len(y) return float(0), len(self.testloader.dataset), {"accuracy": correct/total} # 启动client（在不同终端运行） fl.client.start_numpy_client( server_address="localhost:8080", client=CIFARClient(cid="0", is_malicious=False), )

参数说明：cid必须是字符串（Flower要求），is_malicious控制是否启用攻击；poison_labels()和poison_model()留空待填——这就是你接续2.1节的位置。注意fit()返回的第三个值是metrics字典，此处为空，因global评估由server端evaluate_fn完成。

4. 攻击与防御代码落地：Krum防御+Label-Flipping攻击的完整闭环，附3个必调参数与效果验证方法

现在你已跑通FedAvg baseline，下一步是注入攻击并部署防御。本节以Krum防御对抗Label-Flipping攻击为例，给出可直接运行的完整闭环代码，并指出三个决定效果的参数——它们不是随便写的，而是根据论文公式推导出的实操值。

4.1 Label-Flipping攻击实现：在client.fit()中精准翻转，且只翻训练集标签

# client.py 中补充 poison_labels() 方法 def poison_labels(self): # 获取trainloader的dataset，直接修改其targets（CIFAR-10 targets是list） if hasattr(self.trainloader.dataset, 'dataset'): # Subset情况 base_dataset = self.trainloader.dataset.dataset else: base_dataset = self.trainloader.dataset # 确认是CIFAR-10（10类），否则报错 if not hasattr(base_dataset, 'targets') or len(base_dataset.targets) == 0: raise ValueError("Dataset must have 'targets' attribute") # 只翻转指定比例的样本（论文常用20%） num_samples = len(base_dataset.targets) num_flip = int(0.2 * num_samples) # 20%翻转率 # 随机选索引（固定seed保证可复现） flip_indices = np.random.RandomState(42).choice( num_samples, size=num_flip, replace=False ) # 翻转：将标签y改为(y+1)%10（避免翻到自身） for idx in flip_indices: original_label = base_dataset.targets[idx] base_dataset.targets[idx] = (original_label + 1) % 10 print(f"[Client {self.cid}] Poisoned {num_flip}/{num_samples} labels")

逻辑说明：此方法直接修改dataset.targets，比在dataloader迭代时动态翻转更可靠——因为后者可能被shuffle打乱，导致翻转比例失控。RandomState(42)确保每次运行翻转相同样本，方便对比实验。

4.2 Krum防御实现：server端aggregate()替换为Krum逻辑，注意距离计算维度

# server.py 中替换 aggregate() 函数（替代FedAvg） def krum_aggregate(client_updates): n = len(client_updates) f = 1 # 假设最多1个恶意client（论文默认f=1） m = n - f - 2 # Krum选m个最近的update # 将所有state_dict转为向量便于距离计算 update_vectors = [] for update in client_updates: vec = [] for k in sorted(update.keys()): vec.append(update[k].flatten()) update_vectors.append(torch.cat(vec)) # 计算每对update的欧氏距离平方 scores = torch.zeros(n) for i in range(n): distances = [] for j in range(n): if i != j: dist_sq = torch.sum((update_vectors[i] - update_vectors[j]) ** 2) distances.append(dist_sq.item()) # 取最小的m个距离之和 distances.sort() scores[i] = sum(distances[:m]) # 选score最小的update best_idx = torch.argmin(scores).item() return copy.deepcopy(client_updates[best_idx]) # 在strategy中使用（替换FedAvg） strategy = fl.server.strategy.Strategy( # ... 其他参数同前 # 替换aggregate函数 aggregate_fit=lambda server_round, results, failures: ( krum_aggregate([r[1] for r in results]), # results是(client, parameters)元组列表 {} ), )

参数说明：f是预估恶意client数量，必须≤⌊(n-1)/2⌋，否则Krum失效；m=n-f-2是Krum论文公式，不可随意修改；距离计算用torch.sum((vec_i - vec_j)**2)而非torch.norm()，因后者在GPU上可能有精度误差。

4.3 效果验证三板斧：global acc、client divergence、gradient norm heatmap

光看global accuracy数字不够，必须交叉验证。以下是三个必做的验证步骤：

生成heatmap的代码示例：

# 在server端aggregate前添加 def plot_gradient_heatmap(client_updates, round_num): norms = [] for update in client_updates: # 提取fc.weight的梯度范数（假设key为'fc3.weight'） if 'fc3.weight' in update: norm = torch.norm(update['fc3.weight']).item() else: norm = 0 norms.append(norm) plt.figure(figsize=(8, 2)) plt.imshow([norms], cmap='Reds', aspect='auto') plt.colorbar() plt.title(f"Round {round_num} fc3.weight Gradient Norm") plt.xlabel("Client ID") plt.yticks([]) plt.savefig(f"heatmap_round_{round_num}.png") plt.close()

提示：heatmap比数字更早暴露问题。若攻击后heatmap无变化，立刻检查poison_labels()是否真修改了dataset.targets——打印base_dataset.targets[:10]前后对比。

5. 避坑指南：毕业设计复现中最常踩的5个坑，每个都导致答辩被问住

复现联邦学习攻击防御，90%的问题不是代码写错，而是对联邦学习系统行为的误解。以下5个坑，是我带过17届毕设学生后总结的高频翻车点，每个都附真实现象、根因和解法。

5.1 坑1：client注册失败，报错“Connection reset by peer”——其实是server端口被占，不是网络问题

• 现象：运行fl.client.start_numpy_client()后，client日志卡在Connecting to 0.0.0.0:8080...，server端无client连接记录，几秒后报ConnectionResetError。
• 原因：端口8080已被其他进程占用（如Jupyter Lab、旧的Flower server、Docker容器）。Flower默认不检测端口占用，直接尝试连接，失败后抛出底层socket错误。
• 解决：Linux/Mac执行lsof -i :8080，Windows执行netstat -ano | findstr :8080，杀掉对应PID进程；或改server端口为8081，client同步改server_address="localhost:8081"。

5.2 坑2：global accuracy始终为0.00%——client上传的参数根本没被server加载

• 现象：server日志显示INFO flower 10 clients connected，但global_evaluate_fn中net.load_state_dict(state_dict)后，模型预测全错，accuracy恒为0。
• 原因：client端get_parameters()返回的参数顺序与server端net.state_dict().keys()不一致。PyTorch 2.0+中state_dict().keys()顺序受模型定义顺序影响，若client和server用不同脚本定义Net，keys顺序可能不同，导致zip(keys, parameters)错位。
• 解决：在server和client中强制统一keys顺序：

  # server.py 和 client.py 中都加 ordered_keys = sorted(net.state_dict().keys()) # 排序确保一致 def get_parameters(self, config): params = [val.cpu().numpy() for k, val in net.state_dict().items() if k in ordered_keys] return params # load时也按ordered_keys params_dict = zip(ordered_keys, parameters)

5.3 坑3：防御后accuracy比baseline还低——Krum误杀了良性client

• 现象：开启Krum后，global accuracy从88%掉到72%，且server日志显示每轮都选中同一个client（如cid=3）。
• 原因：Krum的m=n-f-2参数设置错误。当n=5个client，f=1时，m=5-1-2=2，但若恶意client恰好位于数据分布边缘（如label翻转后梯度方向异常），Krum可能连续选中它——因为它的“距离和”意外最小。
• 解决：降低f值（如设f=0），或增加client数量（n≥10），使统计更鲁棒；或改用RFA（geometric median对异常值更鲁棒）。切勿强行调高f。

5.4 坑4：attack生效但defense无反应——防御代码根本没被执行

• 现象：Label-Flipping后global accuracy掉到65%，但启用Krum后仍是65%，曲线完全重合。
• 原因：Flower的Strategy类中，aggregate_fit函数未被正确覆盖。常见错误是复制了FedAvg源码但忘了删掉super().aggregate_fit()调用，导致实际执行的是父类FedAvg而非你的Krum。
• 解决：在自定义strategy中彻底重写aggregate_fit，不要继承FedAvg，直接继承fl.server.strategy.Strategy基类；或确认super().aggregate_fit()被注释掉。

5.5 坑5：复现论文结果差10%——忽略了论文的client heterogeneity设置

• 现象：论文报告Krum在20%攻击下acc=85%，你复现只有75%。
• 原因：论文使用Non-IID数据划分（如Dirichlet分布α=0.1），而你用IID（随机均分）。Non-IID下恶意client更容易被识别，IID下所有client相似，Krum区分度下降。
• 解决：用torch.utils.data.random_split无法模拟Non-IID，必须用sklearn.model_selection.train_test_split按类别分层抽样，或使用torchvision.datasets的Subset配合np.random.dirichlet生成client数据比例。示例：

  # 按Dirichlet分布划分CIFAR-10 from sklearn.model_selection import train_test_split alpha = 0.1 n_clients = 10 class_counts = [1000] * 10 # CIFAR-10每类1000训练样本 dirichlet_dist = np.random.dirichlet([alpha] * n_clients, size=10) # 10类×10client

6. 毕业设计答辩前的终极验证：用3个命令生成可展示的证据链，让导师一眼信服你真复现了

答辩时，导师最想看的不是代码，而是证据链：攻击确实发生了、防御确实起效了、结果确实可复现。以下三个命令，生成三份材料，构成闭环证据。我带的学生用这套方法，100%通过预答辩。

6.1 命令1：生成攻击生效证据——client本地accuracy与global accuracy的剪刀差图

# 运行attack-only实验（不启用Krum） python server.py --strategy "fedavg" --attack "label_flip" --malicious_ratio 0.2 # 日志中提取每轮数据，存为attack_log.csv # 用以下脚本生成对比图 import pandas as pd import matplotlib.pyplot as plt log = pd.read_csv("attack_log.csv") # 列：round, client_acc_cid0, client_acc_cid1, ..., global_acc plt.figure(figsize=(10,6)) for cid in range(10): plt.plot(log['round'], log[f'client_acc_cid{cid}'], alpha=0.6, label=f'Client {cid}') plt.plot(log['round'], log['global_acc'], 'k-', linewidth=2, label='Global Accuracy') plt.xlabel('Round') plt.ylabel('Accuracy (%)') plt.title('Label-Flipping Attack: Local vs Global Accuracy Divergence') plt.legend() plt.grid(True) plt.savefig('attack_divergence.png')

为什么有效：图中若出现“多条client曲线分散，global曲线居中下移”，证明攻击成功制造了client间差异——这是防御的前提。若所有client曲线重合，说明攻击未生效。

6.2 命令2：生成防御生效证据——Krum选中的client ID历史记录表

# 修改server.py，在krum_aggregate()末尾添加日志 best_idx = torch.argmin(scores).item() print(f"Round {server_round}: Krum selected client {best_idx} (score={scores[best_idx]:.2f})") # 运行defense实验，日志存为krum_log.txt # 提取并生成表格 import re with open("krum_log.txt") as f: lines = f.readlines() rounds = [] selected = [] for line in lines: m = re.search(r"Round (\d+): Krum selected client (\d+)", line) if m: rounds.append(int(m.group(1))) selected.append(int(m.group(2))) df = pd.DataFrame({'Round': rounds, 'Selected_Client_ID': selected}) df.to_csv('krum_selection.csv', index=False)

为什么有效：表格证明Krum不是固定选某个client，而是动态响应——若连续10轮都选cid=0，说明它可能是恶意client，但防御机制仍在工作（选它是因为它最“中心”）。导师扫一眼就知道你真跑了Krum。

6.3 命令3：生成可复现性证据——requirements.txt + seed设置 + hash校验

# 1. 锁定环境 pip freeze > requirements.txt # 2. 在所有random操作前加seed（client.py, server.py） torch.manual_seed(42) np.random.seed(42) # 3. 对关键输出生成hash import hashlib with open("global_accuracy_history.txt", "rb") as f: hash_val = hashlib.md5(f.read()).hexdigest() print(f"Result Hash: {hash_val}") # 例如：a1b2c3d4e5f6...

为什么有效：答辩时，你只需说：“导师，这是我的requirements.txt、所有seed设为42、最终accuracy文件的MD5是a1b2c3d4…，您在任意机器上pip install -r后运行，结果完全一致。”——这比解释100行代码更有说服力。

最后说句实在的：联邦学习攻击防御的毕业设计

本文还有配套的精品资源，点击获取