企业官网建设流程全解析

简介：本资源是一套面向高校计算机类专业本科生与研究生的毕业设计级实践项目，聚焦联邦学习在教育数据场景下的落地应用——学生成绩预测。项目采用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. 系统核心架构与联邦学习方案选型

一个完整的联邦学习系统，远不止“训练一个模型”那么简单。它需要一套严谨的架构来协调参与者、保障通信安全、处理异构数据。我们的系统设计主要包含以下几个核心组件：

1. 中央协调服务器：这是系统的大脑。它负责初始化全局模型、选择参与每一轮训练的客户端（学校）、接收并聚合客户端上传的模型更新、评估全局模型性能，并将更新后的模型分发给客户端。在Python中，我们可以用Flask或FastAPI轻松搭建一个RESTful API服务器来实现这些功能。
2. 客户端：即各个数据持有方（学校）。每个客户端实例拥有自己的本地数据集。它的职责是：从服务器下载最新的全局模型；用本地数据对模型进行若干轮训练（称为本地训练周期）；计算模型参数的更新量（如梯度差值）；将更新量（或加密后的更新量）上传给服务器。
3. 通信协议与安全模块：这是联邦学习的生命线。我们必须确保客户端与服务器之间传输的模型更新不会被恶意第三方窃取或篡改，同时还要防止服务器从更新中反推出原始数据（隐私攻击）。常见的方案包括同态加密、差分隐私等。在原型阶段，我们可以使用SSL/TLS进行传输加密，并引入差分隐私噪声来初步保护隐私。
4. 可视化平台：基于Streamlit构建的前端界面。它实时从中央服务器拉取训练状态、模型指标和预测结果，并以图表、进度条、数据表格等直观形式展现给用户。

在联邦学习的具体算法选型上，最经典、应用最广的是FedAvg。它的流程非常直观：

• 服务器初始化一个全局模型 $w_0$。
• 在每一轮通信回合 $t$ 中：
  • 服务器随机选择一部分客户端 $S_t$。
  • 服务器将当前全局模型 $w_t$ 发送给每个选中的客户端。
  • 每个客户端 $k$ 用本地数据训练模型，得到本地更新 $w_t^{k}$。
  • 每个客户端将 $w_t^{k}$ 上传至服务器。
  • 服务器聚合所有更新：$w_{t+1} = \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 python=3.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_samples=200, bias=0.0, noise=20.0): """ 为单个客户端生成模拟数据。 特征X可能包括：平均学习时间、作业提交率、课堂互动次数、前期测验成绩等。 目标y是期末成绩。 通过bias参数模拟不同学校的整体水平差异。 """ np.random.seed(42 + client_id) # 确保可复现，同时每个客户端不同 # 生成基本特征 X, y = make_regression(n_samples=num_samples, n_features=5, noise=noise, random_state=client_id) # 为特征赋予实际意义 feature_names = ['study_hours', 'assignment_rate', 'interaction', 'prev_score1', 'prev_score2'] df = pd.DataFrame(X, columns=feature_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, bias=10.0, noise=15.0) # 重点学校，成绩偏高，数据质量好（噪声低） client_data['school_b'] = generate_client_data(2, bias=0.0, noise=25.0) # 普通学校 client_data['school_c'] = generate_client_data(3, bias=-5.0, noise=30.0) # 基础薄弱学校，成绩偏低，数据噪声大

隐私化处理：在联邦学习中，原始数据本身不离开客户端，这已经是最强的隐私保护。但我们还需要保护上传的“模型更新”。一种简单有效的方法是应用差分隐私。我们可以在客户端本地训练后，给要上传的梯度向量添加精心校准的拉普拉斯噪声或高斯噪声。

def add_laplace_noise(gradients, epsilon=0.1, sensitivity=1.0): """向梯度添加拉普拉斯噪声以实现差分隐私。""" noise = np.random.laplace(loc=0.0, scale=sensitivity/epsilon, size=gradients.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_dim=5): 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 x

5.2 客户端本地训练函数

每个客户端需要能够用本地数据训练模型，并返回更新后的模型参数（或梯度）。

def client_train(model, train_loader, epochs=5, lr=0.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(), lr=lr) 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 {epoch+1}, 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_weights

5.4 训练循环模拟

将以上部分组合起来，模拟多轮联邦学习过程。

def simulate_federated_learning(clients_data_dict, num_rounds=10, clients_per_round=2): """ 模拟联邦学习训练循环。 :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()), size=clients_per_round, replace=False) print(f"Selected 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, epochs=3) 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(f"Global 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_idx+1) 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="联邦学习成绩预测平台", layout="wide") 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"学校_{i+1}" df = generate_client_data(i, bias=bias, noise=20.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_width=True) st.metric("学生人数", len(df), f"平均成绩: {df['final_score'].mean():.1f}分") with col2: st.subheader("特征与成绩分布") fig = make_subplots(rows=2, cols=2, subplot_titles=('学习时间 vs 成绩', '作业提交率 vs 成绩', '前期成绩1 vs 成绩', '特征相关性')) # 散点图1 fig.add_trace(go.Scatter(x=df['study_hours'], y=df['final_score'], mode='markers', name='学习时间'), row=1, col=1) # 散点图2 fig.add_trace(go.Scatter(x=df['assignment_rate'], y=df['final_score'], mode='markers', name='提交率'), row=1, col=2) # 散点图3 fig.add_trace(go.Scatter(x=df['prev_score1'], y=df['final_score'], mode='markers', name='前期成绩1'), row=2, col=1) # 热力图 corr = df.corr().round(2) fig.add_trace(go.Heatmap(z=corr.values, x=corr.columns, y=corr.columns, text=corr.values, texttemplate="%{text}", colorscale='RdBu', zmid=0), row=2, col=2) fig.update_layout(height=600, showlegend=False) st.plotly_chart(fig, use_container_width=True) 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_value=0.0001, max_value=0.1, value=0.01, step=0.001, format="%.4f") use_dp = st.checkbox("启用差分隐私保护 (会降低精度)", value=False) dp_epsilon = st.slider("隐私预算 ε (越小越隐私)", 0.1, 5.0, 1.0, 0.1, disabled=not use_dp) with col2: st.subheader("操作") if st.button("🚀 开始联邦训练", type="primary", use_container_width=True): 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_rounds=num_rounds, clients_per_round=clients_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(x=rounds, y=history['global_loss'], mode='lines+markers', name='全局模型损失', line=dict(width=3))) fig1.update_layout(title='全局模型损失随训练轮次的变化', xaxis_title='通信轮次', yaxis_title='损失 (MSE)', template='plotly_white') st.plotly_chart(fig1, use_container_width=True) 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(x=valid_rounds, y=valid_losses, mode='lines+markers', name=client_name)) fig2.update_layout(title='各客户端测试损失变化', xaxis_title='通信轮次', yaxis_title='损失 (MSE)', template='plotly_white') st.plotly_chart(fig2, use_container_width=True) 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()), size=clients_per_round, replace=False) 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(data=go.Heatmap( z=df_participation['Selected'].values.reshape(len(rounds), -1), x=list(st.session_state.client_data_info.keys()), y=rounds, colorscale=[[0, 'lightgray'], [1, 'royalblue']], showscale=False, text=df_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_width=True)

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("预测期末成绩", type="primary"): input_tensor = torch.tensor([[study_hours, assignment_rate, interaction, prev_score1, prev_score2]], dtype=torch.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, prediction+8):.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_width=True) st.metric("全局测试集平均MSE损失", f"{avg_global_loss:.4f}", f"RMSE: {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 性能与隐私优化方向

1. 模型压缩：在客户端和服务器之间传输完整模型参数可能带宽消耗大。可以考虑使用梯度稀疏化、量化或知识蒸馏来减少通信负载。
2. 高级聚合算法：基础的FedAvg对非独立同分布数据敏感。可以研究实现FedProx（添加近端项约束本地更新，防止客户端漂移）、SCAFFOLD（使用控制变量减少客户端差异）等更鲁棒的算法。
3. 个性化联邦学习：我们的目标是得到一个全局通用的模型。但在教育场景下，每个学校可能最终想要一个更适合自己特色的模型。可以探索在联邦学习框架下生成个性化模型的方法。
4. 增强隐私保护：我们只实现了简单的差分隐私。工业级应用需要考虑安全聚合，即服务器在无法解密单个客户端更新的情况下完成聚合。这需要结合同态加密或安全多方计算技术。

7.3 常见问题与排查技巧实录

在实际开发和运行中，你可能会遇到以下问题：

踩坑心得：

• 状态管理是关键：Streamlit脚本在每次交互后都会从头到尾重新执行。必须使用st.session_state来持久化存储模型、历史数据等关键变量，否则训练结果会丢失。
• 理解“数据非独立同分布”：这是联邦学习项目成败的核心。你的模拟数据必须足够“坏”，才能测试出算法的有效性。如果所有客户端数据分布一模一样，那联邦学习就和集中式训练没区别了。
• 可视化驱动开发：在Streamlit中，边写代码边看界面效果是非常高效的开发方式。多利用st.write()、st.dataframe()和st.json()来调试中间变量。
• 从简单开始：先让最简单的FedAvg在理想数据上跑通，再加入差分隐私、非独立同分布数据、更复杂的模型等高级特性。迭代开发，步步为营。

本文还有配套的精品资源，点击获取