简介这是一份基于深度强化学习构建的网络入侵检测系统完整源代码并附带实验数据集。项目以异步优势 actor critic 算法为核心覆盖数据清洗与格式转换、智能体交互环境、策略网络训练、实时监控等模块可用于计算机、信息安全、人工智能、大数据等专业的毕业设计、课程设计或期末大作业也适合有一定基础的开发者在此之上做功能扩展。资源包共五十个文件主要包含源代码脚本、文本说明、数据文件等类型另附图表与一键启动脚本整体约十九兆目录结构清晰方便按模块查阅学习。目前已有三百四十三人学习下载代码经过运行验证并带有完整的训练集与测试集可支撑不同入侵类型的检测实验和结果对比。无论是希望快速入门深度强化学习在网络防御中的应用还是需要一套可运行的基线系统这份资料都具备较高的参考与二次开发价值。1. 基于深度强化学习的网络入侵检测到底在解决什么网络入侵检测系统的传统方案有两条路一条靠规则库用 Snort 或 Suricata 的签名表匹配已知攻击特征碰到变种和加密流量往往直接哑火另一条靠监督式机器学习在训练集上刷 AUC 很好看但换到真实流量里要重训、要调阈值漏报和误报的平衡仍需手工反复折腾。基于深度强化学习的方案把检测问题换了个思路不再让模型“记住攻击长得什么样”而是把流量交互变成试错过程让智能体自己学会什么情况下该放行、什么情况下该告警把业务目标直接写进奖励函数里收敛。这套 Python 源码完整覆盖了环境封装、DQN 智能体、训练主循环并附带预处理好的数据集适合有 Python 基础、想在网络侧把深度强化学习真正落到实处的安全工程师阅读。以下是我复现这类项目的完整过程和踩坑记录。2. 为什么偏偏是深度强化学习状态、动作与奖励的建模逻辑2.1 静态模型吃不到的“长尾攻击”正是 DRL 的切入点先放结论用深度强化学习做网络入侵检测目标不是把 Benchmark 刷到 99.99% 的准确率而是把人工调阈值、调特征权重的琐碎工作替换成“奖励函数设计”。监督学习框架里模型只看当前样本的特征和标签学到的本质是“这个特征组合以前被标成攻击所以下次大概率也是”。这类模型最大的问题就是静态攻击者会变特征分布会漂移决策边界不会自己跟着调整。强化学习的差异在于把检测当成序列决策问题。智能体不是判完一条就收工它会在当前流量反馈的基础上调整后续策略让长期累计收益最大。网络流量天然有很强的时序习惯攻击有扫描、渗透、回传几个阶段前序行为往往能预示后续行为。这种长尾时序信息恰好是监督模型不擅长、而强化学习擅长利用的。所以这一类项目选深度强化学习不是为了炫技是冲着“决策闭环可演化”来的。2.2 NIDS 里四个关键映射环境、状态、动作、奖励要把深度强化学习套到网络入侵检测上首先得把概念映射清楚。我一般会先画一张表做映射不是为了学术严谨而是为了后面写代码时每一行都知道自己对应的是什么强化学习要素NIDS 中的含义常见实现环境带有反馈信号的流量样本流封装一批会话特征每次暴露一条样本状态当前连接或会话的特征向量数值特征标准化必要时拼接前几条样本做时序窗口动作检测决策的输出空间0 表示放行1 表示告警奖励检测质量与业务代价的组合正确告警给正分误报和漏报给负分终止条件一批样本遍历结束当前 episode 跑完整个数据文件这里最容易被忽略的是动作空间。很多初次接触的人会把动作设计成“哪一类攻击”实际工程里这么做会把问题复杂化。入侵检测的核心诉求是“这个流量要不要引起安全人员注意”二值决策就足够。想做攻击分类是下游威胁情报系统的事不要在检测这个环节里硬塞多分类。2.3 算法选型DQN、DDPG、PPO 怎么选深度强化学习的算法家族很大但能用在入侵检测这种场景里的其实就几个算法适用动作空间用在 NIDS 里的判断DQN离散小动作空间二分类决策最稳定先跑通的首选Double DQN离散小动作空间解决 DQN 的 Q 值过估计训练更平稳DDPG连续动作空间很少用除非把告警阈值设计成连续输出PPO离散或连续都行适合把状态序列拉长、需要策略平滑更新的场景只有“告警 / 放行”两个动作时我不建议一上来就上 PPO。动作空间小DQN 的经验回放机制能让训练更好收敛网络结构简单出问题也好排查。PPO 的优势在连续控制和大动作空间里才明显用在二值动作上属于杀鸡用牛刀还要多调 GAE 系数和 clip 范围。我复现这类项目时起步阶段都是 DQN 或 Double DQN跑通之后再做 PPO 对照实验。3. 跑通最小系统Python 依赖、数据预处理和首次训练3.1 先把环境装好Python 版本与依赖清单压缩包解压后我习惯先建一个干净的虚拟环境避免和系统 Python 打架。项目依赖主要是 PyTorch、pandas、numpy、scikit-learn 和 gymnasium。gymnasium 是 OpenAI Gym 的维护分支新代码建议直接用它API 兼容性好后面做环境封装时会省很多事。# Python 3.8 以上都可以建议用 3.10 python -m venv nids_env source nids_env/bin/activate # 先装 CPU 版本跑通特征维度高或样本量大再换 GPU 版本 pip install torch pip install pandas numpy scikit-learn gymnasium如果你还在 Python 入门阶段建议先跑几个 gymnasium 自带的小环境熟悉一下 reset、step、reward 的交互逻辑再进来看这个项目。这块环境交互的语法不熟后面封装 NIDS 环境会容易看晕。3.2 解压后的目录结构长什么样压缩包里的内容各家整理风格不同但跑这类项目常见的目录形态是这样nids_drl/ ├── agent/ │ ├── dqn_agent.py # DQN 智能体网络、训练、动作选择 │ └── replay_buffer.py # 经验回放 ├── env/ │ └── nids_env.py # gymnasium 环境封装 ├── data/ │ ├── raw/ # 原始数据集不建议直接改动 │ └── processed/ # 预处理后的 CSV训练直接读这里 ├── train.py # 训练入口 ├── evaluate.py # 评估入口 └── config.yaml # 超参数配置data 目录里放的数据集来源通常是 NSL-KDD 或 UNSW-NB15 的某个切片。这类数据集的共同特点是特征列多、混合类型数值加类别、标签列做二分类映射。如果你的压缩包里是原始格式按下面的预处理流程过一遍即可如果已经是处理好的 CSV直接跳到 3.4 节开始训练。3.3 数据预处理从原始 CSV 到强化学习能用的状态向量预处理是整个流水线里最不该省的一步。很多强化学习训练不收敛问题不在算法而是喂进去的特征没有标准化、类别列没有编码。下面这段脚本是我每次都会先跑的通用的预处理流程# preprocess.py import pandas as pd import numpy as np from sklearn.preprocessing import LabelEncoder, StandardScaler def load_and_preprocess(raw_csv, save_csv): df pd.read_csv(raw_csv) # 约定最后一列是标签列先取出来单独处理 label_col df.columns[-1] y df[label_col].map(lambda x: 0 if str(x).strip().lower() normal else 1) df df.drop(columns[label_col]) # 分类特征做标签编码不能直接给神经网络 for col in df.select_dtypes(include[object]).columns: df[col] LabelEncoder().fit_transform(df[col].astype(str)) # 数值特征做标准化避免量纲差异淹没模型 num_cols df.select_dtypes(include[np.number]).columns.tolist() scaler StandardScaler() df[num_cols] scaler.fit_transform(df[num_cols]) # 把标签拼回去输出给训练用的 CSV df[label] y.values df.to_csv(save_csv, indexFalse) return scaler, df if __name__ __main__: scaler, _ load_and_preprocess(data/raw/train_raw.csv, data/processed/train.csv)这段代码有两个关键点一是标签必须在标准化之前取出来否则标签列会被 StandardScaler 当成数值特征一起缩放二是 scaler 对象要保存到本地文件推理阶段用同一套均值方差来做 transform不能用新数据的统计量重新 fit。我通常用 pickle 或 joblib 把 scaler 存下来这是后面做真实流量评估时的后悔药——流数据分布和训练集不可能完全一致scaler 不一致会直接放大检测误差。3.4 首次训练最小命令和观察指标数据准备好了之后看下项目里有没有 config.yaml。没有的话自己建一个把最基本的参数写进去env: reward_weights: tp: 1.0 # 正确告警 fp: -0.5 # 误报 fn: -2.0 # 漏报 tn: 0.1 # 正确放行 agent: lr: 0.001 gamma: 0.99 epsilon: 1.0 epsilon_min: 0.05 epsilon_decay: 0.995 train: episodes: 50 batch_size: 64 target_update_freq: 10然后跑训练入口python train.py --data data/processed/train.csv --config config.yaml第一次训练别指望指标完美。我判断训练是否正常启动重点看三件事episode 内累计奖励有没有整体向上的趋势epsilon 是否在按预期衰减训练日志里有没有出现 NaN。只要这三件事正常说明环境封装、数据读取、智能体更新环路是通的。后面再花时间调优千万不要一开始就纠结检测率数字。4. 核心实现环境封装、DQN 智能体与训练主循环4.1 把流量样本包装成 gymnasium 环境整个项目的核心是环境封装。强化学习环境要回答三个问题当前状态是什么下一步能做什么做了之后获得什么反馈下面的代码是我自定义的 NIDSEnv把每条样本当作一个时间步的状态# env/nids_env.py import numpy as np import gymnasium as gym from gymnasium import spaces class NIDSEnv(gym.Env): 网络入侵检测环境。 每一步暴露一条样本的特征智能体决策是否发出告警。 def __init__(self, features, labels, reward_weightsNone): super().__init__() self.features np.array(features, dtypenp.float32) self.labels np.array(labels, dtypenp.int64) self.reward_weights reward_weights or { tp: 1.0, fp: -0.5, fn: -2.0, tn: 0.1 } # 状态就是单条样本的特征向量观测空间是连续值 self.observation_space spaces.Box( low-np.inf, highnp.inf, shape(self.features.shape[1],) ) # 动作 0 表示放行1 表示告警 self.action_space spaces.Discrete(2) self._index 0 def reset(self, *, seedNone, optionsNone): super().reset(seedseed) self._index 0 return self.features[self._index], {} def step(self, action): true_label self.labels[self._index] w self.reward_weights if action 1 and true_label 1: reward w[tp] # 正确告警 elif action 1 and true_label 0: reward w[fp] # 误报 elif action 0 and true_label 1: reward w[fn] # 漏报 else: reward w[tn] # 正确放行 self._index 1 terminated self._index len(self.features) - 1 truncated False if terminated: # 环境终止时返回零向量占位保证 API 一致 next_obs np.zeros_like(self.features[0]) else: next_obs self.features[self._index] return next_obs, reward, terminated, truncated, {}这里要特别说明奖励的设计逻辑。我把漏报的惩罚设成 -2.0误报是 -0.5因为在实际安全运营里一条漏掉的攻击可能造成几小时甚至几天的横向移动而一次误报只是让安全员多点一次“忽略”。正确放行给 0.1是为了让智能体在大量正常流量的场景里不至于因为“什么都不做没有收益”而倾向于随意告警。这个权重矩阵是第一个要调的超参数后面会说它带来的典型问题。4.2 DQN 网络结构与经验回放环境就绪后接下来是 DQN 智能体。网络结构不需要复杂三层全连接加 ReLU 就够处理特征向量类型的状态表示。经验回放缓冲区的作用是打乱样本间的时间相关性让神经网络更新满足独立同分布的假设# agent/dqn_agent.py import random from collections import deque import torch import torch.nn as nn import torch.optim as optim import numpy as np class QNet(nn.Module): def __init__(self, input_dim, output_dim, hidden_dim128): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim) ) def forward(self, x): return self.net(x) class DQNAgent: def __init__(self, input_dim, output_dim, lr1e-3, gamma0.99, epsilon1.0, epsilon_min0.05, epsilon_decay0.995): self.q_net QNet(input_dim, output_dim) self.target_net QNet(input_dim, output_dim) self.target_net.load_state_dict(self.q_net.state_dict()) self.optimizer optim.Adam(self.q_net.parameters(), lrlr) self.gamma gamma self.epsilon epsilon self.epsilon_min epsilon_min self.epsilon_decay epsilon_decay self.replay_buffer deque(maxlen20000) def act(self, state, eval_modeFalse): # 训练阶段 epsilon-greedy 探索评估阶段直接取 argmax if not eval_mode and np.random.rand() self.epsilon: return np.random.randint(0, 2) with torch.no_grad(): q_values self.q_net(torch.FloatTensor(state).unsqueeze(0)) return int(q_values.argmax(dim1).item()) def remember(self, state, action, reward, next_state, done): self.replay_buffer.append((state, action, reward, next_state, done)) def update(self, batch_size64): if len(self.replay_buffer) batch_size: return 0.0 batch random.sample(self.replay_buffer, batch_size) states torch.FloatTensor([x[0] for x in batch]) actions torch.LongTensor([x[1] for x in batch]).unsqueeze(1) rewards torch.FloatTensor([x[2] for x in batch]) next_states torch.FloatTensor([x[3] for x in batch]) dones torch.BoolTensor([x[4] for x in batch]) # 当前 Q 值 q_pred self.q_net(states).gather(1, actions).squeeze(1) # 目标 Q 值用 target_net 计算避免自举造成的震荡 q_next self.target_net(next_states).max(dim1).values.detach() q_target rewards self.gamma * q_next * (~dones) loss nn.MSELoss()(q_pred, q_target) self.optimizer.zero_grad() loss.backward() self.optimizer.step() return loss.item() def update_target(self): self.target_net.load_state_dict(self.q_net.state_dict()) def decay_epsilon(self): self.epsilon max(self.epsilon * self.epsilon_decay, self.epsilon_min)target_net 的引入是 DQN 训练稳定的关键。它不参与梯度更新只负责计算目标 Q 值每隔一定步数把 q_net 的参数复制过去。如果不这样做模型每次更新都用自己刚算出来的值当目标Q 值会陷入正反馈发散Loss 越跑越高。4.3 训练主循环与关键超参数说明训练主循环的骨架比较固定每个 episode 重置环境、按 epsilon-greedy 选动作、存经验、批量更新网络、周期更新目标网络。下面是 train.py 的核心逻辑# train.py 主循环简化版 def train(env, agent, episodes50, batch_size64, target_update_freq10): for ep in range(episodes): obs, _ env.reset() total_reward 0.0 done False step_count 0 while not done: action agent.act(obs) next_obs, reward, terminated, truncated, _ env.step(action) done terminated or truncated agent.remember(obs, action, reward, next_obs, done) loss agent.update(batch_size) obs next_obs total_reward reward step_count 1 if step_count % target_update_freq 0: agent.update_target() agent.decay_epsilon() print(fepisode{ep}, reward{total_reward:.2f}, fepsilon{agent.epsilon:.3f}, loss{loss:.4f})几个关键超参数的作用和调整方向值得记住参数作用起步值调优方向lr控制 Q 网络参数更新步长0.001不收敛时降到 0.0003gamma折扣因子决定远期奖励的重要程度0.99告警时效性强可降到 0.95epsilon_decay探索率衰减速度0.995衰减太快会过早进入纯利用错过攻击模式replay_buffer 容量经验池大小20000样本类别不均时加大到 50000batch_size每次梯度更新使用的样本数64训练不稳时加大到 128target_update_freq目标网络同步频率10波动大时可以降低到 20如果项目代码里用的是 Double DQN改动点只在 update 函数用 q_net 选动作、用 target_net 算 Q 值其余部分完全相同。建议把 DQN 跑通后再切 Double DQN 做对比你会发现 Q 值过估计带来的误报问题有明显改善。5. 入坑排查检测率上不去的 5 个典型问题5.1 智能体学会了“全放行”奖励函数被钻了空子现象训练日志里累计奖励稳步上升但测试集上检测率接近 0误报也很少。看起来训练“成功”了实际上智能体找到了最偷懒的策略——所有流量都放行。原因数据集里正常样本占比很高时全放行的期望奖励未必是负的。比如正常样本占 90%漏报惩罚是 -2正确放行是 0.1那么全放行每个样本的期望奖励是 0.9×0.1 0.1×(-2) -0.11。如果漏报惩罚不够重这个负数可能比乱告警造成的期望损失还“划算”智能体就学不到告警行为。解决把漏报惩罚调大。我一般会把 fn 压到 -5 以下必要时按类别比例加权。调完奖励矩阵后最关键的是别只看训练奖励要在测试集上算 F1。只盯训练 reward 会把这个“全放行”策略误判成好模型。5.2 训练 Loss 像玄学一样震荡学习率与 epsilon 衰减不匹配现象Loss 忽高忽低reward 曲线大幅摆动甚至出现 Q 值输出到几万的情况。训练跑了二三十个 episode模型好像失忆了一样。原因学习率太大导致参数更新幅度不稳定epsilon 衰减太快模型过早放弃探索陷入局部最优target_net 更新太频繁目标值本身在剧烈变化。这三条可以独立导致震荡更多时候是叠加作用。解决先换一组保守的参数。我的起步组合是 lr0.0003、epsilon_decay0.998、target_update_freq20、batch_size128。这组参数牺牲了收敛速度换来的是一眼看得到的稳定曲线。如果还想再稳一点就上 Double DQNQ 值过估计带来的正向偏差能被明显压住。调参这一步最忌讳一次动好几个参数改一个、跑一轮、看一个否则出了问题说不清是谁引起的。5.3 少数类攻击检测率趋近于零类别不均衡问题现象测试集里多数攻击类别比如 DoS检测率尚可但少数类攻击比如 U2R 这类样本量极少的攻击几乎全部漏检。训练时的平均奖励还挺正常。原因少数类样本在经验回放里出现的频次太少每次采样到它们的概率接近零神经网络根本没有机会针对这类样本做有效的梯度更新。强化学习的奖励函数是按样本平均的少数类贡献的梯度被多数类淹没。解决最直接的做法是在预处理阶段对少数类做上采样复制少数类样本让它们在批次里占据合理比例。另外在构造批次时可以按标签分层采样保证每个 batch 里都有少数类。用专业一点的方案还可以上优先经验回放PER把那些让智能体“猜错”的样本放在更大概率被采样到的位置。验证时务必用 F1 而不是 accuracyaccuracy 在不均衡数据上会给出虚高假象。5.4 灾难性遗忘训练后期把早期学会的攻击又忘了现象训练到中间某个 episode 时检测率不错继续训练后反而下降尤其是一开始学到的那几类攻击又开始漏报。感觉模型学会了后面的把前面的忘了。原因网络参数在持续更新新学到的策略会覆盖旧策略。如果把整份训练数据当流一样反复过每个 episode 的采样顺序不同模型容易被最近批次的数据带偏。经验回放虽然缓解了时序相关性但缓冲区容量有限早期样本不断被弹出。解决两个手段配合使用。一个是扩大回放缓冲区到 50000 以上让早期经验在更长时间内保持存在感另一个是训练过程中定期保存 checkpoint并在固定的验证集上评估最后选择验证指标最好的 checkpoint 而不是最后一个。工程上这一条极其重要很多最终效果差的项目都是直接拿最后一个模型上线的。5.5 测试集指标好看、真实流量翻车特征分布漂移现象在测试集上 F1 有 0.92但放到真实网络环境里跑误报陡增正常业务流量被大量告警。明明是同一个模型换个场景就崩。原因两个层面的问题叠加。第一预处理时 scaler 用的是训练集的均值和方差真实流量的特征分布比训练集宽标准化后数值落到了模型没见过的范围第二强化学习训练时环境暴露的是数据集里的样本这些样本是离线抓取的、相对“干净”真实流量里的噪声和冗余字段会干扰状态表示。解决推理阶段必须保存并使用训练时的 scaler不能重新 fit。在数据链路层面给推理特征做范围截断超出训练分布的部分压缩到边界值。更实际的做法是在推理端做二次确认比如叠加滑动窗口规则单条告警不直接触发连续多条告警才真正发出这样能把偶发性的分布漂移噪声过滤掉。真实流量环境里的长期稳定运行要靠定期用新数据重训模型来维持这不是一次训练能解决的。6. 把模型从离线搬到在线两个验证门槛和一个降低误报的技巧训练脚本跑通只是开始真正上线前我有两个强制门槛。第一个是复现验证固定随机种子、固定推理模式、固定数据顺序重复三次评估指标波动超过一个百分点就要排查。实现上就是把下面这段放在评估脚本最前面# evaluate.py 开头固定随机种子 import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) set_seed()第二个门槛是评估指标要用混淆矩阵说话不能拿训练时的 reward 当业绩。训练奖励反映的是“智能体有没有学会获取奖励的技巧”不代表检测能力。我一般会输出一列测试集混淆矩阵算好 precision、recall、F1再对比一个 Random 策略和全放行策略的基线。如果 DQN 连这两个 trivial baseline 都打不过说明奖励函数设计有问题不值得花时间上线。在线推理阶段我最常用也最推荐的一个技巧是滑动窗口二次确认。单个样本的告警决策波动太大真实网络里一条正常访问可能因为特征分布边缘触发误报。把 DQN 的原始输出累积到一个窗口里连续多次判断为攻击才真正发出告警误报率能降一个量级# detector.pyDQN 原始输出到最终告警的滑动窗口 class SlidingWindowDetector: def __init__(self, window20, threshold3): self.decisions [] self.window window self.threshold threshold def update(self, dqn_alert: bool) - bool: self.decisions.append(dqn_alert) if len(self.decisions) self.window: self.decisions.pop(0) # 窗口内告警次数达到阈值才真正触发 return sum(self.decisions) self.thresholdwindow 和 threshold 的比例决定了告警灵敏度和误报的取舍。window20、threshold3 意味着攻击行为需要短时间内在多个样本上持续触发这对扫描和渗透类的多步攻击很有效但对单包攻击可能不敏感。我在实际使用中会保留两组配置一组给边界防护设备用一组给内网流量审计用。最后说一个我自己栽过的坑早期我以为强化学习训练出来的模型天然适应在线环境结果发现真实流量里一个很小的字段空值就能让环境封装报错。后来我养成了一个习惯每次做完范化推理都先跑一遍空值和异常特征注入测试确保模型在坏数据下也能退化成“安全侧”的保守判断——宁可漏报一条也比整个检测进程崩溃强。希望这些能帮你在跑通这个项目时少走几步弯路。本文还有配套的精品资源点击获取