实现图像数据概念漂移检测应用

查看源文件

概述

图像数据的概念漂移(Concept Drift)是AI学习领域的一种重要数据现象,表现为在线推理的图像数据(实时分布) 与训练阶段(历史分布)不一致,也称作Out-of-Distribution(OOD)。例如,基于MNIST数据集进行训练获得神经网络模型,但实际的测试数据为CIFAR-10 数据环境,那么,CIFAR-10数据集则是OOD样本。

本示例提出一种检测图像数据分布变化的方法,整体流程如下:

  1. 加载公开数据集或使用用户自定义数据。

  2. 加载神经网络模型。

  3. 初始化图像概念漂移类参数。

  4. 获得最优概念漂移检测阈值。

  5. 执行概念漂移检测函数。

  6. 查看结果。

准备环节

确保已经正确安装了MindSpore。如果没有,可以通过MindSpore安装页面进行安装。

准备数据集

示例中用到公开图像数据集MNIST和CIFAR-10。

导入Python库&模块

在使用前,需要导入需要的Python库。

import numpy as np
import mindspore as ms
from mindarmour.utils import LogUtil
from mindspore import nn
from examples.common.networks.lenet5.lenet5_net_for_fuzzing import LeNet5
from mindarmour.reliability import OodDetectorFeatureCluster

加载数据

  1. 将MNIST数据集作为训练集ds_train,这里ds_train只包含image数据,不包含label。

  2. 将MNIST和CIFAR-10的混合数据集作为测试集ds_test,这里ds_test只包含image数据,不包含label。

  3. 将另一组MNIST和CIFAR-10的混合数据集作为验证样本,记作ds_eval,这里ds_eval只包含image数据,不包含label。ds_eval另行标记,其中非OOD样本标记为0,OOD样本标记为1,ds_eval的标记单独记作ood_label

ds_train = np.load('/dataset/concept_train_lenet.npy')
ds_test = np.load('/dataset/concept_test_lenet2.npy')
ds_eval = np.load('/dataset/concept_test_lenet1.npy')

ds_train(numpy.ndarray): 训练集,只包含image数据。

ds_test(numpy.ndarray): 测试集,只包含image数据。

ds_eval(numpy.ndarray): 验证集,只包含image数据。

加载神经网络模型

利用训练集ds_train以及ds_train所对应的分类label,训练神经网络LeNet,并加载模型。这里,我们直接导入已训练好的模型文件。

此处的label区别于前文提到的ood_labellabel表示样本的分类标签,ood_label表示样本是否属于OOD的标签。

ckpt_path = '../../dataset/trained_ckpt_file/checkpoint_lenet-10_1875.ckpt'
net = LeNet5()
load_dict = ms.load_checkpoint(ckpt_path)
ms.load_param_into_net(net, load_dict)
model = ms.Model(net)

ckpt_path(str): 模型文件路径。

需要重点说明的是,为了利用神经网络提取特定层的特征输出,需要在神经网络构建过程中增加特征提取和命名神经层的功能。 layer用于命名神经网络层,可以通过下述方法改造神经网络模型,命名各层神经网络,并获取特征输出值。

  1. 导入TensorSummary模块;

  2. 初始化函数__init__中增加 self.summary = TensorSummary()

  3. 在神经网络各层构造函数之后,增加self.summary('name', x)

由于本用例算法内部使用sklearn中的KMeans函数进一步进行特征聚类分析,要求KMeans的输入数据维度为二维。因此,以LeNet为例,我们提取倒数五层的full-connection层和relu层的特征,其数据维度满足KMeans要求。

LeNet神经网络构造过程如下:

from mindspore import nn
from mindspore.common.initializer import TruncatedNormal
from mindspore.ops import TensorSummary

def conv(in_channels, out_channels, kernel_size, stride=1, padding=0):
    """Wrap conv."""
    weight = weight_variable()
    return nn.Conv2d(in_channels, out_channels, kernel_size=kernel_size, stride=stride, padding=padding,
                     weight_init=weight, has_bias=False, pad_mode="valid")

def fc_with_initialize(input_channels, out_channels):
    """Wrap initialize method of full connection layer."""
    weight = weight_variable()
    bias = weight_variable()
    return nn.Dense(input_channels, out_channels, weight, bias)

def weight_variable():
    """Wrap initialize variable."""
    return TruncatedNormal(0.05)

class LeNet5(nn.Cell):
    """
    Lenet network
    """
    def __init__(self):
        super(LeNet5, self).__init__()
        self.conv1 = conv(1, 6, 5)
        self.conv2 = conv(6, 16, 5)
        self.fc1 = fc_with_initialize(16*5*5, 120)
        self.fc2 = fc_with_initialize(120, 84)
        self.fc3 = fc_with_initialize(84, 10)
        self.relu = nn.ReLU()
        self.max_pool2d = nn.MaxPool2d(kernel_size=2, stride=2)
        self.flatten = nn.Flatten()
        self.summary = TensorSummary()

    def construct(self, x):
        """
        construct the network architecture
        Returns:
            x (tensor): network output
        """
        x = self.conv1(x)

        x = self.relu(x)

        x = self.max_pool2d(x)

        x = self.conv2(x)

        x = self.relu(x)

        x = self.max_pool2d(x)

        x = self.flatten(x)

        x = self.fc1(x)
        self.summary('8', x)

        x = self.relu(x)
        self.summary('9', x)

        x = self.fc2(x)
        self.summary('10', x)

        x = self.relu(x)
        self.summary('11', x)

        x = self.fc3(x)
        self.summary('output', x)
        return x

初始化图像概念漂移检测模块

导入概念漂移检测模块,并初始化。

detector = OodDetectorFeatureCluster(model, ds_train, n_cluster=10, layer='output[:Tensor]')

model(Model): 神经网络模型,由训练集ds_train和其分类标签训练所得。

ds_train(numpy.ndarray): 训练集,只包含image数据。

n_cluster(int): 特征聚类数目。

layer(str): 神经网络用于提取特征的层的名称。

需要注意的是,OodDetectorFeatureCluster初始化时,参数layer后需要加上[:Tensor]后缀。 例如,某神经网络层命名为name,那么layer='name[:Tensor]'。实例layer='output[:Tensor]'中使用的是LeNet最后一层layeroutput的特征, 即layer='output[:Tensor]'。另外,由于算法内部使用到sklearn中的KMeans函数进行特征聚类分析,要求KMeans的输入数据维度为二维, 因此layer提取到的特征需要是二维数据,如上文LeNet示例中的full-connection层和relu层。

获取最优概念漂移检测阈值

基于验证集ds_eval 和其OOD标签ood_label,获得最优概念漂移检测阈值。

这里验证集ds_eval可人为构造,例如由50%的MNIST数据集和50%的CIFAR-10数据集组成,因此,OOD标签ood_label可以得知前50%标签值为0,后50%标签值为1。

num = int(len(ds_eval) / 2)
ood_label = np.concatenate((np.zeros(num), np.ones(num)), axis=0)  # ID data = 0, OOD data = 1
optimal_threshold = detector.get_optimal_threshold(ood_label, ds_eval)

ds_eval(numpy.ndarray): 验证集,只包含image数据。
ood_label(numpy.ndarray): 验证集ds_eval的OOD标签,非OOD样本标记为0,OOD样本标记为1。

当然,如果用户很难获得ds_eval和OOD标签ood_labeloptimal_threshold值可以人为灵活设定,optimal_threshold值是[0,1]之间的浮点数。

执行概念漂移检测

result = detector.ood_predict(optimal_threshold, ds_test)

ds_test(numpy.ndarray): 测试集,只包含image数据。
optimal_threshold(float): 最优阈值。可通过执行detector.get_optimal_threshold(ood_label, ds_eval)获得。 但如果用户很难获得ds_eval和OOD标签ood_labeloptimal_threshold值可以人为灵活设定,optimal_threshold值是[0,1]之间的浮点数。

查看结果

print(result)

result(numpy.ndarray): 由元素0和1构成的一维数组,对应了ds_test的OOD检测结果。 例如ds_test是由5个MNIST和5个CIFAR-10数据组成的数据集,那么检测结果为[0,0,0,0,0,1,1,1,1,1]。