机器学习流量识别(pytorch+NSL-KDD+多分类建模)

本文主要实现以下功能，会提供完整的可运行的代码以及解释为什么这么设计。文章不会收费，若被限制查看，请私信我。
 

使用 NSL-KDD 数据集的CSV文件进行流量攻击检测，使用机器学习算法实现流量攻击检测，使用pytorch框架+采用多分类问题进行建模，进行模型训练和评估，采用四大评估指标，使得KDDTest+测试集的准确率达到98%以上。

本文旨在帮助0基础小白快速构建一个机器学习的项目，也是我初入人工智能领域的第一个脚步的记录。

在我看来。机器学习说白了就是一个劲的给电脑投喂数据，建立一个数据-->分类的映射函数，也就是所谓的模型，没有见过的数据电脑也不容易分辨出来，只会把它分类为未知或待处理这种类型。

事先声明，如果不把测试集的数据合并到训练集里，单纯使用训练集训练，然后测试集测试，采用我这个方法肯定达不到98%正确率以上，因为测试集里有相当一部分是训练集里没有的数据类型，会拉低总体正确率。

用本文代码实现功能或觉得对你有所帮助的请点个赞谢谢。

一、概念详解

这些概念旨在让你看懂这篇文章到底用了什么技术，没有写的很详细，因为不影响实现功能，只要大概指导这些名词值的是什么东西就可以了，可以做的时候回顾一下。

（一）CSV文件

CSV文件：逗号分隔值，是一种纯文本文件，用于存储和交换表格数据，以纯文本方式存储数据，每一行代表一条记录，各个字段用逗号分隔。

（二）NSL-KDD数据集

NSL-KDD数据集：网络入侵检测领域的标准数据集，每条记录包含41个特征和1个标签总共是42个字段，数据集分为训练集和测试集
特征分类：基本特征、内容特征、时间统计特征、主机统计特征
攻击类型分类：拒绝服务攻击、远程主机未授权访问、本地用户未授权提取、探测与扫描攻击
训练集：KDDTrain+.txt,包含123973条数据，正常样本6734，攻击样本119239
测试集：KDDTest+.txt，22544条数据，正常1797条，攻击20747条

数据集的CSV文件是.txt命名，但是内容是CSV格式，早期是这样命名的，能被CSV解析工具解析就行了，不用太在意

（三）机器学习算法

常用的机器学习算法：
监督学习算法（有标注数据（带标签））：
        1. 分类算法：逻辑回归、K近邻、决策树、随机森林、支持向量机、神经网络
        2. 回归算法：线性回归、岭回归、Lasso回归
无监督学习算法（无标注数据（不带标签））：
        1. 聚类算法：K均值聚类、层次聚类、DBSCAN
        2. 降维算法：主成分分析、t-SNE
半监督学习算法（部分数据标注）：标签传播
强化学习算法（通过交互学习）：Q学习、深度Q网络
集成学习方法（组合多个模型）：Boosting、Bagging

（四）多分类问题

多分类问题：是一种监督学习算法的问题，核心特点是将样本划分到三个或更多的类别中。
比如给定的NSL-KDD数据集，每一条数据有41个特征和1个标签，模型需要学习一个映射函数，将新输入的样本准确分配到三个或更多的互斥类别里，每个样本只能属于一个类别，所以模型的输出就是K个类别上的概率分布问题。
多分类问题的评估指标：
准确率：正确分类的样本数占总样本数的比例
精确率：模型预测为该类的样本中，实际属于该类的比例
混淆矩阵：展示各类别预测正确 / 错误的数量
宏平均和微平均：宏平均：计算单个类别的指标后取平均；微平均：将所有类别样本合并计算指标

用多分类问题建模：将现实中的分类需求转化为机器学习任务，通过算法让模型学会区分三个或更多类别的样本。即通过算法构造一个模型，这个模型可以把NSL-KDD的每一条数据样本划分为三个或以上的类别。

（五）四大评估指标

四大评估指标
1、准确率
    正确分类的样本数占总样本数的比例
2、精确率
针对某一类别，模型预测为该类的样本中，实际属于该类的比例，衡量模型对某类的 “误判率”
模型预测100个B类，里面80个对的，20个其他类，精确率80%
3、召回率
模型正确预测为该类的样本数占该类实际样本数的比例，衡量模型对某类的“漏判率
100个B类里，模型预测对80个，20个判为其他类，召回率80%
4、F1分数
精确率和召回率的调和平均，综合衡量模型在某类别上的性能
B类的精确率为85%，召回率为90%，则F1B=2*（0.85*0.9）/（0.85+0.9）≈87.4%。两者都高时，F1才高

（六）pytorch框架

Pytorch框架：开源的深度学习框架
核心特点
1.动态计算图：运行时构建计算图，一边运行代码一边构建
2.张量计算：支持张量操作（矩阵乘法、卷积、规约），无缝衔接NumPy
3.自动微分
4.模块化设计：torch.nn：提供定义的神经网络层和损失函数；torch.potim:实现优化算法 ；torch.utils.data:包含数据加载器(DateLoader)和数据集类(Dataset),简化数据预处理。

常用工具库
1.torchvision：提供预训练模型、常用数据集和图像处理工具
2.torchtext：自然语言处理，包含文本预处理、词向量和预训练模型
3.torchaudio：音频处理

使用流程
1.定义模型
通过继承torch.nn.Moudle创建自定义网络
2.数据准备 
使用torch.utils.data.Dataset和DataLoader加载和批处理数据。
3.训练模型
定义损失函数、优化器，迭代训练数据
4.评估与推理
使用训练好的模型进行预测或评估

二、环境配置

安装一个vscode，在vscode的扩展里安装python解释器，用于运行python文件

接下来你可以选择在电脑里安装python3.9以上的版本，或者使用conda配置虚拟环境后再在conda虚拟环境里安装Python3.9以上的版本。这两种方法没区别，一个是在windows的环境里一个是在虚拟环境而已，看你喜欢哪个，我建议都试试，多扩展眼界。

不出意外你会安装vscode扩展里的中文翻译器，点击左上角查看-->终端，打开终端，在你的工作环境底下安装所需依赖：pip install pandas numpy torch torchvision scikit-learn matplotlib seaborn flask joblib tqdm pyyaml

可能会有些依赖缺失了，因为代码是我缝缝补补挺多次的，到时候有啥依赖问题下载就行了，下载的慢就在pip下载语句后面添加：-i https://pypi.tuna.tsinghua.edu.cn/simple       使用国内源下载。

至于NSL-KDD的数据集，EDGE浏览器随便搜会有git仓库提供下载的，我这里提供一个，不保证以后会不会失效：GitHub - HoaNP/NSL-KDD-DataSet

pandas：用于加载NSL-KDD数据集、特征工程和数据预处理 numpy：处理数组形式的特征数据、模型输入输出的数值计算 torch：构建多分类神经网络模型、定义模型结构、训练和推理过程 torchvision：图像化流量分析 scikit-learn：数据预处理的流水线构建、模型评估指标计算、和pytorch模型结果对比 matplotlib：绘制曲线图 seaborn：绘制矩阵 flask：创建流量分类预测API joblib：保存训练好的pytorch模型 tqdm：模型训练、数据处理中显示当前进度 pyyaml：引用yaml文件

三、代码目录框架

目录结构如下，照着目录结构一个个创建文件，然后用vscode打开文件夹即可

traffic-classification-pytorch

---->config---->model_config.yaml

---->data---->把NSL-KDD的KDDTest+.txt和KDDTrain+.txt放进来

---->models---->label_mapping.pkl    lightgbm_model.txt    preprocessor.pkl    traffic_model.pth

---->notebooks---->data_analysis.ipynb

---->reports---->confusion_matrix.png    classification_report.txt    

---->src---->_pycache_      data_loader.py    evaluator.py    model.py   trainer.py    utils.py

---->main.py

四、具体实现

除了下面提到的文件，上面框架里的其它文件都不用管，会自动生成的，代码中包含了大量注释，当然我也比较懒，就不在每个文件前描述了，嫌麻烦的直接复制粘贴运行，报错了再慢慢看注释即可

（一）main.py

import os
import torch
#导入src目录下的文件中的函数
from src.data_loader import load_data, prepare_data 
from src.model import get_model
from src.trainer import train_model, save_model
from src.evaluator import evaluate_model
from src.utils import load_config, ensure_dirs
"""X为特征，Y为标签"""def main():# 确保目录存在，不存在的话直接就在函数里创建ensure_dirs(['models', 'reports'])# 加载配置config = load_config()epochs = config.get('epochs', 50)#从配置字典中获取训练轮数（epochs），若不存在则使用默认值 50。lr = config.get('learning_rate', 0.001)#同上# 加载数据print("加载数据...")#starttrain_df, test_df = load_data('data', combine_and_split=True, test_size=0.2, random_state=42)# 合并并重新划分# train_df, test_df = load_data('data')#从data路径读取数据集并拆分为训练集和测试集，函数返回两个元素的列表"""预计输出：训练集形状: (125973, 42)测试集形状: (22544, 42)。label列单独被提取，"""# 准备数据，多分类处理，返回8个值。将原始数据转换为模型可用的格式。print("数据预处理...")(train_loader, test_loader, X_train, y_train, #pytorch数据加载器、特征矩阵（输入）、标签向量（输出)、预处理流水线、标签到索引的映射字典X_test, y_test, preprocessor, label_to_idx) = prepare_data(train_df, test_df)# 获取模型，准备模型训练所需的参数并获取模型实例input_dim = X_train.shape[1]#获取特征维度（输入层大小），获取特征矩阵的列数。类别特征编码，比如service类别有70种特征，被独热编码为70列。num_classes = len(set(y_train))#获取类别数量（输出层大小）,计算唯一标签的数量,set(y_train)：将标签转换为集合（去重）。23个不同类别的标签。print(f"输入维度: {input_dim}, 类别数: {num_classes}")model = get_model(input_dim, num_classes)# get_model() 函数创建模型# 训练模型print("训练模型...")device = torch.device("cuda" if torch.cuda.is_available() else "cpu")#设备配置，若GPU可用则用GPU，否则使用cpu   torch.device()：创建一个表示设备的对象model, best_acc = train_model(                  #函数返回值model:训练好的pytorch模型      best_acc:浮点数，验证准确率model, train_loader, test_loader, num_classes, #调用 train_model 函数，传入模型、数据加载器和训练参数epochs=epochs, lr=lr#num_classes:分类任务的类别数   epochs：训练轮数    lr：学习率，控制模型参数更新的幅度)# 保存模型save_model(model)# 评估模型print("评估模型...")metrics = evaluate_model(model, test_loader, y_test, label_to_idx, device)# 检查准确率是否达到98%if metrics['accuracy'] >= 0.98:print(f"成功达到目标准确率: {metrics['accuracy']:.4f}")else:print(f"当前准确率: {metrics['accuracy']:.4f}, 未达到98%目标")print("建议调整模型结构或超参数")# 保存评估结果with open('reports/classification_report.txt', 'w') as f:f.write("===== 评估指标 =====\n")f.write(f"准确率 (Accuracy): {metrics['accuracy']:.4f}\n")f.write(f"宏平均精确率 (Macro Precision): {metrics['macro_precision']:.4f}\n")f.write(f"宏平均召回率 (Macro Recall): {metrics['macro_recall']:.4f}\n")f.write(f"宏平均F1分数 (Macro F1): {metrics['macro_f1']:.4f}\n")f.write(f"宏平均ROC-AUC: {metrics['roc_auc_macro']:.4f}\n")f.write("\n===== 各类别评估指标 =====\n")for label, metrics_dict in metrics['class_metrics'].items():f.write(f"{label}:\n")f.write(f"  精确率: {metrics_dict['precision']:.4f}\n")f.write(f"  召回率: {metrics_dict['recall']:.4f}\n")f.write(f"  F1分数: {metrics_dict['f1']:.4f}\n")f.write(f"  ROC-AUC: {metrics_dict['roc_auc']:.4f}\n")if __name__ == "__main__":main()

（二）data_loader.py 

#数据加载与预处理
import os
from sklearn.model_selection import train_test_split
import torch
import joblib
import pandas as pd
import numpy as npfrom sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from torch.utils.data import Dataset, DataLoader
from sklearn.preprocessing import OneHotEncoder, MinMaxScalerclass NSLKDDDataset(Dataset):#NSLKDDDataset继承自 PyTorch 的torch.utils.data.Dataset基类。封装特征和标签，提供统一的数据访问接口"""NSL-KDD数据集加载器"""def __init__(self, features, labels):#将 NumPy 数组转换为 PyTorch 张量self.features = torch.FloatTensor(features)#特征转为 32 位浮点数。# 将 Pandas Series 转换为 NumPy 数组if isinstance(labels, pd.Series):labels = labels.valuesself.labels = torch.LongTensor(labels)#标签转为 64 位整数（多分类要求）. torch.LongTensor() 不直接支持 Series,所以在上一步添加转换def __len__(self):#返回数据集大小（样本数）return len(self.labels)def __getitem__(self, idx):#支持通过索引访问样本。返回元组(features, label)，适配 PyTorch 模型输入return self.features[idx], self.labels[idx]def load_data(data_dir, combine_and_split=True, test_size=0.2, random_state=42):"""从data_dir目录加载Train训练集和Test测试集"""# 定义数据集的列名columns = ['protocol_type', 'service', 'flag', 'src_bytes', 'dst_bytes','land', 'wrong_fragment', 'urgent', 'hot', 'num_failed_logins','logged_in', 'num_compromised', 'root_shell', 'su_attempted','num_root', 'num_file_creations', 'num_shells', 'num_access_files','num_outbound_cmds', 'is_host_login', 'is_guest_login', 'count','srv_count', 'serror_rate', 'srv_serror_rate', 'rerror_rate','srv_rerror_rate', 'same_srv_rate', 'diff_srv_rate','srv_diff_host_rate', 'dst_host_count', 'dst_host_srv_count','dst_host_same_srv_rate', 'dst_host_diff_srv_rate','dst_host_same_src_port_rate', 'dst_host_srv_diff_host_rate','dst_host_serror_rate', 'dst_host_srv_serror_rate','dst_host_rerror_rate', 'dst_host_srv_rerror_rate', 'label', 'duration']# 构建文件路径，加载训练集和测试集train_file = os.path.join(data_dir, 'KDDTrain+.txt')test_file = os.path.join(data_dir, 'KDDTest+.txt')# 读取数据，header=None表示文件无表头，names指定列名train_df = pd.read_csv(train_file, header=None, names=columns)test_df = pd.read_csv(test_file, header=None, names=columns)if combine_and_split:# 合并数据集combined_data = pd.concat([train_df, test_df], axis=0)print(f"合并后数据集大小: {len(combined_data)} 条记录")# 强制转换duration为数值类型combined_data['duration'] = pd.to_numeric(combined_data['duration'], errors='coerce')# 清洗label列combined_data['label'] = combined_data['label'].str.split(',').str[0]# 重新排列列顺序，将duration移到第一列columns = ['duration'] + columns[:-1]combined_data = combined_data[columns]# 手动划分数据X = combined_data.drop('label', axis=1)y = combined_data['label']X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=random_state)train_df = pd.concat([X_train, y_train], axis=1)test_df = pd.concat([X_test, y_test], axis=1)else:# 强制转换duration为数值类型train_df['duration'] = pd.to_numeric(train_df['duration'], errors='coerce')test_df['duration'] = pd.to_numeric(test_df['duration'], errors='coerce')# 清洗label列train_df['label'] = train_df['label'].str.split(',').str[0]test_df['label'] = test_df['label'].str.split(',').str[0]# 重新排列列顺序，将duration移到第一列columns = ['duration'] + columns[:-1]train_df = train_df[columns]test_df = test_df[columns]print(f"训练集形状: {train_df.shape}")print(f"测试集形状: {test_df.shape}")return train_df, test_dfdef create_preprocessor():"""创建数据预处理流水线"""# 定义特征类型categorical_features = ['protocol_type', 'service', 'flag']#分类特征，需转换为数值,三分类numerical_features = [ #数值特征，需标准化'duration', 'src_bytes', 'dst_bytes', 'land', 'wrong_fragment', 'urgent', 'hot', 'num_failed_logins', 'logged_in', 'num_compromised','root_shell', 'su_attempted', 'num_root', 'num_file_creations','num_shells', 'num_access_files', 'count', 'srv_count', 'serror_rate','srv_serror_rate', 'rerror_rate', 'srv_rerror_rate', 'same_srv_rate','diff_srv_rate', 'srv_diff_host_rate', 'dst_host_count', 'dst_host_srv_count', 'dst_host_same_srv_rate', 'dst_host_diff_srv_rate','dst_host_same_src_port_rate', 'dst_host_srv_diff_host_rate','dst_host_serror_rate', 'dst_host_srv_serror_rate','dst_host_rerror_rate', 'dst_host_srv_rerror_rate']# 创建预处理流水线，确保数据格式的一致性和模型输入的有效性preprocessor = Pipeline([ #Pipeline（）将多个数据处理步骤串联为单一对象，[('步骤名', 转换器), ...]。('column_transformer', ColumnTransformer([# ColumnTransformer()对不同列应用不同的转换器。下面的参数remainder='drop'：丢弃未指定的列('cat', OneHotEncoder(drop='first', handle_unknown='ignore'), categorical_features),# OneHotEncoder()将分类特征转换为二进制向量。protocol_type（TCP/UDP/ICMP）→ 3 个二进制列，drop='first'：删除第一个类别，handle_unknown='ignore'：忽略未知类别。默认输出稀疏矩阵('num', MinMaxScaler(), numerical_features)#数值特征处理，保留数据的比例，X_scaled = (X - X_min) / (X_max - X_min)], remainder='drop')),])return preprocessor
#实现多分类任务，保留原始标签类别
def prepare_data(train_df, test_df, save_preprocessor=True):"""准备训练和测试数据"""# 提取特征和标签 (多分类标签)。X为特征，Y为标签X_train = train_df.drop('label', axis=1)#从train_fd中删除label列，axis=1删除列，axis=0删除行y_train = train_df['label']#从train_df中单独提取label列X_test = test_df.drop('label', axis=1)y_test = test_df['label']# 标签编码，输出形式为整数索引如0、1。# 获取唯一标签类别并排序，sorted()保证映射顺序确定性，用于双向转换all_labels = sorted(y_train.unique())#提取训练集中所有唯一的标签值，按字母顺序排序label_to_idx = {label: i for i, label in enumerate(all_labels)}#构建字典，将每个标签映射到一个唯一的整数索引# 转换标签为数字# y_train = y_train.map(label_to_idx).values# y_test = y_test.map(label_to_idx).valuesy_train = y_train.map(label_to_idx).fillna(0).astype(int)  # 处理缺失标签y_test = y_test.map(label_to_idx).fillna(0).astype(int)# 创建并拟合预处理流水线preprocessor = create_preprocessor()#函数返回一个pipeline对象，包含数值特征的标准化和分类特征的编码X_train_processed = preprocessor.fit_transform(X_train)#在训练集上拟合预处理参数（如均值、标准差），并应用转换X_test_processed = preprocessor.transform(X_test)#使用训练集的参数转换测试集（避免数据泄露）# 保存预处理流水线，确保新数据的预处理方式与训练数据一致if save_preprocessor:joblib.dump(preprocessor, 'models/preprocessor.pkl')#使用joblib库将预处理流水线化为二进制文件joblib.dump(label_to_idx, 'models/label_mapping.pkl')# 创建PyTorch数据集和数据加载器train_dataset = NSLKDDDataset(X_train_processed, y_train)test_dataset = NSLKDDDataset(X_test_processed, y_test)train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)#每次加载 64 条样本.shuffle=True：训练集随机打乱，提高泛化能力test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)#shuffle=False：测试集保持顺序，便于结果比对。return (train_loader, test_loader, X_train_processed, y_train, X_test_processed, y_test, preprocessor, label_to_idx)

（三）evaluator.py 

#模型评估
import torch
import joblib
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
import matplotlib.font_manager as fmfrom sklearn.metrics import confusion_matrix
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_scoredef evaluate_model(model, test_loader, y_true, label_to_idx, device):"""评估模型性能"""model.eval()#设置模型为评估模式all_preds = []all_probs = []all_labels = []with torch.no_grad(): # 关闭梯度计算for features, labels in test_loader:features, labels = features.to(device), labels.to(device)outputs = model(features)_, preds = torch.max(outputs, 1)#将数据从 GPU 转回 CPU 并转换为 NumPy 数组all_preds.extend(preds.cpu().numpy())all_probs.extend(torch.softmax(outputs, 1).cpu().numpy())all_labels.extend(labels.cpu().numpy())all_probs = np.array(all_probs)all_preds = np.array(all_preds)all_labels = np.array(all_labels)# 获取测试集中实际出现的类别unique_classes = np.unique(all_labels)num_classes = len(label_to_idx)# 计算四大评估指标accuracy = accuracy_score(all_labels, all_preds)#准确率# 处理多分类的精确率、召回率和F1分数precision = precision_score(all_labels, all_preds, average=None,zero_division=1)#zero_division=0（默认）：分母为 0 时设为 0，发出警告     zero_division=1：分母为 0 时设为 1，不发出警告recall = recall_score(all_labels, all_preds, average=None,zero_division=1)f1 = f1_score(all_labels, all_preds, average=None,zero_division=1)# 计算宏平均和微平均precision_macro = precision_score(all_labels, all_preds, average='macro',zero_division=1)#计算每个类别的指标后取算术平均，适合类别不平衡场景recall_macro = recall_score(all_labels, all_preds, average='macro',zero_division=1)f1_macro = f1_score(all_labels, all_preds, average='macro',zero_division=1)precision_micro = precision_score(all_labels, all_preds, average='micro',zero_division=1)#将所有类别样本合并计算指标，适合关注整体性能场景recall_micro = recall_score(all_labels, all_preds, average='micro',zero_division=1)f1_micro = f1_score(all_labels, all_preds, average='micro',zero_division=1)# 计算ROC-AUC (多分类情况)roc_auc = {}for class_idx in unique_classes:# 对每个类别计算One-vs-Rest的AUCy_onehot = np.zeros((len(all_labels), num_classes))y_onehot[np.arange(len(all_labels)), all_labels] = 1try:roc_auc[class_idx] = roc_auc_score(y_onehot[:, class_idx], all_probs[:, class_idx])except:roc_auc[class_idx] = 0.5  # 处理只有一类的情况roc_auc_macro = np.mean(list(roc_auc.values())) if roc_auc else 0.5# 打印评估指标print("\n===== 评估指标 =====")print(f"准确率 (Accuracy): {accuracy:.4f}")print(f"宏平均精确率 (Macro Precision): {precision_macro:.4f}")print(f"宏平均召回率 (Macro Recall): {recall_macro:.4f}")print(f"宏平均F1分数 (Macro F1): {f1_macro:.4f}")print(f"微平均精确率 (Micro Precision): {precision_micro:.4f}")print(f"微平均召回率 (Micro Recall): {recall_micro:.4f}")print(f"微平均F1分数 (Micro F1): {f1_micro:.4f}")print(f"宏平均ROC-AUC: {roc_auc_macro:.4f}")# 打印每个类别的指标idx_to_label = {v: k for k, v in label_to_idx.items()}print("\n===== 各类别评估指标 =====")for class_idx in unique_classes:class_name = idx_to_label.get(class_idx, f"未知类别_{class_idx}")print(f"{class_name}:")# 找到该类别在评估指标数组中的位置metric_idx = np.where(unique_classes == class_idx)[0][0]print(f"  精确率: {precision[metric_idx]:.4f}")print(f"  召回率: {recall[metric_idx]:.4f}")print(f"  F1分数: {f1[metric_idx]:.4f}")print(f"  ROC-AUC: {roc_auc.get(class_idx, 0.5):.4f}")# 处理测试集中未出现的类别missing_classes = set(label_to_idx.values()) - set(unique_classes)if missing_classes:print("\n===== 测试集中未出现的类别 =====")for class_idx in missing_classes:class_name = idx_to_label.get(class_idx, f"未知类别_{class_idx}")print(f"{class_name}:")print(f"  精确率: 0.0000")print(f"  召回率: 0.0000")print(f"  F1分数: 0.0000")print(f"  ROC-AUC: 0.5000")# 绘制混淆矩阵（需要调整以处理实际类别）plot_confusion_matrix(all_labels, all_preds, idx_to_label, unique_classes)return {'accuracy': accuracy,'macro_precision': precision_macro,'macro_recall': recall_macro,'macro_f1': f1_macro,'micro_precision': precision_micro,'micro_recall': recall_micro,'micro_f1': f1_micro,'roc_auc_macro': roc_auc_macro,'class_metrics': {idx_to_label[class_idx]: {'precision': precision[np.where(unique_classes == class_idx)[0][0]] if class_idx in unique_classes else 0.0,'recall': recall[np.where(unique_classes == class_idx)[0][0]] if class_idx in unique_classes else 0.0,'f1': f1[np.where(unique_classes == class_idx)[0][0]] if class_idx in unique_classes else 0.0,'roc_auc': roc_auc.get(class_idx, 0.5)} for class_idx in label_to_idx.values()}}def plot_confusion_matrix(y_true, y_pred, label_dict, unique_classes):# 自动查找系统中已安装的中文字体chinese_fonts = []for font in fm.fontManager.ttflist:# 检查字体是否支持中文（通过名称包含中文或支持CJK字符）if "SimHei" in font.name or "Microsoft YaHei" in font.name or any(ord(c) > 127 for c in font.name):chinese_fonts.append(font.name)if chinese_fonts:plt.rcParams["font.family"] = chinese_fontsprint(f"使用字体: {chinese_fonts[0]}")else:plt.rcParams["font.family"] = ["sans-serif"]print("警告：未找到中文字体，使用默认字体")"""绘制混淆矩阵（支持实际出现的类别）"""cm = confusion_matrix(y_true, y_pred)# 确保标签顺序与实际出现的类别一致labels = [label_dict[i] for i in sorted(unique_classes)]plt.figure(figsize=(12, 10))sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=labels, yticklabels=labels)plt.xlabel('预测标签')plt.ylabel('真实标签')plt.title('混淆矩阵')plt.tight_layout()plt.savefig('reports/confusion_matrix.png')plt.close()

（四）model.py

#pytorch模型定义
import torch
import torch.nn as nn
import torch.nn.functional as Fclass TrafficClassifier(nn.Module):"""网络流量分类模型"""def __init__(self, input_dim, num_classes):super(TrafficClassifier, self).__init__()# 定义神经网络层self.fc1 = nn.Linear(input_dim, 256)#四层全连接神经网络（fc1-fc4），输入维度input_dim→输出 256，完成输入特征到隐藏层的映射self.bn1 = nn.BatchNorm1d(256)       #3个批量归一化层（bn1-bn3），对每个隐藏层的输出进行归一化，加速训练并提高稳定性self.fc2 = nn.Linear(256, 128)      #256→128，进一步提取特征self.bn2 = nn.BatchNorm1d(128)self.fc3 = nn.Linear(128, 64)       #128→64，降低特征维度self.bn3 = nn.BatchNorm1d(64)self.fc4 = nn.Linear(64, num_classes)#64→num_classes，输出分类结果（23 个类别）self.dropout = nn.Dropout(0.2)  # Dropout 层（防止过拟合）0.3 可调整为 0.2 或 0.4，视过拟合情况而定def forward(self, x):"""前向传播"""x = F.relu(self.bn1(self.fc1(x)))x = self.dropout(x)x = F.relu(self.bn2(self.fc2(x)))x = self.dropout(x)x = F.relu(self.bn3(self.fc3(x)))x = self.fc4(x)return xdef get_model(input_dim, num_classes):"""获取模型实例"""model = TrafficClassifier(input_dim, num_classes)return model

（五） trainer.py

#模型训练
import os
import torch
import numpy as np
import torch.nn as nn
import torch.optim as optimfrom tqdm import tqdm
from torch.optim.lr_scheduler import ReduceLROnPlateaudef train_model(model, train_loader, val_loader, num_classes, epochs=50, lr=0.001):"""训练模型"""device = torch.device("cuda" if torch.cuda.is_available() else "cpu")model = model.to(device) #模型迁移至指定设备，模型将在GPU或CPU上运行。返回迁移后的模型# 定义损失函数和优化器criterion = nn.CrossEntropyLoss()#损失函数，计算模型预测与真实标签之间的损失optimizer = optim.Adam(model.parameters(), lr=lr)#优化器，使用 Adam 优化算法，训练初期收敛快，model.parameters()参数为需要优化的模型参数scheduler = ReduceLROnPlateau(optimizer, 'min', patience=5)#学习调度器， 当指标（如损失或准确率）停滞时，自动降低学习率#min代表指标越小越好，patience验证集损失连续 5 轮未下降，学习率将降低。verbose=True参数不支持，pytorch版本低于1.10# 记录最佳验证准确率best_val_acc = 0.0#记录验证集上的最高准确率，记录最佳状态可防止使用性能退化的模型best_model_weights = None#保存达到该准确率时的模型参数。for epoch in range(epochs):# 训练模式model.train()train_loss = 0.0correct = 0total = 0for features, labels in tqdm(train_loader, desc=f"Epoch {epoch+1}/{epochs}"):# 数据到设备features, labels = features.to(device), labels.to(device)# 前向传播optimizer.zero_grad() # 清空梯度，防止梯度累加，确保每次迭代的梯度仅来自当前批次outputs = model(features)  # 模型预测loss = criterion(outputs, labels)# 计算损失# 反向传播和优化loss.backward()# 计算梯度，从损失函数反向传播，计算每个可训练参数的梯度optimizer.step() # 更新参数，根据优化器算法（Adam）更新模型参数train_loss += loss.item()_, predicted = outputs.max(1)   #max(1)返回每一行的最大值及索引total += labels.size(0)correct += predicted.eq(labels).sum().item()# 验证模式model.eval()#关闭 Dropout 和 BatchNorm 的训练模式。Dropout 在验证时不随机丢弃神经元。BatchNorm 使用训练集的统计量（均值、方差）val_loss = 0.0val_correct = 0val_total = 0with torch.no_grad():#停止梯度计算，节省内存并加速计算。验证阶段无需更新参数，无需梯度信息for features, labels in val_loader:features, labels = features.to(device), labels.to(device)outputs = model(features)loss = criterion(outputs, labels)val_loss += loss.item()_, predicted = outputs.max(1)val_total += labels.size(0)val_correct += predicted.eq(labels).sum().item()# 计算平均损失和准确率train_loss = train_loss / len(train_loader)train_acc = 100.0 * correct / totalval_loss = val_loss / len(val_loader)val_acc = 100.0 * val_correct / val_total# 基于验证损失调整学习率，每轮验证后，若验证损失未下降，学习率降低，步长变小，精细化的调整参数scheduler.step(val_loss)print(f"Epoch {epoch+1}, Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, "f"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%")# 保存最佳模型if val_acc > best_val_acc:best_val_acc = val_accbest_model_weights = model.state_dict()print(f"保存最佳模型，验证准确率: {best_val_acc:.2f}%")# 加载最佳模型权重if best_model_weights:model.load_state_dict(best_model_weights)return model, best_val_accdef save_model(model, path='models/traffic_model.pth'):"""保存模型"""torch.save(model.state_dict(), path)print(f"模型已保存至 {path}")

（六）utils.py

#工具函数定义
import os
import yaml
import joblibdef load_config(config_path='config/model_config.yaml'):    #用于加载YAML格式的配置文件，config_path参数指定文件路径，不传入参数也可，会默认读取该文件"""加载配置文件"""with open(config_path, 'r',encoding='utf-8') as f:   #以读取方式打开文件，将文件对象赋值给变量f。操作完文件后自动关闭，无需f.close()config = yaml.safe_load(f)  #使用safe_load方法解析文件内容，解析后转化为字典赋值给config变量。依赖yaml库。return configdef ensure_dirs(dirs):  #传入一个目录路径的列表"""确保目录存在"""for dir_path in dirs:   #循环遍历dirs列表下的每个路径if not os.path.exists(dir_path):    #使用os.path.exists()函数检查目录是否存在，不存在返回falseos.makedirs(dir_path)   #不存在则创建目录

（七）model_config.yaml

epochs: 5   #训练模型时，数据被遍历5轮次
learning_rate: 0.001  #学习率，控制模型参数更新的幅度。
batch_size: 64  #每次迭代使用64条数据组成一个批次。1W条数据，需要迭代约157次=10000/64#一般先固定batch_size，调整learning_rate，观察损失函数曲线，若早期收敛停滞，可增大epochs或调整learning_rate
#损失函数：衡量模型预测结果与真实目标之间差异。收敛越慢代表差异越大。收敛越快代表差异逐步缩小。损失函数结果越小越接近真实值。
#优化模型就是最小化损失函数

五、核心流程梳理

（一）核心流程梳理

加载模型训练配置→训练次数、学习率、数据批次

        ↓

加载数据集→定义列名→构建文件路径→读取文件→数值转换

       ↓

数据预处理→提取特征和标签→标签编码→创建保存流水线→创建pytorch数据集和数据加载器

       ↓

获取模型→获取维度和类别数→构造四层全连接神经网络层和三层批量归一层，根据输入维度，实现输入特征到隐藏层的映射，在隐藏层批量归一处理，完成输入特征到类别的映射。

       ↓

训练模型→设备配置→损失函数与优化器定义→学习率调度器→训练循环

       ↓                 ↓                        ↓                               ↓                        ↓

保存模型    GPU/CPU      交叉熵、Adam                 调整学习效率  根据配置次数

       ↓

评估模型→数据准备→批量预测→指标计算→ROC-AUC计算→结果输出→绘图

       ↓

保存评估结果

（二）学习率

学习率（通常用α或lr表示）是梯度下降算法中更新模型参数的步长。控制模型在参数空间中搜索最优解的 "步长"         ---> 缩放梯度，控制实际步长

梯度是损失函数对参数的偏导数向量         --->提供更新方向和基础步长建议

步长是参数在每次迭代中实际更新的距离         ---> 最终决定参数更新的幅度

损失函数计算模型预测与真实标签之间的损失

结合一下，梯度方向就是是损失函数增长最快的方向，梯度大小代表函数变化的陡峭程度。

而学习率控制参数更新的幅度，将梯度方向的理论上步长缩放为实际更新步长

调整学习率--->调整梯度--->控制步长--->步长大损失函数发散，步长小训练时间久

粗略理解为学习率控制机器学习的实际效率，如减小损失、减小学习时间。

六、答疑

1.这个项目使用了哪些pytorch框架的基本用法

模型定义：四层全连接层、三层批量归一化层、dropout层

数据处理：将numpy数组转为pytorch张量，使用torch的方法封装数据集和创建数据加载器

模型训练：训练过程使用pytorch的方法来进行选择设备、定义损失函数和优化器、反向传播、更新模型参数等。

模型评估：使用torch的方法关闭梯度计算和获取预测结果

2.使用什么类型的具体的机器学习算法，在哪个位置

使用的深度学习算法中的全连接神经网络，在model.py文件中

3.采用什么多分类问题建模，分为几类，在哪里分类的

建模方式：全连接神经网络

分类数量：由数据集的唯一标签确定，这个项目中是23类

在model.py最后一层全连接层输出的维度那里

4.采用什么模型来进行训练和评估

四层全连接神经网络模型

5.四大评估指标是怎么计算的，如何计算的

准确率、精确率、召回率和 F1 分数，在evaluator.py，sklearn.metrics库的相关函数

6.项目整体流程，简述一下

查看第五点核心流程梳理

7.反向传播和前向传播

前向传播是神经网络中信息从输入层到输出层的流动过程

       ↓

计算损失

       ↓

反向传播：从输出层开始，根据链式法则，将损失值反向传播到网络的每一层，计算每个权重和偏置对损失的梯度。

参数更新

8.dropout层，有啥用

Dropout 层是一种正则化技术，用于防止神经网络过拟合。过拟合是指模型在训练数据上表现良好，但在测试数据上表现不佳的现象。

就是防止训练的数据曲线过于贴近训练集。每次前向传播时，每个神经元都有一定的概率（通常是 0.2 - 0.5）被丢弃

9.全连接神经网络

全连接神经网络（Fully Connected Neural Network，简称 FCN），也称为多层感知机（Multilayer Perceptron，简称 MLP），是一种最基本的神经网络结构。

每一层的每个神经元都与下一层的所有神经元相连，因此称为全连接

网络通常由输入层、若干个隐藏层和输出层组成。输入层接收原始数据，隐藏层对数据进行特征提取和转换，输出层输出最终的预测结果。

四层：输入层→隐藏层1→隐藏层2→输出层

10.sklearn.metrics库

是Scikit-learn 库中的一个模块，提供了各种用于评估机器学习模型性能的指标和工具

11.宏平均和微平均

宏平均：分别计算每个类别的评估指标（如精确率、召回率、F1 分数等），然后对这些指标取算术平均值。宏平均平等对待每个类别，不考虑类别之间的样本数量差异

微平均：将所有类别的样本合并在一起，计算总的评估指标。微平均更关注整体的性能