简介本资源是一套面向高校计算机、人工智能及相关专业学生与教师的毕业设计级项目聚焦于隐私保护场景下的学生成绩预测问题基于联邦学习框架实现多客户端协同建模并通过Streamlit构建交互式可视化分析界面。资源包含完整可运行Python源码、真实模拟的学生成绩CSV数据集、多种联邦优化算法如FedRep、FedProx、Ditto、Scaffold等实现模块及训练日志文件覆盖本地训练、模型聚合、性能评估全流程。压缩包共67个文件含18个核心Python脚本含主程序、模型定义、采样与训练工具、7个CSV数据与结果记录文件、11个备份文件.zbak、1个知识拓展资料ZIP及说明文档整体大小为6.02MB。目前已有111人学习下载项目答辩获95分高分代码经实测调试无报错附带清晰目录结构与模块化设计便于理解联邦学习通信机制、快速复现实验或在此基础上扩展新算法与数据场景。1. 项目缘起当成绩预测遇上数据隐私在高校的教学管理工作中学业成绩预测一直是个既重要又敏感的话题。教务部门希望能提前识别学业困难的学生以便进行精准帮扶学生也希望能对自己的学业走向有个大致的预判。传统的做法要么是基于单个学生的历史数据进行建模预测能力有限要么是汇集全校所有学生的数据到一个中心服务器进行集中式机器学习。后者听起来效果会更好但问题也随之而来学生的成绩、选课、出勤等数据属于高度敏感的个人隐私将全校数据集中处理不仅面临巨大的数据泄露风险也触碰了日益严格的数据保护法规如GDPR、国内的个人信息保护法的红线。这就形成了一个典型的“数据孤岛”困境数据有价值但不能汇聚。我最近完成的一个项目正是为了解决这个矛盾。我们尝试用联邦学习Federated Learning技术构建一个高校学生学业成绩预测系统。它的核心思想是“数据不动模型动”我们不需要把各院系、各班级的学生数据收集到一个中心而是让机器学习模型“走”到数据所在的地方比如各院系的服务器或安全环境进行训练只将训练得到的模型参数更新而非原始数据加密后传回中心进行聚合。这样在保护数据隐私的前提下我们依然能够利用分散在各处的数据训练出一个强大的全局预测模型。为了让这个听起来有些抽象的技术能够被教务老师或相关研究人员直观地理解和使用我选择了Python作为实现语言并用Streamlit快速搭建了一个交互式可视化界面。整个项目从联邦学习算法实现、模拟数据集构建到前端界面开发形成了一套完整的、可复现的解决方案。下面我就把这个项目的实现思路、关键技术细节、踩过的坑以及最终效果毫无保留地分享出来。2. 联邦学习在不汇聚数据的前提下实现协同建模在深入代码之前我们必须先搞清楚联邦学习到底是如何工作的以及为什么它适合学业成绩预测这个场景。这决定了我们整个系统的架构设计。2.1 联邦学习的基本范式与我们的选择联邦学习主要有三种范式横向联邦学习、纵向联邦学习和联邦迁移学习。针对高校成绩预测的场景我们面临的数据特点是参与方例如计算机学院、经管学院、文学院拥有的学生样本不同学生ID不同但特征空间是相似的大家都有“上学期平均绩点”、“出勤率”、“图书馆借阅次数”等特征。这完美契合了横向联邦学习的适用条件。横向联邦学习也叫基于样本的联邦学习其经典流程以FedAvg算法为例如下中心服务器初始化服务器初始化一个全局机器学习模型比如一个神经网络并将初始模型参数下发给所有客户端各院系。客户端本地训练各客户端在本地用自己的私有数据对接收到的全局模型进行训练例如进行几个轮次的梯度下降。客户端上传更新训练完成后各客户端将计算得到的模型参数更新或梯度加密后上传给中心服务器。服务器聚合更新服务器收集到所有客户端的更新后对其进行加权平均例如根据各客户端的数据量加权得到新的全局模型参数。迭代服务器将聚合后的新全局模型参数再次下发给客户端重复步骤2-4直至模型收敛。这个过程就像是一个“分布式模型锻造厂”数据始终留在本地流通的只是模型的“锻造配方”参数从而从根本上保护了数据隐私。2.2. 为什么是FedAvg算法在众多联邦学习算法中我选择了最经典、最稳定的FedAvgFederated Averaging算法作为核心。原因有三简单有效其核心思想加权平均直观易懂在非独立同分布数据上也有不错的鲁棒性非常适合作为入门实现和概念验证。资源友好相比一些需要复杂加密通信或多次交互的算法FedAvg对客户端和服务器端的计算、通信资源要求相对较低。社区支持好有大量的开源实现和研究资料可供参考遇到问题容易排查。对于学业成绩预测这个具体任务我们本质上是在解决一个回归问题预测具体分数或分类问题预测是否挂科、成绩等级。因此我们需要为联邦学习框架选择一个合适的本地模型。考虑到学生特征可能包含线性关系和复杂非线性关系我选择了多层感知机作为本地模型的基础结构。它比简单的线性模型强大又比深度CNN/RNN更轻量适合在可能资源有限的院系服务器上运行。3. 从零构建系统核心模块实现详解理论清晰后我们进入实战环节。整个项目我采用模块化设计主要分为三大部分模拟数据集生成、联邦学习算法核心、以及Streamlit可视化应用。3.1 模拟数据集的精心设计真实的学生数据难以获取且敏感因此构建一个贴近现实的模拟数据集至关重要。一个好的模拟数据不仅能验证算法还能帮助我们理解数据分布对联邦学习的影响。我设计的数据生成逻辑考虑了以下因素特征工程包含了可能影响成绩的多种特征如previous_gpa: 上学期平均绩点连续值核心特征。attendance_rate: 出勤率连续值。library_visits: 月度图书馆访问次数整数。assignment_submitted: 作业提交率连续值。extracurricular_hours: 每周课外活动小时数连续值可能与成绩负相关。department_id: 院系编码类别特征用于模拟不同院系的数据分布差异。目标变量final_score 根据上述特征通过一个带有随机噪声的公式生成模拟真实成绩。关键引入数据异构性这是联邦学习的核心挑战。我故意让不同“院系”客户端的数据分布不同。例如院系A的学生previous_gpa普遍较高extracurricular_hours较少成绩公式的权重偏向于基础学业。院系B的学生previous_gpa中等但library_visits和assignment_submitted权重更高模拟更看重学习过程的院系。院系C的数据量较少且噪声更大模拟一个小规模或数据质量较差的参与方。import numpy as np import pandas as pd from sklearn.model_selection import train_test_split def generate_student_data(num_samples, dept_id, seed42): np.random.seed(seed dept_id) data {} # 根据院系ID调整数据分布 base_gpa 3.0 dept_id * 0.2 # 不同院系基础GPA不同 data[previous_gpa] np.random.normal(base_gpa, 0.5, num_samples).clip(0, 4.0) data[attendance_rate] np.random.beta(8, 2, num_samples) # 大部分出勤率高 data[library_visits] np.random.poisson(10 dept_id, num_samples) data[assignment_submitted] np.random.beta(9, 1, num_samples) data[extracurricular_hours] np.random.exponential(5, num_samples) data[department_id] dept_id # 生成最终成绩一个简单的线性组合 噪声权重因院系而异 weights { 0: [0.5, 0.15, 0.1, 0.2, -0.05], # 院系0的权重 1: [0.4, 0.2, 0.15, 0.2, -0.1], # 院系1的权重 2: [0.6, 0.1, 0.05, 0.15, -0.02] # 院系2的权重 } w weights[dept_id % 3] linear_component (w[0] * (data[previous_gpa] / 4.0 * 100) w[1] * (data[attendance_rate] * 100) w[2] * data[library_visits] w[3] * (data[assignment_submitted] * 100) w[4] * data[extracurricular_hours]) noise np.random.normal(0, 5, num_samples) # 加入随机噪声 data[final_score] linear_component noise data[final_score] data[final_score].clip(0, 100) # 限制在0-100分 df pd.DataFrame(data) # 添加一个二分类标签例如是否及格60 df[pass_fail] (df[final_score] 60).astype(int) return df # 为3个客户端生成数据 client_data {} for i in range(3): client_data[fclient_{i}] generate_student_data(num_samples200, dept_idi) # 划分训练集和测试集 train_df, test_df train_test_split(client_data[fclient_{i}], test_size0.2, random_state42) client_data[fclient_{i}_train] train_df client_data[fclient_{i}_test] test_df注意这个数据生成器是项目的基石。在实际演示中你可以通过调整weights字典和分布参数来直观展示数据异构性如何影响联邦学习的收敛速度和最终模型性能。这是理解联邦学习与传统集中式学习差异的关键。3.2 联邦学习核心引擎的实现接下来是实现FedAvg算法。我将其封装在一个FederatedLearningServer类中职责是协调整个训练过程。import copy import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset # 首先定义我们的本地神经网络模型 class StudentGradePredictor(nn.Module): def __init__(self, input_dim5, hidden_dim64, output_dim1): super(StudentGradePredictor, self).__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.2), nn.Linear(hidden_dim, hidden_dim//2), nn.ReLU(), nn.Linear(hidden_dim//2, output_dim) ) def forward(self, x): return self.net(x) class FederatedLearningServer: def __init__(self, global_model, client_names): self.global_model global_model self.client_names client_names # 记录每个客户端的数据量用于加权平均 self.client_sample_sizes {} self.history {loss: [], accuracy: []} def aggregate(self, client_models_updates, client_sample_sizes): FedAvg聚合根据数据量加权平均模型参数 total_size sum(client_sample_sizes.values()) global_dict self.global_model.state_dict() # 初始化一个归零的聚合字典 for key in global_dict.keys(): global_dict[key] torch.zeros_like(global_dict[key]) # 加权求和 for client_name, update in client_models_updates.items(): client_size client_sample_sizes[client_name] for key in global_dict.keys(): global_dict[key] update[key] * (client_size / total_size) # 加载聚合后的参数到全局模型 self.global_model.load_state_dict(global_dict) return self.global_model def train_one_round(self, clients_data_loaders, local_epochs3, lr0.01): 执行一轮联邦学习训练 client_updates {} self.client_sample_sizes {} for client_name, (train_loader, _) in clients_data_loaders.items(): # 1. 将当前全局模型副本发送给客户端 local_model copy.deepcopy(self.global_model) local_model.train() optimizer optim.SGD(local_model.parameters(), lrlr) criterion nn.MSELoss() # 回归任务用均方误差损失 # 2. 客户端本地训练 for epoch in range(local_epochs): for batch_x, batch_y in train_loader: optimizer.zero_grad() predictions local_model(batch_x).squeeze() loss criterion(predictions, batch_y) loss.backward() optimizer.step() # 3. 计算客户端模型与初始全局模型的差值作为“更新” global_dict self.global_model.state_dict() local_dict local_model.state_dict() update {key: local_dict[key] - global_dict[key] for key in global_dict.keys()} client_updates[client_name] update # 记录该客户端的数据量用训练集大小 self.client_sample_sizes[client_name] len(train_loader.dataset) # 4. 服务器聚合更新 self.aggregate(client_updates, self.client_sample_sizes) # 5. 可选评估本轮聚合后全局模型的性能 avg_loss, avg_acc self.evaluate_global(clients_data_loaders) self.history[loss].append(avg_loss) self.history[accuracy].append(avg_acc) print(fRound completed. Avg Loss: {avg_loss:.4f}, Avg Acc: {avg_acc:.4f}) return avg_loss, avg_acc def evaluate_global(self, clients_data_loaders): 在各自客户端的测试集上评估全局模型 self.global_model.eval() total_loss 0 total_correct 0 total_samples 0 criterion nn.MSELoss() with torch.no_grad(): for client_name, (_, test_loader) in clients_data_loaders.items(): for batch_x, batch_y in test_loader: predictions self.global_model(batch_x).squeeze() loss criterion(predictions, batch_y) total_loss loss.item() * batch_x.size(0) # 计算一个简单的准确率预测分数与真实分数差值在10分内算对 correct (torch.abs(predictions - batch_y) 10).sum().item() total_correct correct total_samples batch_x.size(0) avg_loss total_loss / total_samples if total_samples 0 else 0 avg_acc total_correct / total_samples if total_samples 0 else 0 return avg_loss, avg_acc这个服务器类实现了FedAvg的核心逻辑。其中有一个设计细节值得注意train_one_round方法中客户端上传的是模型参数的差值local_dict[key] - global_dict[key]而不是训练后的完整模型参数。这在数学上与直接上传模型参数并加权平均是等价的但有时在实现上更清晰也便于未来扩展为差分隐私联邦学习在更新上添加噪声。3.3 客户端数据准备与训练循环服务器准备好了我们需要为每个客户端准备数据加载器并启动训练循环。def prepare_dataloaders(client_data_dict, feature_cols, target_colfinal_score, batch_size32): 为每个客户端准备PyTorch DataLoader loaders {} for client_name, data_dict in client_data_dict.items(): if _train in client_name: base_name client_name.replace(_train, ) train_df data_dict test_df client_data_dict.get(f{base_name}_test, pd.DataFrame()) # 提取特征和标签 X_train torch.FloatTensor(train_df[feature_cols].values) y_train torch.FloatTensor(train_df[target_col].values) X_test torch.FloatTensor(test_df[feature_cols].values) if not test_df.empty else torch.FloatTensor() y_test torch.FloatTensor(test_df[target_col].values) if not test_df.empty else torch.FloatTensor() # 创建数据集和数据加载器 train_dataset TensorDataset(X_train, y_train) train_loader DataLoader(train_dataset, batch_sizebatch_size, shuffleTrue) test_dataset TensorDataset(X_test, y_test) if len(X_test) 0 else None test_loader DataLoader(test_dataset, batch_sizebatch_size) if test_dataset else None loaders[base_name] (train_loader, test_loader) return loaders # 准备数据 feature_columns [previous_gpa, attendance_rate, library_visits, assignment_submitted, extracurricular_hours] client_loaders prepare_dataloaders(client_data, feature_columns) # 初始化全局模型和服务器 global_model StudentGradePredictor(input_dimlen(feature_columns)) server FederatedLearningServer(global_model, list(client_loaders.keys())) # 进行多轮联邦训练 num_rounds 20 for round_idx in range(num_rounds): print(f\n Federated Round {round_idx 1}/{num_rounds} ) server.train_one_round(client_loaders, local_epochs2, lr0.02)运行这段代码你会在控制台看到每一轮训练后全局模型在全体客户端测试集上的平均损失和准确率。随着轮次增加损失应该下降准确率上升这证明联邦学习是有效的。4. 可视化与交互用Streamlit打造用户界面算法跑通了但黑盒子的输出对用户不友好。Streamlit的魔力在于它能让数据科学家用纯Python脚本快速创建出漂亮的Web应用。我们的目标是创建一个界面让用户能直观地看到联邦学习的过程和结果。4.1 应用布局与状态管理首先设计应用的主要页面布局并利用Streamlit的session_state来管理应用状态如训练是否开始、当前轮次等避免每次交互都从头开始。import streamlit as st import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots import time # 页面配置 st.set_page_config(page_title联邦学习成绩预测系统, layoutwide) st.title( 高校学生学业成绩预测系统 - 联邦学习演示) # 初始化session state保存关键状态 if server not in st.session_state: st.session_state.server None if client_loaders not in st.session_state: st.session_state.client_loaders None if training_history not in st.session_state: st.session_state.training_history {loss: [], accuracy: [], round: []} if is_training not in st.session_state: st.session_state.is_training False # 创建侧边栏用于控制和参数配置 with st.sidebar: st.header(⚙️ 联邦学习参数配置) num_clients st.slider(客户端数量院系, min_value2, max_value5, value3, step1) num_rounds st.slider(联邦训练轮数, min_value5, max_value50, value20, step5) local_epochs st.slider(客户端本地训练轮数, min_value1, max_value10, value2, step1) learning_rate st.slider(学习率, min_value0.001, max_value0.1, value0.02, step0.005, format%.3f) batch_size st.slider(批次大小, min_value16, max_value128, value32, step16) col1, col2 st.columns(2) with col1: btn_generate_data st.button( 生成模拟数据, use_container_widthTrue) with col2: btn_start_training st.button( 开始联邦训练, typeprimary, use_container_widthTrue, disabledst.session_state.server is None) st.divider() st.header( 数据洞察) if st.session_state.client_loaders: selected_client st.selectbox(选择查看的客户端, list(st.session_state.client_loaders.keys()))4.2 数据生成与探索性分析当用户点击“生成模拟数据”时我们调用之前写好的数据生成函数并将数据可视化让用户直观感受不同客户端的数据分布差异。# 主界面区域 tab1, tab2, tab3 st.tabs([ 数据概览, 训练过程, 模型预测]) with tab1: st.header(模拟学生数据集概览) if btn_generate_data: with st.spinner(正在生成模拟数据...): # 调用之前的数据生成函数生成指定数量的客户端数据 client_data {} for i in range(num_clients): df generate_student_data(num_samples200, dept_idi, seed42i) client_data[f院系_{i}] df # 准备DataLoader并存入session state feature_cols [previous_gpa, attendance_rate, library_visits, assignment_submitted, extracurricular_hours] loaders prepare_dataloaders_for_streamlit(client_data, feature_cols, batch_size) # 需要一个适配streamlit的版本 st.session_state.client_loaders loaders st.session_state.server None # 重置服务器 st.session_state.training_history {loss: [], accuracy: [], round: []} st.success(f已为 {num_clients} 个院系生成模拟数据) if st.session_state.client_loaders: # 展示数据样本 sample_client list(st.session_state.client_loaders.keys())[0] # 我们需要从原始DataFrame展示这里假设我们把原始df也存下来了 # 在实际代码中需要将df也存入session_state if raw_client_data in st.session_state: st.dataframe(st.session_state.raw_client_data[sample_client].head(10), use_container_widthTrue) # 使用Plotly绘制多客户端数据分布对比图 fig make_subplots(rows2, cols2, subplot_titles(上学期GPA分布, 出勤率分布, 图书馆访问次数, 最终成绩分布)) colors [blue, red, green, purple, orange] for idx, (client_name, df) in enumerate(st.session_state.raw_client_data.items()): color colors[idx % len(colors)] # GPA分布 fig.add_trace(go.Histogram(xdf[previous_gpa], nameclient_name, opacity0.7, marker_colorcolor, legendgroupclient_name), row1, col1) # 出勤率分布 fig.add_trace(go.Histogram(xdf[attendance_rate], nameclient_name, opacity0.7, marker_colorcolor, legendgroupclient_name, showlegendFalse), row1, col2) # 图书馆访问次数 fig.add_trace(go.Box(ydf[library_visits], nameclient_name, marker_colorcolor, legendgroupclient_name, showlegendFalse), row2, col1) # 成绩分布 fig.add_trace(go.Violin(ydf[final_score], nameclient_name, marker_colorcolor, legendgroupclient_name, showlegendFalse), row2, col2) fig.update_layout(height700, title_text各院系客户端数据分布对比, barmodeoverlay) fig.update_xaxes(title_textGPA, row1, col1) fig.update_xaxes(title_text出勤率, row1, col2) fig.update_yaxes(title_text次数, row2, col1) fig.update_yaxes(title_text分数, row2, col2) st.plotly_chart(fig, use_container_widthTrue) st.caption(**解读**通过上图可以清晰看到不同院系客户端的学生数据在特征分布上存在差异数据异构性这正是联邦学习需要解决的核心问题。)4.3 动态训练过程可视化这是整个演示最精彩的部分。当用户点击“开始联邦训练”后我们需要在界面上动态地展示训练过程包括损失/准确率曲线、全局模型权重变化等。with tab2: st.header(联邦学习训练过程) if btn_start_training and st.session_state.client_loaders: st.session_state.is_training True # 初始化全局模型和服务器 global_model StudentGradePredictor(input_dimlen(feature_cols)) server FederatedLearningServer(global_model, list(st.session_state.client_loaders.keys())) st.session_state.server server # 创建进度条和动态图表占位符 progress_bar st.progress(0) status_text st.empty() chart_placeholder_loss st.empty() chart_placeholder_acc st.empty() # 模拟训练循环并更新界面 for round_idx in range(num_rounds): status_text.text(f正在进行第 {round_idx 1} / {num_rounds} 轮联邦训练...) # 执行一轮训练 avg_loss, avg_acc server.train_one_round( st.session_state.client_loaders, local_epochslocal_epochs, lrlearning_rate ) # 记录历史 st.session_state.training_history[loss].append(avg_loss) st.session_state.training_history[accuracy].append(avg_acc) st.session_state.training_history[round].append(round_idx 1) # 更新进度条 progress_bar.progress((round_idx 1) / num_rounds) # 实时更新损失和准确率曲线 fig_loss go.Figure() fig_loss.add_trace(go.Scatter( xst.session_state.training_history[round], yst.session_state.training_history[loss], modelinesmarkers, name全局模型平均损失, linedict(colorfirebrick, width2) )) fig_loss.update_layout(title联邦训练损失下降曲线, xaxis_title训练轮次, yaxis_title平均损失 (MSE)) chart_placeholder_loss.plotly_chart(fig_loss, use_container_widthTrue) fig_acc go.Figure() fig_acc.add_trace(go.Scatter( xst.session_state.training_history[round], yst.session_state.training_history[accuracy], modelinesmarkers, name全局模型预测准确率, linedict(colorroyalblue, width2), yaxisy )) fig_acc.update_layout(title联邦训练准确率提升曲线, xaxis_title训练轮次, yaxis_title准确率 (误差10分)) chart_placeholder_acc.plotly_chart(fig_acc, use_container_widthTrue) time.sleep(0.3) # 稍微延迟让动画可见 st.session_state.is_training False progress_bar.empty() status_text.success(f联邦训练完成共进行 {num_rounds} 轮。最终准确率: {st.session_state.training_history[accuracy][-1]:.2%}) elif not st.session_state.client_loaders: st.warning(请先在左侧栏点击「生成模拟数据」。) else: # 如果已有训练历史则静态展示 if st.session_state.training_history[round]: col1, col2 st.columns(2) with col1: st.metric(最终损失, f{st.session_state.training_history[loss][-1]:.4f}) with col2: st.metric(最终准确率, f{st.session_state.training_history[accuracy][-1]:.2%}) # 绘制完整的训练历史图表 fig make_subplots(specs[[{secondary_y: True}]]) fig.add_trace(go.Scatter(xst.session_state.training_history[round], yst.session_state.training_history[loss], name损失, linedict(colorred)), secondary_yFalse) fig.add_trace(go.Scatter(xst.session_state.training_history[round], yst.session_state.training_history[accuracy], name准确率, linedict(colorblue)), secondary_yTrue) fig.update_layout(title联邦学习训练历史, xaxis_title训练轮次) fig.update_yaxes(title_text损失 (MSE), secondary_yFalse) fig.update_yaxes(title_text准确率, secondary_yTrue) st.plotly_chart(fig, use_container_widthTrue)4.4 模型预测与对比分析在最后一个标签页我们提供交互功能让用户输入一个虚拟学生的特征分别用联邦学习训练出的全局模型和仅用单个客户端数据训练的本地模型进行预测直观对比效果。with tab3: st.header(模型预测与对比) if st.session_state.server and st.session_state.server.global_model: st.subheader(输入学生特征进行预测) col1, col2, col3 st.columns(3) with col1: prev_gpa st.slider(上学期平均绩点 (0-4), 0.0, 4.0, 3.2, 0.1) attendance st.slider(出勤率 (%), 0.0, 100.0, 85.0, 1.0) / 100.0 with col2: library st.slider(月度图书馆访问次数, 0, 50, 12, 1) assignment st.slider(作业提交率 (%), 0.0, 100.0, 90.0, 1.0) / 100.0 with col3: extracurricular st.slider(每周课外活动小时数, 0.0, 30.0, 8.0, 0.5) dept st.selectbox(所属院系用于选择本地模型, list(st.session_state.client_loaders.keys())) input_features torch.FloatTensor([[prev_gpa, attendance, library, assignment, extracurricular]]) # 使用全局模型预测 global_model st.session_state.server.global_model global_model.eval() with torch.no_grad(): pred_global global_model(input_features).item() st.metric( 联邦全局模型预测成绩, f{pred_global:.1f} 分) # 使用所选客户端的本地模型预测假设我们保存了最后一轮本地模型 # 这里为了演示我们临时用该客户端数据从头训练一个本地模型作为对比 st.subheader(与本地模型对比) if st.button(训练并对比本地模型, keytrain_local): with st.spinner(f正在使用「{dept}」的数据训练本地模型...): # 获取该客户端数据 train_loader, _ st.session_state.client_loaders[dept] local_model StudentGradePredictor(input_dim5) criterion nn.MSELoss() optimizer optim.SGD(local_model.parameters(), lr0.02) local_model.train() for epoch in range(10): # 本地训练10个epoch for batch_x, batch_y in train_loader: optimizer.zero_grad() pred local_model(batch_x).squeeze() loss criterion(pred, batch_y) loss.backward() optimizer.step() local_model.eval() with torch.no_grad(): pred_local local_model(input_features).item() col1, col2 st.columns(2) with col1: st.metric( 联邦全局模型预测, f{pred_global:.1f} 分) with col2: st.metric( 仅用本地数据训练的模型预测, f{pred_local:.1f} 分) st.info(**对比分析**联邦全局模型聚合了所有院系的数据知识通常比仅用单一院系数据训练的本地模型具有更好的泛化能力和预测稳定性尤其是在面对数据分布不同的学生时。) else: st.info(请先完成联邦学习训练以启用预测功能。)5. 部署、优化与踩坑实录将代码跑起来只是第一步要让这个系统真正可用还需要考虑部署和优化。在这个过程中我遇到了几个典型问题。5.1 模型收敛不稳定与调参心得在最初的版本中联邦模型的损失曲线波动很大准确率提升缓慢。经过排查问题出在几个方面学习率过大联邦学习中客户端本地更新多次后才聚合相当于放大了学习率。我将服务器下发的学习率调低为传统集中式训练的1/5到1/10后收敛稳定了许多。客户端本地训练轮数local_epochs是一个关键超参数。设置太小如1模型学不到东西设置太大如10每个客户端会朝着自己本地数据的最优点“跑偏”导致客户端漂移聚合困难。通常设置在2-5之间是个不错的起点。数据标准化各客户端数据特征尺度不同如GPA是0-4图书馆次数是0-50会严重影响模型训练。必须在每个客户端本地分别进行特征标准化如Z-Score但切记标准化参数均值、方差不能共享以模拟真实的隐私保护场景。实操技巧在prepare_dataloaders函数中我为每个客户端单独计算了训练集的均值和标准差并用它们来标准化该客户端的训练集和测试集。这比使用全局统计量更符合联邦学习的隐私前提。5.2 Streamlit应用性能与状态管理陷阱Streamlit的编程模型是“脚本从头到尾执行”每次交互都会重新运行整个脚本。这带来了两个挑战训练状态丢失点击按钮开始训练后下一轮循环时session_state里的模型和状态可能因为脚本重跑而重置。我的解决方案是将所有关键对象server,client_loaders,training_history都存入st.session_state并在脚本开头检查它们是否存在。长时间训练阻塞联邦训练20轮可能需要几十秒这会阻塞Streamlit界面导致“运行中”的旋转图标一直转用户体验差。对于真正的演示可以考虑使用st.spinner和st.progress如上文代码所示提供视觉反馈。异步或分步执行将每一轮训练包装成一个生成器函数配合st.empty()占位符和time.sleep()进行分步更新模拟实时效果。对于极耗时的任务应使用st.cache_data缓存结果或考虑用后台线程执行。5.3 从模拟到现实的鸿沟本项目使用的是精心设计的模拟数据。但在真实场景中挑战更大客户端掉线与异构性真实院系服务器可能离线或响应慢。需要实现健壮的通信机制、超时处理和客户端选择策略。安全与隐私增强基础的FedAvg仅提供“算法层面”的隐私。生产系统需要考虑加入差分隐私在客户端更新上添加噪声或安全聚合使用密码学协议确保服务器无法看到单个客户端的更新来提供更强的隐私保障。模型个性化训练出的全局模型可能对所有客户端“一视同仁”但某些院系可能希望模型更贴合自己的数据分布。可以在联邦学习框架上集成个性化技术让每个客户端在全局模型的基础上进行微调。这个项目从构思到实现让我对联邦学习的理论、工程实现和可视化演示有了更深的体会。它不仅仅是一个算法更是一套解决“数据隐私与价值利用”矛盾的工程框架。用Streamlit将其可视化则大大降低了理解门槛使得这项前沿技术能够被更多非技术背景的决策者所理解和接受。代码虽已开源但其中关于数据异构性的模拟、联邦平均的实现细节以及Streamlit状态管理的技巧都是经过多次调试才稳定下来的希望能为你的类似项目提供一个坚实的起点。本文还有配套的精品资源点击获取
基于联邦学习与Streamlit的高校成绩预测系统：隐私保护下的协同建模实践
简介本资源是一套面向高校计算机、人工智能及相关专业学生与教师的毕业设计级项目聚焦于隐私保护场景下的学生成绩预测问题基于联邦学习框架实现多客户端协同建模并通过Streamlit构建交互式可视化分析界面。资源包含完整可运行Python源码、真实模拟的学生成绩CSV数据集、多种联邦优化算法如FedRep、FedProx、Ditto、Scaffold等实现模块及训练日志文件覆盖本地训练、模型聚合、性能评估全流程。压缩包共67个文件含18个核心Python脚本含主程序、模型定义、采样与训练工具、7个CSV数据与结果记录文件、11个备份文件.zbak、1个知识拓展资料ZIP及说明文档整体大小为6.02MB。目前已有111人学习下载项目答辩获95分高分代码经实测调试无报错附带清晰目录结构与模块化设计便于理解联邦学习通信机制、快速复现实验或在此基础上扩展新算法与数据场景。1. 项目缘起当成绩预测遇上数据隐私在高校的教学管理工作中学业成绩预测一直是个既重要又敏感的话题。教务部门希望能提前识别学业困难的学生以便进行精准帮扶学生也希望能对自己的学业走向有个大致的预判。传统的做法要么是基于单个学生的历史数据进行建模预测能力有限要么是汇集全校所有学生的数据到一个中心服务器进行集中式机器学习。后者听起来效果会更好但问题也随之而来学生的成绩、选课、出勤等数据属于高度敏感的个人隐私将全校数据集中处理不仅面临巨大的数据泄露风险也触碰了日益严格的数据保护法规如GDPR、国内的个人信息保护法的红线。这就形成了一个典型的“数据孤岛”困境数据有价值但不能汇聚。我最近完成的一个项目正是为了解决这个矛盾。我们尝试用联邦学习Federated Learning技术构建一个高校学生学业成绩预测系统。它的核心思想是“数据不动模型动”我们不需要把各院系、各班级的学生数据收集到一个中心而是让机器学习模型“走”到数据所在的地方比如各院系的服务器或安全环境进行训练只将训练得到的模型参数更新而非原始数据加密后传回中心进行聚合。这样在保护数据隐私的前提下我们依然能够利用分散在各处的数据训练出一个强大的全局预测模型。为了让这个听起来有些抽象的技术能够被教务老师或相关研究人员直观地理解和使用我选择了Python作为实现语言并用Streamlit快速搭建了一个交互式可视化界面。整个项目从联邦学习算法实现、模拟数据集构建到前端界面开发形成了一套完整的、可复现的解决方案。下面我就把这个项目的实现思路、关键技术细节、踩过的坑以及最终效果毫无保留地分享出来。2. 联邦学习在不汇聚数据的前提下实现协同建模在深入代码之前我们必须先搞清楚联邦学习到底是如何工作的以及为什么它适合学业成绩预测这个场景。这决定了我们整个系统的架构设计。2.1 联邦学习的基本范式与我们的选择联邦学习主要有三种范式横向联邦学习、纵向联邦学习和联邦迁移学习。针对高校成绩预测的场景我们面临的数据特点是参与方例如计算机学院、经管学院、文学院拥有的学生样本不同学生ID不同但特征空间是相似的大家都有“上学期平均绩点”、“出勤率”、“图书馆借阅次数”等特征。这完美契合了横向联邦学习的适用条件。横向联邦学习也叫基于样本的联邦学习其经典流程以FedAvg算法为例如下中心服务器初始化服务器初始化一个全局机器学习模型比如一个神经网络并将初始模型参数下发给所有客户端各院系。客户端本地训练各客户端在本地用自己的私有数据对接收到的全局模型进行训练例如进行几个轮次的梯度下降。客户端上传更新训练完成后各客户端将计算得到的模型参数更新或梯度加密后上传给中心服务器。服务器聚合更新服务器收集到所有客户端的更新后对其进行加权平均例如根据各客户端的数据量加权得到新的全局模型参数。迭代服务器将聚合后的新全局模型参数再次下发给客户端重复步骤2-4直至模型收敛。这个过程就像是一个“分布式模型锻造厂”数据始终留在本地流通的只是模型的“锻造配方”参数从而从根本上保护了数据隐私。2.2. 为什么是FedAvg算法在众多联邦学习算法中我选择了最经典、最稳定的FedAvgFederated Averaging算法作为核心。原因有三简单有效其核心思想加权平均直观易懂在非独立同分布数据上也有不错的鲁棒性非常适合作为入门实现和概念验证。资源友好相比一些需要复杂加密通信或多次交互的算法FedAvg对客户端和服务器端的计算、通信资源要求相对较低。社区支持好有大量的开源实现和研究资料可供参考遇到问题容易排查。对于学业成绩预测这个具体任务我们本质上是在解决一个回归问题预测具体分数或分类问题预测是否挂科、成绩等级。因此我们需要为联邦学习框架选择一个合适的本地模型。考虑到学生特征可能包含线性关系和复杂非线性关系我选择了多层感知机作为本地模型的基础结构。它比简单的线性模型强大又比深度CNN/RNN更轻量适合在可能资源有限的院系服务器上运行。3. 从零构建系统核心模块实现详解理论清晰后我们进入实战环节。整个项目我采用模块化设计主要分为三大部分模拟数据集生成、联邦学习算法核心、以及Streamlit可视化应用。3.1 模拟数据集的精心设计真实的学生数据难以获取且敏感因此构建一个贴近现实的模拟数据集至关重要。一个好的模拟数据不仅能验证算法还能帮助我们理解数据分布对联邦学习的影响。我设计的数据生成逻辑考虑了以下因素特征工程包含了可能影响成绩的多种特征如previous_gpa: 上学期平均绩点连续值核心特征。attendance_rate: 出勤率连续值。library_visits: 月度图书馆访问次数整数。assignment_submitted: 作业提交率连续值。extracurricular_hours: 每周课外活动小时数连续值可能与成绩负相关。department_id: 院系编码类别特征用于模拟不同院系的数据分布差异。目标变量final_score 根据上述特征通过一个带有随机噪声的公式生成模拟真实成绩。关键引入数据异构性这是联邦学习的核心挑战。我故意让不同“院系”客户端的数据分布不同。例如院系A的学生previous_gpa普遍较高extracurricular_hours较少成绩公式的权重偏向于基础学业。院系B的学生previous_gpa中等但library_visits和assignment_submitted权重更高模拟更看重学习过程的院系。院系C的数据量较少且噪声更大模拟一个小规模或数据质量较差的参与方。import numpy as np import pandas as pd from sklearn.model_selection import train_test_split def generate_student_data(num_samples, dept_id, seed42): np.random.seed(seed dept_id) data {} # 根据院系ID调整数据分布 base_gpa 3.0 dept_id * 0.2 # 不同院系基础GPA不同 data[previous_gpa] np.random.normal(base_gpa, 0.5, num_samples).clip(0, 4.0) data[attendance_rate] np.random.beta(8, 2, num_samples) # 大部分出勤率高 data[library_visits] np.random.poisson(10 dept_id, num_samples) data[assignment_submitted] np.random.beta(9, 1, num_samples) data[extracurricular_hours] np.random.exponential(5, num_samples) data[department_id] dept_id # 生成最终成绩一个简单的线性组合 噪声权重因院系而异 weights { 0: [0.5, 0.15, 0.1, 0.2, -0.05], # 院系0的权重 1: [0.4, 0.2, 0.15, 0.2, -0.1], # 院系1的权重 2: [0.6, 0.1, 0.05, 0.15, -0.02] # 院系2的权重 } w weights[dept_id % 3] linear_component (w[0] * (data[previous_gpa] / 4.0 * 100) w[1] * (data[attendance_rate] * 100) w[2] * data[library_visits] w[3] * (data[assignment_submitted] * 100) w[4] * data[extracurricular_hours]) noise np.random.normal(0, 5, num_samples) # 加入随机噪声 data[final_score] linear_component noise data[final_score] data[final_score].clip(0, 100) # 限制在0-100分 df pd.DataFrame(data) # 添加一个二分类标签例如是否及格60 df[pass_fail] (df[final_score] 60).astype(int) return df # 为3个客户端生成数据 client_data {} for i in range(3): client_data[fclient_{i}] generate_student_data(num_samples200, dept_idi) # 划分训练集和测试集 train_df, test_df train_test_split(client_data[fclient_{i}], test_size0.2, random_state42) client_data[fclient_{i}_train] train_df client_data[fclient_{i}_test] test_df注意这个数据生成器是项目的基石。在实际演示中你可以通过调整weights字典和分布参数来直观展示数据异构性如何影响联邦学习的收敛速度和最终模型性能。这是理解联邦学习与传统集中式学习差异的关键。3.2 联邦学习核心引擎的实现接下来是实现FedAvg算法。我将其封装在一个FederatedLearningServer类中职责是协调整个训练过程。import copy import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset # 首先定义我们的本地神经网络模型 class StudentGradePredictor(nn.Module): def __init__(self, input_dim5, hidden_dim64, output_dim1): super(StudentGradePredictor, self).__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.2), nn.Linear(hidden_dim, hidden_dim//2), nn.ReLU(), nn.Linear(hidden_dim//2, output_dim) ) def forward(self, x): return self.net(x) class FederatedLearningServer: def __init__(self, global_model, client_names): self.global_model global_model self.client_names client_names # 记录每个客户端的数据量用于加权平均 self.client_sample_sizes {} self.history {loss: [], accuracy: []} def aggregate(self, client_models_updates, client_sample_sizes): FedAvg聚合根据数据量加权平均模型参数 total_size sum(client_sample_sizes.values()) global_dict self.global_model.state_dict() # 初始化一个归零的聚合字典 for key in global_dict.keys(): global_dict[key] torch.zeros_like(global_dict[key]) # 加权求和 for client_name, update in client_models_updates.items(): client_size client_sample_sizes[client_name] for key in global_dict.keys(): global_dict[key] update[key] * (client_size / total_size) # 加载聚合后的参数到全局模型 self.global_model.load_state_dict(global_dict) return self.global_model def train_one_round(self, clients_data_loaders, local_epochs3, lr0.01): 执行一轮联邦学习训练 client_updates {} self.client_sample_sizes {} for client_name, (train_loader, _) in clients_data_loaders.items(): # 1. 将当前全局模型副本发送给客户端 local_model copy.deepcopy(self.global_model) local_model.train() optimizer optim.SGD(local_model.parameters(), lrlr) criterion nn.MSELoss() # 回归任务用均方误差损失 # 2. 客户端本地训练 for epoch in range(local_epochs): for batch_x, batch_y in train_loader: optimizer.zero_grad() predictions local_model(batch_x).squeeze() loss criterion(predictions, batch_y) loss.backward() optimizer.step() # 3. 计算客户端模型与初始全局模型的差值作为“更新” global_dict self.global_model.state_dict() local_dict local_model.state_dict() update {key: local_dict[key] - global_dict[key] for key in global_dict.keys()} client_updates[client_name] update # 记录该客户端的数据量用训练集大小 self.client_sample_sizes[client_name] len(train_loader.dataset) # 4. 服务器聚合更新 self.aggregate(client_updates, self.client_sample_sizes) # 5. 可选评估本轮聚合后全局模型的性能 avg_loss, avg_acc self.evaluate_global(clients_data_loaders) self.history[loss].append(avg_loss) self.history[accuracy].append(avg_acc) print(fRound completed. Avg Loss: {avg_loss:.4f}, Avg Acc: {avg_acc:.4f}) return avg_loss, avg_acc def evaluate_global(self, clients_data_loaders): 在各自客户端的测试集上评估全局模型 self.global_model.eval() total_loss 0 total_correct 0 total_samples 0 criterion nn.MSELoss() with torch.no_grad(): for client_name, (_, test_loader) in clients_data_loaders.items(): for batch_x, batch_y in test_loader: predictions self.global_model(batch_x).squeeze() loss criterion(predictions, batch_y) total_loss loss.item() * batch_x.size(0) # 计算一个简单的准确率预测分数与真实分数差值在10分内算对 correct (torch.abs(predictions - batch_y) 10).sum().item() total_correct correct total_samples batch_x.size(0) avg_loss total_loss / total_samples if total_samples 0 else 0 avg_acc total_correct / total_samples if total_samples 0 else 0 return avg_loss, avg_acc这个服务器类实现了FedAvg的核心逻辑。其中有一个设计细节值得注意train_one_round方法中客户端上传的是模型参数的差值local_dict[key] - global_dict[key]而不是训练后的完整模型参数。这在数学上与直接上传模型参数并加权平均是等价的但有时在实现上更清晰也便于未来扩展为差分隐私联邦学习在更新上添加噪声。3.3 客户端数据准备与训练循环服务器准备好了我们需要为每个客户端准备数据加载器并启动训练循环。def prepare_dataloaders(client_data_dict, feature_cols, target_colfinal_score, batch_size32): 为每个客户端准备PyTorch DataLoader loaders {} for client_name, data_dict in client_data_dict.items(): if _train in client_name: base_name client_name.replace(_train, ) train_df data_dict test_df client_data_dict.get(f{base_name}_test, pd.DataFrame()) # 提取特征和标签 X_train torch.FloatTensor(train_df[feature_cols].values) y_train torch.FloatTensor(train_df[target_col].values) X_test torch.FloatTensor(test_df[feature_cols].values) if not test_df.empty else torch.FloatTensor() y_test torch.FloatTensor(test_df[target_col].values) if not test_df.empty else torch.FloatTensor() # 创建数据集和数据加载器 train_dataset TensorDataset(X_train, y_train) train_loader DataLoader(train_dataset, batch_sizebatch_size, shuffleTrue) test_dataset TensorDataset(X_test, y_test) if len(X_test) 0 else None test_loader DataLoader(test_dataset, batch_sizebatch_size) if test_dataset else None loaders[base_name] (train_loader, test_loader) return loaders # 准备数据 feature_columns [previous_gpa, attendance_rate, library_visits, assignment_submitted, extracurricular_hours] client_loaders prepare_dataloaders(client_data, feature_columns) # 初始化全局模型和服务器 global_model StudentGradePredictor(input_dimlen(feature_columns)) server FederatedLearningServer(global_model, list(client_loaders.keys())) # 进行多轮联邦训练 num_rounds 20 for round_idx in range(num_rounds): print(f\n Federated Round {round_idx 1}/{num_rounds} ) server.train_one_round(client_loaders, local_epochs2, lr0.02)运行这段代码你会在控制台看到每一轮训练后全局模型在全体客户端测试集上的平均损失和准确率。随着轮次增加损失应该下降准确率上升这证明联邦学习是有效的。4. 可视化与交互用Streamlit打造用户界面算法跑通了但黑盒子的输出对用户不友好。Streamlit的魔力在于它能让数据科学家用纯Python脚本快速创建出漂亮的Web应用。我们的目标是创建一个界面让用户能直观地看到联邦学习的过程和结果。4.1 应用布局与状态管理首先设计应用的主要页面布局并利用Streamlit的session_state来管理应用状态如训练是否开始、当前轮次等避免每次交互都从头开始。import streamlit as st import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots import time # 页面配置 st.set_page_config(page_title联邦学习成绩预测系统, layoutwide) st.title( 高校学生学业成绩预测系统 - 联邦学习演示) # 初始化session state保存关键状态 if server not in st.session_state: st.session_state.server None if client_loaders not in st.session_state: st.session_state.client_loaders None if training_history not in st.session_state: st.session_state.training_history {loss: [], accuracy: [], round: []} if is_training not in st.session_state: st.session_state.is_training False # 创建侧边栏用于控制和参数配置 with st.sidebar: st.header(⚙️ 联邦学习参数配置) num_clients st.slider(客户端数量院系, min_value2, max_value5, value3, step1) num_rounds st.slider(联邦训练轮数, min_value5, max_value50, value20, step5) local_epochs st.slider(客户端本地训练轮数, min_value1, max_value10, value2, step1) learning_rate st.slider(学习率, min_value0.001, max_value0.1, value0.02, step0.005, format%.3f) batch_size st.slider(批次大小, min_value16, max_value128, value32, step16) col1, col2 st.columns(2) with col1: btn_generate_data st.button( 生成模拟数据, use_container_widthTrue) with col2: btn_start_training st.button( 开始联邦训练, typeprimary, use_container_widthTrue, disabledst.session_state.server is None) st.divider() st.header( 数据洞察) if st.session_state.client_loaders: selected_client st.selectbox(选择查看的客户端, list(st.session_state.client_loaders.keys()))4.2 数据生成与探索性分析当用户点击“生成模拟数据”时我们调用之前写好的数据生成函数并将数据可视化让用户直观感受不同客户端的数据分布差异。# 主界面区域 tab1, tab2, tab3 st.tabs([ 数据概览, 训练过程, 模型预测]) with tab1: st.header(模拟学生数据集概览) if btn_generate_data: with st.spinner(正在生成模拟数据...): # 调用之前的数据生成函数生成指定数量的客户端数据 client_data {} for i in range(num_clients): df generate_student_data(num_samples200, dept_idi, seed42i) client_data[f院系_{i}] df # 准备DataLoader并存入session state feature_cols [previous_gpa, attendance_rate, library_visits, assignment_submitted, extracurricular_hours] loaders prepare_dataloaders_for_streamlit(client_data, feature_cols, batch_size) # 需要一个适配streamlit的版本 st.session_state.client_loaders loaders st.session_state.server None # 重置服务器 st.session_state.training_history {loss: [], accuracy: [], round: []} st.success(f已为 {num_clients} 个院系生成模拟数据) if st.session_state.client_loaders: # 展示数据样本 sample_client list(st.session_state.client_loaders.keys())[0] # 我们需要从原始DataFrame展示这里假设我们把原始df也存下来了 # 在实际代码中需要将df也存入session_state if raw_client_data in st.session_state: st.dataframe(st.session_state.raw_client_data[sample_client].head(10), use_container_widthTrue) # 使用Plotly绘制多客户端数据分布对比图 fig make_subplots(rows2, cols2, subplot_titles(上学期GPA分布, 出勤率分布, 图书馆访问次数, 最终成绩分布)) colors [blue, red, green, purple, orange] for idx, (client_name, df) in enumerate(st.session_state.raw_client_data.items()): color colors[idx % len(colors)] # GPA分布 fig.add_trace(go.Histogram(xdf[previous_gpa], nameclient_name, opacity0.7, marker_colorcolor, legendgroupclient_name), row1, col1) # 出勤率分布 fig.add_trace(go.Histogram(xdf[attendance_rate], nameclient_name, opacity0.7, marker_colorcolor, legendgroupclient_name, showlegendFalse), row1, col2) # 图书馆访问次数 fig.add_trace(go.Box(ydf[library_visits], nameclient_name, marker_colorcolor, legendgroupclient_name, showlegendFalse), row2, col1) # 成绩分布 fig.add_trace(go.Violin(ydf[final_score], nameclient_name, marker_colorcolor, legendgroupclient_name, showlegendFalse), row2, col2) fig.update_layout(height700, title_text各院系客户端数据分布对比, barmodeoverlay) fig.update_xaxes(title_textGPA, row1, col1) fig.update_xaxes(title_text出勤率, row1, col2) fig.update_yaxes(title_text次数, row2, col1) fig.update_yaxes(title_text分数, row2, col2) st.plotly_chart(fig, use_container_widthTrue) st.caption(**解读**通过上图可以清晰看到不同院系客户端的学生数据在特征分布上存在差异数据异构性这正是联邦学习需要解决的核心问题。)4.3 动态训练过程可视化这是整个演示最精彩的部分。当用户点击“开始联邦训练”后我们需要在界面上动态地展示训练过程包括损失/准确率曲线、全局模型权重变化等。with tab2: st.header(联邦学习训练过程) if btn_start_training and st.session_state.client_loaders: st.session_state.is_training True # 初始化全局模型和服务器 global_model StudentGradePredictor(input_dimlen(feature_cols)) server FederatedLearningServer(global_model, list(st.session_state.client_loaders.keys())) st.session_state.server server # 创建进度条和动态图表占位符 progress_bar st.progress(0) status_text st.empty() chart_placeholder_loss st.empty() chart_placeholder_acc st.empty() # 模拟训练循环并更新界面 for round_idx in range(num_rounds): status_text.text(f正在进行第 {round_idx 1} / {num_rounds} 轮联邦训练...) # 执行一轮训练 avg_loss, avg_acc server.train_one_round( st.session_state.client_loaders, local_epochslocal_epochs, lrlearning_rate ) # 记录历史 st.session_state.training_history[loss].append(avg_loss) st.session_state.training_history[accuracy].append(avg_acc) st.session_state.training_history[round].append(round_idx 1) # 更新进度条 progress_bar.progress((round_idx 1) / num_rounds) # 实时更新损失和准确率曲线 fig_loss go.Figure() fig_loss.add_trace(go.Scatter( xst.session_state.training_history[round], yst.session_state.training_history[loss], modelinesmarkers, name全局模型平均损失, linedict(colorfirebrick, width2) )) fig_loss.update_layout(title联邦训练损失下降曲线, xaxis_title训练轮次, yaxis_title平均损失 (MSE)) chart_placeholder_loss.plotly_chart(fig_loss, use_container_widthTrue) fig_acc go.Figure() fig_acc.add_trace(go.Scatter( xst.session_state.training_history[round], yst.session_state.training_history[accuracy], modelinesmarkers, name全局模型预测准确率, linedict(colorroyalblue, width2), yaxisy )) fig_acc.update_layout(title联邦训练准确率提升曲线, xaxis_title训练轮次, yaxis_title准确率 (误差10分)) chart_placeholder_acc.plotly_chart(fig_acc, use_container_widthTrue) time.sleep(0.3) # 稍微延迟让动画可见 st.session_state.is_training False progress_bar.empty() status_text.success(f联邦训练完成共进行 {num_rounds} 轮。最终准确率: {st.session_state.training_history[accuracy][-1]:.2%}) elif not st.session_state.client_loaders: st.warning(请先在左侧栏点击「生成模拟数据」。) else: # 如果已有训练历史则静态展示 if st.session_state.training_history[round]: col1, col2 st.columns(2) with col1: st.metric(最终损失, f{st.session_state.training_history[loss][-1]:.4f}) with col2: st.metric(最终准确率, f{st.session_state.training_history[accuracy][-1]:.2%}) # 绘制完整的训练历史图表 fig make_subplots(specs[[{secondary_y: True}]]) fig.add_trace(go.Scatter(xst.session_state.training_history[round], yst.session_state.training_history[loss], name损失, linedict(colorred)), secondary_yFalse) fig.add_trace(go.Scatter(xst.session_state.training_history[round], yst.session_state.training_history[accuracy], name准确率, linedict(colorblue)), secondary_yTrue) fig.update_layout(title联邦学习训练历史, xaxis_title训练轮次) fig.update_yaxes(title_text损失 (MSE), secondary_yFalse) fig.update_yaxes(title_text准确率, secondary_yTrue) st.plotly_chart(fig, use_container_widthTrue)4.4 模型预测与对比分析在最后一个标签页我们提供交互功能让用户输入一个虚拟学生的特征分别用联邦学习训练出的全局模型和仅用单个客户端数据训练的本地模型进行预测直观对比效果。with tab3: st.header(模型预测与对比) if st.session_state.server and st.session_state.server.global_model: st.subheader(输入学生特征进行预测) col1, col2, col3 st.columns(3) with col1: prev_gpa st.slider(上学期平均绩点 (0-4), 0.0, 4.0, 3.2, 0.1) attendance st.slider(出勤率 (%), 0.0, 100.0, 85.0, 1.0) / 100.0 with col2: library st.slider(月度图书馆访问次数, 0, 50, 12, 1) assignment st.slider(作业提交率 (%), 0.0, 100.0, 90.0, 1.0) / 100.0 with col3: extracurricular st.slider(每周课外活动小时数, 0.0, 30.0, 8.0, 0.5) dept st.selectbox(所属院系用于选择本地模型, list(st.session_state.client_loaders.keys())) input_features torch.FloatTensor([[prev_gpa, attendance, library, assignment, extracurricular]]) # 使用全局模型预测 global_model st.session_state.server.global_model global_model.eval() with torch.no_grad(): pred_global global_model(input_features).item() st.metric( 联邦全局模型预测成绩, f{pred_global:.1f} 分) # 使用所选客户端的本地模型预测假设我们保存了最后一轮本地模型 # 这里为了演示我们临时用该客户端数据从头训练一个本地模型作为对比 st.subheader(与本地模型对比) if st.button(训练并对比本地模型, keytrain_local): with st.spinner(f正在使用「{dept}」的数据训练本地模型...): # 获取该客户端数据 train_loader, _ st.session_state.client_loaders[dept] local_model StudentGradePredictor(input_dim5) criterion nn.MSELoss() optimizer optim.SGD(local_model.parameters(), lr0.02) local_model.train() for epoch in range(10): # 本地训练10个epoch for batch_x, batch_y in train_loader: optimizer.zero_grad() pred local_model(batch_x).squeeze() loss criterion(pred, batch_y) loss.backward() optimizer.step() local_model.eval() with torch.no_grad(): pred_local local_model(input_features).item() col1, col2 st.columns(2) with col1: st.metric( 联邦全局模型预测, f{pred_global:.1f} 分) with col2: st.metric( 仅用本地数据训练的模型预测, f{pred_local:.1f} 分) st.info(**对比分析**联邦全局模型聚合了所有院系的数据知识通常比仅用单一院系数据训练的本地模型具有更好的泛化能力和预测稳定性尤其是在面对数据分布不同的学生时。) else: st.info(请先完成联邦学习训练以启用预测功能。)5. 部署、优化与踩坑实录将代码跑起来只是第一步要让这个系统真正可用还需要考虑部署和优化。在这个过程中我遇到了几个典型问题。5.1 模型收敛不稳定与调参心得在最初的版本中联邦模型的损失曲线波动很大准确率提升缓慢。经过排查问题出在几个方面学习率过大联邦学习中客户端本地更新多次后才聚合相当于放大了学习率。我将服务器下发的学习率调低为传统集中式训练的1/5到1/10后收敛稳定了许多。客户端本地训练轮数local_epochs是一个关键超参数。设置太小如1模型学不到东西设置太大如10每个客户端会朝着自己本地数据的最优点“跑偏”导致客户端漂移聚合困难。通常设置在2-5之间是个不错的起点。数据标准化各客户端数据特征尺度不同如GPA是0-4图书馆次数是0-50会严重影响模型训练。必须在每个客户端本地分别进行特征标准化如Z-Score但切记标准化参数均值、方差不能共享以模拟真实的隐私保护场景。实操技巧在prepare_dataloaders函数中我为每个客户端单独计算了训练集的均值和标准差并用它们来标准化该客户端的训练集和测试集。这比使用全局统计量更符合联邦学习的隐私前提。5.2 Streamlit应用性能与状态管理陷阱Streamlit的编程模型是“脚本从头到尾执行”每次交互都会重新运行整个脚本。这带来了两个挑战训练状态丢失点击按钮开始训练后下一轮循环时session_state里的模型和状态可能因为脚本重跑而重置。我的解决方案是将所有关键对象server,client_loaders,training_history都存入st.session_state并在脚本开头检查它们是否存在。长时间训练阻塞联邦训练20轮可能需要几十秒这会阻塞Streamlit界面导致“运行中”的旋转图标一直转用户体验差。对于真正的演示可以考虑使用st.spinner和st.progress如上文代码所示提供视觉反馈。异步或分步执行将每一轮训练包装成一个生成器函数配合st.empty()占位符和time.sleep()进行分步更新模拟实时效果。对于极耗时的任务应使用st.cache_data缓存结果或考虑用后台线程执行。5.3 从模拟到现实的鸿沟本项目使用的是精心设计的模拟数据。但在真实场景中挑战更大客户端掉线与异构性真实院系服务器可能离线或响应慢。需要实现健壮的通信机制、超时处理和客户端选择策略。安全与隐私增强基础的FedAvg仅提供“算法层面”的隐私。生产系统需要考虑加入差分隐私在客户端更新上添加噪声或安全聚合使用密码学协议确保服务器无法看到单个客户端的更新来提供更强的隐私保障。模型个性化训练出的全局模型可能对所有客户端“一视同仁”但某些院系可能希望模型更贴合自己的数据分布。可以在联邦学习框架上集成个性化技术让每个客户端在全局模型的基础上进行微调。这个项目从构思到实现让我对联邦学习的理论、工程实现和可视化演示有了更深的体会。它不仅仅是一个算法更是一套解决“数据隐私与价值利用”矛盾的工程框架。用Streamlit将其可视化则大大降低了理解门槛使得这项前沿技术能够被更多非技术背景的决策者所理解和接受。代码虽已开源但其中关于数据异构性的模拟、联邦平均的实现细节以及Streamlit状态管理的技巧都是经过多次调试才稳定下来的希望能为你的类似项目提供一个坚实的起点。本文还有配套的精品资源点击获取
