核电站故障诊断增量学习：经验回放+EWC防遗忘实战

简介：深度学习模型在工业场景中持续迭代时，常因新旧数据分布冲突而出现灾难性遗忘，导致旧任务识别能力骤降。增量学习通过让模型在保持旧知识的同时吸收新类别，成为解决该问题的关键技术路径。经验回放从数据层面保留旧样本进行“复习”，EWC（弹性权重巩固）则从参数层面约束重要权重漂移，两者互补化解遗忘困境。该方案在核电站故障诊断等工业场景中价值显著，既能保障蒸汽发生器传热管破裂等既有故障的准确率，又能快速学习新增故障类型。文章从增量学习原理、方案选型到核心代码实现，完整展示一套可落地的工业智能诊断系统改造路径，为模型持续更新提供工程参考。 上个月我在做核电站故障诊断模型的迭代时，碰到了一个特别现实的困境：某个机组新增了一种此前从未见过的故障模式，拿老模型去测，直接被当成正常工况漏掉了；可是把新故障样本混进旧数据重新训练，模型对蒸汽发生器传热管破裂这类老故障的识别率又掉了近一半。这个现象就是增量学习里最典型的灾难性遗忘，它挡住的不只是模型更新，而是整个工业智能诊断系统能否持续迭代的问题。

我最后落地的方案，是经验回放加EWC组合：经验回放负责“数据级复习”，EWC负责“参数级保护”，两者合起来把核电站故障诊断模型改造成了一套支持增量学习的诊断系统。整套流程跑下来，老故障的识别准确率能稳定在93%以上，新增故障也能在几个epoch内被顺利学会。这篇文章把从方案选型、系统设计、核心代码实现到调参踩坑的完整路径都写出来，代码可以直接拿去改，没有保留。

1. 为什么核电站故障诊断要搞增量学习

1.1 真实场景：模型会“过时”

很多人一听到核电站故障诊断，第一反应是“这种系统不是一次性训练好就能用很多年吗？”实际上根本不是这样。核电站设计寿命普遍在40到60年，DCS控制系统、传感器、执行机构等设备会随着技术改造不断升级，设备老化会带来新的异常模式，运维人员也会在日常积累中不断定义出新的故障类型。

举个例子，某个阀门在运行三年后出现了一种间歇性卡涩，这种卡涩的特性在最初的故障定义里根本没有。你拿部署了一年多的诊断模型去判断，它大概率会输出“正常”或者“未知”，因为它的输出空间里压根没有这个类别。这时候你有几个选择：全量重训、拿旧数据+新数据重新训练所有类别、或者用增量学习。全量重训的问题在于历史数据不一定完整保存，而且重新标注、重新训练、重新测试的成本很高，对核电站这种数据合规要求严格的场景来说，频繁动模型很不受运维欢迎。

增量学习的价值就在于：旧知识不丢，新知识能进。模型在原有故障识别能力不下降的前提下，把新出现的故障类型学进来，不需要从头再来。这个能力对工业场景的重要性，怎么强调都不过分。

1.2 灾难性遗忘的机理

灾难性遗忘（Catastrophic Forgetting）是神经网络在增量学习任务中的经典问题。它的本质原因，是神经网络参数共享导致的“覆盖效应”。

我说一个通俗的类比：你把神经网络想象成一块白板，旧任务在上面写了一部分内容。学习新任务的时候，新任务也往这块白板上写东西，写的过程中把旧内容给擦掉了。由于同一个参数要同时服务新旧两个任务，新任务的梯度更新会强制改变这些参数的值，而旧任务又不在训练过程中提供纠错信号，于是旧任务的知识就被“冲”掉了。

我在早期没用任何保护方案的时候做过一次实验：Task 1训练好蒸汽发生器传热管破裂、小破口失水、主泵故障三类故障和正常运行状态，在测试集上准确率98%。随后在Task 2加入主蒸汽管道破裂这个新类别，只用新数据训练，训练完成后回到Task 1的测试集上测，准确率直接掉到52%。52%是什么概念？跟抛硬币差不多，等于这个模型对安全事故的识别能力报废了。

这也是工业场景对灾难性遗忘特别敏感的原因：核电站故障诊断里，漏检一个老故障的代价远大于暂时学不会一个新故障。你要是为了学新故障把老故障忘了，运维根本不敢用你的系统。

2. 方案选型：经验回放和EWC怎么配

2.1 经验回放：数据级“复习”

经验回放（Experience Replay）的思路非常直接：把旧任务里的代表性样本存一部分下来，训练新任务的时候，把这些旧样本混合进每个batch一起训练。模型一边学新东西，一边“复习”旧知识，自然就不容易遗忘。

这个思路在强化学习里用得很多，后来被迁移到增量学习里。它的优势很明显：实现简单，不需要动训练框架，改改数据加载逻辑就行；对类别平衡也有帮助，工业故障数据往往不均衡，通过重放可以人为让旧类别保持一定比例；最重要的是它提供的是“真实记忆”，模型看到的是旧数据的真实特征分布，而不是一个抽象约束。

代价是存储和访问成本。你不可能把所有历史数据都存下来，每个类别存多少、哪些样本代表性强，都需要设计。有些数据场景因为合规要求，旧数据不能长期保留在本地，经验回放没法成为唯一手段。

2.2 EWC：参数级“钉钉子”

EWC（Elastic Weight Consolidation，弹性权重巩固）是另一条路线：不碰数据，从参数层面做保护。它的核心逻辑是用Fisher信息矩阵评估每个网络参数对旧任务的重要性，然后在学习新任务时，对重要参数的更新施加惩罚，让它们尽量少动。

最终损失函数的形式是：

L_new + (λ/2) * Σ F_i * (θ_i - θ*_i)^2

这里的F_i就是Fisher信息矩阵的对角元素，衡量第i个参数对旧任务的重要性；θ*_i是旧任务训练收敛后第i个参数的取值；λ是超参数，控制约束强度。直观理解：重要参数被一根“弹性绳”拴在旧任务确定的位置，新任务想让它们大幅偏移很费力，而那些不重要的参数可以自由更新。

EWC的好处是不需要保存任何旧数据，特别适合数据不能留存的场景。缺点是要在旧任务训练完成后额外算一遍Fisher矩阵，而且λ很难一次调准。更关键的问题是，Fisher信息矩阵只能反映“参数对旧任务重要”，它没法帮模型复习旧任务的具体输出模式，所以单独用EWC时遗忘率仍然不够低。

2.3 为什么两个要组合用

我试过只加经验回放、只加EWC、两个都用，效果差距很明显。经验回放是数据级手段，相当于让模型“记得旧样本长什么样”；EWC是参数级手段，相当于让模型“记住旧参数的位置”。两者是从不同层面解决问题，组合起来不是简单的加法，而是互补：回放应对旧数据分布被兼顾不到的盲区，EWC应对重放样本覆盖不够的参数漂移。

在实际项目中，如果只靠经验回放，当回放缓冲区太小或者新任务数据量和旧任务差异较大时，遗忘率会迅速上升；如果只靠EWC，遇到新旧任务特征分布差异较大的情况，单靠参数约束又限制不住模型的内部表示漂移。两个一起用的时候，模型一方面通过重放样本获得旧任务的“真实记忆”，另一方面通过EWC保证参数不会跑太远，哪怕回放样本没有覆盖到所有特征空间，参数层面的约束也能兜住底。

3. 系统整体设计与实现

3.1 增量学习主流程

整个系统的训练流程我用六步来描述，这样比较直观。

1. 初始训练：用Task 1数据训练一个基座故障诊断模型，得到一组收敛的参数。
2. 计算并保存Fisher信息矩阵：在Task 1训练完成后，用Task 1的全部数据计算每个参数的Fisher信息，同时保存一份参数快照θ*。
3. 构建经验回放缓冲区：从Task 1数据中按类别抽取代表性样本存入缓冲区，每个类别设置容量上限。
4. 扩展输出层：当Task 2数据到达，新增故障类别时，在模型输出层增加对应数量的节点，旧类别节点参数原样复制。
5. 增量训练：每个batch同时包含新任务数据和从回放缓冲区采样的旧数据，损失函数叠加EWC惩罚项。
6. 更新知识库：训练完成后，更新Fisher信息矩阵（与旧Fisher合并），把新类别的代表性样本存入缓冲区，并更新参数快照。

这个流程对后续第三个、第四个任务同样适用，每次新任务到来自动重复4到6步。

3.2 故障诊断骨干网络设计

模型结构上，我选择的是1D CNN加全连接分类头。输入是多传感器时间窗口，shape为[batch, n_channels, seq_len]，其中n_channels对应不同的传感器通道（一回路压力、稳压器水位、蒸汽发生器水位、主泵转速、各管段流量等），seq_len是滑窗长度。

骨干网络由三层1D卷积加BatchNorm加ReLU加MaxPooling组成，最后用全局平均池化把特征压成一维向量，再进全连接层分类。这个结构虽然简单，但在工业时序故障诊断场景里非常够用。你也可以换成LSTM或者Transformer，但增量学习场景下RNN类结构在计算Fisher时梯度稳定性差一些，调试成本高，第一次做增量系统建议先用1D CNN把流程跑通。

输出层的处理是增量学习里一个关键点。新类别加入后，原来的全连接层out_features必须扩展，我会在代码部分详细展开。

3.3 增量任务序列设计

为了验证系统的增量能力，我设计了三个任务序列。每个任务对应的故障类别如下表所示：

样本数据来自DCS组态仿真器，覆盖稳态功率运行、变负荷运行等不同工况。每个样本是24个传感器通道、128个时间步的滑动窗口，滑动步长16，标签由运维专家根据事故规程标注。如果你手头没有核电仿真数据，先用田纳西伊士曼过程的公开数据跑通代码也是可以的，流程完全一样。

4. 核心代码实现

这部分的代码是我在实际项目中反复打磨过的版本，直接按顺序拼起来就能跑。

4.1 经验回放缓冲区实现

缓冲区我按类别存储，每个类别设上限。新增样本时如果该类别已满，就随机替换掉一条旧样本。采样时按类别均匀抽，保证每个类别的重放比例均衡。这样实现的好处是简单可靠，不会出现某个旧类别被淹没的情况。

import random import torch class ReplayBuffer: def __init__(self, capacity_per_class=200): self.capacity = capacity_per_class self.buffer = {} # class_id -> list of (x, y) def add(self, x, y): """接收一批样本，按类别存入缓冲区""" for xi, yi in zip(x, y): ci = int(yi.item()) if ci not in self.buffer: self.buffer[ci] = [] if len(self.buffer[ci]) < self.capacity: self.buffer[ci].append((xi.detach().cpu(), yi.detach().cpu())) else: # 容量已满，随机替换一条旧样本 idx = random.randrange(len(self.buffer[ci])) self.buffer[ci][idx] = (xi.detach().cpu(), yi.detach().cpu()) def sample(self, batch_size): """均匀从所有已有类别中采样，保证类别平衡""" categories = list(self.buffer.keys()) if len(categories) == 0: return None, None per_class = max(1, batch_size // len(categories)) items = [] for c in categories: items.extend(random.sample(self.buffer[c], min(per_class, len(self.buffer[c])))) random.shuffle(items) items = items[:batch_size] xs = torch.stack([it[0] for it in items]) ys = torch.tensor([it[1] for it in items]) return xs, ys

这里有个操作细节：存入缓冲区前把数据移到CPU上保存，因为工业场景中数据量可能很大，长期驻留在GPU显存里不现实。每类200个样本，24通道128步长的float32数据，一个类别大约2.5MB，存几十类也不会有压力。

4.2 Fisher信息矩阵计算

Fisher信息矩阵的工程实现，有两点必须注意。第一，要在旧任务训练“收敛后”计算，不能随便拿一个中间状态算。第二，要逐样本计算梯度再平方，不能用整个batch的平均梯度去平方，否则会严重低估Fisher值，导致约束失效。

def compute_fisher(model, dataloader, device): """ 计算模型backbone参数的Fisher信息矩阵（对角近似）。 必须在任务训练收敛后调用。 """ model.eval() params = [p for p in model.backbone.parameters() if p.requires_grad] fisher = [torch.zeros_like(p) for p in params] total_samples = 0 for x, y in dataloader: x, y = x.to(device), y.to(device) batch_size = x.size(0) # 逐样本计算梯度，避免batch平均造成信息抵消 for i in range(batch_size): model.zero_grad() out = model(x[i:i+1]) loss = torch.nn.functional.nll_loss( torch.log_softmax(out, dim=1), y[i:i+1] ) loss.backward() for idx, p in enumerate(params): if p.grad is not None: fisher[idx] += p.grad.data ** 2 total_samples += batch_size fisher = [f / total_samples for f in fisher] return fisher

这个实现的数学含义是：在模型当前参数下，用每个样本的负对数似然梯度平方去估计Fisher信息矩阵的对角元素。严格来说Fisher的定义是梯度向量外积的期望，但在深度学习中用样本梯度平方做对角近似是主流做法，EWC原文也是这么处理的。

4.3 EWC损失函数

EWC的损失项实现很简单，就是把Fisher值和当前参数与旧参数之差的平方做逐元素乘法，然后求和再乘λ/2。

def ewc_penalty(model, fisher, old_params, lambda_ewc): """ 计算EWC弹性惩罚项。 只对backbone参数施加约束，新扩展的输出层不参与。 """ if fisher is None or old_params is None: return 0.0 penalty = 0.0 params = [p for p in model.backbone.parameters() if p.requires_grad] for p, f, op in zip(params, fisher, old_params): if f is not None: penalty += torch.sum(f * (p - op) ** 2) return (lambda_ewc / 2.0) * penalty

我只对backbone参数做EWC约束，输出层（全连接分类头）不做。原因有两个：一是输出层在扩展新类别时参数数量会变，维护老参数快照和Fisher对齐很麻烦；二是新任务对输出层的适当更新有助于适应小样本类别，强行约束输出层反而会让新类别学不动。旧类别输出节点的保护，交给经验回放就够了。

4.4 增量训练主循环

这是整个系统的核心调度逻辑。每个任务到来时先扩展输出层，然后构造训练循环，每个batch同时用新数据和回放数据计算联合损失，再叠加EWC惩罚，最后统一反向传播。

def expand_output_layer(model, num_new_classes, std=0.05): """扩展分类头，旧节点参数原样复制，新节点小方差初始化""" old_fc = model.fc old_classes = old_fc.out_features new_fc = torch.nn.Linear(old_fc.in_features, old_classes + num_new_classes) with torch.no_grad(): new_fc.weight[:old_classes] = old_fc.weight.data new_fc.bias[:old_classes] = old_fc.bias.data # 新节点初始化标准差不能太大，否则会干扰softmax归一化 new_fc.weight[old_classes:] = torch.randn(num_new_classes, old_fc.in_features) * std new_fc.bias[old_classes:] = 0.0 model.fc = new_fc return model def incremental_train(model, tasks, config, device): replay = ReplayBuffer(capacity_per_class=config['buffer_per_class']) fisher = None old_params = None for task_id, task in enumerate(tasks): print(f"===== Task {task_id + 1} =====") # 1. 首个任务直接训练；后续任务先扩展输出层 if task_id > 0: model = expand_output_layer(model, task['num_new_classes']) # 2. 构建优化器，增量阶段学习率要比初始训练小 optimizer = torch.optim.Adam(model.parameters(), lr=config['lr']) model.train() for epoch in range(config['epochs']): for x_new, y_new in task['train_loader']: x_new, y_new = x_new.to(device), y_new.to(device) # 从回放缓冲区采样旧数据 x_replay, y_replay = replay.sample(config['batch_size']) optimizer.zero_grad() # 新数据损失 out_new = model(x_new) loss_new = torch.nn.functional.cross_entropy(out_new, y_new) # 回放数据损失 loss_replay = 0.0 if x_replay is not None: x_replay = x_replay.to(device) y_replay = y_replay.to(device) out_replay = model(x_replay) loss_replay = config['replay_weight'] * torch.nn.functional.cross_entropy( out_replay, y_replay ) # EWC惩罚项 loss_ewc = ewc_penalty(model, fisher, old_params, config['lambda_ewc']) loss = loss_new + loss_replay + loss_ewc loss.backward() optimizer.step() # 3. 任务训练完成后，更新经验回放缓冲区 replay.add(task['representative_x'], <p> <a href="https://download.csdn.net/download/weixin_44603934/92467429" 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>