☰
CNN+LSTM网络流量检测实战:从模型搭建到数据闭环
2026/10/14 13:06:27 网站建设 项目流程

简介:一套基于卷积神经网络与长短期记忆网络相结合的网络流量检测系统源码,主要面向计算机、人工智能、通信工程等专业在校学生和老师,适用于课程设计、毕业设计、项目立项初期演示等场景,也适合具备初步Python编程基础、希望深入学习深度学习应用的新手。压缩包内共有6个文件,整体体积仅为5KB,虽然不大但结构完整,其中包含5个Python脚本和1个Markdown说明文档,脚本分别承担数据读取、数据预处理、模型结构定义、训练与测试以及主程序控制等任务,说明文档则清晰交代了项目结构与运行步骤,方便快速跑通。目前该资源已有1307人浏览学习,代码在真实环境中经过运行验证,功能可靠。借助这份源码,使用者能够完整掌握卷积网络与循环网络融合建模的基本流程,走通从原始流量数据准备、特征处理、模型搭建到测试评估的完整链路,还可以在此基础上调整网络层数、修改超参数或替换数据集,灵活适配自己的课程设计、毕业设计或创新实践项目。

1. 网络流量检测系统:课设选型先看这套 CNN+LSTM 的实现

网络流量检测系统做课设,最尴尬的局面不是代码跑不动,而是跑出个 98% 的准确率,答辩时一问数据怎么进模型的、每层输出是什么维度,答不上来。这套基于CNN+LSTM的流量检测Python源码提供的是一个完整的检测闭环:预处理、序列化、模型搭建、训练、测试五个环节分开放在独立的 py 文件里,模型用一维卷积抓短时模式、LSTM 建模跨包的时序关系,最后输出每个样本的类别概率。适合正在写课设或毕设的同学直接复现,也适合跑过一些模型但想对照一份规范实现来修正自己流程的从业者。整个项目加上说明文档共六个文件,代码里关键参数的取值都带着语义,不是随手填的。

2. CNN 与 LSTM 的分工:model.py 里的模型结构与 input_shape 约定

先把这份资源里最关键的 model.py 讲清楚。它负责模型组装,是整个系统里最容易变成“黑匣子”的文件,但它的输入和输出其实非常直观:输入是一批定长的流量会话矩阵,输出是每个会话属于各流量类别的概率分布。

2.1 选型逻辑:为什么流量检测常用 CNN 前置、LSTM 后置

课程设计里流量检测最常用的数据组织方式是:一条会话拆成若干个连续的包,每个包抽出若干统计特征,于是这条会话就变成形状为“时间步 × 特征维度”的二维矩阵。这个矩阵同时包含两种信息:

  • 空间/局部信息:相邻几个包的特征组合模式,比如连续的短包、固定间隔的握手包;
  • 时序信息:一个完整会话中不同阶段的先后依赖,扫描、爆破这类行为会阶段性地反复出现。

纯 LSTM 在这类任务上的通病是:输入序列很长的时候,较早的包对最后判别的贡献容易被稀释,训练速度也慢;纯一维 CNN 的问题则是感受野固定,跨窗口的长程关系只能靠堆层数,而流量会话短则几十个包、长则上百个包,堆太深会导致训练样本根本不够支撑。把 CNN 放前面、LSTM 放后面,其实是各取所长:卷积层先在短窗口内把特征提纯、压缩时间步数,后面的 LSTM 只管串联已经压缩好的模式序列,训练负担小很多。

结构局部特征提取时序建模在流量检测中的典型问题
纯 CNN强弱固定感受野,长会话跨窗口依赖难建模
纯 LSTM弱强序列长时训练慢,早期信息易被稀释
CNN + LSTM强强超参数更多,两个模块的维度需要对齐

2.2 model.py 核心代码:一套能直接跑出数值结果的组装方式

这份资源里的模型定义,核心结构和下面这段代码一致,不同小版本可能合并写法,但关键层不会变:

# model.py(核心结构示意) from tensorflow.keras.models import Sequential from tensorflow.keras.layers import ( Conv1D, MaxPooling1D, LSTM, Dropout, Dense ) def build_model(input_shape, num_classes): """ input_shape: (seq_len, feature_dim),不含 batch 维 num_classes: 类别数,内部用 softmax 输出概率 """ model = Sequential([ # 一维卷积:沿时间步滑动,提取相邻包的局部模式 Conv1D(filters=64, kernel_size=3, activation='relu', input_shape=input_shape), # 池化:把时间步数压缩一半,保留最强响应 MaxPooling1D(pool_size=2), # 防止卷积层过拟合 Dropout(0.2), # LSTM:读入压缩后的模式序列,只输出最后一个时间步的隐状态 LSTM(units=128, return_sequences=False), Dropout(0.3), # 分类头 Dense(units=64, activation='relu'), Dense(units=num_classes, activation='softmax') ]) model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy']) return model

逻辑说明:这段代码最关键的是 input_shape 的写法。Keras 里默认不写 batch 维,所以 input_shape=(seq_len, feature_dim) 表示每个样本是一个 seq_len 行的二维矩阵;Conv1D 沿着第一维“时间步”做滑动窗口卷积,每个卷积核能同时看到相邻 kernel_size 个时间步在全部特征维度上的联合取值。MaxPooling1D 把序列长度从 seq_len 压缩为 seq_len/2,这相当于告诉模型“相邻模式里取最强的一个就够”。之后 LSTM 的输入形状还是 (time_steps, channels),只是 time_steps 已经被池化减半,它只需要串联这些模式,不需要处理原始包级别的长序列。

参数说明:filters=64 表示第一层有 64 个卷积核,每个核负责一种局部模式的检测,常见范围在 32~128 之间;kernel_size=3 指卷积窗口覆盖 3 个连续时间步,太小捕捉不到突发模式、太大会模糊边界;LSTM units=128 是隐状态维度,代表模型记住“此前模式”的容量;Dropout 放在卷积池化之后和全连接之前两个位置,作用分别是抑制卷积特征过拟合和抑制分类头过拟合。

参数默认值常见调参范围改动方向
filters6432 ~ 128表达容量不够时调大
kernel_size33 ~ 5依赖更长跨步模式时调大
pool_size22 ~ 4序列太长时可调大
LSTM units12864 ~ 256特征维度高时调大
Dropout(conv)0.20.1 ~ 0.3过拟合时调大
Dropout(fc)0.30.2 ~ 0.5过拟合时调大

2.3 input_shape 约定:模型由 model.py 定义,取值范围却由预处理决定

这是整个系统里最容易翻车的地方。model.py 里 build_model 的 input_shape 不是随便填的,它必须和 data_preprocess.py 的输出对齐。比如预处理把每条会话截断成固定 20 个时间步,每个时间步抽 8 个字段,那么 input_shape 就强制为 (20, 8)。改任何一边,另一边不更新,都会直接报维度不匹配的错。

数据形状流转可以概括成一张表:

阶段数据形状说明
原始会话长度不定每个包的特征数量一致,包的数量不一致
预处理后(样本数, 20, 8)定长截断或补齐
Conv1D 输出(样本数, 10, 64)时间步被池化减半,特征维度升到 64
LSTM 输出(样本数, 128)只取最后一个隐状态
softmax 输出(样本数, K)K 为类别数

我一般会在写 model.py 之前先用一行代码打印预处理结果的形状,确认之后再去填参数:

print(f"X shape: {X.shape}") # 期望输出如 (样本数, 20, 8) print(f"y shape: {y.shape}") # 期望输出如 (样本数, K)

拿到这个结果再填 input_shape,基本不会错。常见做法是先预处理、再写模型文件;很多人反过来先搭网络,再去改数据脚本,结果反复被 shape 错误打断,浪费大量时间。

3. 数据闭环:预处理、加载与训练参数怎么对齐

这份资源的编排很清楚:data_preprocess.py 产出数值样本,data_load.py 负责分批喂给模型,train_and_test.py 跑训练和测试,main.py 把它们串成一条链。下面按执行顺序讲每个文件的职责。

3.1 预处理要做三件事:定长、归一化、标签映射

data_preprocess.py 的职责是把原始流量记录变成模型能吃的形式。原始数据通常是 pcap 文件解析出来的包记录,每条会话的包数量不固定。模型需要定长矩阵,所以预处理第一步是截断或补齐:超过上限的直接截断,不足的用 0 填到统一长度。

# data_preprocess.py(核心逻辑示意) import numpy as np def build_sequences(session_groups, seq_len=20, feature_cols=8): X, y = [], [] for session in session_groups: # session 形状是 (该会话包数, 特征维度) if session.shape[0] > seq_len: session = session[:seq_len] # 超长截断 mat = np.zeros((seq_len, session.shape[1])) # 不足补零 mat[:session.shape[0], :] = session X.append(mat) y.append(session.label) return np.stack(X), np.array(y)

这里把每条会话统一成 seq_len 行。截断方向要注意:常见做法是保留会话前 seq_len 个包,因为攻击行为往往在会话早期有握手特征;如果你的数据里攻击特征偏后段,则要考虑保留尾部。代码里 feature_cols 是预留的参数,实际字段数量由原始数据决定,但最终必须和 model.py 里的 feature_dim 保持一致。

归一化要在样本切分之后单独做。也就是先用训练集计算每个特征维度的最小值、最大值,再用同一组参数去缩放验证集和测试集。把归一化放在切分之前做,会让测试集的分布信息混进训练过程,指标虚高,答辩时很容易被问住。归一化后的数值建议直接保存成 npy 文件,标签单独存一份,并额外保存一个 label_names 列表,后面加载数据和推理阶段都会反复用到。

参数说明:seq_len=20 是每个样本的时间步数,决定了一条会话保留多少包的信息;字段选择上“少而全”比“多而杂”好,连续字段全部参与归一化,类别字段要在进模型之前做编码。

3.2 数据加载:写一个生成器而不是一次性全量载入

流量数据集预处理完往往是数万乘数十的矩阵,直接全部载入内存也能跑,但处理更大规模 pcap 时内存会爆。data_load.py 在这个资源里就是解决这个问题的:按 batch 读取,训练时每个 epoch 打乱顺序,验证时保持固定顺序。

# data_load.py(核心逻辑示意) class SequenceDataLoader: def __init__(self, X, y, batch_size=64, shuffle=True, seed=42): self.X = X.astype('float32') self.y = y self.batch_size = batch_size self.shuffle = shuffle self.index = np.arange(len(X)) if shuffle: np.random.seed(seed) np.random.shuffle(self.index) def __iter__(self): for start in range(0, len(self.index), self.batch_size): ids = self.index[start:start + self.batch_size] yield self.X[ids], self.y[ids]

逻辑说明:iter每次返回一个 batch,也就是一批 (样本矩阵, 标签矩阵);y 如果是 one-hot 编码,就可以直接进 categorical_crossentropy。shuffle 打开时先对索引整体打乱,再按 batch 切分,保证每个 batch 内类别分布相对均匀。seed 固定后,同一数据集每次加载的划分顺序一致,这在做实验对比时非常重要。

参数说明:batch_size 默认 64,内存或显存吃紧时降到 32;显存充足可以试 128,但流量检测任务里性能并不会因此明显提升,batch_size 太大反而容易让模型记住多数类。seed 一旦在第一次实验时确定,后面全部固定,不要每轮换。

3.3 训练主循环:早停、模型保存、测试集评估

train_and_test.py 是整套系统真正执行的地方。训练过程除了常规的 fit,还要考虑两个细节:早停和只保存最优模型。流量检测的验证集指标容易抖动,最后一轮的 loss 往往不是最小的。

# train_and_test.py(核心逻辑示意) from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint def run_training(model, train_loader, val_loader, epochs=50, patience=5): callbacks = [ EarlyStopping(monitor='val_loss', patience=patience, restore_best_weights=True), ModelCheckpoint('best_model.h5', monitor='val_loss', save_best_only=True) ] history = model.fit( train_loader, validation_data=val_loader, epochs=epochs, callbacks=callbacks, verbose=1 ) return history

逻辑说明:EarlyStopping 盯着验证集 loss,patience=5 表示连续 5 个 epoch 没有下降就让训练提前终止,restore_best_weights 会在结束后把模型权重回滚到验证集最好的那一步;ModelCheckpoint 同时在磁盘上保存最优权重,这样就算训练中途被手动中断,best_model.h5 也能用于后续测试。这两个回调是课设代码里最容易被漏掉的细节,但在答辩时是明显的加分项。

参数说明:epochs=50 是上限,实际很可能在 20 轮左右就触发早停;patience=5 太小容易在一段抖动中错过后面的下降,太大则浪费训练时间,一般取 5~10。学习率如果没在参数列表里显式给出,建议在 model.compile 的 optimizer 里设置,比如 Adam 配 learning_rate=0.001,调不动时降到 0.0005。

训练完成后,main.py 会加载 best_model.h5 对测试集做预测,并输出 accuracy。但只打印一个 accuracy 远远不够,第 5 章会给一套更完整的评估组合。

参数推荐值说明
学习率0.001,瓶颈时 0.0005Adam 优化器常用起点
batch_size64显存受限时降到 32
epochs50早停通常会在 20 轮左右生效
patience5验证集抖动的容忍度
数据切分训练:验证:测试 = 6:2:2固定随机种子后再切

4. 避坑笔记:跑通这套流量检测代码后最常见的五个问题

这一章不是泛泛的故障列表,是我在几个课设评测和数据复现里真实遇到的问题。每条按“现象、原因、解决”三步写,对照自己的日志就能快速定位。

4.1 现象:accuracy 很高,但 F1 惨不忍睹

很多同学第一次跑完打印 accuracy=0.92,高高兴兴准备截图。再打印 F1 却发现只有 0.4。原因很简单:流量数据天然不均衡,正常类可能占 80% 以上,模型只要全预测成正常类,accuracy 就有 80 多分。准确率这个指标在这种类别分布下根本没有说服力。

解决:训练时不要只盯 accuracy。两个地方要改:一是在 model.compile 里用加权交叉熵,把少数类每个样本的 loss 权重提高;二是模型保存和早停的监控指标改成 F1 或 AUC。如果资源里只打印 accuracy,就在训练结束后把混淆矩阵打出来,先看少数类的召回率再下结论。

提示:全预测成多数类时,混淆矩阵里除多数类那一列外其他列都是 0,这是最容易识别的失败模式。

4.2 现象:loss 反复波动,训练一段时间完全不下降

用定长补齐之后,padding 出来的零向量会被 LSTM 当成真实输入处理。如果补齐时把零放在序列开头,LSTM 会先读大量全零向量,梯度几乎为零,等于模型在一大段无信息的输入上白白消耗容量。

解决:补齐方向统一放序列末尾,让真实包紧挨着开头。更彻底的方案是训练时记录每个样本的真实长度,LSTM 层开启 masking 或手动把真实长度之后的时间步掩码掉。如果只是课设,先改 padding 方向,大多数情况下 loss 就能稳定下来。

4.3 现象:训练中途内存或显存爆掉

整个预处理矩阵一次性堆进内存,同时 DataLoader 又不在 batch 读取时释放中间副本,数十万样本的流量数据很容易让机器直接卡死。

解决:data_preprocess.py 的输出只保留训练、验证、测试三个数组,不要再额外保存多个中间副本;DataLoader 按 batch 读取后及时释放临时变量。显存不够就先减 batch_size,再不行把 seq_len 从 20 降到 16,虽然牺牲信息量,但能让流程先通。

4.4 现象:训练时表现好,换一批新数据预测完全不对

这是最隐蔽的坑。推理阶段往往只有 model.predict 而没有在前面接预处理。如果训练数据做过 MinMax 归一化,但推理时直接喂原始数值,模型收到的分布完全不一样,输出概率自然错乱。

原因就是归一化参数没有跨阶段保存。解决:data_preprocess.py 在训练阶段把每个特征维度的 min 和 max 存成 npy 或直接 pickle 保存;推理脚本先加载这套参数,对输入做同样的缩放,再进模型。标签也一样,label_names 要固定,防止训练集里出现过的类别在推理时因为编码顺序变化而错位。

4.5 现象:同一个项目跑两次,指标差一个百分点以上

不固定随机种子时,训练数据切分、batch 打乱顺序、初始化权重三次随机过程不同,结果自然会有波动。有人因此把参数调到“恰好在某一次实验里表现好”,这类参数并不可靠。

解决:在 main.py 或 train_and_test.py 开头集中固定三个随机源。

import random, numpy as np, tensorflow as tf random.seed(42) np.random.seed(42) tf.random.set_seed(42)

同时数据切分要在固定 seed 下先做,保存切分索引,之后每一次实验都用同一份验证集和测试集。这样才算公平对比,指标波动会小很多。

5. 从“能跑”到“能答辩”:混淆矩阵与分片召回率怎么用

5.1 一组指标覆盖不了所有问题,一张混淆矩阵可以

训练结束后的标准动作不是抄 accuracy 进报告,而是先把测试集完整过一遍,打印分类报告和混淆矩阵。

# 读取测试集并恢复保存过的标签名 test_y = np.load('test_y.npy') y_pred = model.predict(test_X).argmax(axis=1) from sklearn.metrics import classification_report, confusion_matrix print(classification_report(test_y, y_pred, target_names=label_names, digits=4)) cm = confusion_matrix(test_y, y_pred) np.save('confusion_matrix.npy', cm)

classification_report 会一次给出每个类别的精确率、召回率、F1 和支持样本数。把 digits=4 改成 2 可以缩短输出,但报告里保留四位小数更有说服力,也方便和已有基准做对比。

5.2 再进一步:按类别分片统计,导出误判样本

准确率只体现整体占比,无法说明哪一类被漏掉;反而是每类各自的召回率能直接暴露模型漏掉了哪种攻击。按类别切片统计后,如果某类召回率低于 60%,就要去查漏判样本集中在哪些特征范围内。

import pandas as pd err_idx = np.where(y_pred != test_y)[0] hard_samples = pd.DataFrame({ 'sample_id': err_idx.tolist(), 'true_label': np.array(label_names)[test_y[err_idx]], 'pred_label': np.array(label_names)[y_pred[err_idx]] }) hard_samples.to_csv('hard_samples.csv', index=False)

导出的 CSV 可以直接在报告里引用,说明“误判的某类样本多数特征分布与正常类重叠”这类结论,比“模型准确率 98%”有说服力得多。如果样本量够,还可以按特征维度做简单的分组统计,看误判样本在某个字段上的均值是否显著偏移。

我从那以后每次拿到一套新数据,不管数据量多小,第一步都不是搭模型或调参数,而是先写一个统计脚本,确认类别分布、特征范围、缺失值数量这三个基线,再谈训练。这套流程帮我避开过好几个看起来指标很好、实际上已经翻车的方案。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询