联邦学习实战：基于PyTorch与Streamlit构建隐私保护的学生成绩预测系统

简介本资源是一套面向高校计算机类专业本科生与研究生的毕业设计级实践项目聚焦联邦学习在教育数据场景下的落地应用——学生成绩预测。项目采用Python实现多客户端协同训练框架集成FedRep、FedProx、Ditto、Scaffold等主流联邦优化算法并通过Streamlit构建可交互的可视化分析平台解决跨院系/班级数据孤岛下的模型共建问题。压缩包含67个文件涵盖18个核心Python模块含训练主逻辑、模型定义、通信辅助函数、7个CSV格式的真实与模拟学生成绩数据集、11个备份文件.zbak及28个编译缓存文件.pyc整体体积仅2.26MB轻量易部署。已有75人下载学习提供完整可运行代码、调试验证通过的实验结果含准确率曲线、混淆矩阵图、清晰分层的模块化架构与详细README说明特别适合课程设计、毕设选题与联邦学习入门实践者快速掌握算法实现与系统集成要点。1. 项目概述为什么我们需要一个“不共享数据”的预测系统在传统的教育数据分析场景里如果你想构建一个学生成绩预测模型最直接的做法是什么没错就是把所有学校、所有班级的学生数据——比如历次考试成绩、出勤率、作业完成情况、甚至家庭背景信息——统统收集到一个中心服务器上然后训练一个统一的机器学习模型。这个思路听起来很高效但实际操作起来几乎是一个“不可能完成的任务”。数据隐私和安全法规比如GDPR、国内的《个人信息保护法》像一道道高墙让学校之间、甚至班级之间的数据共享变得异常敏感和困难。每个数据孤岛都像一座戒备森严的城堡里面藏着宝贵的“知识矿石”但我们却无法将它们熔炼在一起。这就是“联邦学习”登场的时候了。它不是一个具体的算法而是一种颠覆性的机器学习范式。简单来说联邦学习的核心思想是“数据不动模型动”。我们不再把原始数据汇集到中心而是把初始的预测模型比如一个神经网络分发到各个数据持有方例如各个学校的服务器。每个学校用自己的本地数据在本地训练这个模型得到模型的更新通常是梯度或参数更新量。然后这些“更新”被加密上传到一个中央服务器。中央服务器的工作就是安全地聚合这些来自各地的模型更新融合成一个更强大、更通用的“全局模型”再分发给所有参与方。整个过程原始数据始终留在本地从未离开过数据所有者的控制范围。我们这个项目——“联邦学习驱动的学生成绩预测系统”正是为了解决上述痛点而生。它旨在构建一个既能利用多源数据提升预测精度又能严格保护各方数据隐私的实用系统。我们选择Python作为实现语言得益于其丰富的机器学习生态如PyTorch, TensorFlow Federated。而Streamlit则为我们提供了一个极其高效的方式将复杂的联邦学习流程和结果转化为一个交互式、可视化的Web应用平台。老师或教育管理者无需理解底层代码通过浏览器就能直观地看到模型训练过程、各参与方的贡献、以及最终的预测效果。这不仅仅是技术演示更是一个面向真实教育场景的、具备可操作性的解决方案原型。2. 系统核心架构与联邦学习方案选型一个完整的联邦学习系统远不止“训练一个模型”那么简单。它需要一套严谨的架构来协调参与者、保障通信安全、处理异构数据。我们的系统设计主要包含以下几个核心组件中央协调服务器这是系统的大脑。它负责初始化全局模型、选择参与每一轮训练的客户端学校、接收并聚合客户端上传的模型更新、评估全局模型性能并将更新后的模型分发给客户端。在Python中我们可以用Flask或FastAPI轻松搭建一个RESTful API服务器来实现这些功能。客户端即各个数据持有方学校。每个客户端实例拥有自己的本地数据集。它的职责是从服务器下载最新的全局模型用本地数据对模型进行若干轮训练称为本地训练周期计算模型参数的更新量如梯度差值将更新量或加密后的更新量上传给服务器。通信协议与安全模块这是联邦学习的生命线。我们必须确保客户端与服务器之间传输的模型更新不会被恶意第三方窃取或篡改同时还要防止服务器从更新中反推出原始数据隐私攻击。常见的方案包括同态加密、差分隐私等。在原型阶段我们可以使用SSL/TLS进行传输加密并引入差分隐私噪声来初步保护隐私。可视化平台基于Streamlit构建的前端界面。它实时从中央服务器拉取训练状态、模型指标和预测结果并以图表、进度条、数据表格等直观形式展现给用户。在联邦学习的具体算法选型上最经典、应用最广的是FedAvg。它的流程非常直观服务器初始化一个全局模型 $w_0$。在每一轮通信回合 $t$ 中服务器随机选择一部分客户端 $S_t$。服务器将当前全局模型 $w_t$ 发送给每个选中的客户端。每个客户端 $k$ 用本地数据训练模型得到本地更新 $w_t^{k}$。每个客户端将 $w_t^{k}$ 上传至服务器。服务器聚合所有更新$w_{t1} \sum_{k \in S_t} \frac{n_k}{n} w_t^{k}$其中 $n_k$ 是客户端k的数据量$n$ 是所选客户端总数据量。重复上述过程直到模型收敛。注意FedAvg假设各个客户端的数据是独立同分布的但现实中不同学校的学生数据分布可能差异巨大有的重理科有的重文科这被称为“数据非独立同分布”。这是联邦学习中的核心挑战之一可能导致全局模型在某些客户端上表现很差。在后续实现中我们需要关注这一点。为什么选择FedAvg作为起点因为它概念清晰实现相对简单非常适合作为我们项目的基石。在Streamlit平台上我们可以清晰地展示每一轮中哪些客户端被选中、它们的本地数据量、以及它们对全局模型的贡献权重让整个“联邦”过程透明化。3. 开发环境搭建与核心库依赖详解工欲善其事必先利其器。在开始编码之前一个稳定、隔离的Python环境至关重要。我强烈推荐使用conda或venv创建虚拟环境避免包版本冲突。# 使用 conda 创建环境 conda create -n fl-grade-prediction python3.9 conda activate fl-grade-prediction # 或者使用 venv python -m venv fl-venv # Windows: fl-venv\Scripts\activate # Linux/Mac: source fl-venv/bin/activate接下来是安装核心依赖库。我们的项目主要涉及机器学习、联邦学习框架和Web可视化。# 基础数据处理与科学计算 pip install numpy pandas scikit-learn # 深度学习框架这里以PyTorch为例更灵活 # 请根据你的CUDA版本访问PyTorch官网获取安装命令例如 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 联邦学习框架我们使用PySyft的一个简化实践版或直接实现FedAvg。 # 为了教学清晰我们选择自己实现核心逻辑但可以安装一些辅助库。 # pip install syft (注PySyft安装可能较复杂原型阶段可暂缓) # 可视化与Web应用核心 pip install streamlit plotly matplotlib seaborn # Web服务器框架用于中央服务器 pip install flask关键库选型理由PyTorch vs TensorFlow我选择PyTorch是因为它的动态图机制在研究和原型开发阶段更加灵活直观调试方便。对于联邦学习这种需要频繁修改训练逻辑的场景PyTorch的友好性更高。TensorFlow Federated虽然是为联邦学习而生但学习曲线稍陡且生态相对较新。Streamlit它是快速构建数据科学Web应用的“神器”。你几乎可以用纯Python脚本创建出包含交互控件、图表、表格的完整应用无需前端知识。这对于我们快速展示联邦学习过程来说效率是决定性的。Flask作为中央服务器的后端它轻量、易用足够处理我们模型聚合和分发的HTTP请求。一个常见的坑是版本冲突。特别是torch和torchvision的版本需要匹配且与你的Python版本、CUDA版本兼容。如果遇到问题先去官方文档核对版本矩阵。我个人的经验是在项目初期就使用pip freeze requirements.txt记录所有依赖的精确版本便于复现环境。4. 数据模拟与隐私化处理实践真实的学生成绩数据涉及隐私我们无法获取。因此构建一个贴近现实的模拟数据集是第一步也是检验我们系统逻辑的关键。我们需要模拟多个客户端学校每个客户端的数据具有不同的分布特点。import numpy as np import pandas as pd from sklearn.datasets import make_regression from sklearn.model_selection import train_test_split def generate_client_data(client_id, num_samples200, bias0.0, noise20.0): 为单个客户端生成模拟数据。 特征X可能包括平均学习时间、作业提交率、课堂互动次数、前期测验成绩等。 目标y是期末成绩。 通过bias参数模拟不同学校的整体水平差异。 np.random.seed(42 client_id) # 确保可复现同时每个客户端不同 # 生成基本特征 X, y make_regression(n_samplesnum_samples, n_features5, noisenoise, random_stateclient_id) # 为特征赋予实际意义 feature_names [study_hours, assignment_rate, interaction, prev_score1, prev_score2] df pd.DataFrame(X, columnsfeature_names) # 对特征进行缩放和偏移使其更符合实际范围 df[study_hours] df[study_hours] * 5 15 # 平均15-20小时 df[assignment_rate] (df[assignment_rate] * 0.2 0.7).clip(0,1) # 提交率70%上下 # 添加学校偏差模拟“名校”或“普通学校”的整体水平差异 y y bias # 将成绩映射到0-100分区间 y (y - y.min()) / (y.max() - y.min()) * 60 40 # 大致在40-100分之间 df[final_score] y return df # 生成3个客户端的数据具有不同偏差 client_data {} client_data[school_a] generate_client_data(1, bias10.0, noise15.0) # 重点学校成绩偏高数据质量好噪声低 client_data[school_b] generate_client_data(2, bias0.0, noise25.0) # 普通学校 client_data[school_c] generate_client_data(3, bias-5.0, noise30.0) # 基础薄弱学校成绩偏低数据噪声大隐私化处理在联邦学习中原始数据本身不离开客户端这已经是最强的隐私保护。但我们还需要保护上传的“模型更新”。一种简单有效的方法是应用差分隐私。我们可以在客户端本地训练后给要上传的梯度向量添加精心校准的拉普拉斯噪声或高斯噪声。def add_laplace_noise(gradients, epsilon0.1, sensitivity1.0): 向梯度添加拉普拉斯噪声以实现差分隐私。 noise np.random.laplace(loc0.0, scalesensitivity/epsilon, sizegradients.shape) return gradients noise这里的epsilon是隐私预算值越小隐私保护越强但添加的噪声越大模型精度可能下降。sensitivity是函数梯度计算的敏感度需要根据模型和数据集进行估计。在实际部署中需要和数据所有者学校共同确定可接受的epsilon值在隐私和效用之间取得平衡。实操心得模拟数据时故意制造客户端之间的“数据非独立同分布”非常重要。比如让A校的“学习时间”特征与成绩相关性更强而B校的“前期成绩”特征预测力更强。这能更好地测试联邦学习算法在异构数据下的鲁棒性。我们的bias和noise参数就是用于此目的。5. 联邦学习模型设计与PyTorch实现我们预测学生成绩是一个回归任务因此选择一个合适的模型。一个简单的多层感知机就足以作为起点。关键在于实现本地训练和联邦平均的逻辑。5.1 定义神经网络模型import torch import torch.nn as nn import torch.optim as optim class GradePredictor(nn.Module): def __init__(self, input_dim5): super(GradePredictor, self).__init__() self.fc1 nn.Linear(input_dim, 64) self.relu nn.ReLU() self.dropout nn.Dropout(0.2) # 防止过拟合 self.fc2 nn.Linear(64, 32) self.fc3 nn.Linear(32, 1) # 输出层预测一个分数值 def forward(self, x): x self.relu(self.fc1(x)) x self.dropout(x) x self.relu(self.fc2(x)) x self.fc3(x) return x5.2 客户端本地训练函数每个客户端需要能够用本地数据训练模型并返回更新后的模型参数或梯度。def client_train(model, train_loader, epochs5, lr0.01): 在客户端本地数据上训练模型。 device torch.device(cuda if torch.cuda.is_available() else cpu) model.to(device) model.train() # 设置为训练模式 criterion nn.MSELoss() # 回归任务使用均方误差损失 optimizer optim.SGD(model.parameters(), lrlr) for epoch in range(epochs): running_loss 0.0 for data, target in train_loader: data, target data.to(device), target.to(device) optimizer.zero_grad() output model(data) loss criterion(output, target.view(-1, 1)) loss.backward() optimizer.step() running_loss loss.item() # print(f Epoch {epoch1}, Loss: {running_loss/len(train_loader):.4f}) # 返回训练后的模型状态字典 return model.state_dict()5.3 联邦平均核心逻辑这是中央服务器的核心职能。它接收来自多个客户端的模型参数按数据量进行加权平均。def federated_averaging(client_weights_list, client_sizes): 执行联邦平均。 :param client_weights_list: 列表每个元素是一个客户端模型的状态字典state_dict :param client_sizes: 列表每个客户端的数据样本数量 :return: 聚合后的全局模型状态字典 total_size sum(client_sizes) averaged_weights {} # 初始化平均权重字典结构取自第一个客户端 for key in client_weights_list[0].keys(): averaged_weights[key] torch.zeros_like(client_weights_list[0][key]) # 加权求和 for client_weights, size in zip(client_weights_list, client_sizes): weight size / total_size for key in averaged_weights.keys(): averaged_weights[key] weight * client_weights[key] return averaged_weights5.4 训练循环模拟将以上部分组合起来模拟多轮联邦学习过程。def simulate_federated_learning(clients_data_dict, num_rounds10, clients_per_round2): 模拟联邦学习训练循环。 :param clients_data_dict: 字典键为客户端名值为(DataLoader_train, DataLoader_test) :param num_rounds: 通信轮数 :param clients_per_round: 每轮选择的客户端数量 # 初始化全局模型 global_model GradePredictor() global_state global_model.state_dict() history {round: [], global_loss: [], client_losses: {}} for round_idx in range(num_rounds): print(f\n Round {round_idx 1} ) # 1. 选择客户端 selected_clients np.random.choice(list(clients_data_dict.keys()), sizeclients_per_round, replaceFalse) print(fSelected clients: {selected_clients}) client_weights [] client_sizes [] # 2. 每个选中的客户端进行本地训练 for client in selected_clients: train_loader, _ clients_data_dict[client] # 下载全局模型 local_model GradePredictor() local_model.load_state_dict(global_state) # 本地训练 updated_weights client_train(local_model, train_loader, epochs3) client_weights.append(updated_weights) client_sizes.append(len(train_loader.dataset)) # 3. 服务器聚合联邦平均 new_global_state federated_averaging(client_weights, client_sizes) global_model.load_state_dict(new_global_state) global_state new_global_state # 4. 评估全局模型在所有客户端的测试集上 global_model.eval() total_loss 0 total_samples 0 criterion nn.MSELoss() client_losses_this_round {} with torch.no_grad(): for client, (_, test_loader) in clients_data_dict.items(): client_loss 0 for data, target in test_loader: output global_model(data) loss criterion(output, target.view(-1, 1)) client_loss loss.item() * data.size(0) avg_client_loss client_loss / len(test_loader.dataset) client_losses_this_round[client] avg_client_loss total_loss client_loss total_samples len(test_loader.dataset) avg_global_loss total_loss / total_samples print(fGlobal Model Loss this round: {avg_global_loss:.4f}) for client, loss in client_losses_this_round.items(): print(f - Client {client} test loss: {loss:.4f}) # 记录历史 history[round].append(round_idx1) history[global_loss].append(avg_global_loss) for client in clients_data_dict.keys(): if client not in history[client_losses]: history[client_losses][client] [] history[client_losses][client].append(client_losses_this_round.get(client, None)) return global_model, history这段代码构成了我们联邦学习系统的核心引擎。你可以看到数据始终没有离开clients_data_dict这个模拟的本地环境只有模型的参数state_dict在流动。6. Streamlit可视化平台开发全流程Streamlit的魅力在于你可以像写脚本一样构建应用。我们将创建一个多页面的应用展示联邦学习的全过程。6.1 应用骨架与侧边栏导航首先我们建立应用的主框架和导航。# app.py import streamlit as st import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots import sys import os # 假设我们的联邦学习模拟代码在一个叫 fl_simulation.py 的模块里 sys.path.append(os.path.dirname(__file__)) from fl_simulation import simulate_federated_learning, generate_client_data, prepare_dataloaders st.set_page_config(page_title联邦学习成绩预测平台, layoutwide) st.title( 联邦学习驱动的学生成绩预测系统) # 初始化session_state用于存储跨页面的状态 if global_model not in st.session_state: st.session_state.global_model None if training_history not in st.session_state: st.session_state.training_history None if client_data_info not in st.session_state: st.session_state.client_data_info {} # 侧边栏导航 st.sidebar.header(导航) page st.sidebar.radio(选择页面, [ 系统总览, 数据模拟与探索, ⚙️ 联邦训练控制台, 训练过程可视化, 成绩预测与评估]) # 在侧边栏添加一些全局控制参数 st.sidebar.header(全局配置) num_clients st.sidebar.slider(模拟学校数量, 2, 5, 3) clients_per_round st.sidebar.slider(每轮参与学校数, 1, num_clients, 2) num_rounds st.sidebar.slider(联邦训练轮数, 5, 50, 20)6.2 数据模拟与探索页面这个页面让用户直观看到我们模拟的、具有差异化的各校数据。if page 数据模拟与探索: st.header(客户端数据模拟) if st.button(生成/刷新模拟数据): with st.spinner(正在生成各学校模拟数据...): client_dfs {} for i in range(num_clients): bias (i - num_clients//2) * 8.0 # 制造差异 client_name f学校_{i1} df generate_client_data(i, biasbias, noise20.0 i*5) client_dfs[client_name] df st.session_state.client_data_info[client_name] { samples: len(df), avg_score: df[final_score].mean(), bias: bias } st.session_state.client_dfs client_dfs st.success(数据生成完成) if client_dfs in st.session_state: selected_client st.selectbox(选择要查看的学校, list(st.session_state.client_dfs.keys())) df st.session_state.client_dfs[selected_client] col1, col2 st.columns(2) with col1: st.subheader(f{selected_client} 数据概览) st.dataframe(df.describe(), use_container_widthTrue) st.metric(学生人数, len(df), f平均成绩: {df[final_score].mean():.1f}分) with col2: st.subheader(特征与成绩分布) fig make_subplots(rows2, cols2, subplot_titles(学习时间 vs 成绩, 作业提交率 vs 成绩, 前期成绩1 vs 成绩, 特征相关性)) # 散点图1 fig.add_trace(go.Scatter(xdf[study_hours], ydf[final_score], modemarkers, name学习时间), row1, col1) # 散点图2 fig.add_trace(go.Scatter(xdf[assignment_rate], ydf[final_score], modemarkers, name提交率), row1, col2) # 散点图3 fig.add_trace(go.Scatter(xdf[prev_score1], ydf[final_score], modemarkers, name前期成绩1), row2, col1) # 热力图 corr df.corr().round(2) fig.add_trace(go.Heatmap(zcorr.values, xcorr.columns, ycorr.columns, textcorr.values, texttemplate%{text}, colorscaleRdBu, zmid0), row2, col2) fig.update_layout(height600, showlegendFalse) st.plotly_chart(fig, use_container_widthTrue) st.caption(**观察**不同学校的特征分布和与成绩的相关性可能存在差异这正是联邦学习要处理的‘数据非独立同分布’挑战。)6.3 联邦训练控制台页面这是系统的“驾驶舱”用户可以启动、控制训练过程。elif page ⚙️ 联邦训练控制台: st.header(联邦训练控制中心) if client_dfs not in st.session_state: st.warning(请先在‘数据模拟与探索’页面生成数据。) else: col1, col2 st.columns([2,1]) with col1: st.subheader(训练参数) local_epochs st.slider(客户端本地训练轮数, 1, 10, 3) learning_rate st.number_input(学习率, min_value0.0001, max_value0.1, value0.01, step0.001, format%.4f) use_dp st.checkbox(启用差分隐私保护 (会降低精度), valueFalse) dp_epsilon st.slider(隐私预算 ε (越小越隐私), 0.1, 5.0, 1.0, 0.1, disablednot use_dp) with col2: st.subheader(操作) if st.button( 开始联邦训练, typeprimary, use_container_widthTrue): with st.spinner(f正在进行联邦训练共{num_rounds}轮...): # 准备数据加载器 client_loaders {} for name, df in st.session_state.client_dfs.items(): train_loader, test_loader prepare_dataloaders(df) client_loaders[name] (train_loader, test_loader) # 调用模拟训练函数 global_model, history simulate_federated_learning( client_loaders, num_roundsnum_rounds, clients_per_roundclients_per_round ) st.session_state.global_model global_model st.session_state.training_history history st.success(联邦训练完成) st.balloons() # 显示训练状态摘要 if st.session_state.training_history: st.subheader(最新训练摘要) last_round st.session_state.training_history[round][-1] last_loss st.session_state.training_history[global_loss][-1] col_a, col_b, col_c st.columns(3) col_a.metric(训练总轮数, last_round) col_b.metric(最终全局损失, f{last_loss:.4f}) # 计算相比第一轮的提升 if len(st.session_state.training_history[global_loss]) 1: improvement (st.session_state.training_history[global_loss][0] - last_loss) / st.session_state.training_history[global_loss][0] * 100 col_c.metric(损失下降, f{improvement:.1f}%)6.4 训练过程可视化页面这是整个平台的精华用动态图表展示联邦学习的核心过程。elif page 训练过程可视化: st.header(训练过程动态分析) if st.session_state.training_history is None: st.info(训练历史为空请先启动训练。) else: history st.session_state.training_history rounds history[round] tab1, tab2, tab3 st.tabs([ 全局损失曲线, 各客户端损失对比, ⚖️ 客户端贡献分析]) with tab1: fig1 go.Figure() fig1.add_trace(go.Scatter(xrounds, yhistory[global_loss], modelinesmarkers, name全局模型损失, linedict(width3))) fig1.update_layout(title全局模型损失随训练轮次的变化, xaxis_title通信轮次, yaxis_title损失 (MSE), templateplotly_white) st.plotly_chart(fig1, use_container_widthTrue) st.markdown( **解读**理想的曲线应随着训练轮次增加而稳步下降最终趋于平缓。如果曲线剧烈波动或上升可能意味着学习率过高、客户端数据差异过大或每轮选择的客户端太少。 ) with tab2: fig2 go.Figure() for client_name, client_losses in history[client_losses].items(): # 客户端可能在某些轮次未被选中损失为None valid_rounds [r for r, l in zip(rounds, client_losses) if l is not None] valid_losses [l for l in client_losses if l is not None] fig2.add_trace(go.Scatter(xvalid_rounds, yvalid_losses, modelinesmarkers, nameclient_name)) fig2.update_layout(title各客户端测试损失变化, xaxis_title通信轮次, yaxis_title损失 (MSE), templateplotly_white) st.plotly_chart(fig2, use_container_widthTrue) st.markdown( **解读**此图反映了全局模型在各个学校本地测试集上的表现。在数据非独立同分布下某些客户端的损失可能始终较高。这是联邦学习中的“客户漂移”问题也是后续优化的方向如FedProx算法。 ) with tab3: # 这里可以模拟展示每轮各客户端被选中的情况及其数据量权重 st.subheader(客户端参与情况模拟) # 这是一个简化的模拟展示实际项目中应从训练日志中提取真实数据 participation_data [] for r in rounds: # 模拟随机选择 selected np.random.choice(list(st.session_state.client_data_info.keys()), sizeclients_per_round, replaceFalse) for client in st.session_state.client_data_info.keys(): participation_data.append({ Round: r, Client: client, Selected: 1 if client in selected else 0, Weight: st.session_state.client_data_info[client][samples] if client in selected else 0 }) df_participation pd.DataFrame(participation_data) fig3 go.Figure(datago.Heatmap( zdf_participation[Selected].values.reshape(len(rounds), -1), xlist(st.session_state.client_data_info.keys()), yrounds, colorscale[[0, lightgray], [1, royalblue]], showscaleFalse, textdf_participation[Weight].values.reshape(len(rounds), -1), texttemplate%{text}, textfont{size:10} )) fig3.update_layout(title每轮客户端选择与数据量权重蓝色表示被选中数字为权重, xaxis_title客户端学校, yaxis_title通信轮次) st.plotly_chart(fig3, use_container_widthTrue)6.5 成绩预测与评估页面最后我们提供一个交互界面让用户可以使用训练好的全局模型进行预测并评估模型效果。elif page 成绩预测与评估: st.header(模型预测与性能评估) if st.session_state.global_model is None: st.warning(暂无训练好的模型请先完成训练。) else: model st.session_state.global_model model.eval() st.subheader(单样本成绩预测) col1, col2 st.columns(2) with col1: study_hours st.slider(每周学习时间 (小时), 5.0, 40.0, 20.0, 0.5) assignment_rate st.slider(作业提交率, 0.0, 1.0, 0.8, 0.05) interaction st.slider(课堂互动指数, -2.0, 2.0, 0.0, 0.1) with col2: prev_score1 st.slider(期中考试成绩, 0.0, 100.0, 70.0, 1.0) prev_score2 st.slider(平时测验平均分, 0.0, 100.0, 75.0, 1.0) if st.button(预测期末成绩, typeprimary): input_tensor torch.tensor([[study_hours, assignment_rate, interaction, prev_score1, prev_score2]], dtypetorch.float32) with torch.no_grad(): prediction model(input_tensor).item() st.metric(预测期末成绩, f{prediction:.1f} 分) # 给出一个简单的解释区间 st.info(f根据模型预测该学生的期末成绩预计在 **{max(0, prediction-8):.0f} ~ {min(100, prediction8):.0f}** 分之间仅供参考。) st.subheader(模型在全体测试集上的评估) if st.button(运行全局评估): # 这里需要重新加载所有客户端的测试集进行评估 total_loss 0 total_samples 0 eval_results [] criterion nn.MSELoss() for client_name, (_, test_loader) in st.session_state.client_loaders.items(): # 假设loaders已保存 client_loss 0 for data, target in test_loader: output model(data) loss criterion(output, target.view(-1, 1)) client_loss loss.item() * data.size(0) avg_loss client_loss / len(test_loader.dataset) eval_results.append({Client: client_name, Test Loss (MSE): avg_loss, RMSE: np.sqrt(avg_loss)}) total_loss client_loss total_samples len(test_loader.dataset) avg_global_loss total_loss / total_samples df_eval pd.DataFrame(eval_results) st.dataframe(df_eval.style.format({Test Loss (MSE): {:.4f}, RMSE: {:.2f}}), use_container_widthTrue) st.metric(全局测试集平均MSE损失, f{avg_global_loss:.4f}, fRMSE: {np.sqrt(avg_global_loss):.2f}) st.caption(**RMSE均方根误差** 可以理解为模型预测成绩与真实成绩的平均差距单位分这个值越小越好。)通过这五个页面我们构建了一个从数据模拟、训练控制、过程可视化到预测评估的完整闭环。Streamlit的交互组件滑块、按钮、选择框和Plotly的动态图表让整个联邦学习过程变得清晰可见。7. 部署、优化与常见问题排查7.1 本地运行与部署在项目根目录下运行以下命令即可启动Streamlit应用streamlit run app.py浏览器会自动打开http://localhost:8501。对于生产环境部署可以考虑使用Docker容器化然后部署到云服务器如AWS EC2, Google Cloud Run, 或国内的阿里云ECS上。Streamlit也提供了原生的云分享服务Streamlit Community Cloud可以一键部署。7.2 性能与隐私优化方向模型压缩在客户端和服务器之间传输完整模型参数可能带宽消耗大。可以考虑使用梯度稀疏化、量化或知识蒸馏来减少通信负载。高级聚合算法基础的FedAvg对非独立同分布数据敏感。可以研究实现FedProx添加近端项约束本地更新防止客户端漂移、SCAFFOLD使用控制变量减少客户端差异等更鲁棒的算法。个性化联邦学习我们的目标是得到一个全局通用的模型。但在教育场景下每个学校可能最终想要一个更适合自己特色的模型。可以探索在联邦学习框架下生成个性化模型的方法。增强隐私保护我们只实现了简单的差分隐私。工业级应用需要考虑安全聚合即服务器在无法解密单个客户端更新的情况下完成聚合。这需要结合同态加密或安全多方计算技术。7.3 常见问题与排查技巧实录在实际开发和运行中你可能会遇到以下问题问题现象可能原因排查与解决思路全局模型损失不下降甚至上升。1. 客户端学习率过高。2. 每轮参与客户端太少或客户端数据差异极大。3. 本地训练轮数过多导致客户端模型偏离全局模型太远客户端漂移。1. 调低lr(如从0.01到0.001)。2. 增加clients_per_round或检查模拟数据偏差是否设置得过于极端。3. 减少本地训练轮数local_epochs或尝试FedProx算法。某个客户端的损失始终远高于其他客户端。该客户端的数据分布与其他客户端差异过大非独立同分布问题。1. 这是联邦学习的固有挑战。可以检查该客户端模拟数据的bias和noise。2. 考虑为该客户端分配更小的聚合权重或采用个性化联邦学习方案。Streamlit应用运行缓慢特别是训练时页面卡死。联邦训练是计算密集型任务会阻塞Streamlit的主线程。1. 将训练任务放入后台线程或使用st.spinner包裹。2. 对于演示减少num_rounds、clients_per_round和数据量。3. 考虑将训练逻辑移出Streamlit作为一个独立的服务通过API调用。模拟数据特征与成绩看不出相关性。make_regression生成的数据线性关系强但添加偏移和缩放后可能被掩盖。检查数据生成函数中的缩放参数。确保bias和noise在合理范围内。可以手动构造更有逻辑关系的特征。“启用差分隐私”后模型精度急剧下降。隐私预算epsilon设置过小添加的噪声过大。逐步增大epsilon值例如从1.0到5.0在隐私和模型效用之间寻找平衡点。需要对噪声的尺度有理论估算。踩坑心得状态管理是关键Streamlit脚本在每次交互后都会从头到尾重新执行。必须使用st.session_state来持久化存储模型、历史数据等关键变量否则训练结果会丢失。理解“数据非独立同分布”这是联邦学习项目成败的核心。你的模拟数据必须足够“坏”才能测试出算法的有效性。如果所有客户端数据分布一模一样那联邦学习就和集中式训练没区别了。可视化驱动开发在Streamlit中边写代码边看界面效果是非常高效的开发方式。多利用st.write()、st.dataframe()和st.json()来调试中间变量。从简单开始先让最简单的FedAvg在理想数据上跑通再加入差分隐私、非独立同分布数据、更复杂的模型等高级特性。迭代开发步步为营。本文还有配套的精品资源点击获取