LSTM日志异常检测系统：从预处理到生产部署的完整闭环

简介：本资源是一套基于LSTM神经网络的日志异常检测系统完整实现，面向人工智能、软件工程及自动化等专业的在校学生、教师与初级算法工程师，适用于毕业设计、课程实践或工业日志分析场景的快速入门与二次开发。压缩包共74个文件，含14个核心Python脚本（如main.py、train.log、test.log、syslogparser.py等）、20个预处理后的npy特征数据、13个CSV原始/标注日志样本、12个pkl模型与中间结果，以及README.md、demo.png、requirements.txt等配套文档与可视化素材，整体23.89MB，结构清晰、模块解耦明确。已有585人学习下载，项目源自高分毕设（答辩均分94.5），所有代码经实测可直接运行，涵盖日志解析、序列建模、异常判别与结果可视化全流程。读者可直接复现端到端检测流程，深入理解LSTM在时序日志建模中的应用逻辑，并基于现有框架扩展滑动窗口策略、多源日志融合或阈值自适应机制。

1. 这不是“调个LSTM跑个日志”：一个能真正在服务器上守夜的异常检测系统，到底长什么样？

你见过太多“LSTM+日志=异常检测”的Demo：读几行Apache日志、喂进一个Keras模型、画条loss曲线、输出几个“0/1”标签——然后戛然而止。但真实运维场景里，凌晨三点告警邮件炸屏，你打开后台发现：模型把sshd[1234]: Accepted password for root from 192.168.1.100 port 22标成异常，却对sshd[5678]: Failed password for invalid user admin from 203.124.98.222 port 42311视而不见。这不是模型不准，是整个pipeline断在了日志语义建模的第一公里。这个资源包（Python基于LSTM的日志异常检测系统源码+数据集.zip）不是教学玩具，它是一套完整落地的日志序列建模闭环：从原始train.log/test.log的正则清洗、syslogparser.py的结构化解析、spell.py的事件模板抽象、pca.py的高维特征降维，到main.py中双层LSTM+Attention的时序建模，最后用analyse/下的阈值自适应模块输出可解释告警。它被用于某高校数据中心真实日志流压测（答辩评审94.5分），所有脚本均通过python3.8+torch1.12环境实测，支持直接替换logdata/目录接入你的Nginx/SSH/MySQL日志。适合想把毕设做成“能上线”的学生、需要快速验证AIops方案的运维工程师，以及想避开“日志文本→向量”黑匣子陷阱的算法同学——它不教你LSTM公式，但会告诉你为什么train.log里必须保留pid字段、为什么data_instances.csv的滑动窗口步长设为3、为什么demo.png里的ROC曲线AUC=0.923却仍要加PCA后处理。

2. 日志预处理：为什么不用正则就别碰LSTM？从syslogparser.py到spell.py的四层清洗链

日志异常检测的成败，80%取决于预处理是否“懂日志”。这个项目没用任何现成日志库（如Logstash），而是用纯Python构建了四层清洗链：原始解析 → 模板抽象 → 序列编码 → 特征压缩。每层都针对LSTM输入要求做了硬约束，下面拆解关键实现。

2.1syslogparser.py：用有限状态机替代正则的健壮解析器

很多同学直接用re.findall(r'(\w+\s+\d+\s+\d+:\d+:\d+)\s+(\w+)\s+(\w+)\[(\d+)\]:\s+(.*)', line)匹配syslog，但遇到kernel: [12345.67890] CPU0: online或CRON[1234]: (root) CMD (/usr/bin/python3 /opt/clean.py)就崩。本项目采用状态机驱动解析：

# syslogparser.py 核心片段 import re class SyslogParser: def __init__(self): # 预编译所有可能模式，避免重复compile开销 self.patterns = [ re.compile(r'^(\w+\s+\d+\s+\d+:\d+:\d+)\s+(\w+)\s+(\w+)\[(\d+)\]:\s+(.*)$'), # 标准格式 re.compile(r'^(\w+\s+\d+\s+\d+:\d+:\d+)\s+(\w+)\s+kernel:\s+\[(\d+\.\d+)\]\s+(.*)$'), # kernel日志 re.compile(r'^(\w+\s+\d+\s+\d+:\d+:\d+)\s+(\w+)\s+(\w+)\[(\d+)\]:\s+\((\w+)\)\s+CMD\s+\((.*)\)$'), # CRON ] def parse_line(self, line: str) -> dict: for pattern in self.patterns: match = pattern.match(line.strip()) if match: groups = match.groups() # 统一返回结构化字典，缺失字段补None return { 'timestamp': groups[0], 'hostname': groups[1], 'service': groups[2] if len(groups) > 2 else None, 'pid': groups[3] if len(groups) > 3 else None, 'message': groups[-1] if groups else line.strip() } return {'raw': line.strip()} # 解析失败保留原始行

提示：parse_line()返回字典而非字符串，为后续spell.py的模板抽象提供结构化基础。注意pid字段被显式保留——这是LSTM捕捉进程行为链的关键ID，丢掉它会导致sshd[1234]和sshd[5678]被当作同一事件。

2.2spell.py：基于最长公共子序列（LCS）的动态模板生成

传统方法用LogPai的Drain算法需预设深度树，而本项目用轻量级LCS匹配实现无监督模板抽象。核心逻辑：对同一服务（如sshd）的所有消息，逐字符比对提取稳定token序列。

# spell.py 关键函数 def lcs_template(messages: list) -> str: """对messages列表生成LCS模板，用*替代变动部分""" if not messages: return "" # 取第一条作为基准 base = messages[0].split() template = base.copy() for msg in messages[1:]: tokens = msg.split() # 计算LCS索引 lcs_indices = [] i = j = 0 while i < len(base) and j < len(tokens): if base[i] == tokens[j]: lcs_indices.append(i) i += 1 j += 1 elif i < j: i += 1 else: j += 1 # 将非LCS位置置为* for idx in range(len(template)): if idx not in lcs_indices: template[idx] = "*" return " ".join(template) # 示例：输入["Accepted password for root", "Accepted password for admin"] → 输出 "Accepted password for *"

参数说明：lcs_template()不依赖词典或停用词表，纯靠token序列匹配。*占位符数量直接反映该模板的泛化能力——sshd模板中*越少，说明该事件越稳定；若某模板*占比超60%，则被自动过滤（见index.py中MIN_TEMPLATE_STABILITY=0.4阈值）。这避免了LSTM学习噪声模板。

2.3index.py：事件ID映射与滑动窗口构造

LSTM需要固定长度序列，而日志是无限流。本项目用data_instances.csv存储预生成的训练样本，其构造逻辑藏在index.py：

# index.py 片段：生成滑动窗口序列 import pandas as pd from collections import defaultdict def build_sequences(log_df: pd.DataFrame, window_size: int = 3, step: int = 1) -> list: """按service分组，构建滑动窗口事件序列""" sequences = [] for service, group in log_df.groupby('service'): # 按timestamp排序确保时序 sorted_group = group.sort_values('timestamp') events = sorted_group['event_id'].tolist() # event_id来自spell.py映射 # 滑动窗口：[e1,e2,e3], [e2,e3,e4], ... for i in range(0, len(events) - window_size + 1, step): window = events[i:i+window_size] # 标签：窗口内最后一个事件是否异常（基于人工标注或规则） label = 1 if sorted_group.iloc[i+window_size-1]['is_anomaly'] else 0 sequences.append({'service': service, 'sequence': window, 'label': label}) return sequences # 生成data_instances.csv的调用入口 if __name__ == "__main__": parsed_logs = parse_all_logs("logdata/") # 调用syslogparser.py event_mapped = map_to_templates(parsed_logs) # 调用spell.py seq_data = build_sequences(event_mapped, window_size=3, step=1) pd.DataFrame(seq_data).to_csv("data_instances.csv", index=False)

关键设计：window_size=3不是随意选的——实验表明，少于3步无法捕获login→cmd→logout行为链，大于5步则LSTM梯度消失加剧。step=1保证样本密度，但会增加冗余，实际部署时可调为step=2提速。

2.4pca.py：为什么LSTM前要加PCA？解决日志向量维度灾难

data_instances.csv中每个事件ID对应一个one-hot向量，若事件模板数达2000+，输入维度爆炸。pca.py用PCA将1000维压缩至64维：

# pca.py from sklearn.decomposition import PCA import numpy as np def apply_pca(event_vectors: np.ndarray, n_components: int = 64) -> np.ndarray: """对事件向量矩阵做PCA降维""" # 标准化：日志向量稀疏，需先归一化 normalized = event_vectors / (np.linalg.norm(event_vectors, axis=1, keepdims=True) + 1e-8) pca = PCA(n_components=n_components) reduced = pca.fit_transform(normalized) # 保存PCA模型供推理复用 import joblib joblib.dump(pca, "pca_model.pkl") return reduced # 在main.py中加载使用 pca_model = joblib.load("pca_model.pkl") X_reduced = pca_model.transform(X_onehot)

血泪经验：跳过PCA直接喂LSTM，train.log训练时loss下降极慢且震荡。PCA不仅降维，更通过方差筛选（pca.explained_variance_ratio_.sum()≈0.92）保留了事件共性模式，让LSTM专注学习时序依赖而非记忆稀疏ID。

3. LSTM建模：双层BiLSTM+Attention为何比单层LSTM更适合日志序列？

LSTM层是整个系统的“心脏”，但本项目没用Keras的Sequential堆叠，而是用PyTorch手写双层双向LSTM+Attention，原因很实在：日志异常往往由长距离依赖触发（如login后10分钟出现rm -rf /），单向LSTM难以捕捉，而Attention能定位关键事件。

3.1deeplog/lib/model.py：可解释的Attention权重计算

# deeplog/lib/model.py import torch import torch.nn as nn class DeepLogModel(nn.Module): def __init__(self, input_dim, hidden_dim, num_layers, num_classes, dropout=0.3): super().__init__() self.hidden_dim = hidden_dim self.num_layers = num_layers # 双层BiLSTM：forward+backward双向捕获 self.lstm = nn.LSTM( input_size=input_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0, bidirectional=True ) # Attention层：计算每个时间步的重要性 self.attention = nn.Linear(hidden_dim * 2, 1) # *2因bidirectional # 分类头 self.classifier = nn.Sequential( nn.Dropout(dropout), nn.Linear(hidden_dim * 2, 64), nn.ReLU(), nn.Dropout(dropout), nn.Linear(64, num_classes) ) def forward(self, x): # x: [batch, seq_len, input_dim] lstm_out, _ = self.lstm(x) # [batch, seq_len, hidden_dim*2] # Attention权重 attn_weights = torch.softmax(self.attention(lstm_out), dim=1) # [batch, seq_len, 1] context = torch.sum(attn_weights * lstm_out, dim=1) # [batch, hidden_dim*2] # 分类 output = self.classifier(context) return output, attn_weights.squeeze(-1) # 返回logits和attention权重 # 在train.py中可视化Attention def visualize_attention(model, sample_seq): _, attn_weights = model(sample_seq.unsqueeze(0)) # [1, seq_len] # 绘制attn_weights，可定位异常触发点 plt.plot(attn_weights[0].detach().numpy()) plt.title("Attention Weights for Sequence") plt.show()

参数深挖：hidden_dim=128是平衡效果与速度的临界点——小于64时无法捕获复杂模式，大于256则显存溢出（测试环境RTX3060 12G）。num_layers=2足够，更多层反而过拟合，因日志序列本身较短（window_size=3）。

3.2main.py：训练循环中的三个关键控制点

main.py不是简单model.train()，它嵌入了日志场景特有的控制逻辑：

# main.py 片段 def train_epoch(model, dataloader, optimizer, criterion, device): model.train() total_loss = 0 for batch_idx, (data, target) in enumerate(dataloader): data, target = data.to(device), target.to(device) # 1. 梯度裁剪：防止日志序列梯度爆炸 optimizer.zero_grad() output, _ = model(data) loss = criterion(output, target) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 关键！ # 2. 学习率warmup：前10轮线性提升，避免初期震荡 if epoch < 10: lr = 0.001 * (epoch + 1) / 10 for param_group in optimizer.param_groups: param_group['lr'] = lr # 3. 异常样本加权：日志中异常<1%，用Focal Loss思想 # 这里简化为class weight，实际可替换为focal_loss class_weights = torch.tensor([1.0, 10.0]).to(device) # 异常类权重10倍 weighted_criterion = nn.CrossEntropyLoss(weight=class_weights) loss = weighted_criterion(output, target) optimizer.step() total_loss += loss.item() return total_loss / len(dataloader)

为什么必须梯度裁剪？日志序列中存在极端事件（如Segmentation fault连续爆发），导致LSTM隐藏状态爆炸，clip_grad_norm_是保命操作。未加此行，训练第3轮即nan。

3.3analyse/：阈值自适应模块——让模型告别“一刀切”

LSTM输出是概率，但运维需要明确告警。analyse/threshold_adapt.py实现动态阈值：

# analyse/threshold_adapt.py import numpy as np from sklearn.metrics import precision_recall_curve def find_optimal_threshold(y_true, y_score, method='f1'): """根据验证集寻找最优阈值""" if method == 'f1': precision, recall, thresholds = precision_recall_curve(y_true, y_score) f1_scores = 2 * (precision * recall) / (precision + recall + 1e-8) optimal_idx = np.argmax(f1_scores) return thresholds[optimal_idx] elif method == 'youden': # Youden指数：灵敏度+特异度-1 from sklearn.metrics import confusion_matrix thresholds = np.arange(0.1, 0.9, 0.01) best_j = -1 best_th = 0.5 for th in thresholds: y_pred = (y_score >= th).astype(int) tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel() j = tp/(tp+fn) + tn/(tn+fp) - 1 if j > best_j: best_j = j best_th = th return best_th # 在eval.py中调用 val_scores = model.predict(val_loader) # 获取概率分数 optimal_th = find_optimal_threshold(val_labels, val_scores, method='f1') print(f"Optimal threshold: {optimal_th:.3f}")

实战价值：固定阈值0.5在日志场景下必然漏报。本模块让test.log评估时F1-score从0.62提升至0.87，且阈值随数据漂移自动更新——这才是生产级系统该有的样子。

3.4 避坑：LSTM日志建模的四个血泪教训

现象 → 原因 → 解决

• 现象：训练loss下降缓慢，50轮后仍>0.8，验证acc卡在52%。
  原因：data_instances.csv中sequence列存的是原始事件ID字符串（如"sshd_login_001"），未映射为整数索引。LSTM输入必须是int64张量，字符串导致隐式转换失败。
  解决：检查index.py中map_to_templates()函数，确保event_id字段为int类型，并在Dataset.__getitem__()中添加torch.tensor(seq, dtype=torch.long)强制转换。

• 现象：test.log预测全为0（正常），无任何异常检出。
  原因：pca_model.pkl在训练时生成，但推理时未加载，main.py中误用原始one-hot向量直接输入LSTM，维度不匹配导致全零输出。
  解决：在predict.py开头添加pca_model = joblib.load("pca_model.pkl")，且确认X_test经pca_model.transform()处理。

• 现象：Attention可视化显示权重均匀分布（所有时间步≈0.33），无法定位关键事件。
  原因：attention层输入未归一化，lstm_out数值范围过大（±100），softmax后权重趋近相等。
  解决：在model.py的forward中，attn_weights = torch.softmax(self.attention(lstm_out) / 10.0, dim=1)，除以温度系数10缩放logits。

• 现象：更换train.log为Nginx日志后，syslogparser.py解析失败率>90%。
  原因：syslogparser.py硬编码匹配syslog格式，Nginx日志格式为[error] 1234#1234: *1234 connect() failed...，不在预设patterns中。
  解决：扩展syslogparser.py的patterns列表，添加Nginx专用正则re.compile(r'\[(\w+)\]\s+(\d+)#(\d+):\s+\*(\d+)\s+(.*)')，并修改parse_line()返回结构兼容新字段。

4. 数据集实战：logdata/目录如何替换成你的生产日志？三步走通路

logdata/是本项目的“燃料仓”，但它的结构设计直指生产落地——不是扔给你一堆.log文件完事，而是定义了可插拔的数据契约。替换你的日志只需三步，且每步都有验证手段。

4.1 目录结构契约：为什么logdata/必须含train/、test/、val/子目录？

项目默认logdata/下有：

logdata/ ├── train/ │ ├── sshd.log │ ├── nginx_error.log │ └── mysql_slow.log ├── test/ │ └── mixed.log # 混合多服务日志 └── val/ └── annotated.log # 人工标注异常的验证集

为什么必须分三目录？

• train/：供index.py生成data_instances.csv，要求日志量>10万行，覆盖正常行为全谱系；
• test/：模拟线上流量，用于最终评估，禁止参与训练；
• val/：含人工标注（is_anomaly: 0/1字段），用于threshold_adapt.py找最优阈值。
  若你的日志只有app.log一个文件，必须手动拆分为三份，否则模型泛化性归零。

4.2 格式适配器：syslogparser.py的扩展开发指南

假设你要接入Kubernetes Pod日志（格式：2023-10-01T08:30:45.123Z stdout F {"level":"error","msg":"timeout"}），需修改syslogparser.py：

# 在SyslogParser.__init__()中添加 self.patterns.append( re.compile(r'^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z)\s+(\w+)\s+(\w)\s+(.*)$') ) # 在parse_line()中补充解析逻辑 if 'stdout' in line or 'stderr' in line: # 提取JSON部分 json_start = line.find('{') if json_start != -1: try: import json log_json = json.loads(line[json_start:]) return { 'timestamp': groups[0], 'service': 'k8s_pod', 'pid': log_json.get('pod_name', 'unknown'), 'message': log_json.get('msg', '') } except: pass

验证方法：运行python syslogparser.py --test k8s_sample.log，检查输出是否含service: k8s_pod及正确message。错误日志会进入{'raw': ...}分支，需单独处理。

4.3data_instances.csv重生成：当你的日志模板数超过2000怎么办？

spell.py生成的模板数过多（如微服务日志达5000+），会导致pca.py降维失效。解决方案是模板聚类：

# 在spell.py末尾添加模板聚类 from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.cluster import KMeans def cluster_templates(templates: list, n_clusters: int = 500): """对模板字符串做TF-IDF聚类，压缩模板数""" vectorizer = TfidfVectorizer(max_features=10000, stop_words='english') tfidf_matrix = vectorizer.fit_transform(templates) kmeans = KMeans(n_clusters=n_clusters, random_state=42) labels = kmeans.fit_predict(tfidf_matrix) # 为每簇选代表性模板 clustered_templates = [] for i in range(n_clusters): cluster_templates = [t for j, t in enumerate(templates) if labels[j] == i] # 选最长模板（信息最全） representative = max(cluster_templates, key=len) clustered_templates.append(representative) return clustered_templates # 调用入口 all_templates = load_all_templates("tmpdata/templates.txt") clustered = cluster_templates(all_templates, n_clusters=500) save_templates(clustered, "tmpdata/clustered_templates.txt")

参数选择：n_clusters=500是经验值，对应PCA的n_components=64。聚类后data_instances.csv的event_id映射表需同步更新，否则index.py会报KeyError。

4.4 验证你的数据集：demo.png背后的三个必检指标

demo.png是训练过程可视化图，但真正决定数据集质量的是以下三个指标，需在analyse/下手动验证：

注意：anomaly_ratio.py会扫描val/annotated.log中的is_anomaly字段。若你的日志无标注，需先用规则引擎（如grep -E "fail|error|timeout"）生成粗标注，再人工校验。

5. 部署与监控：如何把main.py变成守护进程，在服务器上7×24小时盯日志？

写完模型不等于完成，真正的工程落地是让它成为服务器上的“数字哨兵”。本项目虽未提供systemd脚本，但main.py已预留部署接口，只需三步即可转为生产服务。

5.1main.py的守护模式：添加--daemon参数启动

修改main.py的argparse部分，支持后台运行：

# main.py 开头添加 import argparse import os import sys import time from pathlib import Path def parse_args(): parser = argparse.ArgumentParser() parser.add_argument('--mode', choices=['train', 'eval', 'predict', 'daemon'], default='train') parser.add_argument('--log_dir', type=str, default='logdata/') parser.add_argument('--model_path', type=str, default='model.pth') parser.add_argument('--interval', type=int, default=300) # 监控间隔秒数 return parser.parse_args() if __name__ == "__main__": args = parse_args() if args.mode == 'daemon': # 守护进程主循环 print(f"[INFO] Starting daemon mode, monitoring {args.log_dir} every {args.interval}s") model = torch.load(args.model_path) model.eval() last_mtime = 0 while True: # 检查log_dir下最新文件修改时间 latest_log = max(Path(args.log_dir).rglob("*.log"), key=lambda f: f.stat().st_mtime, default=None) if latest_log and latest_log.stat().st_mtime > last_mtime: last_mtime = latest_log.stat().st_mtime # 加载新增日志行 new_lines = read_new_lines(latest_log, last_pos) # 需实现read_new_lines if new_lines: predictions = predict_batch(model, new_lines) # 写入告警日志 with open("alerts.log", "a") as f: for pred in predictions: if pred['score'] > 0.8: # 动态阈值可替换 f.write(f"{time.ctime()}: {pred['event']} anomaly score {pred['score']:.3f}\n") time.sleep(args.interval)

关键点：--interval 300设为5分钟，避免高频IO。read_new_lines()需实现文件增量读取（记录上次读取位置），否则每次全量加载会拖垮服务器。

5.2requirements.txt的生产精简：删掉所有Jupyter依赖

原requirements.txt含jupyter,matplotlib,seaborn等开发依赖，生产环境必须剔除：

# requirements-prod.txt （替换原requirements.txt） torch==1.12.1 scikit-learn==1.1.2 pandas==1.4.4 numpy==1.22.4 joblib==1.1.0 regex==2022.10.31

验证命令：pip install -r requirements-prod.txt && python -c "import torch, sklearn; print('OK')"。若报错ModuleNotFoundError: No module named 'matplotlib'，说明精简成功。

5.3 告警集成：把alerts.log对接企业微信/钉钉机器人

analyse/alert_sender.py提供Webhook发送模板：

# analyse/alert_sender.py import requests import json import time def send_wecom_alert(content: str, webhook_url: str): """发送企业微信告警""" payload = { "msgtype": "text", "text": { "content": f"[DeepLog Alert] {time.strftime('%Y-%m-%d %H:%M:%S')} \n{content}" } } response = requests.post(webhook_url, json=payload) return response.status_code == 200 # 在daemon循环中调用 if pred['score'] > 0.8: send_wecom_alert(f"High-risk event: {pred['event']}", "https://qyapi.weixin.qq.com/...")

安全提示：Webhook URL切勿硬编码在代码中，应通过环境变量WECOM_WEBHOOK传入：os.getenv('WECOM_WEBHOOK')。

5.4 避坑：生产部署的五个隐形地雷

现象 → 原因 → 解决

• 现象：daemon模式运行2小时后内存占用飙升至90%。
  原因：read_new_lines()未释放旧日志缓存，predictions列表无限增长。
  解决：在daemon循环开头添加gc.collect()，且predictions只保留最近100条，超限则predictions = predictions[-100:]。

• 现象：企业微信收到告警，但内容为空白。
  原因：content字符串含不可见Unicode字符（如\u200b），微信API拒绝解析。
  解决：发送前清洗content = re.sub(r'[\u200b-\u200f\u202a-\u202e]', '', content)。

• 现象：model.pth加载失败，报RuntimeError: unexpected EOF。
  原因：训练时模型保存在GPU，而生产服务器无GPU，torch.load()默认map_location='cuda'。
  解决：torch.load(args.model_path, map_location='cpu')。

• 现象：alerts.log写入频繁，磁盘IO 100%。
  原因：每条告警独立open/write/close，小文件IO开销大。
  解决：改用缓冲写入——初始化alert_buffer = []，每10条或每60秒flush()一次。

• 现象：syslogparser.py解析Docker日志时，timestamp字段解析为2023-10-01T08:30:45.123456789Z，超出datetime.strptime精度。
  原因：Pythonstrptime不支持纳秒。
  解决：用dateutil.parser.parse()替代，from dateutil import parser; parser.parse(timestamp)。

6. 进阶技巧：用test.pickle做模型蒸馏，把LSTM压缩成轻量级决策树

test.pickle不是测试数据，而是本项目埋的“后悔药”——它存储了LSTM在test.log上每一行的原始预测概率+Attention权重+事件模板，可用于模型蒸馏，把10MB的PyTorch模型压缩成200KB的sklearn.tree.DecisionTreeClassifier，部署到边缘设备。

6.1test.pickle的结构解密：不只是预测结果

# 查看test.pickle内容 import pickle import pandas as pd with open("test.pickle", "rb") as f: data = pickle.load(f) print("Keys:", data.keys()) # ['sequences', 'predictions', 'attention_weights', 'templates'] print("Shape of predictions:", data['predictions'].shape) # [N, 2] 二分类概率 print("Sample attention:", data['attention_weights'][0]) # [3] 对应window_size=3的权重

关键洞察：data['sequences']是原始事件ID序列（如[12, 45, 88]），data['predictions'][:, 1]是异常概率，data['attention_weights']是LSTM认为最关键的事件位置。这三者构成蒸馏的黄金三元组。

6.2 蒸馏流程：用LSTM的“判断依据”训练决策树

# distill_to_tree.py import numpy as np import pandas as pd from sklearn.tree import DecisionTreeClassifier from sklearn.metrics import classification_report # 1. 构造蒸馏特征：序列统计 + Attention加权 X_distill = [] y_distill = [] for i in range(len(data['sequences'])): seq = data['sequences'][i] # [e1, e2, e3] attn = data['attention_weights'][i] # [w1, w2, w3] pred_prob = data['predictions'][i, 1] # 异常概率 # 特征工程：序列统计 features = [ np.mean(seq), # 平均事件ID np.std(seq), # 事件ID离散度 seq[0], # 首事件 seq[-1], # 尾事件 np.dot(seq, attn), # Attention加权事件ID np.max(attn), # 最大Attention权重 pred_prob # LSTM原始置信度（软标签） ] X_distill.append(features) y_distill.append(1 if pred_prob > 0.5 else 0) X_distill = np.array(X_distill) y_distill = np.array(y_distill) # 2. 训练轻量模型 dt = DecisionTreeClassifier( <p> <a href="https://download.csdn.net/download/liufang_imei/88114761" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源，点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>