简介：基于联邦学习的高校学生成绩预测项目，面向计算机、人工智能及相关专业学生、教师与从业者，可服务于毕业设计、课程设计、课程大作业或联邦学习科研入门。项目代码经过调试，可稳定运行，并借助Streamlit搭建可视化平台，便于直观查看各联邦学习算法在成绩预测任务中的精度、损失变化与对比效果。压缩包共55个文件，主体为18个Python源码和28个编译缓存pyc，另有7个CSV数据集与实验记录、1份说明文档及1张混淆矩阵图，整体大小2.25MB；目录按模型、数据处理、训练工具等模块划分，结构清晰，便于按需修改和二次开发。资源中涉及FedProx、FedRep、Ditto、APFL等联邦学习变体，配套的样本数据可帮助理解不同数据分布对模型效果的影响。已有670人学习下载。对想快速复现联邦学习全流程、完成成绩预测实践项目或撰写相关毕设的读者，具有较高的学习借鉴与扩展价值。
1. 这是一个把数据留在本地又能一起训练的方案：高校成绩预测的联邦学习实践
做过高校数据项目的工程师都知道，最头疼的不是模型选型，而是数据根本凑不齐：教务处在A系统、各学院在B系统，甚至涉及学生隐私不能直接导出汇总。这个标题给的方案很直接——用联邦学习把成绩预测这件事拆成“各方本地训练、只传模型参数”，最后再用Streamlit把结果可视化。它适合两类人：一类是校内做数据平台的工程师，需要在不触碰原始数据的前提下跑通成绩预警模型；另一类是刚入门联邦学习的Python开发，想找一个软硬件门槛都不高的完整样例。接下来我按自己复现这个项目的思路，把它拆成数据、训练、可视化和避坑四段讲清楚。
2. 数据怎么拆才是联邦：按客户端切分成绩数据集，先还原真实数据孤岛
2.1 为什么不需要把数据集中到一份CSV里：横向联邦的基本假设
联邦学习里的“联邦”不是指把数据合并，而是指参与训练的每一方都各自持有一份数据，彼此不交换原始记录，只交换模型参数。高校场景里，最常见的划分方式是横向联邦：每个学院/年级/教务系统的数据特征相同（学号、课程、出勤、成绩等），但样本不同。这正好对应教务系统里“不同学院各自存储本院学生成绩”的现状。
那个标题压缩包里带的“数据集”，在我见过的类似开源项目里，通常会整理成一份完整成绩表的Excel或CSV，同时附上按学院或按批次切分好的多个子文件。你拿到手后不要急着直接把整份数据丢进模型，第一步要做的是模拟“每个客户端只看到自己那份数据”的状态。这样才能让后面的联邦训练有意义——如果你把所有数据都集中在一起训练，那和普通深度学习没有区别，联邦学习这个环节就变成了摆设。
2.2 用pandas按“客户端ID”切分数据：最小可复现的纵向拆法
我一般用下面这段Python脚本把原始成绩表拆成多个客户端数据分片，并留出一份服务端不参与训练的测试集。注意，这个拆法不是按行随机打乱，而是按客户端属性（比如学院）分组，这样才能还原真实场景。
import pandas as pd from sklearn.model_selection import train_test_split # 原始成绩表：包含 student_id, college, course, attendance, score 等列 df = pd.read_excel("student_scores.xlsx") # 按学院字段分组，模拟每个学院为一个独立客户端 clients = {} for college, group in df.groupby("college"): # 每个客户端内部再划分本地训练集和本地验证集（本地验证集可选） local_train, local_val = train_test_split( group, test_size=0.2, random_state=42, stratify=group["score_label"] ) clients[college] = { "train": local_train.reset_index(drop=True), "val": local_val.reset_index(drop=True), } # 单独留出一份全局测试集：从全体数据中抽取，模拟“服务端持有的公共评估数据” _, global_test = train_test_split( df, test_size=0.15, random_state=42, stratify=df["score_label"] ) global_test.to_csv("global_test.csv", index=False)这段脚本的关键逻辑是先按college字段分组，每个组就是一个客户端。train_test_split里用了stratify按score_label分层，目的是保证每个客户端本地训练集和验证集的成绩等级分布与原分组基本一致。random_state=42保证可复现。如果原始数据没有college字段，也可以用grade（年级）或major（专业）来分组，只要保证“客户端之间数据没有交集”即可。
这里强调一个容易忽略的点：服务端那份global_test.csv不是用来训练的，它只用来在每轮联邦训练后统一评估全局模型的精度。真实联邦场景里服务端可能没有这些标签，但做实验验证联邦算法效果时，保留一个全局测试集是标准做法，否则你无法判断聚合出的模型到底有没有变好。
2.3 特征工程：把成绩预测变成二分类或多分类，比回归更贴近实际用途
我见过的成绩预测项目普遍分成两类：预测具体分数（回归）和预测“是否挂科/是否优秀”（分类）。标题里没写具体任务，但按高校学生工作场景，“风险预警”比“预测85.3分”实用得多。所以建议你在数据预处理阶段把成绩转换为等级标签，比如0-59 为不及格、60-79 为合格、80-100 为优秀。这样后面的联邦分类模型训练更稳定，也更容易向非技术人员解释。
import pandas as pd import numpy as np df = pd.read_csv("global_test.csv") # 这里以全局测试集为例 # 连续特征 numeric_cols = ["attendance_rate", "homework_avg", "midterm_score", "gpa"] # 类别特征 categorical_cols = ["course_type", "weekday_schedule"] # 生成分类标签 df["score_label"] = pd.cut( df["final_score"], bins=[0, 60, 80, 100], labels=[0, 1, 2] # 0: 不及格, 1: 合格, 2: 优秀 ) # 数值特征归一化 for col in numeric_cols: df[col] = (df[col] - df[col].mean()) / (df[col].std() + 1e-8) # 类别特征独热编码 df = pd.get_dummies(df, columns=categorical_cols, drop_first=True) # 最终特征列 feature_cols = [c for c in df.columns if c not in ["student_id", "final_score", "score_label"]] X = df[feature_cols].values.astype(np.float32) y = df["score_label"].values.astype(np.int64)pd.cut是核心操作，bins=[0,60,80,100]默认左开右闭，注意0分会被排除在下限之外，实际数据里不会有0分的话没问题，但保险起见可以改成bins=[-0.1, 60, 80, 100]。数值特征做Z-Score归一化，联邦学习里每个客户端应该使用各自的均值和标准差，而不是全量数据的，否则相当于间接共享了统计信息。类别特征用独热编码，如果类别过多（比如课程名），建议先做词频筛选，只保留出现次数前20的类别，避免维度爆炸。
有的同学会直接把原始final_score作为回归目标，然后在客户端之间比较MSE。这也不是不行，但分类任务在联邦训练下更容易收敛，而且可以配合流式可视化做“红色预警人数”这类指标，和Streamlit搭起来展示效果更好。
3. 联邦平均FedAvg：从零跑通核心训练循环，参数不调整就翻车
3.1 本地模型：一个两层全连接网络就够，不要一上来就上Transformer
高校成绩数据通常只有几百到几千条，客户端再一分，每个客户端可能只有几百条。这种规模下，一个两层全连接网络已经足够。如果你用PyTorch实现，我推荐下面这个结构：输入维度是特征数，隐藏层32维，输出3类，激活函数用ReLU，最后加一层Softmax（训练时不需要先Softmax，直接用CrossEntropyLoss）。
import torch import torch.nn as nn import torch.optim as optim class ScoreNet(nn.Module): def __init__(self, input_dim, hidden_dim=32, num_classes=3): super(ScoreNet, self).__init__() self.net = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, num_classes), ) def forward(self, x): return self.net(x)隐藏层选32维是经验值。成绩特征经过独热编码后通常有20到30维，32维隐藏层不会过拟合，也不会欠拟合。如果客户端数量特别多（比如超过20个），可以减到16维，减少每轮要上传的参数量。注意联邦学习通信成本是主要瓶颈，模型越小越省。
3.2 服务端聚合：不要用全量平均，用户数差异大的时候要加权
联邦平均（FedAvg）最朴素的实现就是：每个客户端本地训练若干个epoch后，把模型参数上传，服务端对所有客户端的参数做加权平均。权重一般是每个客户端拥有的样本数占总样本数的比例。如果你用最简单的算术平均（每个客户端权重一样），在某个学院数据特别多、另一个学院数据特别少时，小客户端的信息会被稀释，模型明显偏向大数据客户端。
完整的训练循环要拆成服务端脚本和客户端脚本。为了演示方便，我在一个文件里模拟多客户端进程，实际部署时客户端之间不通信，只各自连接服务端。
import copy import torch from torch.utils.data import DataLoader, TensorDataset def train_client(model, loader, epochs=3, lr=0.01): model.train() optimizer = optim.Adam(model.parameters(), lr=lr) loss_fn = nn.CrossEntropyLoss() for epoch in range(epochs): for xb, yb in loader: optimizer.zero_grad() out = model(xb) loss = loss_fn(out, yb) loss.backward() optimizer.step() return model.state_dict() def fedavg(client_models, client_sizes): # 返回加权平均后的全局模型参数 avg_model = copy.deepcopy(client_models[0]) total_size = sum(client_sizes) for key in avg_model.keys(): weighted_sum = sum( model[key] * (size / total_size) for model, size in zip(client_models, client_sizes) ) avg_model[key] = weighted_sum return avg_model # 假设 clients_data 是一个列表，每个元素是 (DataLoader, 样本数) global_model = ScoreNet(input_dim=X.shape[1]) for round_idx in range(10): client_weights = [] client_sizes = [] for loader, size in clients_data: client_model = ScoreNet(input_dim=X.shape[1]) client_model.load_state_dict(global_model.state_dict()) trained_state = train_client(client_model, loader, epochs=3, lr=0.01) client_weights.append(trained_state) client_sizes.append(size) # 服务端聚合 global_model.load_state_dict(fedavg(client_weights, client_sizes)) # 每轮后在全局测试集上评估 ...这段代码里最关键的是train_client中的model.load_state_dict(global_model.state_dict())，它把上一轮的全局模型同步给每个客户端再开始本地训练。lr=0.01是Adam常用默认值，但联邦学习里每个客户端只跑3个epoch，如果学习率太大容易让客户端参数偏离全局初始点太远，聚合后全局模型反而震荡。我做过实验，把lr设在0.005到0.02之间，比0.1稳定得多。
fedavg里的加权平均逻辑，注意client_sizes是每个客户端的本地样本数，服务端理论上不知道客户端原始数据，但知道样本数量是允许的（联邦学习协议里客户端会上报数量用于加权）。如果你连样本数量都不想暴露，可以改成统一权重，但效果会差一些，这是一个隐私与效果的权衡点。
3.3 灾难性遗忘很容易在联邦训练中悄然出现
很多跑到轮次中后段的同学发现精度掉回去了，第一反应是调大epoch，结果越调越差。这里要提一个热词是“灾难性遗忘 联邦学习”——本地模型在连续多轮本地迭代后，如果学习率设置不当，会遗忘掉全局模型带来的共性知识。尤其是客户端本地数据分布与全局分布差异较大时，每个客户端倾向于把模型拉到自己的局部最优，聚合后又拉回来，反复拉扯就会造成震荡。
缓解办法有几种，我常用的是降低本地epoch，从5降到2；同时降低学习率到0.005。另一个办法是加“近端项”，就是在本地损失函数里加一个约束，让本地模型的参数不要偏离全局模型太远，也就是Proximal FedAvg（FedProx）的核心思想。如果标题里源码实现的是FedAvg，你也可以自行改成FedProx，改动很小，把CrossEntropyLoss换成CrossEntropyLoss + mu * ||w_local - w_global||^2即可。这个技巧在真实项目中经常能救回两三成的精度。
4. Streamlit可视化：把联邦训练过程变成可交互的预测面板
4.1 为什么用Streamlit而不直接用Gradio：这是“数据看板”不是“模型演示”
标题里明确写了“Streamlit搭建可视化平台”，不要换成Gradio。两者定位完全不同：Gradio适合做一个交互Demo给用户试模型，但你要展示联邦训练过程中的全局精度曲线、各客户端的损失变化、预测结果分布，这些是数据分析场景。Streamlit的生态本来就是围绕数据应用设计的，可以直接用st.line_chart、st.dataframe展示训练历史。Gradio的图表能力没有那么顺手。
4.2 在Streamlit中加载训练产物并展示联邦指标
训练完成之后，把全局模型保存为model.pth，把每轮的评估日志保存为training_log.csv。下面是Streamlit应用的核心代码，把这三部分组合成一个可视化面板。
import streamlit as st import pandas as pd import torch st.set_page_config(page_title="高校成绩预测联邦平台", layout="wide") st.title("基于联邦学习的高校成绩预测平台") @st.cache_resource def load_model(): model = ScoreNet(input_dim=42) # 维度要与训练时一致 model.load_state_dict(torch.load("global_model.pth", map_location="cpu")) model.eval() return model log_df = pd.read_csv("training_log.csv") col1, col2 = st.columns(2) with col1: st.subheader("全局模型精度") st.line_chart(log_df[["round", "global_acc"]].set_index("round")) with col2: st.subheader("各客户端损失") st.line_chart(log_df.set_index("round")[["client1_loss", "client2_loss"]]) st.subheader("输入学生特征进行预测") attendance = st.slider("出勤率（%）", 0.0, 100.0, 80.0) homework_avg = st.slider("作业平均分", 0.0, 100.0, 70.0) midterm = st.number_input("期中成绩", 0, 100, 60) course_type = st.selectbox("课程类型", ["专业课", "公共课"]) if st.button("预测"): model = load_model() # 构造输入特征向量，注意顺序要与训练特征列一致 # 这里使用简化的特征构造，实际请参考你的特征工程脚本 feature = torch.tensor([[attendance / 100.0, homework_avg / 100.0, midterm / 100.0]], dtype=torch.float32) with torch.no_grad(): pred = model(feature) label = int(torch.argmax(pred, dim=1)) st.success(f"预测结果：{'不及格' if label == 0 else '合格' if label == 1 else '优秀'}")st.cache_resource用来缓存加载的模型，避免每次点击按钮都重新读权重文件。st.line_chart直接吃DataFrame，不需要额外绘图库。这里注意特征构造要与训练时完全对齐，我为了示例只用了三个手填特征，真实项目里需要把所有特征都做同样的归一化和独热编码后才能喂给模型。一个省事的做法是在训练时把特征列顺序保存到一个feature_order.json里，Streamlit加载后动态构造输入向量，避免硬编码。
4.3 在Streamlit中模拟联邦训练的实时可视化
如果你不想等离线训练完再展示，可以把联邦训练循环直接搬进Streamlit的缓存或进度条里。常见做法是每轮聚合后，把global_acc追加到st.session_state["log"]，用st.empty()刷新图表。这样平台上可以一边跑训练一边更新图表，效果更像一个“联邦训练监控台”。下面是片段：
if st.button("开始联邦训练"): progress = st.progress(0) chart_placeholder = st.empty() log = [] for round_idx in range(20): # 执行一轮训练与聚合，代码见第3章 ... acc = evaluate(global_model, global_test_loader) log.append({"round": round_idx, "global_acc": acc}) chart_placeholder.line_chart(pd.DataFrame(log).set_index("round")) progress.progress((round_idx + 1) / 20)st.empty()配合line_chart可以做原位更新，每次刷新都不会在页面上堆积旧图表。这个技巧比生成GIF或保存图片再展示要方便得多。需要注意的是，训练中每个客户端都要在子线程里跑，否则Streamlit的主线程会被阻塞，界面会卡死。实际项目中我一般把训练逻辑放进一个独立的trainer.py，用st.session_state缓存训练状态，避免重连后模型丢失。
5. 避坑：联邦学习成绩预测的5个常见翻车点与排查清单
5.1 客户端数据分布不一致，导致聚合后的模型反而比单客户端还差
现象：每个客户端本地验证精度都还行，但全局模型精度却不超过0.5，甚至接近随机猜。
原因：这是联邦学习最典型的非独立同分布问题。成绩数据天然和学院强相关，比如某学院全是高数课，另一个学院全是体育课，特征分布差异巨大。直接FedAvg会把彼此冲突的参数平均掉，模型两边都不讨好。
解决：先做数据分布可视化，把每个客户端的标签分布打印出来。如果差异过大，第一步不是换算法，而是调整客户端划分方式，尽量让每个客户端包含多个学院的混合数据（不过这与真实场景冲突）。如果想保留真实划分，就把本地epoch降到1，学习率降到0.005，并在聚合时给每个客户端模型乘以一个缩放因子（比如0.8），让本地更新更温和。另一个有效方案是改用FedAvg的变体FedProx，在本地损失中加入“离全局模型不要太远”的惩罚项。
5.2 全局测试集泄漏到客户端训练集里，精度虚高
现象：训练过程中验证精度一路涨到0.95，但部署到实际新学生数据时预测完全不准。
原因：这是初学者最容易犯的错误——切分客户端时直接用train_test_split全量随机切分，没有按客户端分组。结果同一个学生可能一部分记录在客户端A的训练集里，另一部分记录在全局测试集里，模型等于提前看到了测试答案。
解决：严格按照第2章的做法，先按学院/学号分组，再在组内划分。特别注意：如果一个学生有多个学期的记录，要按学生ID分组而不是按行分组，防止同一人的不同学期数据被拆到训练集和测试集。判断泄漏的方法是查看测试集里有没有训练集中出现过的student_id，可以在切分后执行一次集合求交集。
5.3 Streamlit刷新页面后模型状态被重置
现象：点击一次预测按钮没问题，刷新页面后模型又得重新加载，或者需要重新上传数据集。
原因：Streamlit的执行模型是“每次交互都重新运行整个脚本”，变量默认不会保留。你把模型加载写在顶层，每次输入变化都会重新load_state_dict，如果没用st.cache_resource，模型文件会被重复读入内存，页面越来越卡。
解决：把所有耗资源且不依赖交互参数的初始化放进@st.cache_resource或@st.cache_data。对于训练过程中的轮次记录，使用st.session_state保存，比如st.session_state["training_log"]。此外，如果模型文件很大，加载时务必指定map_location="cpu"，避免Streamlit在无GPU环境上报错。
5.4 不同客户端上的特征列不一致，聚合时报张量shape不匹配
现象：聚合时state_dict的key完全一致，但某个key的shape对不上，报错信息类似size mismatch。
原因：不同客户端在本地做独热编码时，类别集合不同。比如客户端A的课程类型有“专业课、公共课、选修课”三列，客户端B少了一个“选修课”，独热编码后特征维度就不一致。本地模型输入层维度也就不同，参数自然无法直接平均。
解决：在做数据预处理时，所有客户端必须使用同一个特征生成规则。常见做法是在服务端预先定义好全量特征模板（比如所有可能的课程类型），然后每个客户端按这个模板做对齐，缺失的列补0。这个规则同样适用于归一化：每个客户端用各自均值标准差归一化没问题，但特征列顺序必须是同一个。我在代码里会把特征列顺序写死成一个列表，并且在训练前打印每个客户端的input_dim做校验。
5.5 训练曲线震荡不收敛，看起来像“情绪化”的折线
现象：全局精度在不同轮次之间大起大落，这轮0.75下轮0.4再下轮0.7。
原因：常见原因有三个。一是每个客户端本地epoch太多，造成客户端漂移；二是学习率太大，聚合后在局部最优点之间跳来跳去；三是每个客户端只返回了一次迭代的结果，服务端又做了硬平均，缺少类似Momentum的平滑机制。
解决：优先调小lr和epochs。如果还震荡，在服务端聚合时引入上一次全局模型的动量：new_global = 0.9 * old_global + 0.1 * avg_this_round。这么做虽然会减慢新数据的贡献速度，但曲线会稳定很多。另外建议每轮训练后都强制torch.cuda.empty_cache()，防止显存碎片让训练时快时慢，影响精度统计的准确性。
6. 进阶：给联邦模型加差分隐私，并做集中式/本地式/联邦式的三方对比
如果你打算把这个项目写进论文或作为部门的技术选型依据，只跑通FedAvg还不够，至少要做两个进阶动作。第一个是加入差分隐私：在客户端向服务端上传梯度之前，对梯度做裁剪并添加高斯噪声。下面是一个极简实现片段：
def add_noise_to_gradients(model, clip_c=1.0, noise_scale=0.1): for param in model.parameters(): param.grad = torch.clamp(param.grad, -clip_c, clip_c) noise = torch.randn_like(param.grad) * noise_scale param.grad.add_(noise)调用时机在train_client里的每次optimizer.step()之前。注意clip_c太小会抑制学习，noise_scale太大会让模型完全随机。实际我一般先做无噪声基线，再按0.1步长调噪声，观察全局测试集的精度折损，选一个折损在5%以内的噪声尺度和裁剪阈值。
第二个动作是做一个三方对比实验：集中式训练（所有数据汇总）、本地独立训练（每个客户端各自训练不聚合）、联邦训练。用同一个全局测试集评估三个模型的精度和红线指标（比如挂科召回率）。我习惯把结果画在一个Streamlit页面的分组柱状图里，这样能直观说明联邦学习“损失少量精度换取隐私保护”的性价比。真实结果往往是：集中式精度最高，联邦式比本地式高5到10个百分点，这个数字就是方案的落地价值。
最后给你一个我的个人习惯：任何联邦学习项目先跑通10轮小实验，打印每轮每个客户端的参数范数变化，再上线全量数据。参数范数突变的那一轮，往往是客户端本地学习率要调低的信号。这套流程我已经用在了好几个内部数据平台项目上，每次都能提前发现一些训练玄学问题。希望帮到你。
本文还有配套的精品资源，点击获取
