CFLOW-AD: Real-Time Unsupervised Anomaly Detection with Localization via Conditional Normalizing Flows

1、Background

虽然最近提出针对此类数据设置的模型在准确性指标上取得了很高的成绩，但它们的复杂性限制了实时处理的能力。

CFLOW-AD由一个经过判别式预训练的编码器组成，后面跟着一个多尺度生成解码器，后者明确估计编码特征的似然性。

该方法得到了一个计算和内存高效的模型： CFLOW-AD在相同的输入设置下，比之前的最先进模型快10倍，小10倍。

2、Method

pseudo-code

# 输入: 训练数据 Dtrain, 测试数据 Dtest
# 输出: 异常检测和定位的结果

# Step 1: 特征提取
def feature_extraction(image):
    # 使用预训练的CNN作为编码器
    encoder = pretrained_cnn()
    feature_maps = encoder(image)
    return feature_maps

# Step 2: 多尺度特征金字塔池化
def pyramid_pooling(feature_maps):
    pooled_features = []
    for scale in scales:
        pooled_feature = pool(feature_maps[scale])
        pooled_features.append(pooled_feature)
    return pooled_features

# Step 3: 条件归一化流
def conditional_normalizing_flows(pooled_features, conditions):
    decoders_outputs = []
    for k, pooled_feature in enumerate(pooled_features):
        decoder = train_cflow_decoder(pooled_feature.shape)
        likelihood = decoder(pooled_feature, conditions[k])
        decoders_outputs.append(likelihood)
    return decoders_outputs

# Step 4: 多尺度融合
def multi_scale_fusion(decoders_outputs):
    anomaly_map = sum(upsample(decoder_output) for decoder_output in decoders_outputs)
    return anomaly_map

# Step 5: 阈值处理和异常定位
def thresholding(anomaly_map, threshold):
    binary_map = anomaly_map > threshold
    return binary_map

# 训练阶段
def train(model, Dtrain):
    for image in Dtrain:
        features = feature_extraction(image)
        conditions = get_conditions(features)
        model = update_model(features, conditions)
    return model

# 测试阶段
def test(model, Dtest):
    anomalies = []
    for image in Dtest:
        features = feature_extraction(image)
        conditions = get_conditions(features)
        decoders_outputs = conditional_normalizing_flows(features, conditions)
        anomaly_map = multi_scale_fusion(decoders_outputs)
        threshold = determine_threshold(anomaly_map)
        binary_map = thresholding(anomaly_map, threshold)
        anomalies.append(binary_map)
    return anomalies

# 主函数
def main(Dtrain, Dtest):
    model = train(Dtrain)
    test_results = test(model, Dtest)
    return test_results

# 运行模型
Dtrain = load_training_data()
Dtest = load_testing_data()
results = main(Dtrain, Dtest)

3、Experiments

4、Conclusion

提出了一种结合多尺度特征提取和条件归一化流的异常检测CFLOW-AD模型