自编码器异常检测训练：如何用自定义损失函数融入异常数据

论文算法翻译 #

有论文提出了一种将异常数据点融入训练过程的算法：
算法1：自编码二分类器

while 未收敛 do:
  从数据集中采样小批量数据 {(x1, y1), · · · , (xK , yK )}
  使用融入x标签的自定义损失函数计算θ的梯度gθ
  用gθ对θ进行SGD更新
end while

核心问题解决思路 #

自编码器训练时输入输出确实是X，但Keras支持在训练时将标签作为额外参数传入自定义损失函数，推理阶段仅需输入X即可，完全不需要标签。下面结合你给出的代码，给出具体实现方案：

具体实现步骤 #

1. 自定义带标签的损失函数

根据论文思路，损失函数需要对正常样本（y=0）和异常样本（y=1）设置差异化逻辑：比如让正常样本的重构误差尽可能小，异常样本的重构误差尽可能大（或施加更高惩罚）。示例如下：

def custom_loss(y_true, y_pred, labels):
    # y_true是原始输入X，y_pred是重构后的X
    reconstruction_loss = tf.keras.losses.mean_squared_error(y_true, y_pred)
    # 对异常样本（labels=1）放大损失权重，这里权重可根据需求调整
    weighted_loss = reconstruction_loss * (1 + labels * 1)
    return tf.reduce_mean(weighted_loss)

2. 修改模型类，支持训练时传入标签

重写模型的train_step方法，让它能接收标签并调用自定义损失：

class AnomalyDetector(tf.keras.Model):
    def __init__(self):
        super(AnomalyDetector, self).__init__()
        self.encoder = tf.keras.Sequential([
            tf.keras.layers.Dense(32, activation="relu"),
            tf.keras.layers.Dense(16, activation="relu"),
            tf.keras.layers.Dense(8, activation="relu")
        ]);
        
        self.decoder = tf.keras.Sequential([
            tf.keras.layers.Dense(16, activation="relu"),
            tf.keras.layers.Dense(32, activation="relu"),
            tf.keras.layers.Dense(140, activation="sigmoid")
        ]);
    
    def call(self, x):
        encoded = self.encoder(x)
        decoded = self.decoder(encoded)
        return decoded
    
    def train_step(self, data):
        # data格式为(X, (X, y))：输入是X，目标是(X, 标签y)
        x, (y_true, labels) = data
        
        with tf.GradientTape() as tape:
            y_pred = self(x, training=True)
            loss = custom_loss(y_true, y_pred, labels)
        
        # 更新模型权重
        trainable_vars = self.trainable_variables
        gradients = tape.gradient(loss, trainable_vars)
        self.optimizer.apply_gradients(zip(gradients, trainable_vars))
        
        return {"loss": loss}

3. 训练模型

训练时传入(X_train, (X_train, y_train))，其中y_train是样本标签（0=正常，1=异常）：

autoencoder = AnomalyDetector()
autoencoder.compile(optimizer="adam")
# 假设X_train是特征数据，y_train是对应标签
autoencoder.fit(X_train, (X_train, y_train), epochs=50, batch_size=32)

4. 推理阶段（无需标签）

直接输入测试数据X_test，通过重构误差判断是否异常：

# 生成重构结果
reconstructed = autoencoder.predict(X_test)
# 计算每个样本的重构误差
mse = tf.keras.losses.mean_squared_error(X_test, reconstructed)
# 设置阈值，超过则判定为异常
threshold = 0.01
anomalies = mse > threshold

关键说明 #

训练时标签仅用于调整损失权重，不会作为模型输入特征的一部分，因此推理阶段完全不需要标签，完全符合你的需求。损失函数的具体逻辑可以根据论文公式或任务需求进一步调整。

内容的提问来源于stack exchange，提问作者BovineScatologist