当前位置：首页 > news >正文

基于类内类间优化的元学习少样本故障诊断方法

news 2025/10/24 11:08:28

基于类内类间优化的元学习少样本故障诊断方法详解

1. 引言

在工业设备故障诊断领域，深度学习技术虽然取得了显著成果，但通常需要大量标注数据进行模型训练。然而在实际工业场景中，收集足够的故障数据往往十分困难，这严重限制了智能诊断方法的应用。本文提出了一种创新的类内类间优化元学习（MLIIO）方法，旨在通过有限的训练数据构建有效的基于度量的故障分类器。

2. 方法原理

2.1 整体框架概述

MLIIO方法基于度量学习元学习（MML）框架，通过设计新型损失函数来优化特征空间。整体诊断流程如图2所示：

该方法的核心创新在于设计了两个互补的损失函数组件：

类内聚合损失（IAL）：促使同类样本特征聚集在一起
类间判别损失（IDL）：强制不同类样本特征保持较大间隔

2.2 类间判别损失（IDL）

传统三元组损失在工业场景中存在局限性，因为单个样本容易受到噪声干扰。MLIIO创新性地使用类别原型间的距离来表示类间差异：

IDL公式：

总IDL：

其中d(⋅,⋅)为平方欧氏距离，α为强制间隔，[⋅]+=max(0,⋅)。

该方法的关键优势在于利用原型间的最小距离来表示类间差异，如图1所示，即使锚点样本距离正类原型较近但绝对距离较大时，仍能有效优化特征空间：

2.3 类内聚合损失（IAL）

IAL借鉴中心损失思想，但针对元学习框架进行了改进：

IAL公式：

总IAL：

与传统中心损失不同，IAL使用支持集原型cm作为聚合中心，更适合元学习的任务式训练范式。

2.4 整体损失函数

MLIIO的总体损失函数为两个损失的加权和：

其中λ>0为平衡超参数，根据两个损失的数值范围进行调整。

3. 创新点分析

3.1 理论创新

原型间距离度量：首次使用类别原型间的最小距离来表示类间差异，克服了传统三元组损失对噪声敏感的问题
元学习适配的聚合损失：改进了中心损失，使其更适合元学习的 episodic 训练机制
双重约束优化：同时优化类内紧凑性和类间可分性，形成完整的特征空间优化框架

3.2 技术优势

抗噪声能力强：使用原型而非单个样本计算类间差异
训练稳定性高：避免了MAML方法的梯度不稳定问题
计算效率优：相比两阶段优化的MAML，训练过程更简洁高效

4. 算法实现

4.1 元学习训练流程

MLIIO采用典型的元学习episodic训练机制，算法流程如下：

算法1：MLIIO训练过程

输入：训练数据集𝒟_train，超参数λ, α
输出：优化后的模型参数φ*1: 随机初始化模型参数φ
2: for 每个训练epoch do
3:     for 每个episode中的任务𝒟^μ do
4:         从𝒟^μ中随机采样支持集𝒮^μ和查询集𝒬^μ
5:         计算每个类别的原型c_m ← (1/|𝒮_m|)∑f_φ(x_i)
6:         计算总体损失ℒ_total ← ℒ_ial + λℒ_idl
7:         通过梯度下降更新参数：φ ← φ - η∇ℒ_total
8:     end for
9: end for
10: return φ*

4.2 网络架构设计

MLIIO使用的嵌入网络架构如表II所示：

核心组件：

4个卷积模块（Conv-BN-ReLU-MaxPool）
特征维度从600压缩到30
适用于振动信号和电流信号处理

4.3 关键代码实现

基于PyTorch的核心代码实现如下：

import torch
import torch.nn as nn
import torch.nn.functional as Fclass MLIIOLoss(nn.Module):def __init__(self, lambda_param=5e-5, alpha=20):super(MLIIOLoss, self).__init__()self.lambda_param = lambda_paramself.alpha = alphadef forward(self, features, labels, prototypes):"""features: 查询集样本特征 [batch_size, feature_dim]labels: 查询集样本标签 [batch_size]prototypes: 各类别原型 [num_classes, feature_dim]"""batch_size, feature_dim = features.shapenum_classes = prototypes.shape[0]# 计算类内聚合损失（IAL）ial_loss = 0.0for i in range(num_classes):class_mask = (labels == i)if class_mask.sum() > 0:class_features = features[class_mask]class_center = prototypes[i].unsqueeze(0)# 计算样本到原型的距离distances = torch.norm(class_features - class_center, dim=1)ial_loss += distances.mean()# 计算类间判别损失（IDL）idl_loss = 0.0# 计算原型间距离矩阵proto_distances = torch.cdist(prototypes, prototypes, p=2)# 将对角线元素设为无穷大，避免自身比较proto_distances.fill_diagonal_(float('inf'))for i in range(num_classes):# 找到当前类别到其他类别的最小距离min_inter_dist = proto_distances[i].min()class_mask = (labels == i)if class_mask.sum() > 0:class_features = features[class_mask]class_center = prototypes[i].unsqueeze(0)# 计算样本到本类原型的距离intra_dists = torch.norm(class_features - class_center, dim=1)# 计算IDL损失class_idl_loss = F.relu(intra_dists - min_inter_dist + self.alpha)idl_loss += class_idl_loss.mean()total_loss = ial_loss + self.lambda_param * idl_lossreturn total_loss, ial_loss, idl_lossclass PrototypicalNetwork(nn.Module):def __init__(self, input_dim=600, hidden_dims=[300, 100], output_dim=30):super(PrototypicalNetwork, self).__init__()layers = []prev_dim = input_dimfor hidden_dim in hidden_dims:layers.extend([nn.Linear(prev_dim, hidden_dim),nn.BatchNorm1d(hidden_dim),nn.ReLU(),nn.Dropout(0.2)])prev_dim = hidden_dimlayers.append(nn.Linear(prev_dim, output_dim))self.encoder = nn.Sequential(*layers)def forward(self, x):return self.encoder(x)def compute_prototypes(self, support_features, support_labels):"""计算各类别的原型（特征均值）"""unique_labels = torch.unique(support_labels)prototypes = []for label in unique_labels:class_mask = (support_labels == label)class_features = support_features[class_mask]class_prototype = class_features.mean(dim=0)prototypes.append(class_prototype)return torch.stack(prototypes)def episodic_training(model, train_tasks, num_episodes=1000):"""episodic训练循环"""optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)criterion = MLIIOLoss()for episode in range(num_episodes):# 采样一个训练任务support_data, query_data = sample_episode(train_tasks)support_inputs, support_labels = support_dataquery_inputs, query_labels = query_data# 前向传播support_features = model(support_inputs)query_features = model(query_inputs)# 计算原型prototypes = model.compute_prototypes(support_features, support_labels)# 计算损失total_loss, ial_loss, idl_loss = criterion(query_features, query_labels, prototypes)# 反向传播optimizer.zero_grad()total_loss.backward()optimizer.step()if episode % 100 == 0:print(f'Episode {episode}: Total Loss: {total_loss.item():.4f}, 'f'IAL: {ial_loss.item():.4f}, IDL: {idl_loss.item():.4f}')