【MindSpore易点通】模型测试和验证

1 模型测试

在训练完成之后，需要测试模型在测试集上的表现。依据模型评估方式的不同，分以下两种情况

1.评估方式在MindSpore中已实现

MindSpore中提供了多种Metrics方式：Accuracy、Precision、Recall、F1、TopKCategoricalAccuracy、Top1CategoricalAccuracy、Top5CategoricalAccuracy、MSE、MAE、Loss 。在测试中调用MindSpore已有的评估函数，需要定义一个dict，包含要使用的评估方式，并在定义model时传入，后续调用model.eval()会返回一个dict，内容即为metrics的指标和结果。

...def test_net(network, model, test_data_path, test_batch):

"""define the evaluation method"""

print("============== Start Testing ==============")

# load the saved model for evaluation

param_dict = load_checkpoint("./train_resnet_cifar10-1_390.ckpt")

#load parameter to the network

load_param_into_net(network, param_dict)

#load testing dataset

ds_test = create_dataset(test_data_path, do_train=False,

batch_size=test_batch)

acc = model.eval(ds_test, dataset_sink_mode=False)

print("============== test result:{} ==============".format(acc))

if __name__ == "__main__":

...

net = resnet()

loss = nn.loss.SoftmaxCrossEntropyWithLogits(sparse=True,

reduction='mean')

opt = nn.SGD(net.trainable_params(), LR_ORI, MOMENTUM_ORI, WEIGHT_DECAY)

metrics = {

'accuracy': nn.Accuracy(),

'loss': nn.Loss()

}

model = Model(net, loss, opt, metrics=metrics)

test_net(net, model_constructed, TEST_PATH, TEST_BATCH_SIZE)

2.评估方式在MindSpore中没有实现

如果MindSpore中的评估函数不能满足要求，可参考accuracy.py 通过继承Metric基类完成Metric定义之后，并重写clear,updata,eval三个方法即可。通过调用model.predict()接口，得到网络输出后，按照自定义评估标准计算结果。

下面以计算测试集精度为例，实现自定义Metrics：

class AccuracyV2(EvaluationBase):
    def __init__(self, eval_type='classification'):
        super(AccuracyV2, self).__init__(eval_type)
        self.clear()
 
    def clear(self):
        """Clears the internal evaluation result."""
        self._correct_num = 0
        self._total_num = 0
 
    def update(self, output_y, label_input):
        y_pred = self._convert_data(output_y)
        y = self._convert_data(label_input)
        indices = y_pred.argmax(axis=1)
        results = (np.equal(indices, y) * 1).reshape(-1)        
        self._correct_num += results.sum()
        self._total_num += label_input.shape[0]
 
    def eval(self):
        if self._total_num == 0:
            raise RuntimeError('Accuary can not be calculated')
        return self._correct_num / self._total_num
        def test_net(network, model, test_data_path, test_batch):
    """define the evaluation method"""
    print("============== Start Testing ==============")
 # Load the saved model for evaluation
    param_dict = load_checkpoint("./train_resnet_cifar10-1_390.ckpt")
 # Load parameter to the network
    load_param_into_net(network, param_dict)
 # Load testing dataset
    ds_test = create_dataset(test_data_path, do_train=False,
                            batch_size=test_batch)
    metric = AccuracyV2()
    metric.clear()
    for data, label in ds_test.create_tuple_iterator():
        output = model.predict(data)
        metric.update(output, label)
    results = metric.eval()
    print("============== New Metric:{} ==============".format(results))
    if __name__ == "__main__":
...
    net = resnet()
    loss = nn.loss.SoftmaxCrossEntropyWithLogits(sparse=True,
                                        reduction='mean')
    opt = nn.SGD(net.trainable_params(), LR_ORI, MOMENTUM_ORI, WEIGHT_DECAY)
    model_constructed = Model(net, loss, opt)
    test_net(net, model_constructed, TEST_PATH, TEST_BATCH_SIZE)

2 边训练边验证

在训练的过程中，在验证集上测试模型的效果。目前MindSpore有两种方式。

1、交替调用model.train()和model.eval() ，实现边训练边验证。

...def train_and_val(model, dataset_train, dataset_val, steps_per_train,
                    epoch_max, evaluation_interval):
    config_ck = CheckpointConfig(save_checkpoint_steps=steps_per_train,
                                 keep_checkpoint_max=epoch_max)
    ckpoint_cb = ModelCheckpoint(prefix="train_resnet_cifar10",
                                 directory="./", config=config_ck)
    model.train(evaluation_interval, dataset_train,
            callbacks=[ckpoint_cb, LossMonitor()], dataset_sink_mode=True)
    acc = model.eval(dataset_val, dataset_sink_mode=False)
    print("============== Evaluation:{} ==============".format(acc))
if __name__ == "__main__":
    ...
    ds_train, steps_per_epoch_train = create_dataset(TRAIN_PATH,
            do_train=True, batch_size=TRAIN_BATCH_SIZE, repeat_num=1)
    ds_val, steps_per_epoch_val = create_dataset(VAL_PATH, do_train=False,
                batch_size=VAL_BATCH_SIZE, repeat_num=1)
    net = resnet()
    loss = nn.loss.SoftmaxCrossEntropyWithLogits(sparse=True,
                                    reduction='mean')
    opt = nn.SGD(net.trainable_params(), LR_ORI, MOMENTUM_ORI, WEIGHT_DECAY)
    metrics = {
    'accuracy': nn.Accuracy(),
    'loss': nn.Loss()
    }
    net = Model(net, loss, opt, metrics=metrics)
for i in range(int(EPOCH_MAX / EVAL_INTERVAL)):
train_and_val(net, ds_train, ds_val, steps_per_epoch_train,
                        EPOCH_MAX, EVAL_INTERVAL)
 
2、MindSpore通过调用model.train接口，在callbacks中传入自定义的EvalCallBack实例，进行训练并验证。
 
class EvalCallBack(Callback):
 
    def __init__(self, model, eval_dataset, eval_epoch, result_evaluation):
        self.model = model
        self.eval_dataset = eval_dataset
        self.eval_epoch = eval_epoch
        self.result_evaluation = result_evaluation
 
    def epoch_end(self, run_context):
        cb_param = run_context.original_args()
        cur_epoch = cb_param.cur_epoch_num
        if cur_epoch % self.eval_epoch == 0:
            acc = self.model.eval(self.eval_dataset, dataset_sink_mode=False)
            self.result_evaluation["epoch"].append(cur_epoch)
            self.result_evaluation["acc"].append(acc["accuracy"])
            self.result_evaluation["loss"].append(acc["loss"])
            print(acc)
 
if __name__ == "__main__":
    ...
    ds_train, steps_per_epoch_train = create_dataset(TRAIN_PATH,
        do_train=True, batch_size=TRAIN_BATCH_SIZE, repeat_num=REPEAT_SIZE)
    ds_val, steps_per_epoch_val = create_dataset(VAL_PATH, do_train=False,
                batch_size=VAL_BATCH_SIZE, repeat_num=REPEAT_SIZE)
    net = resnet()
    loss = nn.loss.SoftmaxCrossEntropyWithLogits(sparse=True,
                                            reduction='mean')
    opt = nn.SGD(net.trainable_params(), LR_ORI, MOMENTUM_ORI, WEIGHT_DECAY)
    metrics = {
    'accuracy': nn.Accuracy(),
    'loss': nn.Loss()
    }
    net = Model(net, loss, opt, metrics=metrics)
result_eval = {"epoch": [], "acc": [], "loss": []}
    eval_cb = EvalCallBack(net, ds_val, EVAL_PER_EPOCH, result_eval)
    net.train(EPOCH_MAX, ds_train,
            callbacks=[ckpoint_cb, LossMonitor(), eval_cb],
            dataset_sink_mode=True, sink_size=steps_per_epoch_train)

3 样例代码使用说明

本文的样例代码是一个Resnet50在Cifar10上训练的分类网络，采用datasets.Cifar10Dataset接口读取二进制版本的CIFAR-10数据集，因此下载CIFAR-10 binary version (suitable for C programs)，并在代码中配置好数据路径。

启动命令：

python xxx.py --data_path=xxx --epoch_num=xxx

运行脚本，可以看到网络输出结果：

详细代码请前往MindSpore论坛进行下载：华为云论坛_云计算论坛_开发者论坛_技术论坛-华为云