简介这是一套面向计算机、人工智能及相关专业学生与开发者的联邦学习实战项目聚焦高校学生成绩预测场景通过多客户端协同训练在保护数据隐私的前提下完成建模并借助Streamlit搭建可视化交互平台。资源包共55个文件以18个py源码与28个pyc编译文件为核心辅以7个csv数据集、1张混淆矩阵png与1份md说明文档压缩包约2.25MB涵盖FedProx、SCAFFOLD、Ditto、APFL、FedRep等多种联邦算法实现及模型、采样、训练工具模块目录结构清晰便于按算法分支对照学习。目前已有218人学习下载。读者可据此获得一套完整可运行的毕设级方案既能直接复现成绩预测流程与可视化界面也能在源码基础上修改算法或迁移至其他联邦学习任务适合课程设计、毕业设计及项目初期立项参考。1. 联邦学习做成绩预测为什么本地能跑、数据不出校也能训高校教务系统里躺着大量成绩、考勤、借阅、消费数据但真正想拿来做挂科预警时往往卡在第一步数据不能出二级学院甚至不能出机房。教务处一句数据安全谁负责项目就黄了。联邦学习Federated Learning解决的正是这个死结——模型参数在各参与方本地更新只把梯度或权重上传聚合原始记录一条都不动。杨强那本《联邦学习》里讲的横向联邦放到高校场景就是每个学院是一个 client各自用本院学生数据训本地模型中心服务器做 FedAvg 聚合得到一个全局成绩预测模型。这套方案适合谁一是高校信息化中心的工程师手里有 Python 基础、想给教务做个预警看板二是做教育数据挖掘的研究生需要一套能复现的联邦学习代码跑实验三是想学 Streamlit 快速搭可视化平台的后端同学。标题里的三个关键词——联邦学习、Python 源代码、Streamlit 可视化平台——对应三件事算法框架、可运行代码、能演示的界面。下面按数据怎么切、模型怎么训、平台怎么搭、坑在哪的顺序讲透代码可以直接抄。2. 联邦成绩预测的数据切分与特征工程把一张成绩表拆成 N 个 client2.1 为什么不能直接把成绩表丢进模型真实教务数据长这样学号、课程号、平时分、期中、期末、总评、是否重修。直接拿来训会踩三个雷。第一挂科样本极少一个班 40 人可能就 3 个不及格正负样本 1:13模型全预测通过也有 92% 准确率这种模型上线就是摆设。第二特征里有大量泄漏项——总评本身就是期末算出来的拿它预测挂科等于作弊必须剔除。第三不同学院课程难度差异巨大计算机学院的高数和文学院的高数不是一回事全局归一化会把分布拉平反而降低本地精度。联邦场景下还要多考虑一层数据怎么分片。横向联邦要求各 client 特征空间一致、样本不同所以每个学院保留相同的特征列只是学生记录不同。如果某个学院只有选修课成绩、另一个只有必修课那就得走纵向联邦复杂度上一个台阶本文不展开。2.2 特征工程的可复现脚本我一般会先做一版集中式基线确认特征有效再拆成联邦。下面这段代码把原始成绩表处理成建模可用的特征矩阵同时生成 client 分片。import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split # 读取原始成绩表假设字段stu_id, college, course_type, usual_score, mid_score, final_score, total_score df pd.read_csv(raw_scores.csv) # 1. 构造标签总评低于60为挂科风险 df[risk] (df[total_score] 60).astype(int) # 2. 剔除泄漏特征total_score 不能进模型 feature_cols [usual_score, mid_score, final_score, course_type_enc] # 3. 课程类型独热编码避免把类别当数值 df pd.get_dummies(df, columns[course_type], prefixcourse_type) feature_cols [c for c in df.columns if c.startswith((usual, mid, final, course_type_))] # 4. 按学院切分 client模拟联邦的 N 个参与方 clients {} for college, group in df.groupby(college): X group[feature_cols].values y group[risk].values # 每个 client 内部再分训练/测试测试集不出本地 X_tr, X_te, y_tr, y_te train_test_split(X, y, test_size0.2, stratifyy, random_state42) scaler StandardScaler().fit(X_tr) # 归一化参数只在本地拟合 clients[college] { X_train: scaler.transform(X_tr), y_train: y_tr, X_test: scaler.transform(X_te), y_test: y_te, scaler: scaler } print(f{college}: 训练 {len(y_tr)} 条, 挂科率 {y_tr.mean():.2%})逻辑说明标签用总评是否低于 60 构造这是教务最关心的口径。total_score必须从特征里删掉否则模型学到的只是总评低所以挂科这种废话。按college分组就是联邦的 client 划分每个学院的StandardScaler独立拟合这一点很关键——如果全局拟合再分发等于把全体数据的均值方差泄露给了中心联邦的隐私前提就破了。参数说明test_size0.2是本地留出集用于评估本地模型不上传。stratifyy保证切分后挂科比例一致小样本场景必加。random_state固定后结果可复现写论文或做对比实验时别省。2.3 类别不平衡的处理边界挂科样本少是常态常见做法是 SMOTE 过采样或给损失函数加权重。联邦场景下我倾向后者因为 SMOTE 在本地生成合成样本各 client 分布差异会被放大聚合时反而引入噪声。用pos_weight给正类加权更稳import torch import torch.nn as nn # 假设本地统计出正负样本比约 1:12 pos_weight torch.tensor([12.0]) criterion nn.BCEWithLogitsLoss(pos_weightpos_weight)pos_weight的含义是正类损失放大倍数设成负正样本比即可。注意这个值每个 client 可能不同聚合时只聚合模型权重不聚合损失配置各 client 保留自己的pos_weight更合理。3. FedAvg 聚合的 Python 实现从本地训练到全局模型下发3.1 联邦平均的核心逻辑与选型理由FedAvg 的思路朴素到有点反直觉让每个 client 用本地数据跑几轮 SGD把得到的权重按样本量加权平均就是新一轮全局模型。公式是 $w_{t1} \sum_{k} \frac{n_k}{n} w_t^k$$n_k$ 是第 k 个 client 的样本数。选它而不是 FedProx 或 FedAdam是因为高校场景 client 数量少通常十几个学院、网络稳定、数据量差异不算极端FedAvg 足够且实现简单出问题好排查。如果某个学院数据量特别小导致权重被淹没再考虑 FedProx 加近端项。3.2 本地训练与聚合的完整代码下面用一个两层 MLP 做成绩预测PyTorch 实现包含本地训练函数和聚合函数。import torch import torch.nn as nn import copy class ScoreNet(nn.Module): def __init__(self, input_dim): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, 32), nn.ReLU(), nn.Dropout(0.3), nn.Linear(32, 16), nn.ReLU(), nn.Linear(16, 1) # 二分类输出 logit ) def forward(self, x): return self.net(x).squeeze(-1) def local_train(model, X, y, epochs5, lr0.01, pos_weight12.0): 单个 client 的本地训练返回更新后的权重 model copy.deepcopy(model) optimizer torch.optim.SGD(model.parameters(), lrlr, momentum0.9) criterion nn.BCEWithLogitsLoss(pos_weighttorch.tensor([pos_weight])) X_t torch.tensor(X, dtypetorch.float32) y_t torch.tensor(y, dtypetorch.float32) model.train() for _ in range(epochs): optimizer.zero_grad() loss criterion(model(X_t), y_t) loss.backward() optimizer.step() return model.state_dict(), len(y) def fed_avg(global_model, client_results): 按样本量加权平均各 client 权重 total sum(n for _, n in client_results) new_state copy.deepcopy(global_model.state_dict()) for key in new_state.keys(): new_state[key] sum(state[key] * (n / total) for state, n in client_results) global_model.load_state_dict(new_state) return global_model逻辑说明local_train里先深拷贝全局模型保证本地更新不污染全局对象。epochs5是本地轮数太小欠拟合、太大过拟合且通信开销高5 到 10 是常见区间。fed_avg按n/total加权样本多的学院话语权大这是 FedAvg 的设计意图但也意味着小学院容易被带偏后面避坑章节会讲怎么处理。参数说明lr0.01配 SGD momentum 0.9联邦场景学习率不宜大因为聚合本身有平滑效果。Dropout(0.3)在小样本上防过拟合如果某学院样本过少可以调高。input_dim由特征列数决定各 client 必须一致否则聚合时形状对不上直接报错。3.3 主训练循环与全局评估把上面的零件串起来跑 20 轮联邦通信def run_federated(clients, input_dim, rounds20): global_model ScoreNet(input_dim) for r in range(rounds): results [] for name, data in clients.items(): state, n local_train(global_model, data[X_train], data[y_train]) results.append((state, n)) global_model fed_avg(global_model, results) # 全局评估各 client 本地测试集上算准确率 accs [] global_model.eval() for name, data in clients.items(): with torch.no_grad(): logits global_model(torch.tensor(data[X_test], dtypetorch.float32)) pred (torch.sigmoid(logits) 0.5).float() acc (pred torch.tensor(data[y_test], dtypetorch.float32)).float().mean() accs.append(acc.item()) print(fRound {r1}: 平均准确率 {np.mean(accs):.4f}) return global_model这里每轮都做全局评估只是为了观察收敛真实部署时测试集也不该集中评估指标应由各 client 上报后汇总。rounds20是通信轮数联邦学习的主要成本在通信轮数越多越贵通常 20 到 50 轮能看到收敛。4. Streamlit 可视化平台把联邦训练结果做成能演示的看板4.1 为什么选 Streamlit 而不是 Gradio热词里 gradio/streamlit 经常一起出现两个都能快速搭界面。成绩预测这个场景我选 Streamlit原因是它更适合多页面、多图表的数据看板侧边栏放 client 选择、主区域放训练曲线和混淆矩阵、底部放学生风险名单这种布局 Gradio 做起来别扭。Gradio 强在单模型输入输出 demo适合快速验证模型不适合做教务看板。如果只是给导师演示模型能跑Gradio 十分钟搞定要交给教务处用Streamlit 更合适。4.2 平台骨架与联邦训练集成下面是一个可运行的 Streamlit 应用骨架包含数据上传、联邦训练触发、结果可视化三块。import streamlit as st import pandas as pd import plotly.express as px import torch st.set_page_config(page_title联邦成绩预测平台, layoutwide) st.sidebar.title(联邦配置) # 1. 侧边栏参数 rounds st.sidebar.slider(通信轮数, 5, 50, 20) local_epochs st.sidebar.slider(本地训练轮数, 1, 20, 5) lr st.sidebar.select_slider(学习率, options[0.001, 0.005, 0.01, 0.05], value0.01) # 2. 数据上传实际部署时各 client 本地读取不上传原始数据 uploaded st.file_uploader(上传成绩表演示用, type[csv]) if uploaded: df pd.read_csv(uploaded) st.write(f共 {len(df)} 条记录{df[college].nunique()} 个学院) # 展示各学院挂科率分布 risk_by_college df.groupby(college).apply(lambda g: (g[total_score] 60).mean()).reset_index() risk_by_college.columns [学院, 挂科率] fig px.bar(risk_by_college, x学院, y挂科率, title各学院挂科率分布) st.plotly_chart(fig, use_container_widthTrue) # 3. 训练按钮与结果展示 if st.button(开始联邦训练): progress st.progress(0) acc_history [] # 这里调用第 3 章的 run_federated实际项目里把 clients 从本地加载 for r in range(rounds): # 伪代码每轮聚合后记录准确率 acc_history.append(0.7 0.2 * (r / rounds)) # 替换为真实评估值 progress.progress((r 1) / rounds) st.success(训练完成) st.line_chart(pd.DataFrame({轮数: range(1, rounds1), 准确率: acc_history}).set_index(轮数))逻辑说明st.sidebar放超参数让演示时能现场调参看效果这是 Streamlit 的杀手锏。file_uploader在真实联邦部署里应该去掉因为原始数据不出本地这里保留只是为了演示方便。st.progress给训练过程反馈联邦训练动辄几分钟没有进度条用户会以为卡死。st.line_chart画收敛曲线比 matplotlib 嵌进去省事。参数说明layoutwide让图表占满宽度看板场景必开。select_slider限定学习率候选值避免用户输入非法值导致训练发散。use_container_widthTrue让 Plotly 图表自适应不然窄屏会溢出。4.3 风险名单与模型解释教务最关心的不是准确率是哪些学生要预警。加一个表格展示高风险学生并给出主要影响特征if st.checkbox(显示高风险学生名单): # 用全局模型对全体学生打分 model.eval() X_all torch.tensor(df[feature_cols].values, dtypetorch.float32) with torch.no_grad(): risk_prob torch.sigmoid(model(X_all)).numpy() df[风险概率] risk_prob high_risk df[df[风险概率] 0.7][[stu_id, college, 风险概率]].sort_values(风险概率, ascendingFalse) st.dataframe(high_risk.head(50)) st.download_button(导出名单, high_risk.to_csv(indexFalse), risk_list.csv)阈值 0.7 是经验值宁可漏报不可误报太多否则辅导员被大量假阳性淹没就不用了。download_button让教务能直接导出 CSV 发给辅导员这个功能比任何花哨图表都实用。5. 联邦成绩预测的避坑清单这 5 个坑我踩过5.1 坑一各 client 特征列不一致导致聚合报错现象训练到聚合那一步抛RuntimeError: size mismatch for net.0.weight。原因某个学院的数据里缺少某门课的成绩列get_dummies后特征维度和其他学院不同。解决在切分 client 之前先确定全局特征列用reindex(columnsglobal_cols, fill_value0)对齐缺失的课成绩填 0 并加一个是否缺失标志列而不是直接删。5.2 坑二小样本 client 被大 client 带偏现象训练几轮后小学院的本地准确率反而下降全局模型在它上面表现很差。原因FedAvg 按样本量加权小学院权重低梯度被大学院主导。解决改用样本量开方加权或对小学院提高本地学习率也可以设权重下限。我一般用 $\sqrt{n_k}$ 加权实测比纯比例稳。5.3 坑三灾难性遗忘让全局模型越训越差现象前几轮准确率上升第 10 轮后突然掉下去。原因联邦学习里的灾难性遗忘——各 client 数据分布差异大模型在轮换拟合不同分布时把之前学的忘了。解决降低本地 epochs增加通信频率或者在损失里加一个近端项让本地模型不要偏离全局太远这就是 FedProx 的思路def local_train_proximal(model, global_model, X, y, mu0.01, epochs5): model copy.deepcopy(model) optimizer torch.optim.SGD(model.parameters(), lr0.01, momentum0.9) criterion nn.BCEWithLogitsLoss(pos_weighttorch.tensor([12.0])) X_t torch.tensor(X, dtypetorch.float32) y_t torch.tensor(y, dtypetorch.float32) global_params [p.detach().clone() for p in global_model.parameters()] for _ in range(epochs): optimizer.zero_grad() loss criterion(model(X_t), y_t) # 近端项惩罚本地参数偏离全局 prox sum(((p - gp) ** 2).sum() for p, gp in zip(model.parameters(), global_params)) (loss mu / 2 * prox).backward() optimizer.step() return model.state_dict(), len(y)mu控制近端强度0.01 到 0.1 之间调太大模型学不动太小等于没加。5.4 坑四Streamlit 每次交互重跑全脚本现象点一下滑块整个联邦训练重新跑一遍等半天。原因Streamlit 的执行模型是自上而下重跑没有状态保持。解决用st.session_state缓存训练结果把训练函数用st.cache_data装饰或者把训练放到按钮回调里只触发一次。滑块调参时只更新展示不重训。5.5 坑五评估指标虚高上线就翻车现象平台显示准确率 95%教务用了一学期发现预警名单一半是错的。原因测试集和训练集同分布且挂科样本太少准确率被多数类主导。解决看 AUC 和召回率不看准确率用时间切分而不是随机切分拿上学期数据训、下学期数据测才接近真实场景。这个坑最隐蔽血泪经验。6. 让联邦成绩预测真正可用的两个进阶技巧第一个技巧是差分隐私加噪。联邦学习只保证原始数据不出本地但梯度仍可能被反推。给上传的权重加高斯噪声sigma取 0.01 到 0.1隐私预算epsilon越小越安全但精度掉得越多。高校场景下我一般加很小的噪声主要是应付合规审查实际隐私风险在教务内网里没那么高。第二个技巧是模型版本管理。联邦训练每轮产生一个全局模型直接覆盖会导致无法回滚。我习惯每轮存一个 checkpoint命名带轮数和时间戳Streamlit 侧边栏加一个模型版本选择器教务可以对比不同轮次的效果。这个习惯救过我一次——某轮聚合后模型突然变差回滚到上一轮就恢复了没有后悔药可吃。验证方法上别只看平台上的曲线。拿一个学院的数据完全留出不参与训练最后用它测全局模型这个留一学院评估比随机切分可信得多。如果留出学院的 AUC 和参与训练的学院差距在 5 个点以内这套联邦方案才算站得住。我自己做这类项目的习惯是先跑通集中式基线确认特征和标签没问题再拆联邦联邦跑通后先不加差分隐私确认精度达标再逐层加约束。顺序反了出了问题根本不知道是哪一层导致的。希望帮到你。本文还有配套的精品资源点击获取