简介针对高校学生成绩预测任务提供基于联邦学习的Python完整实现并借助Streamlit搭建可视化平台。项目包含多类联邦学习算法源码与本地训练脚本涵盖FedRep、Ditto、L2GD、APFL、FedProx、SCAFFOLD等主流方法可对比不同策略下的预测效果附带的CSV数据集与混淆矩阵图像便于验证与展示。资源共55个文件以Python脚本.py及字节码.pyc为主另有7个CSV数据文件、1份说明文档和1张可视化图表整体压缩包约2.25MB结构清晰适合二次开发。该方案为作者毕业设计答辩评审达95分代码经调试可运行对计算机、人工智能等相关专业学生完成课程设计或毕业设计具有较高参考价值。目前已有670人学习下载可作为入门联邦学习与成绩预测实战的优质范例。1. 基于联邦学习进行高校学生成绩预测这份源码和数据集到底解决了什么问题如果一个教务处的老师拿着全校几千条成绩记录来找你想让你做个成绩预警模型但碍于学生隐私和院系数据归属这些数据不能汇总到同一个地方——你手头的答案往往就是“联邦学习”。它让每个院系的数据留在院系本地只把模型参数或梯度上传到一个中心节点做聚合用这种方式训练出全校统一的成绩预测模型。这个标题里打包的正是这样一套完整可跑通的方案python源码负责联邦训练逻辑数据集负责喂给模型Streamlit则把训练过程和预测结果做成一个点开就能看的可视化平台。这套东西解决的核心问题是“数据不出本地但模型却能共享”。适合谁两类人一类是做教育数据挖掘的学生和研究者需要一个能复现联邦学习流程的基线项目另一类是学校信息中心或教务部门的开发想快速评估“联邦学习用于学业预警”到底可不可行。接下来我会照着源码的组织方式从原理、数据划分、训练闭环、可视化搭建一直讲到踩坑点争取让你拿到源码后一个下午能跑通一周内能改成自己的方案。2. 联邦学习成绩预测的原理与选型为什么是“数据不动模型动”2.1 学习成绩预测到底在预测什么拿到一份高校成绩数据集首先要想清楚任务定义。常见做法是把它当回归问题处理目标变量是学生的期末成绩、GPA或者某个关键课程的成绩当你要做“是否预警”时才转成二分类——预测学生会不会挂科或者会不会进入学业警告名单。不要一上来就用分类模型因为成绩本身就是连续值回归能保留更多信息预警阈值后续想怎么划都行。特征方面常见的数据集字段包括出勤率、作业平均分、期中成绩、前序课程成绩、课堂参与度以及一些背景属性。这里最容易出问题的是特征泄漏——比如把期末考试成绩本身或跟期末高度同源的期中考试后段信息放进特征里。经验是只使用“预测时间点之前”一定能拿到的数据比如你要在学期中段做预警那就只能用开学到中段的考勤、作业和一次测验成绩。2.2 FedAvg是怎么工作的联邦学习在这个项目里通常用的是FedAvg联邦平均算法它足够简单也足够说明问题。流程是服务端初始化一个全局模型参数把它分发给参与训练的客户端比如每个院系一个客户端每个客户端用自己的本地数据在本地跑若干轮SGD随机梯度下降然后客户端把更新后的模型权重返回给服务端服务端把这些权重按样本量比例求加权平均得到新的全局模型接着进入下一轮通信。系统学习的过程就是用这种“数据不动模型动”的方式完成的。这里的难点是Non-IID数据分布不同院系的学生基础不同成绩分布差异很大有些客户端数据量还特别少。这种情况下本地模型容易“跑偏”聚合后全局模型可能震荡甚至不收敛。遇到这种情况后面会在避坑章节单独讲怎么处理。2.3 为什么在高校成绩场景里选联邦学习而不是集中训练从技术选型上说集中式训练当然是最省事的——把所有数据拉到一台机器上直接用LightGBM或深度学习模型训。但高校数据的归属权和管理边界很现实A学院的数据不能拷给中心B学院的成绩字段格式跟A还不一样。联邦学习把“数据汇总”换成了“模型汇总”正好避开这个行政和隐私上的坎。实现上这份源码有两种常见的组织方式一种是完全手写FedAvg循环numpy或者PyTorch都行适合理解原理另一种是调用开源框架比如Flower做客户端管理和通信调度。看源码时先确认它属于哪一种。手写版的优点是依赖少、逻辑一目了然改起来快框架版的优点是自带客户端并行和断线重连但要装额外依赖。对跑通流程来说我建议先跑手写版把训练日志看明白后再决定要不要迁移到框架。3. 跑通源码从数据集划分到最小联邦训练闭环3.1 数据集字段与预处理脚本这份源码带的数据集常见做法是以CSV格式存储按院系或班级字段区分客户端归属。典型的字段设置包含学生ID、院系代码、出勤率、作业平均分、期中成绩、前序课程平均绩点以及目标字段最终成绩。数据规模一般在一千到几千条之间够演示用。拿到数据后第一步是加载并做预处理。import pandas as pd from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler df pd.read_csv(student_scores.csv) print(df.head()) print(df[department].value_counts()) # 确认客户端划分依据 # 特征列与目标列分离 feature_cols [attendance_rate, homework_avg, midterm_score, previous_gpa] target_col final_score X df[feature_cols].copy() y df[target_col].copy() # 缺失值用中位数填充成绩数据常有零星空值 X X.fillna(X.median()) # 按院系将数据拆成多客户端再各自划分训练/测试集 clients_data {} for dept, group in df.groupby(department): X_dept group[feature_cols] y_dept group[target_col] X_train, X_test, y_train, y_test train_test_split( X_dept, y_dept, test_size0.2, random_state42 ) clients_data[dept] { X_train: X_train, y_train: y_train, X_test: X_test, y_test: y_test } # 统一做标准化防止出勤率0-100和前序绩点0-4量纲差异过大 scaler StandardScaler() for dept in clients_data: scaler.fit(clients_data[dept][X_train]) clients_data[dept][X_train_scaled] scaler.transform(clients_data[dept][X_train]) clients_data[dept][X_test_scaled] scaler.transform(clients_data[dept][X_test])这里有两个关键参数要注意test_size0.2意味着每个院系都留出20%的数据做本地测试用来评估“单院系模型”和“联邦模型”的差别random_state42保证每次运行数据划分一致否则后面比较实验就没法复现。标准化用的是训练集统计量再去transform测试集这是避免数据泄漏的标准做法。排序上groupby(department)这一步直接决定了后续有几个联邦客户端——院系越多通信开销越大但全局模型的泛化性通常也越好。3.2 把一份全校数据切成Non-IID客户端样本如果源码提供的数据集里没有现成的院系字段就需要自己模拟Non-IID分布。这是联邦学习项目里绕不开的一步因为真实场景中院系之间的成绩分布差异很大而IID独立同分布数据会让联邦学习退化得跟集中式训练几乎一样演示价值就没了。import numpy as np # 按成绩水平把学生排序再分段分配给不同客户端模拟“强院”和“弱院” df_sorted df.sort_values(final_score).reset_index(dropTrue) num_clients 4 shard_size len(df_sorted) // num_clients client_data_map {} for i in range(num_clients): shard df_sorted.iloc[i * shard_size : (i 1) * shard_size] # 每个客户端拿到的是成绩分布不同的一段形成天然的数据偏移 client_data_map[fclient_{i}] shard # 打印每个客户端的目标变量均值确认Non-IID效果 for name, data in client_data_map.items(): print(name, data[final_score].mean())注意这里跟随机切分的区别随机切分每个客户端拿到的都是全量分布的一个抽样模型学到的几乎一样按成绩排序分段切分后客户端A可能全是高分段学生客户端B可能全是低分段学生这时联邦聚合才有意义——它要把这些“各有偏见”的本地模型融合成一个全局模型。打印出的均值就是最直观的确认方式如果均值差异在10分以上说明Non-IID切分有效果。3.3 最小的FedAvg核心实现下面这段代码是联邦训练的核心骨架。它不依赖任何联邦学习框架只用PyTorch的神经网络模块把“本地训练”和“服务端聚合”两个动作拆成独立函数。import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset class ScorePredictor(nn.Module): def __init__(self, input_dim4): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, 32), nn.ReLU(), nn.Linear(32, 16), nn.ReLU(), nn.Linear(16, 1) ) def forward(self, x): return self.net(x).squeeze(-1) def local_train(model, X, y, lr0.01, epochs3): 在单个客户端本地做几轮SGD返回更新后的模型参数 model.train() dataset TensorDataset(torch.tensor(X, dtypetorch.float32), torch.tensor(y.values, dtypetorch.float32)) loader DataLoader(dataset, batch_size16, shuffleTrue) optimizer torch.optim.Adam(model.parameters(), lrlr) loss_fn nn.MSELoss() for _ in range(epochs): for batch_x, batch_y in loader: optimizer.zero_grad() pred model(batch_x) loss loss_fn(pred, batch_y) loss.backward() optimizer.step() return {k: v.clone() for k, v in model.state_dict().items()} def fed_avg(global_state, client_states, client_sizes): 按样本量加权平均各客户端上传的参数 total_size sum(client_sizes) new_state {} for key in global_state: weighted sum(client_states[i][key] * (client_sizes[i] / total_size) for i in range(len(client_states))) new_state[key] weighted return new_state逻辑拆开看local_train在客户端本地拿到模型后只用自己的数据训练几个epochepochs3是保守值数据量小的时候3到5轮就够太多会让本地模型偏离全局方向太少则学不到东西训练完只返回state_dict也就是权重参数而不返回原始数据。fed_avg是FedAvg题眼所在——聚合时样本量大的客户端权重更高这样全局模型不会被某个数据量小的院系带偏。batch_size16对几百条数据是稳妥选择如果客户端数据量少可以降到8。3.4 训练主循环与通信轮次有了上面的两个函数整个联邦训练主循环就很简单初始化全局模型反复执行“下发参数—本地训练—上传参数—聚合更新”。def run_federated_training(clients_data, comm_rounds15, local_epochs3, lr0.01): global_model ScorePredictor() global_state global_model.state_dict() history [] for round_idx in range(comm_rounds): client_states [] client_sizes [] for dept, data in clients_data.items(): # 每个客户端都从当前的全局参数开始训练 model ScorePredictor() model.load_state_dict({k: v.clone() for k, v in global_state.items()}) state local_train( model, data[X_train_scaled], data[y_train], lrlr, epochslocal_epochs ) client_states.append(state) client_sizes.append(len(data[X_train_scaled])) # 聚合所有客户端参数更新全局模型 global_state fed_avg(global_state, client_states, client_sizes) # 用全局模型在全校测试集汇总各客户端测试集上评估RMSE global_model.load_state_dict(global_state) global_model.eval() all_X_test np.vstack([clients_data[d][X_test_scaled] for d in clients_data]) all_y_test np.hstack([clients_data[d][y_test].values for d in clients_data]) with torch.no_grad(): preds global_model(torch.tensor(all_X_test, dtypetorch.float32)).numpy() rmse np.sqrt(np.mean((preds - all_y_test) ** 2)) history.append((round_idx 1, rmse)) print(fround {round_idx 1}, RMSE: {rmse:.4f}) return global_model, history这里有两个参数值得反复调试comm_rounds通信轮数和local_epochs本地训练轮数。通信轮数太少全局模型还没收敛太多后几轮RMSE几乎不动纯浪费算力。经验判断方法跑完后看打印的RMSE序列如果最后几轮下降幅度小于0.01就可以提前停了。lr0.01对Adam来说是常用起点但如果客户端数据量差异极大建议调低到0.005防止某个客户端剧烈更新把全局模型带跑。4. 用Streamlit把训练过程和预测结果变成可视化平台4.1 为什么选Streamlit而不是Gradio联邦学习跑出来的结果如果只停留在终端打印里说服力很弱。标题里特意挂了Streamlit说明作者想让训练过程和预测结果可视化。选Streamlit而不选Gradio是因为这个场景是“数据应用”而不是“模型Demo”需要展示训练曲线、数据分布、特征相关性还要提供一个完整的输入表单做单条预测。Streamlit对pandas和plotly的集成更自然而且页面布局用侧边栏加主区域的模式非常适合展示训练控制项。Gradio更偏“丢一张图进去出结果”的交互做联邦学习监控类页面会别扭一些。4.2 页面骨架侧边栏控参数主区域看曲线把训练封装成带缓存的函数页面加载时跑一次后续所有交互都不再重新训练。这一步很关键否则每点一次按钮模型就重新训一遍页面直接卡死。import streamlit as st import pandas as pd import plotly.graph_objects as go st.set_page_config(page_title联邦学习成绩预测平台, layoutwide) st.title(基于联邦学习的高校学生成绩预测平台) # 侧边栏联邦训练参数 with st.sidebar: st.header(训练参数) comm_rounds st.slider(通信轮数, 5, 50, 15) local_epochs st.slider(本地训练轮数, 1, 10, 3) lr st.number_input(学习率, 0.001, 0.1, 0.01, step0.001) run_button st.button(开始重新训练) # 训练结果只做一次用缓存避免页面刷新重跑 st.cache_resource def train_and_get_history(clients_data, comm_rounds, local_epochs, lr): model, history run_federated_training( clients_data, comm_rounds, local_epochs, lr ) return model, history if run_button: st.cache_resource.clear() st.rerun() model, history train_and_get_history( clients_data, comm_rounds, local_epochs, lr ) # 主区域训练曲线 st.subheader(全局模型训练收敛曲线RMSE) df_history pd.DataFrame(history, columns[round, rmse]) fig go.Figure() fig.add_trace(go.Scatter(xdf_history[round], ydf_history[rmse], nameRMSE)) fig.update_layout(xaxis_title通信轮次, yaxis_titleRMSE, height350) st.plotly_chart(fig, use_container_widthTrue)这里的st.cache_resource是防翻车的核心它让训练只执行一次之后页面做任何交互都不会重复计算。侧边栏的三个控件对应前面代码里的comm_rounds、local_epochs、lr——改参数后点按钮会清缓存并重新训练符合直觉。把训练封装在这个函数里便于后续替换成Flower框架版本。4.3 单条预测输入学生特征输出预测成绩训练完的模型放在页面下方用一个表单接收新学生的中期数据实时算出预测期末成绩。用st.form的另一个好处是按回车或点提交按钮才会触发预测不会因为页面其他部件刷新而反复提交。st.subheader(输入学生中期数据预测期末成绩) with st.form(prediction_form): col1, col2, col3, col4 st.columns(4) with col1: attendance_rate st.number_input(出勤率0-100, 0.0, 100.0, 85.0) with col2: homework_avg st.number_input(作业平均分0-100, 0.0, 100.0, 80.0) with col3: midterm_score st.number_input(期中成绩0-100, 0.0, 100.0, 75.0) with col4: previous_gpa st.number_input(前序课程绩点0-4, 0.0, 4.0, 3.0) submitted st.form_submit_button(预测) if submitted: # 用训练好的全局模型做推理注意与训练时一致的特征顺序 import numpy as np sample np.array([[attendance_rate, homework_avg, midterm_score, previous_gpa]]) sample_scaled scaler.transform(sample) model.eval() with torch.no_grad(): pred model(torch.tensor(sample_scaled, dtypetorch.float32)).item() st.success(f预测期末成绩{pred:.1f} 分)这段代码里最容易出问题的是scaler.transform(sample)——如果训练时用了StandardScaler预测时就必须用同一个已经拟合好的scaler对象转换输入否则输入量纲和模型期望不一致预测结果会差得离谱。st.success只是展示结果的组件换成st.metric更直观但要注意它不支持带单位格式化实际项目里按需取用。4.4 数据分布与特征相关性视图除了训练曲线可视化平台还应该放两张图各院系成绩分布的箱线图以及特征相关性热力图。前者直观展示Non-IID效果后者帮助排查特征是否冗余。import plotly.express as px st.subheader(各院系成绩分布查看Non-IID效果) fig_box px.box( df, xdepartment, yfinal_score, colordepartment, title不同院系的期末成绩分布 ) st.plotly_chart(fig_box, use_container_widthTrue) st.subheader(特征相关性分析) corr df[feature_cols [target_col]].corr() fig_corr go.Figure(go.Heatmap( zcorr.values, xcorr.columns, ycorr.columns, colorscaleRdBu_r, zmin-1, zmax1 )) st.plotly_chart(fig_corr, use_container_widthTrue)箱线图一眼就能看出各院系中位数和四分位距的差异——如果四个院系的箱体几乎重叠说明数据近似IID联邦学习的优势就体现不出来相关性热力图则要重点看midterm_score跟final_score的相关系数如果在0.8以上说明期中成绩已经是极强信号模型预测分数偏高不稀奇真实部署时要警惕这种“好得可疑”的表现。启动命令很简单在项目根目录执行streamlit run app.py --server.port 8501浏览器访问http://localhost:8501就能看到页面。如果端口被占用换成--server.port 8502。第一次启动会下载plotly和streamlit相关依赖这一步建议先建虚拟环境别直接装进系统Python里。5. 避坑清单联邦学习成绩预测的几个常见翻车现场5.1 训练Loss不降或聚合后模型直接“失忆”现象联邦训练跑了二十多轮RMSE几乎一条直线或者前几轮RMSE下降后面突然反弹到初始水平。原因一般有两种一是客户端本地学习率太大各自在本地拟合到不同的局部最优聚合时互相抵消二是Non-IID太严重每个客户端的数据分布差异过大最终全局模型等于一堆模型的平均“和稀泥”。解决先把lr从0.01降到0.005试一轮再把local_epochs从3提到5让每个客户端先充分学到本地模式再聚合。如果还是不行检查客户端之间的样本量差异是否超过10倍——差异太大时改用手动加权平均确认fed_avg里client_sizes传的是训练样本数而不是总样本数这个字段写错会直接让聚合权重失效。5.2 测试RMSE低到不合理结果是特征泄漏现象模型在测试集上RMSE只有5分看起来很漂亮但部署后发现实际预测完全不准。原因是特征里混入了“未来信息”——最常见的坑是把期末成绩前的一次测验或课程总评的平时分直接当成特征或者期中成绩包含了期末考试的范围。还有一个隐蔽情况做数据预处理时用全校数据的均值和标准差去标准化训练集而不是用每个院系自己的训练集统计量这也算一种泄漏。解决回到train_test_split之前把特征列严格限定在预测时间节点前可获得的字段标准化时对每个院系独立fit训练集再transform其测试集。我在3.1的代码里已经按这个逻辑写了但很多改版源码会把scaler.fit放到所有数据上fit一次发现后要立刻改。5.3 Streamlit页面一交互就卡死现象首次加载页面要等很久点击“开始训练”后浏览器转圈多点多刷直接无响应。原因是训练逻辑写在了页面主体里每次页面重跑Streamlit的交互模型是自上而下重新执行脚本都会再次触发训练。尤其当comm_rounds设成50时一次训练几十秒页面就卡几十秒。解决把训练函数套上st.cache_resource让训练只执行一次需要重新训练时显式调用st.cache_resource.clear()再st.rerun()。另外把数据加载pd.read_csv也放进缓存函数里否则每次页面刷新都重新读盘。如果训练时间超过30秒还可以在侧边栏显示st.spinner或进度条让用户知道正在算。5.4 拿分类准确率评估回归任务现象有人把成绩预测做成了“及格/不及格”二分类然后用准确率、F1做评估发现联邦模型准确率95%觉得大功告成。问题在于这个数据集的及格线往往在60分附近而大部分学生的分数都远离分界线准确率天然就高模型学得好不好根本看不出来。解决回归任务统一用RMSE、MAE和R²评估如果确实要做预警分类也应该是先回归预测分数再对预测分数做阈值划分单独报告混淆矩阵、查全率和查准率。尤其查全率漏报比例在学业预警场景比准确率重要得多——你更不希望把真正会挂科的学生漏掉。5.5 客户端数量一多训练时长成倍上涨现象把院系从4个改成20个后训练时间接近线性增长每轮通信都要等最慢的客户端跑完。原因是源码默认串行执行for dept in clients_data客户端之间没有并行。另外每个客户端的DataLoader如果num_workers设置太高默认0也会拖慢整体速度。解决先用ThreadPoolExecutor把各客户端的local_train并行化这是改动最小、收益最明显的方案。需要注意的是本地训练用的是PyTorch多线程时要确认模型实例各自独立不能共享一个model对象。更彻底的方案是把客户端逻辑迁到Flower框架它自带并行调度和状态管理但代价是要改接口适合后续长期迭代。6. 进阶验证怎么证明联邦学习比“各自为政”更值得投入光把联邦模型训出来还不够汇报时别人一定会问“你搞这么复杂比每个院系自己训一个模型强在哪比把所有数据汇总到一起训练差多少”这个问题需要两套对照实验来回答。第一套对照集中式baseline。把全校数据合并用同样的模型结构和超参数训练一个模型忽略隐私约束它的RMSE是“理论上限”。第二套对照每个院系只用本地数据训练自己的模型在各自本地测试集上算RMSE它的均值是“各自为政的基线”。最后把联邦全局模型的RMSE放进去对比。# 集中式baseline全部数据合并训练 all_X_train np.vstack([clients_data[d][X_train_scaled] for d in clients_data]) all_y_train np.hstack([clients_data[d][y_train].values for d in clients_data]) central_model ScorePredictor() dataset TensorDataset(torch.tensor(all_X_train, dtypetorch.float32), torch.tensor(all_y_train, dtypetorch.float32)) loader DataLoader(dataset, batch_size32, shuffleTrue) optimizer torch.optim.Adam(central_model.parameters(), lr0.01) loss_fn nn.MSELoss() for _ in range(50): for batch_x, batch_y in loader: optimizer.zero_grad() loss loss_fn(central_model(batch_x), batch_y) loss.backward() optimizer.step() central_preds central_model(torch.tensor(all_X_test, dtypetorch.float32)).detach().numpy() central_rmse np.sqrt(np.mean((central_preds - all_y_test) ** 2)) print(f集中式训练 RMSE: {central_rmse:.4f})对照结果通常落在这样一个格局里集中式RMSE最低联邦全局模型略高于集中式差距在5%到10%之间是正常水平如果超过20%你就要回去调参数了而单个院系本地模型的RMSE明显更高尤其数据量小的院系可能高出一倍。这个对比能直接回答“值不值得做”联邦学习用少量精度损失换来了数据不出本地的合规性。第二套实验是验证通信轮数和客户端数量的影响我一般跑一个简单网格comm_rounds取[5, 15, 30]本地epochs取[1, 3, 5]把结果整理成对比表然后选拐点位置的参数作为最终配置。在Non-IID程度严重的场景里还可以尝试FedProx在本地损失函数里加一个近端项约束本地模型不要偏离全局模型太远或全局模型下发后做一次本地微调再做预测。对我来说每次拿到新的成绩数据集我都会先跑一遍对照流程确认联邦模型和集中式的差距在可接受范围再做正式评估——这个习惯帮我挡掉了好几次“自认为效果好实则模型压根没学到东西”的翻车。希望这一套从原理到踩坑的流程能帮你更快把这个项目跑起来也少走我当年走过的弯路。本文还有配套的精品资源点击获取