简介这份资源面向计算机、人工智能及网络安全方向的学习者与研究人员提供一套基于动态图神经网络的异常流量检测完整实现方案用于解决传统静态拓扑方法在动态网络环境中准确率与效率不足的问题。压缩包共141个文件约34.94MB以60个Python源码文件为核心辅以56个编译缓存、8个模型权重文件、4份CSV数据集、3份JSON配置及论文PDF、说明文档等覆盖数据预处理、特征提取、模型训练到评估的完整链路。项目通过多层网络结构学习流量时间序列中的节点与边动态特性并借助预训练模型参数支持快速推理。目前已有69人学习下载适合作为毕业设计、课程设计或科研参考。读者可获得可运行的源码、逐行注释、项目说明与论文资料便于理解DGNN工作机制、复现实验并迁移到实际网络监控场景。1. 动态图神经网络做异常流量检测为什么静态模型总在真实网络里翻车线上流量从来不是一张固定的图。早上九点办公网里全是内网横向的 SMB 会话凌晨三点变成几台机器往境外 IP 打小包周末又换成 CDN 回源的长连接。你拿一周前的拓扑训一个 GCN上线第一天 AUC 就掉到 0.6这不是模型不行是图本身在动。动态图神经网络Dynamic Graph Neural NetworkDGNN要解决的就是这件事把「谁在跟谁通信」建成随时间演化的图节点是 IP 或主机边是会话边权是流量统计量然后让模型在时间维度上学习正常与异常的差异。这套方法适合做内网东西向流量检测、DDoS 早期识别、C2 心跳发现也适合已经有一批 NetFlow 或 Zeek 日志、想从规则阈值升级到模型的人。标题里那份 python 源码加论文的组合本质就是给你一条从原始流量到图快照、再到 DGNN 训练和推理的完整链路下面我按自己复现过的顺序把它拆开讲。2. 从 pcap 到动态图数据管线的四个关键决策2.1 为什么选 NetFlow 聚合而不是逐包建图逐包建图听起来最保真但一个千兆口一天就是几亿条边DGNN 的邻居聚合根本跑不动。常见做法是按固定时间窗做流聚合比如 5 秒或 1 分钟一个快照每个快照内把五元组相同的包合并成一条边。窗口太短正常的长连接会被切碎模型学到一堆碎片窗口太长DDoS 的突发特征被平均掉。我一般先用 10 秒窗跑一遍统计看每个快照的边数是否稳定在几千到几万量级这个量级单卡才能吃得下。聚合时至少要保留这些字段源 IP、目的 IP、源端口、目的端口、协议号、包数、字节数、持续时间、TCP 标志位计数。前五个用来确定边的两端后四个是边特征。端口和协议号不要直接当数值喂进去要做 one-hot 或 embedding否则模型会以为 443 比 80 大。import pandas as pd # 假设原始 flow 记录已经带时间戳按 10 秒窗口聚合 def build_edge_features(flow_df, window10s): flow_df[ts] pd.to_datetime(flow_df[ts]) flow_df flow_df.set_index(ts) # 同一窗口内同五元组聚合 grouped flow_df.groupby([ pd.Grouper(freqwindow), src_ip, dst_ip, src_port, dst_port, proto ]).agg( pkt_count(pkt_count, sum), byte_count(byte_count, sum), duration(duration, max), syn_cnt(syn_flag, sum), rst_cnt(rst_flag, sum) ).reset_index() return grouped这段代码的逻辑是先把时间戳设为索引再用pd.Grouper按窗口切分同一窗口内相同五元组的记录合并。pkt_count和byte_count用求和duration取最大值标志位计数求和。参数上window是最需要调的建议从 5s、10s、30s、60s 各跑一遍看后续模型验证集上的表现再定。2.2 节点特征怎么构造才不丢信息边建好了节点特征不能只放一个 IP 字符串。我一般给每个 IP 节点拼三类特征一是该窗口内作为源和作为目的的次数二是收发字节比三是历史窗口的滑动统计比如过去 5 个窗口的平均出度。滑动统计是让模型感知「这个 IP 平时很安静突然开始扫段」的关键。没有历史对比单窗口的绝对数值很难区分正常的高流量和异常的突发。# 为每个窗口的每个 IP 计算节点特征 def build_node_features(edge_df): src_stat edge_df.groupby([window, src_ip]).agg( out_deg(dst_ip, nunique), out_pkt(pkt_count, sum), out_byte(byte_count, sum) ).reset_index().rename(columns{src_ip: ip}) dst_stat edge_df.groupby([window, dst_ip]).agg( in_deg(src_ip, nunique), in_pkt(pkt_count, sum), in_byte(byte_count, sum) ).reset_index().rename(columns{dst_ip: ip}) node_df pd.merge(src_stat, dst_stat, on[window, ip], howouter).fillna(0) # 滑动窗口均值窗口大小 5 node_df node_df.sort_values([ip, window]) for col in [out_deg, out_pkt, out_byte, in_deg, in_pkt, in_byte]: node_df[f{col}_ma5] node_df.groupby(ip)[col].transform( lambda x: x.rolling(5, min_periods1).mean() ) return node_df这里out_deg是该 IP 在窗口内连了多少个不同目的 IP扫描行为会让这个值异常高。_ma5后缀是 5 窗口滑动均值用来给模型提供基线。注意fillna(0)之后要检查一下有没有 IP 只在目的侧出现这类节点如果直接丢会漏掉被攻击目标。2.3 快照序列的切分与标签对齐动态图模型吃的是快照序列不是单张图。假设你用 10 秒窗那 1 小时就是 360 个快照。训练时通常取连续 20 到 50 个快照作为一个样本预测下一个或下一段快照里哪些边是异常的。标签来自你手里的告警日志或人工标注常见做法是把告警时间点前后各扩一个窗口避免边界漏标。def make_sequences(node_df, edge_df, label_df, seq_len30): windows sorted(edge_df[window].unique()) samples [] for i in range(len(windows) - seq_len): seq_windows windows[i:iseq_len] next_window windows[iseq_len] seq_edges edge_df[edge_df[window].isin(seq_windows)] seq_nodes node_df[node_df[window].isin(seq_windows)] # 下一窗口的边标签 next_edges edge_df[edge_df[window] next_window].copy() next_edges next_edges.merge(label_df, on[src_ip,dst_ip], howleft) next_edges[label] next_edges[label].fillna(0) samples.append((seq_nodes, seq_edges, next_edges)) return samplesseq_len是序列长度太小模型看不到演化趋势太大显存吃不消。我一般从 20 开始试显存够就加到 50。标签对齐时howleft保证正常边不会被丢掉fillna(0)把未标注边当负样本但这里有个坑未标注不等于正常后面避坑章节会细说。2.4 图快照的存储格式与加载速度快照多了以后CSV 读写会成为瓶颈。我一般把每个快照存成单独的.npz或.pt节点特征矩阵和边索引分开存。边索引用 COO 格式两行分别是源节点在节点表中的下标和目的节点下标。加载时用内存映射不要一次性全读进内存。import numpy as np import torch def save_snapshot(window, node_feat, edge_index, edge_attr, path): np.savez_compressed( f{path}/{window}.npz, node_featnode_feat.astype(np.float32), edge_indexedge_index.astype(np.int64), edge_attredge_attr.astype(np.float32) ) def load_snapshot(window, path): data np.load(f{path}/{window}.npz) return ( torch.from_numpy(data[node_feat]), torch.from_numpy(data[edge_index]), torch.from_numpy(data[edge_attr]) )np.savez_compressed比 pickle 安全也比 CSV 快一个数量级。edge_index用 int64 是 PyG 的默认要求node_feat和edge_attr转 float32 省显存。如果快照数量超过几千建议再建一个索引文件记录每个快照的节点数和边数加载时按需分配。3. DGNN 模型选型与训练把演化信息真正用起来3.1 为什么在流量场景里 GRU 加 GCN 比纯 Transformer 稳纯 Transformer 做动态图不是不行但流量图的节点数每个窗口都在变位置编码很难对齐而且自注意力的 O(N²) 在几万节点时直接爆显存。我试过在同样数据上跑纯 Transformer验证集 loss 震荡得厉害换成 GRU 沿时间维更新节点状态、GCN 做空间聚合之后收敛曲线平滑很多。常见做法是每个快照先用 GCN 聚合邻居再把聚合后的节点表示送进 GRUGRU 的隐藏状态跨窗口传递。import torch.nn as nn import torch.nn.functional as F from torch_geometric.nn import GCNConv class DGNN(nn.Module): def __init__(self, node_dim, edge_dim, hidden64, num_layers2): super().__init__() self.gcn_layers nn.ModuleList([ GCNConv(node_dim if i 0 else hidden, hidden) for i in range(num_layers) ]) self.edge_encoder nn.Linear(edge_dim, hidden) self.gru nn.GRU(hidden, hidden, batch_firstTrue) self.classifier nn.Linear(hidden * 2, 1) def forward(self, node_feats, edge_indices, edge_attrs): # node_feats: [T, N, node_dim] T, N, _ node_feats.shape h None outputs [] for t in range(T): x node_feats[t] ei edge_indices[t] for gcn in self.gcn_layers: x F.relu(gcn(x, ei)) x x.unsqueeze(0) # [1, N, hidden] out, h self.gru(x, h) outputs.append(out.squeeze(0)) return torch.stack(outputs, dim0)GCNConv的层数不要超过 3流量图里两跳邻居已经能覆盖大部分通信关系再深会过平滑。GRU的隐藏维度跟 GCN 输出保持一致省去投影。classifier里hidden * 2是因为后面要把源节点和目的节点的表示拼接起来判断这条边是否异常。3.2 边分类头的设计源节点和目的节点怎么拼异常检测最终要落到边上所以分类头不能只看单个节点。我一般把边的两个端点表示取出来拼上边特征再过两层 MLP。拼接方式有 concat、hadamard 积、差值三种实测 concat 最稳hadamard 积在稀疏图上容易丢信息。def edge_predict(self, node_emb, edge_index, edge_attr): src, dst edge_index[0], edge_index[1] src_emb node_emb[src] dst_emb node_emb[dst] edge_emb self.edge_encoder(edge_attr) combined torch.cat([src_emb, dst_emb, edge_emb], dim-1) return self.classifier(combined).squeeze(-1)edge_encoder把原始边特征投影到跟节点表示同维避免量纲差异。classifier输出 logit训练时用BCEWithLogitsLoss推理时过 sigmoid 得到概率。如果正负样本极度不均衡可以在 loss 里加pos_weight一般设成负样本数除以正样本数。3.3 训练循环里必须监控的三个量训练 DGNN 不能只看 loss。我一般同时盯验证集 AUC、正样本召回率、以及每个 epoch 的梯度范数。AUC 看整体排序能力召回率看漏报梯度范数看有没有梯度爆炸。流量数据里正样本通常不到 1%AUC 高但召回低是常态这时候要调阈值或加 focal loss。from sklearn.metrics import roc_auc_score, recall_score def train_epoch(model, loader, optimizer, pos_weight): model.train() total_loss 0 for seq_nodes, seq_edges, next_edges in loader: optimizer.zero_grad() node_emb model(seq_nodes, seq_edges, seq_edges) logits model.edge_predict(node_emb[-1], next_edges.edge_index, next_edges.edge_attr) loss F.binary_cross_entropy_with_logits( logits, next_edges.label, pos_weightpos_weight ) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm5.0) optimizer.step() total_loss loss.item() return total_loss / len(loader)clip_grad_norm_的max_norm设 5.0 是经验值梯度再大就说明学习率可能偏高。pos_weight用torch.tensor([neg/pos])传进去。验证时不要用训练阈值要在验证集上扫一遍找最佳 F1 对应的阈值再固定下来用于测试。3.4 推理阶段的滑动窗口与在线更新线上推理不是拿一个固定序列跑一次就完事。我一般维护一个长度为seq_len的滑动窗口队列每来一个新快照就弹出最旧的、压入最新的然后跑一次前向。模型参数可以定期用新数据微调但不要每个窗口都更新否则容易灾难性遗忘。from collections import deque class OnlineDetector: def __init__(self, model, seq_len30): self.model model self.seq_len seq_len self.buffer deque(maxlenseq_len) def update(self, node_feat, edge_index, edge_attr): self.buffer.append((node_feat, edge_index, edge_attr)) if len(self.buffer) self.seq_len: return None seq_nodes [b[0] for b in self.buffer] seq_edges [b[1] for b in self.buffer] with torch.no_grad(): node_emb self.model(seq_nodes, seq_edges, seq_edges) logits self.model.edge_predict( node_emb[-1], edge_index, edge_attr ) return torch.sigmoid(logits)deque的maxlen自动控制窗口长度。推理时torch.no_grad()省显存。返回的概率要跟固定阈值比较阈值来自验证集。如果线上分布漂移明显可以每周用最近数据重新扫一次阈值。4. 避坑与排查复现时最容易翻车的五个地方4.1 未标注边当负样本导致召回率虚低现象是验证集 AUC 0.95 但召回只有 0.3。原因是标签只覆盖了告警涉及的边大量正常边没标注被fillna(0)当成负样本模型学到「没标注就是正常」遇到真正异常时反而不敢报。解决方法是做负样本采样只从确认正常的边里抽或者用 PU learning 的思路给未标注边低权重。4.2 节点表每个窗口重建导致 embedding 对不上现象是训练时 loss 正常推理时同一 IP 的表示每次都不一样。原因是每个快照单独建节点表IP 到下标的映射变了。解决方法是维护全局 IP 字典所有快照共用一套下标新 IP 追加到末尾旧 IP 不删除。4.3 时间窗边界把一条会话切成两半现象是某些正常长连接被反复报异常。原因是 10 秒窗刚好切在会话中间两个窗口各拿到一半包数滑动均值波动大。解决方法是聚合时按会话开始时间归窗或者把窗口边界做重叠比如步长 5 秒、窗长 10 秒。4.4 正负样本比失衡时 AUC 骗人现象是 AUC 很高但实际漏报严重。原因是负样本太多模型只要把少数正样本排前面 AUC 就好看但阈值一卡就漏。解决方法是看 PR 曲线和召回率用 focal loss 或调整pos_weight不要只盯 AUC。4.5 显存不够时盲目减层现象是 OOM 后把 GCN 从 2 层减到 1 层效果掉很多。原因是瓶颈在快照序列长度和节点数不在层数。解决方法是先减seq_len再用邻居采样最后才考虑减层。邻居采样用 PyG 的NeighborLoader每层采 10 到 15 个邻居就够。5. 把模型压到线上量化、阈值扫描与一个可复用的验证脚本模型训完只是第一步能不能上线看推理延迟和阈值稳定性。我一般先做动态量化把 GRU 和 Linear 层转成 int8延迟能降三到四成AUC 掉不到 0.01。量化后再扫一遍阈值因为量化会轻微改变输出分布。import torch.quantization def quantize_model(model): model.eval() quantized torch.quantization.quantize_dynamic( model, {nn.GRU, nn.Linear}, dtypetorch.qint8 ) return quantized def scan_threshold(model, val_loader): model.eval() all_probs, all_labels [], [] with torch.no_grad(): for seq_nodes, seq_edges, next_edges in val_loader: node_emb model(seq_nodes, seq_edges, seq_edges) logits model.edge_predict( node_emb[-1], next_edges.edge_index, next_edges.edge_attr ) all_probs.append(torch.sigmoid(logits)) all_labels.append(next_edges.label) probs torch.cat(all_probs).numpy() labels torch.cat(all_labels).numpy() best_f1, best_th 0, 0.5 for th in np.arange(0.1, 0.9, 0.01): pred (probs th).astype(int) f1 f1_score(labels, pred, zero_division0) if f1 best_f1: best_f1, best_th f1, th return best_th, best_f1quantize_dynamic只量化 GRU 和 LinearGCN 的稀疏操作量化收益不大。scan_threshold在 0.1 到 0.9 之间以 0.01 步长扫返回最佳 F1 和对应阈值。这个脚本我每次换数据都会跑一遍阈值不要写死在代码里放配置文件。验证方法上除了离线指标我习惯再做一个「时间外推」测试用前 70% 时间的数据训练后 30% 测试看性能衰减。如果衰减超过 15%说明模型对时间漂移敏感需要加滑动统计特征或缩短重训周期。这个习惯帮我提前发现过好几次线上翻车比只看随机划分的验证集靠谱得多。希望帮到你。本文还有配套的精品资源点击获取 动态图神经网络DGNN实战：异常流量检测从pcap到线上部署
简介这份资源面向计算机、人工智能及网络安全方向的学习者与研究人员提供一套基于动态图神经网络的异常流量检测完整实现方案用于解决传统静态拓扑方法在动态网络环境中准确率与效率不足的问题。压缩包共141个文件约34.94MB以60个Python源码文件为核心辅以56个编译缓存、8个模型权重文件、4份CSV数据集、3份JSON配置及论文PDF、说明文档等覆盖数据预处理、特征提取、模型训练到评估的完整链路。项目通过多层网络结构学习流量时间序列中的节点与边动态特性并借助预训练模型参数支持快速推理。目前已有69人学习下载适合作为毕业设计、课程设计或科研参考。读者可获得可运行的源码、逐行注释、项目说明与论文资料便于理解DGNN工作机制、复现实验并迁移到实际网络监控场景。1. 动态图神经网络做异常流量检测为什么静态模型总在真实网络里翻车线上流量从来不是一张固定的图。早上九点办公网里全是内网横向的 SMB 会话凌晨三点变成几台机器往境外 IP 打小包周末又换成 CDN 回源的长连接。你拿一周前的拓扑训一个 GCN上线第一天 AUC 就掉到 0.6这不是模型不行是图本身在动。动态图神经网络Dynamic Graph Neural NetworkDGNN要解决的就是这件事把「谁在跟谁通信」建成随时间演化的图节点是 IP 或主机边是会话边权是流量统计量然后让模型在时间维度上学习正常与异常的差异。这套方法适合做内网东西向流量检测、DDoS 早期识别、C2 心跳发现也适合已经有一批 NetFlow 或 Zeek 日志、想从规则阈值升级到模型的人。标题里那份 python 源码加论文的组合本质就是给你一条从原始流量到图快照、再到 DGNN 训练和推理的完整链路下面我按自己复现过的顺序把它拆开讲。2. 从 pcap 到动态图数据管线的四个关键决策2.1 为什么选 NetFlow 聚合而不是逐包建图逐包建图听起来最保真但一个千兆口一天就是几亿条边DGNN 的邻居聚合根本跑不动。常见做法是按固定时间窗做流聚合比如 5 秒或 1 分钟一个快照每个快照内把五元组相同的包合并成一条边。窗口太短正常的长连接会被切碎模型学到一堆碎片窗口太长DDoS 的突发特征被平均掉。我一般先用 10 秒窗跑一遍统计看每个快照的边数是否稳定在几千到几万量级这个量级单卡才能吃得下。聚合时至少要保留这些字段源 IP、目的 IP、源端口、目的端口、协议号、包数、字节数、持续时间、TCP 标志位计数。前五个用来确定边的两端后四个是边特征。端口和协议号不要直接当数值喂进去要做 one-hot 或 embedding否则模型会以为 443 比 80 大。import pandas as pd # 假设原始 flow 记录已经带时间戳按 10 秒窗口聚合 def build_edge_features(flow_df, window10s): flow_df[ts] pd.to_datetime(flow_df[ts]) flow_df flow_df.set_index(ts) # 同一窗口内同五元组聚合 grouped flow_df.groupby([ pd.Grouper(freqwindow), src_ip, dst_ip, src_port, dst_port, proto ]).agg( pkt_count(pkt_count, sum), byte_count(byte_count, sum), duration(duration, max), syn_cnt(syn_flag, sum), rst_cnt(rst_flag, sum) ).reset_index() return grouped这段代码的逻辑是先把时间戳设为索引再用pd.Grouper按窗口切分同一窗口内相同五元组的记录合并。pkt_count和byte_count用求和duration取最大值标志位计数求和。参数上window是最需要调的建议从 5s、10s、30s、60s 各跑一遍看后续模型验证集上的表现再定。2.2 节点特征怎么构造才不丢信息边建好了节点特征不能只放一个 IP 字符串。我一般给每个 IP 节点拼三类特征一是该窗口内作为源和作为目的的次数二是收发字节比三是历史窗口的滑动统计比如过去 5 个窗口的平均出度。滑动统计是让模型感知「这个 IP 平时很安静突然开始扫段」的关键。没有历史对比单窗口的绝对数值很难区分正常的高流量和异常的突发。# 为每个窗口的每个 IP 计算节点特征 def build_node_features(edge_df): src_stat edge_df.groupby([window, src_ip]).agg( out_deg(dst_ip, nunique), out_pkt(pkt_count, sum), out_byte(byte_count, sum) ).reset_index().rename(columns{src_ip: ip}) dst_stat edge_df.groupby([window, dst_ip]).agg( in_deg(src_ip, nunique), in_pkt(pkt_count, sum), in_byte(byte_count, sum) ).reset_index().rename(columns{dst_ip: ip}) node_df pd.merge(src_stat, dst_stat, on[window, ip], howouter).fillna(0) # 滑动窗口均值窗口大小 5 node_df node_df.sort_values([ip, window]) for col in [out_deg, out_pkt, out_byte, in_deg, in_pkt, in_byte]: node_df[f{col}_ma5] node_df.groupby(ip)[col].transform( lambda x: x.rolling(5, min_periods1).mean() ) return node_df这里out_deg是该 IP 在窗口内连了多少个不同目的 IP扫描行为会让这个值异常高。_ma5后缀是 5 窗口滑动均值用来给模型提供基线。注意fillna(0)之后要检查一下有没有 IP 只在目的侧出现这类节点如果直接丢会漏掉被攻击目标。2.3 快照序列的切分与标签对齐动态图模型吃的是快照序列不是单张图。假设你用 10 秒窗那 1 小时就是 360 个快照。训练时通常取连续 20 到 50 个快照作为一个样本预测下一个或下一段快照里哪些边是异常的。标签来自你手里的告警日志或人工标注常见做法是把告警时间点前后各扩一个窗口避免边界漏标。def make_sequences(node_df, edge_df, label_df, seq_len30): windows sorted(edge_df[window].unique()) samples [] for i in range(len(windows) - seq_len): seq_windows windows[i:iseq_len] next_window windows[iseq_len] seq_edges edge_df[edge_df[window].isin(seq_windows)] seq_nodes node_df[node_df[window].isin(seq_windows)] # 下一窗口的边标签 next_edges edge_df[edge_df[window] next_window].copy() next_edges next_edges.merge(label_df, on[src_ip,dst_ip], howleft) next_edges[label] next_edges[label].fillna(0) samples.append((seq_nodes, seq_edges, next_edges)) return samplesseq_len是序列长度太小模型看不到演化趋势太大显存吃不消。我一般从 20 开始试显存够就加到 50。标签对齐时howleft保证正常边不会被丢掉fillna(0)把未标注边当负样本但这里有个坑未标注不等于正常后面避坑章节会细说。2.4 图快照的存储格式与加载速度快照多了以后CSV 读写会成为瓶颈。我一般把每个快照存成单独的.npz或.pt节点特征矩阵和边索引分开存。边索引用 COO 格式两行分别是源节点在节点表中的下标和目的节点下标。加载时用内存映射不要一次性全读进内存。import numpy as np import torch def save_snapshot(window, node_feat, edge_index, edge_attr, path): np.savez_compressed( f{path}/{window}.npz, node_featnode_feat.astype(np.float32), edge_indexedge_index.astype(np.int64), edge_attredge_attr.astype(np.float32) ) def load_snapshot(window, path): data np.load(f{path}/{window}.npz) return ( torch.from_numpy(data[node_feat]), torch.from_numpy(data[edge_index]), torch.from_numpy(data[edge_attr]) )np.savez_compressed比 pickle 安全也比 CSV 快一个数量级。edge_index用 int64 是 PyG 的默认要求node_feat和edge_attr转 float32 省显存。如果快照数量超过几千建议再建一个索引文件记录每个快照的节点数和边数加载时按需分配。3. DGNN 模型选型与训练把演化信息真正用起来3.1 为什么在流量场景里 GRU 加 GCN 比纯 Transformer 稳纯 Transformer 做动态图不是不行但流量图的节点数每个窗口都在变位置编码很难对齐而且自注意力的 O(N²) 在几万节点时直接爆显存。我试过在同样数据上跑纯 Transformer验证集 loss 震荡得厉害换成 GRU 沿时间维更新节点状态、GCN 做空间聚合之后收敛曲线平滑很多。常见做法是每个快照先用 GCN 聚合邻居再把聚合后的节点表示送进 GRUGRU 的隐藏状态跨窗口传递。import torch.nn as nn import torch.nn.functional as F from torch_geometric.nn import GCNConv class DGNN(nn.Module): def __init__(self, node_dim, edge_dim, hidden64, num_layers2): super().__init__() self.gcn_layers nn.ModuleList([ GCNConv(node_dim if i 0 else hidden, hidden) for i in range(num_layers) ]) self.edge_encoder nn.Linear(edge_dim, hidden) self.gru nn.GRU(hidden, hidden, batch_firstTrue) self.classifier nn.Linear(hidden * 2, 1) def forward(self, node_feats, edge_indices, edge_attrs): # node_feats: [T, N, node_dim] T, N, _ node_feats.shape h None outputs [] for t in range(T): x node_feats[t] ei edge_indices[t] for gcn in self.gcn_layers: x F.relu(gcn(x, ei)) x x.unsqueeze(0) # [1, N, hidden] out, h self.gru(x, h) outputs.append(out.squeeze(0)) return torch.stack(outputs, dim0)GCNConv的层数不要超过 3流量图里两跳邻居已经能覆盖大部分通信关系再深会过平滑。GRU的隐藏维度跟 GCN 输出保持一致省去投影。classifier里hidden * 2是因为后面要把源节点和目的节点的表示拼接起来判断这条边是否异常。3.2 边分类头的设计源节点和目的节点怎么拼异常检测最终要落到边上所以分类头不能只看单个节点。我一般把边的两个端点表示取出来拼上边特征再过两层 MLP。拼接方式有 concat、hadamard 积、差值三种实测 concat 最稳hadamard 积在稀疏图上容易丢信息。def edge_predict(self, node_emb, edge_index, edge_attr): src, dst edge_index[0], edge_index[1] src_emb node_emb[src] dst_emb node_emb[dst] edge_emb self.edge_encoder(edge_attr) combined torch.cat([src_emb, dst_emb, edge_emb], dim-1) return self.classifier(combined).squeeze(-1)edge_encoder把原始边特征投影到跟节点表示同维避免量纲差异。classifier输出 logit训练时用BCEWithLogitsLoss推理时过 sigmoid 得到概率。如果正负样本极度不均衡可以在 loss 里加pos_weight一般设成负样本数除以正样本数。3.3 训练循环里必须监控的三个量训练 DGNN 不能只看 loss。我一般同时盯验证集 AUC、正样本召回率、以及每个 epoch 的梯度范数。AUC 看整体排序能力召回率看漏报梯度范数看有没有梯度爆炸。流量数据里正样本通常不到 1%AUC 高但召回低是常态这时候要调阈值或加 focal loss。from sklearn.metrics import roc_auc_score, recall_score def train_epoch(model, loader, optimizer, pos_weight): model.train() total_loss 0 for seq_nodes, seq_edges, next_edges in loader: optimizer.zero_grad() node_emb model(seq_nodes, seq_edges, seq_edges) logits model.edge_predict(node_emb[-1], next_edges.edge_index, next_edges.edge_attr) loss F.binary_cross_entropy_with_logits( logits, next_edges.label, pos_weightpos_weight ) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm5.0) optimizer.step() total_loss loss.item() return total_loss / len(loader)clip_grad_norm_的max_norm设 5.0 是经验值梯度再大就说明学习率可能偏高。pos_weight用torch.tensor([neg/pos])传进去。验证时不要用训练阈值要在验证集上扫一遍找最佳 F1 对应的阈值再固定下来用于测试。3.4 推理阶段的滑动窗口与在线更新线上推理不是拿一个固定序列跑一次就完事。我一般维护一个长度为seq_len的滑动窗口队列每来一个新快照就弹出最旧的、压入最新的然后跑一次前向。模型参数可以定期用新数据微调但不要每个窗口都更新否则容易灾难性遗忘。from collections import deque class OnlineDetector: def __init__(self, model, seq_len30): self.model model self.seq_len seq_len self.buffer deque(maxlenseq_len) def update(self, node_feat, edge_index, edge_attr): self.buffer.append((node_feat, edge_index, edge_attr)) if len(self.buffer) self.seq_len: return None seq_nodes [b[0] for b in self.buffer] seq_edges [b[1] for b in self.buffer] with torch.no_grad(): node_emb self.model(seq_nodes, seq_edges, seq_edges) logits self.model.edge_predict( node_emb[-1], edge_index, edge_attr ) return torch.sigmoid(logits)deque的maxlen自动控制窗口长度。推理时torch.no_grad()省显存。返回的概率要跟固定阈值比较阈值来自验证集。如果线上分布漂移明显可以每周用最近数据重新扫一次阈值。4. 避坑与排查复现时最容易翻车的五个地方4.1 未标注边当负样本导致召回率虚低现象是验证集 AUC 0.95 但召回只有 0.3。原因是标签只覆盖了告警涉及的边大量正常边没标注被fillna(0)当成负样本模型学到「没标注就是正常」遇到真正异常时反而不敢报。解决方法是做负样本采样只从确认正常的边里抽或者用 PU learning 的思路给未标注边低权重。4.2 节点表每个窗口重建导致 embedding 对不上现象是训练时 loss 正常推理时同一 IP 的表示每次都不一样。原因是每个快照单独建节点表IP 到下标的映射变了。解决方法是维护全局 IP 字典所有快照共用一套下标新 IP 追加到末尾旧 IP 不删除。4.3 时间窗边界把一条会话切成两半现象是某些正常长连接被反复报异常。原因是 10 秒窗刚好切在会话中间两个窗口各拿到一半包数滑动均值波动大。解决方法是聚合时按会话开始时间归窗或者把窗口边界做重叠比如步长 5 秒、窗长 10 秒。4.4 正负样本比失衡时 AUC 骗人现象是 AUC 很高但实际漏报严重。原因是负样本太多模型只要把少数正样本排前面 AUC 就好看但阈值一卡就漏。解决方法是看 PR 曲线和召回率用 focal loss 或调整pos_weight不要只盯 AUC。4.5 显存不够时盲目减层现象是 OOM 后把 GCN 从 2 层减到 1 层效果掉很多。原因是瓶颈在快照序列长度和节点数不在层数。解决方法是先减seq_len再用邻居采样最后才考虑减层。邻居采样用 PyG 的NeighborLoader每层采 10 到 15 个邻居就够。5. 把模型压到线上量化、阈值扫描与一个可复用的验证脚本模型训完只是第一步能不能上线看推理延迟和阈值稳定性。我一般先做动态量化把 GRU 和 Linear 层转成 int8延迟能降三到四成AUC 掉不到 0.01。量化后再扫一遍阈值因为量化会轻微改变输出分布。import torch.quantization def quantize_model(model): model.eval() quantized torch.quantization.quantize_dynamic( model, {nn.GRU, nn.Linear}, dtypetorch.qint8 ) return quantized def scan_threshold(model, val_loader): model.eval() all_probs, all_labels [], [] with torch.no_grad(): for seq_nodes, seq_edges, next_edges in val_loader: node_emb model(seq_nodes, seq_edges, seq_edges) logits model.edge_predict( node_emb[-1], next_edges.edge_index, next_edges.edge_attr ) all_probs.append(torch.sigmoid(logits)) all_labels.append(next_edges.label) probs torch.cat(all_probs).numpy() labels torch.cat(all_labels).numpy() best_f1, best_th 0, 0.5 for th in np.arange(0.1, 0.9, 0.01): pred (probs th).astype(int) f1 f1_score(labels, pred, zero_division0) if f1 best_f1: best_f1, best_th f1, th return best_th, best_f1quantize_dynamic只量化 GRU 和 LinearGCN 的稀疏操作量化收益不大。scan_threshold在 0.1 到 0.9 之间以 0.01 步长扫返回最佳 F1 和对应阈值。这个脚本我每次换数据都会跑一遍阈值不要写死在代码里放配置文件。验证方法上除了离线指标我习惯再做一个「时间外推」测试用前 70% 时间的数据训练后 30% 测试看性能衰减。如果衰减超过 15%说明模型对时间漂移敏感需要加滑动统计特征或缩短重训周期。这个习惯帮我提前发现过好几次线上翻车比只看随机划分的验证集靠谱得多。希望帮到你。本文还有配套的精品资源点击获取