从模型集成到任务闭环：一次遥测异常检测系统重构

7 月底开始整理遥测数据分析系统时，我原以为工作主要是替换几种异常检测模型。真正动手以后才发现，模型只占其中一部分。更棘手的问题散落在模型前后：训练和推理的数据处理不一致，长时间缺失被直接插值，一个任务拆成多条 Kafka 消息后可能永远收不齐，metadata 也在不同模块里被反复复制和兼容。

这次重构从异常检测开始，最后一路改到了 Java 查询、Kafka 传输、Python 任务收集、GPU 调度和结果入库。本文把这段过程记下来，也顺便说明现在这套系统能做什么、还欠什么。

系统在处理什么

项目处理的是多传感器遥测数据。用户选择场景、传感器和时间范围后，Java 服务查询历史数据，把分析任务发给 Python。Python 完成数据整理，再根据任务类型执行异常检测、趋势识别或其他分析，最后将结果写入数据库并通知后端。

把主链路压缩一下，大致是：

历史遥测数据
    │
    ▼
Java 查询与数据库侧降采样
    │
    ▼
按传感器、点数和消息大小分块
    │
    ▼
Kafka 数据任务
    │
    ▼
Python 收集多传感器、多批次消息
    │
    ▼
时间对齐、缺口处理、连续段切分
    │
    ▼
GPU / CPU 算法队列
    │
    ├── 异常检测
    ├── 趋势识别
    └── 分类与关系分析
    │
    ▼
事件化结果、任务状态与 ACK

表面上这是一次请求和一次返回，中间实际上跨了数据库、Java、Kafka、Python 和 GPU 队列。任何一层对任务边界的理解不一致，最后表现出来的往往都是“任务一直处理中”。

先统一异常检测模型

原来的异常检测代码里同时存在 Isolation Forest、LSTM AutoEncoder、Darts 和一些单独的分析器。每种算法都有自己的数据格式、模型加载方式和结果字段。增加一个模型，不只是实现网络，还要重新写训练、推理、注册和入库逻辑。

我先把模型拆成几个比较固定的部分：

• model 只描述网络结构；
• trainer 负责训练、验证、阈值校准和 checkpoint；
• detector 将连续时序窗口转成点级异常分数；
• analyzer 适配系统任务协议，输出统一结果。

在这套接口上接入了 OracleAD、PaAno、DLinear、InvConv，随后又加入 KAN-AD。它们都通过统一的模型注册表加载，缺少 checkpoint 时直接返回错误，不再偷偷使用随机参数继续推理。

这些模型本身不是我提出的。这一阶段主要解决的是工程接入和结果可比性，而不是发明一个新的网络结构。

不同模型的分数怎么融合

几种模型给出的异常分数不在同一个尺度上。一个模型的 0.8 可能已经超过决策边界，另一个模型的 10 仍然正常，直接平均没有意义。

现在先用各模型自己的校准阈值做归一化。对于模型 $m$、时刻 $t$ 和传感器 $j$，定义：

其中 $s_{m,t,j}$ 是原始异常分数，${\tau }_m$ 是该模型的阈值。归一化以后，1 就是每个模型各自的决策边界。融合分数为：

当 $A_t>1$ 时，该时刻被标记为异常。系统还会保存各成员模型的阈值、最大分数、异常点数量和主要贡献传感器。模型一致率只在至少一个模型报告异常的位置计算，否则大量正常点会把一致率轻易推到接近 100%。

集成允许个别成员加载或推理失败，但至少需要两个模型成功。只剩一个模型时继续返回“集成结果”，数字看起来完整，含义却已经变了。

时间缺口不能随便补

遥测数据并不是干净、等间隔的时间序列。通信中断、设备停机和查询范围跨段都会制造缺口。如果把所有数据放到同一条时间轴上直接插值，长缺口中会出现一条平滑但不存在的曲线，模型随后还可能把它当作正常模式。

现在的处理方式分两种：

1. 大缺口直接切段，前后的数据作为两个连续片段分别推理；
2. 小缺口按配置做有限插值，支持 linear、nearest、ffill 和 bfill，并标记填充出来的数据点。

各传感器在进入模型前还要去重、排序和对齐。训练与推理使用相同的数据排列、窗口长度和归一化方式，模型阈值也随 checkpoint 一起保存。这里没有很花哨的东西，但如果这一步不一致，后面的验证误差和异常分数都不太可信。

从异常点变成异常事件

点级分数适合模型计算，不适合用户查询。用户真正关心的是哪段时间发生了什么、严重程度如何、涉及哪些传感器。

因此，模型输出之后还要把相邻异常点合并成事件。数据库中保存事件的开始时间、结束时间、异常分数、置信度、严重级别、模型版本和点级证据。结果既可以按事件查询，也可以继续下钻到具体异常点。

系统还支持按设备状态机切分运行阶段，并将模型结果与场景规则融合。目前规则等级采用 normal < warning < error。例如模型检测到异常，而工艺规则同时越过错误阈值，最终事件可以提升为 error；只有模型信号、规则没有命中时，则保留模型异常及其来源说明。

这套等级不是“模型自动理解了故障有多严重”。模型主要负责找异常，业务规则负责补充场景语义。两者分开保存，后续才有机会解释一次告警到底从哪里来。

真正难修的是 Kafka 任务收集

一个分析任务可能包含多个传感器，单个传感器的数据又可能因为点数或消息大小被拆成多个批次。Java 发送的不是“一条任务消息”，而是一组共享 task_id 的消息。

早期收集逻辑只盯着传感器数量，没有完整处理批次边界。更麻烦的是，如果某个传感器查询失败，Java 一条消息也不发，但 total_sensors 仍然包含它。Python 会一直等那个永远不会出现的传感器。

重构后，Java 即使查询失败也会发送一个带错误信息的空占位批次。Python 按下面这些字段判断任务是否收齐：

task_id
sensor_name
batch_index
total_batches
total_sensors

同一任务使用 task_id 作为 Kafka key，保证消息进入同一个分区。收集器会处理重复批次，合并同一传感器的数据，并在所有传感器的所有批次到齐后提交算法队列。

失败路径也重新整理了一遍：

• 未收齐任务有超时时间，过期后清理缓存并返回失败；
• 已处理任务保留有限大小的去重记录，避免 Kafka 重投造成重复计算；
• ACK 缓存增加 TTL 和容量上限；
• 算法任务重试期间保持运行状态，只在最终放弃时写入 FAILED；
• GPU 队列和重试入队都有容量与超时，队列满时不会把 worker 永久卡住。

这些改动没有改变模型精度，却明显改变了系统的可预测性。以前看到“处理中”很难判断是在等消息、排队还是跑模型；现在任务至少会走向成功、明确失败或超时清理。

GPU 调度也需要边界

Python 端使用有界优先级队列调度分析任务。worker 数量和 GPU 列表可以配置，信号量限制同时进入 GPU 的任务数。当 GPU 不可用时，可以按部署配置回退到 CPU。

调度器记录排队长度、活跃 worker、完成数和失败数，并在积压超过阈值时告警。它算不上复杂的集群调度系统，但比“每来一个 Kafka 消息就直接启动模型”稳得多，至少内存和并发有清楚的上限。

最后才轮到删代码

可靠性逻辑补完以后，kafka_consumer.py 一度变得很难读。merged_payload["metadata"] 在多个模块中反复解析，scene_id 又同时存在 snake case、camel case 和顶层兼容字段。

最后一轮重构把消息收集、任务提交、offset 安全通知和算法处理重新分开。标准字段统一放进 metadata，只在入口处兼容旧 payload。下游模块不再为了读三个状态字段复制整份传感器数据。代码里原本为了“抽象”增加的小包装也尽量去掉，直接传字典，并在数据边界写清楚中文注释。

仓库本身也清理了一遍。旧模块副本、前端模板文档、IDE 配置、临时 CSV/JSON、模型权重和一份六百多兆的模拟数据都不应该长期留在 Git 中。清理提交与业务代码提交分开，正式 PR 分支只挑选需要的代码改动，避免本地精简分支误删主线文档。

这次重构留下了什么

从 7 月 30 日到 8 月中旬，相关功能和重构提交涉及约 1.1 万行新增代码。行数不能说明代码质量，尤其这里还包括模型适配、训练脚本和测试，但它能说明这次工作并不只是替换一个算法名字。

目前留下的主要结果有：

• 五种异常检测模型共享训练、加载、推理和结果协议；
• 模型分数按自身阈值归一化后再融合；
• 时间缺口按大小分别切段或有限填充；
• 异常结果可以定位到时间区间，并保存等级与点级证据；
• Java 到 Python 的多传感器、多批次任务能够闭环；
• 队列、缓存、重试和 ACK 都有明确的容量或超时边界。

还没有解决的事

这套系统离“完成”还很远。

首先，我们没有高质量的真实故障标签。当前更现实的验证方式是用正常运行数据训练，再人工注入尖峰、漂移、均值偏移、噪声增加和数据中断，统计事件召回率、误报次数、检测延迟和异常区间 IoU。这样的实验能验证流程是否大致可用，但不能替代真实现场标签。

其次，系统还没有完成每秒 20 万遥测点的持续压测。现有架构已经有数据库侧降采样、Kafka 解耦、分块传输和有界任务队列，不过“具备这些机制”和“稳定处理 20 万点每秒”不是一回事。后续需要实测吞吐、Kafka Lag、P95/P99 延迟、丢失率和资源占用。

还有一个结构性问题：同一任务都用 task_id 作为 Kafka key，收集逻辑简单了，但超大任务可能形成单分区热点。如果以后把原始实时接入和算法任务都放到这条链路上，就需要拆成不同主题，并重新设计分区与聚合策略。

这次重构至少把问题摆到了明面上。模型不再掩盖数据处理错误，任务也不会因为少一条消息就无限等待。下一步该做的不是继续堆模型，而是补完压测和可重复实验，把系统到底能承受多少数据、异常结果到底有多可信测出来。