基于PyTorch的CNN-LSTM网络流量检测系统实践指南

简介：基于PyTorch的CNN-LSTM网络流量检测系统实现与性能分析，是一份面向高校计算机科学专业教学演示、科研实践及行业算法参考的资源。项目融合卷积神经网络与长短期记忆网络构建混合架构，专门用于网络流量行为识别，利用KDDCUP十百分比抽样数据集完成训练，在PyTorch框架下经过十轮迭代后测试集分类准确率稳定维持在95%以上，可支持毕业设计、课程实验与专业能力提升训练，系统各模块运行稳定，功能实现符合设计预期。压缩包共10个文件，以Python源码（5个py）为主，另含备份文件（3个zbak）、说明文档（1个md）和附加压缩包，整体仅11KB，非常轻量便携。目前已有99人学习。内含完整的模型定义、数据预处理、训练与测试脚本，以及说明文档和备份文件，读者可据此对照理解CNN-LSTM流量检测的实现流程、参数调整思路与数据处理细节，适合快速搭建同类实验环境，仅限非商业性学术用途。

1. 把 CNN-LSTM 用在网络流量检测：这套组合到底解决了什么问题

手里攒了几个 GB 的 pcap，想判断每一条网络流到底是 Web 访问、视频流、文件传输还是恶意样本通信，单靠端口号和包长统计根本扛不住——加密流量越来越多，特征常常藏在时序行为里。基于 PyTorch 的 CNN-LSTM 网络流量检测系统，做的就是让模型自己去读「流」：CNN 负责抓短窗口内的局部模式，LSTM 负责把窗口与窗口之间的先后关系串起来，最后输出一个分类结果。这个方向适合正在做流量分类课题的学生，也适合给安全平台做流量预研的工程师，比起手工特征加树模型，它能把会话级别的时序关系直接交给网络去学，省掉一大半特征工程的体力活。

2. 数据准备：从 pcap 到序列样本的完整流程

2.1 流切分：别把报文一条条喂给模型

流量检测的最小单元不是单个数据包，而是一条「流」。常见做法是按五元组（源 IP、目的 IP、源端口、目的端口、协议号）把 pcap 里的报文分组，并且把反向报文归并到同一条流，因为一次 TCP 会话的请求和响应是同一行为的两半。我这里用 scapy 做了个最小切流，重点看双向流的 key 怎么设计。

from scapy.all import rdpcap, IP, TCP, UDP def flow_key(pkt): if IP not in pkt: return None proto = pkt[IP].proto if TCP in pkt: sport, dport = pkt[TCP].sport, pkt[TCP].dport elif UDP in pkt: sport, dport = pkt[UDP].sport, pkt[UDP].dport else: sport = dport = 0 src, dst = pkt[IP].src, pkt[IP].dst return (src, dst, sport, dport, proto), (dst, src, dport, sport, proto) def split_streams(pcap_path): packets = rdpcap(pcap_path) streams = {} for pkt in packets: res = flow_key(pkt) if res is None: continue fwd, rev = res # 双向流：无论是正向包还是反向包都进入同一条流 if fwd in streams or rev in streams: k = fwd if fwd in streams else rev else: k = fwd streams[k] = [] streams[k].append(pkt) return streams

这段代码的逻辑说明有几层：flow_key同时算出正向 key 和反向 key，这样请求包和响应包能被归到同一个列表里；归并时先看两个方向是否已有流存在，存在就追加，不存在就以正向 key 新建。协议号只取 IP 层的proto，TCP 和 UDP 的端口区分要单独判断，因为 scapy 里TCP in pkt不一定对非 TCP 包成立。

参数层面的考虑：如果你处理的是超大 pcap（几个 GB），rdpcap一次性读进内存会直接吃满内存。我一般会改用PcapReader做流式迭代，每读一个包就更新对应的流缓存，并加一个「空闲超时」：超过 120 秒没有新包的流直接落盘，防止长连接把内存拖垮。这个 120 秒的取值通常参考协议会话习惯，TCP 连接的空闲回收在大部分网关设备上也取这个量级。

流切完并不是所有流都能用。像 TCP 三次握手中的纯 SYN/ACK 报文，序列信息太短，属于噪声样本；UDP 里的 QUIC 流量虽然不靠端口区分连接，但五元组 + 时间窗口仍然能切出可用的流。还有一个容易被忽略的地方：如果 pcap 是从交换机镜像口抓的，同一个流的两端方向都会被镜像，反向归并不做的话，同样的流量在数据集里会出现两份，训练时会造成「同一条流同时出现在训练集和验证集」的数据泄漏，模型指标虚高，上线就原形毕露。

2.2 构造特征序列：时间窗口统计与归一化

CNN-LSTM 的输入是序列，所以要把一条流变成一串固定长度的向量。我这里不直接丢原始报文，而是把流按 0.1 秒的时间窗切成一个个小片段，每个片段统计一批数值，比如包数、字节数、平均包长、到包间隔均值。这样的特征序列在公开的流量分类数据集上属于主流做法，既能压缩数据量，又保留了窗口之间的时序关系。

import numpy as np def extract_sequence(stream_pkts, window_s=0.1, seq_len=50, n_features=6): if not stream_pkts: return None start = float(stream_pkts[0].time) end = float(stream_pkts[-1].time) num_windows = max(1, int((end - start) / window_s) + 1) win_pkt_cnt = np.zeros(num_windows) win_byte_cnt = np.zeros(num_windows) win_len_sum = np.zeros(num_windows) win_tcp_flag = np.zeros(num_windows) idx = 0 for pkt in stream_pkts: w = int((float(pkt.time) - start) / window_s) w = min(w, num_windows - 1) win_pkt_cnt[w] += 1 pkt_len = len(pkt) win_byte_cnt[w] += pkt_len win_len_sum[w] += pkt_len if TCP in pkt and (pkt[TCP].flags & 0x02): # SYN win_tcp_flag[w] += 1 idx += 1 seq = np.zeros((seq_len, n_features)) step = max(1, num_windows // seq_len) for i in range(seq_len): s = i * step e = min(s + step, num_windows) if s >= num_windows: break seq[i, 0] = win_pkt_cnt[s:e].sum() seq[i, 1] = win_byte_cnt[s:e].sum() seq[i, 2] = win_len_sum[s:e].sum() / max(1, win_pkt_cnt[s:e].sum()) seq[i, 3] = win_pkt_cnt[s:e].mean() seq[i, 4] = e - s seq[i, 5] = win_tcp_flag[s:e].sum() return seq

这个函数的逻辑说明：先按时间把整条流切成等宽窗口，每个窗口底层的统计量是「包数、字节数、包长总和、SYN 次数」，然后在取序列时用分块聚合把窗口数压缩到固定的seq_len。这里我写得很直白，实际项目里n_features可以扩展到 10 个以上，常见补充包括：上下行字节比例、ACK 数量、窗口大小均值、TTL 均值等。

参数说明：

归一化放在特征提取之后而不是之前：对每个特征维度独立算 min-max 或 z-score。我踩过的坑是只对整体矩阵做了一次归一化，导致数值范围大、方差大的字节数特征把 SYN 计数特征压得几乎没有梯度。正确做法是按特征列归一化，并在训练集上算好统计量、验证和测试时沿用同一套参数，避免每个 Batch 重新归一化造成数据泄漏。

2.3 标签组织与数据集划分

标签来源通常是 pcap 文件所在的目录名或文件命名，例如ISCX-VPN2016数据集的目录会区分Chat、Streaming、VoIP等类别。建议在数据准备阶段就生成一份meta.csv，字段包含flow_id, label, duration, pkt_count, pcap_file，后续划分训练集和测试集直接基于这份 CSV 做，而不是每次重新遍历 pcap。

我在标签映射上吃过一次亏：类别名直接作为模型输出下标，但字典序和业务优先级不一致，导致混淆矩阵上对不上号。现在我的做法是显式写一个类别到索引的映射文件，比如label_map.json，并在加载数据集时从 JSON 读入，而不是让代码里隐式依赖sorted(set(labels))的结果。

数据集划分要按流而不是按报文，并且要做分层采样。如果某个类别的流数量特别多，直接train_test_split(shuffle=True)会让多类样本在训练集里占比过高，这时用StratifiedShuffleSplit按标签分层更稳。流的切分还有一个与时间相关的注意点：同一个 IP 的流往往会落在相近时间窗口内，如果训练集和测试集按时间切在前后两段，能更接近真实部署效果；按流随机切分则更乐观。我在做性能分析时会把两种切分都跑一遍，报告里分别标注，不然线上效果容易被随机切分的高指标误导。

3. 模型搭建与训练：PyTorch 里的 CNN-LSTM

3.1 网络结构：Conv1d + LSTM + FC 怎么拼

很多人第一次搭 CNN-LSTM 会直接把二维 CNN 里的Conv2d拿来用，然后发现输入维度怎么都对不上。在流量检测场景里输入是(batch, seq_len, n_features)，卷积要沿着时间维做，所以用的是Conv1d，并且要把特征维放到通道位置。下面是一个可以直接跑通的最小结构，我保留了最核心的组件。

import torch import torch.nn as nn class CNNLSTM(nn.Module): def __init__(self, n_features=6, seq_len=50, cnn_out=64, lstm_hidden=64, num_layers=2, n_classes=6): super().__init__() self.conv1 = nn.Conv1d(n_features, 32, kernel_size=3, padding=1) self.conv2 = nn.Conv1d(32, cnn_out, kernel_size=3, padding=1) self.pool = nn.MaxPool1d(kernel_size=2, stride=2) self.lstm = nn.LSTM( input_size=cnn_out, hidden_size=lstm_hidden, num_layers=num_layers, batch_first=True, dropout=0.3, bidirectional=False ) self.dropout = nn.Dropout(0.3) self.fc = nn.Linear(lstm_hidden, n_classes) def forward(self, x): # x: (batch, seq_len, n_features) x = x.permute(0, 2, 1) # (batch, n_features, seq_len) x = torch.relu(self.conv1(x)) x = torch.relu(self.conv2(x)) x = self.pool(x) # (batch, cnn_out, seq_len // 2) x = x.permute(0, 2, 1) # (batch, seq_len // 2, cnn_out) out, _ = self.lstm(x) # (batch, seq_len // 2, lstm_hidden) out = out[:, -1, :] # 取最后一个时间步 out = self.dropout(out) return self.fc(out)

这段代码有两个关键设计。第一个是permute的来回调换：Conv1d要求输入是(batch, channels, length)，而 LSTM 的batch_first=True要求(batch, length, input_size)，所以中间必须把维度换回来。第二个是池化后序列长度变成了seq_len // 2，LSTM 输出的最后一维长度也随之减半，取最后一个时间步的隐藏状态作为整条流的向量表示。如果seq_len是奇数，MaxPool1d之后长度是(seq_len - 1) // 2 + 1，这一点在写模型时很容易算错，建议最外层做一次长度检查。

参数表如下，按一个小规模流量数据集（几万条流）配置的典型值：

这里有一个很多人忽略的点：LSTM 层的dropout参数只有在num_layers > 1时才会生效，单层 LSTM 传dropout=0.3会被 PyTorch 直接忽略。如果你想要单层 LSTM 也要 dropout，得自己在输入或输出位置手动加，或者在数据输入 CNN 之前对特征做随机掩码。

3.2 训练配置：环境、超参与训练循环

在开始训练之前，环境就绪是第一步。我在 Windows 上用 WSL 搭 PyTorch 环境比较多，因为 NVIDIA 驱动在 WSL 里可以直接透传给 Linux 用户态，CUDA 工具链不需要单独装。Anaconda 配置环境的命令通常是：

conda create -n traffic python=3.10 -y conda activate traffic # 先确认本机 CUDA 版本，再选对应的 PyTorch 版本，python 和 pytorch 版本对应关系以官网为准 pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121

PyTorch 和 Python 版本对应关系是玄学重灾区，装完跑import torch报DLPacked或者GLIBCXX错误，大概率是版本不对应。我的习惯是先用python --version确认解释器版本，再按这个版本去选兼容的 PyTorch 轮子。机器驱动暂时起不来时，先在 CPU 环境把模型跑通，再切 GPU，能省下很多排查时间。

数据加载我建议把特征提取后的.npy或.npz直接映射到Dataset，不要在__getitem__里做重计算。

from torch.utils.data import Dataset, DataLoader class TrafficDataset(Dataset): def __init__(self, x_path, y_path): self.x = np.load(x_path) # shape: (num_samples, seq_len, n_features) self.y = np.load(y_path) def __len__(self): return len(self.x) def __getitem__(self, idx): x = torch.from_numpy(self.x[idx]).float() y = torch.tensor(self.y[idx], dtype=torch.long) return x, y train_ds = TrafficDataset("train_x.npy", "train_y.npy") train_loader = DataLoader(train_ds, batch_size=64, shuffle=True, num_workers=2)

训练循环里我习惯加梯度裁剪和早停。LSTM 对梯度爆炸特别敏感，clip_grad_norm_是保底手段；早停则防止模型在验证集上过拟合后继续跑浪费算力。

import torch.optim as optim from torch.optim.lr_scheduler import CosineAnnealingLR model = CNNLSTM() opt = optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4) scheduler = CosineAnnealingLR(opt, T_max=30) criterion = nn.CrossEntropyLoss() best_val_acc = 0.0 patience = 0 for epoch in range(50): model.train() for x, y in train_loader: opt.zero_grad() pred = model(x) loss = criterion(pred, y) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0) opt.step() scheduler.step() # 每个 epoch 后在验证集上评估 val_acc = evaluate(model, val_loader) if val_acc > best_val_acc: best_val_acc = val_acc torch.save(model.state_dict(), "best_model.pt") patience = 0 else: patience += 1 if patience >= 5: print(f"early stop at epoch {epoch}") break

参数设置上，lr=1e-3对流量分类的序列数据来说是一个常见起点，如果验证 loss 震荡得很厉害就降到3e-4；weight_decay=1e-4给全连接层加一点正则；max_norm=5.0是梯度裁剪的阈值，太小会让训练变慢，太大失去裁剪意义。CosineAnnealingLR配上早停在流量这类数据上比固定学习率收敛得更顺，尤其是在 epoch 后半段，loss 容易在小范围内抖动不下降。

4. 避坑：CNN-LSTM 流量检测里的 4 个翻车点

4.1 维度翻车：LSTM 输入必须三维

现象：forward 跑到 LSTM 时报错，信息类似Expected 3D tensor as input, got 2D。

原因：Conv1d 输出经过MaxPool1d后维度已经变成三维权重的最后两维，如果少做一次permute，LSTM 拿到的是(batch, cnn_out, seq_len // 2)，而它要的是(batch, seq_len // 2, cnn_out)。这个顺序反了，PyTorch 会直接拒绝。

解决：在进入 LSTM 前统一断言一下形状，把调试成本降到最低。

assert x.dim() == 3, f"LSTM input must be 3D, got {x.dim()}D" # 预期形状: (batch, seq_len // 2, cnn_out)

我的习惯是在模型初始化的__init__里打一行print(x.shape)，跑第一个 batch 时看每层的形状变化，比报错后脑补维度省时间。

4.2 序列长度翻车：MaxPool 之后长度对不齐

现象：模型在训练集上跑得好好的，验证时换了一批不同时长的流，报 LSTM 输入维度不一致，或者池化后序列长度变成零。

原因：训练时seq_len=50，但池化后50 // 2 = 25是固定的；如果某个流本身的窗口数不足而导致补零逻辑有 bug，或者预测阶段对实时流做了不同的截断，长度就变了。更隐蔽的是 seq_len 为奇数时，池化后长度是(seq_len - 1) // 2 + 1，不是简单的整除。

解决：把序列长度固定写在模型外部，数据加载阶段统一pad或truncate。再一个办法是不要用MaxPool1d，改用GlobalAvgPool1d，直接把长度压成 1，彻底摆脱长度对齐问题。代价是局部时序信息会被平均掉一部分，精度通常略降。

4.3 收敛不稳：LSTM 的梯度爆炸和过拟合

现象：loss 在某个 epoch 突然跳到nan，或者验证准确率停在 60% 死活上不去。

原因：LSTM 的梯度路径长，学习率偏高时梯度爆炸是常态；流量特征里字节数这类数值尺度大，也容易把梯度推向极端。另一个常见原因是bidirectional=True且num_layers=2时参数量翻倍，数据量只有几万条的话模型直接过拟合，验证指标曲线呈锯齿状。

解决：梯度裁剪加上lr从3e-4起步试一试；过拟合则优先加大 dropout、减小 hidden size，再看要不要提前停止。我习惯给 LSTM 的输入特征做一次BatchNorm1d，把不同特征尺度压到接近量级，往往会比单纯调学习率收敛更快。

self.bn = nn.BatchNorm1d(n_features) # 在模型 forward 的 permute 之前对原始输入做归一化

4.4 类别不均衡：少数类被多数类吃掉了

现象：准确率有 92%，但少数类别的 recall 只有 12%，混淆矩阵里这一行几乎全被预测成了多数类。

原因：流量数据集天然不均衡，比如正常 HTTP 流量占总样本的八成，而某类攻击只占 0.5%。CrossEntropyLoss对每个样本平等对待，模型只要把所有样本都预测成多数类就能拿到很高的准确率，少数类的梯度贡献被淹没了。

解决从两个方向下手：数据层面做欠采样或过采样，损失函数层面用带权重的 CrossEntropy 或 Focal Loss。我一般先用类别权重，改动最小：

class_weights = torch.tensor([0.1, 1.0, 2.0, 5.0, 10.0, 20.0]) criterion = nn.CrossEntropyLoss(weight=class_weights)

如果还不行，再用 Focal Loss 把难样本的梯度放大。类别权重需要根据训练集的真实分布计算，通常取n_samples / (n_classes * np.bincount(y))，不要拍脑袋写死，不然少数类权重过大反而把精确率带崩。

5. 性能分析：用什么指标和实验说明模型有效

5.1 混淆矩阵和分类报告：别只看准确率

网络流量检测的评估指标不能只看准确率，这一点做安全场景的人应该有共识。要交出一份能说服人的性能分析，至少需要三样：混淆矩阵、每个类别的精确率/召回率/F1、以及 ROC 曲线或 PR 曲线。边做边看哪个类别被混淆、被漏掉，比只给一个总分有信息量得多。

from sklearn.metrics import confusion_matrix, classification_report import seaborn as sns import matplotlib.pyplot as plt y_true, y_pred = get_preds(model, val_loader) cm = confusion_matrix(y_true, y_pred) sns.heatmap(cm, annot=True, fmt="d", cmap="Blues") plt.xlabel("Predicted") plt.ylabel("True") plt.tight_layout() plt.savefig("confusion_matrix.png", dpi=150) print(classification_report(y_true, y_pred, digits=3))

这里我直接用了 sklearn 的工具，classification_report会自动按类别给出精确率、召回率、F1 和支持样本数，比手写循环统计要快。注意查看时别只看weighted avg，流量不均衡时macro avg更能反映每个类别的平均表现——如果 macro F1 明显低于 weighted F1，说明模型在少数类上偏弱，这正好对应第四章的不均衡坑。

5.2 对比实验与消融：用数字说明 CNN-LSTM 值得用

性能分析的说服力来自对比，而不是单独一个模型跑出多少分。常见做法是设置三组基线：纯 CNN（只看局部窗口）、纯 LSTM（只看时序）、CNN-LSTM（两者都看），分别在同样的数据划分下用同样的训练参数跑。另一个对比维度是窗口长度window_s从 0.05 到 0.2 的变化，看时序分辨率对结果的影响。

一个典型实验设计表格长这样：

这个表的目的是呈现「为什么值得做」：单个 CNN 丢了时序上下文，单个 LSTM 又抓不住窗口内的局部模式，两个叠加在精度上是正向收益。耗时上的增加也要写清楚，因为在线检测系统对延迟敏感，精度换来的收益是否值得，由具体业务决定。

5.3 推理时延与参数量：上线前必须测的两个数

模型参数量和单条推理时延是性能分析里容易被忽略的部分。参数量决定模型部署时占多少内存，时延决定实时检测能不能跟上流量速率。这里给一小段用 PyTorch 统计参数量的代码：

def count_params(model): total = sum(p.numel() for p in model.parameters()) trainable = sum(p.numel() for p in model.parameters() if p.requires_grad) return total, trainable total, trainable = count_params(model) print(f"Total params: {total:,} | Trainable: {trainable:,}")

推理时延的测量要用torch.inference_mode()而不是普通的no_grad，前者会跳过一些不必要的追踪逻辑，更接近部署时的真实状态。测量时输入一个批次的数据，循环跑几百次取均值，避免单次运行受 CPU 冷启动影响。

6. 把模型导成 ONNX：上线前的最后一步

6.1 用 PyTorch 转 ONNX：固定输入维度与 dynamic_axes

模型在训练脚本里验证够了，要上线又不想在服务端跑一个完整 PyTorch 进程时，最省事的路径是把模型转成 ONNX。PyTorch 里自带的torch.onnx.export就能干这件事，关键是把动态轴声明清楚，否则导出模型会被固定成训练时的 batch size 和序列长度。

model.eval() dummy_input = torch.randn(1, 50, 6) torch.onnx.export( model, dummy_input, "traffic_cnn_lstm.onnx", input_names=["flow_input"], output_names=["logits"], dynamic_axes={ "flow_input": {0: "batch", 1: "seq_len"}, "logits": {0: "batch"} }, opset_version=17 )

这里的参数说明：dummy_input的形状要和模型真实输入一致，即(batch, seq_len, n_features)；dynamic_axes里声明第 0 维和第 1 维可变，这样服务端可以接受不同长度的流序列。opset_version决定算子兼容性，PyTorch 2.x 推荐 15~18 之间，版本太低时 LSTM 的算子可能导出失败。

6.2 导出后的输出一致性检查：防止黑匣子翻车

ONNX 导出后不验证就直接接上线，是我吃过亏的地方。模型在 PyTorch 里的输出和执行引擎里的输出可能有细微差异，尤其是在eval模式和training模式切换出错时，Dropout没有关掉，线上跑出来的结果会和训练时不一致。

import onnxruntime as ort import numpy as np session = ort.InferenceSession("traffic_cnn_lstm.onnx", providers=["CPUExecutionProvider"]) np.random.seed(0) test_input = np.random.randn(4, 50, 6).astype(np.float32) pytorch_out = model(torch.from_numpy(test_input)).detach().numpy() onnx_out = session.run(["logits"], {"flow_input": test_input})[0] diff = np.abs(pytorch_out - onnx_out).max() print(f"max diff: {diff:.6e}")

如果 diff 在1e-5量级，说明无损转换成功；如果到1e-2甚至更大，先检查模型是否忘了model.eval()，再用onnxruntime的log_severity_level=3打开告警信息查算子兼容性。另外实践里我习惯把 ONNX 的输入也做成和训练一样的归一化管线，因为 ONNX 模型只是一个计算图，前置特征处理不在里面，忘了归一化相当于线上模型换了套输入分布，效果直接打对折。

这套流程走完后，你的 CNN-LSTM 流量检测系统从数据切流、序列特征、模型训练到性能分析、模型导出，都有了一条能复现的路径。我现在每次做新流量数据集，都会先把 ONNX 导出和一致性检查写在训练脚本之后，而不是等模型调完再补——这个顺序帮我避开了好几次把实验模型直接搬上线的翻车。希望帮到你。

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