一文详解 ATK Loss 论文复现与代码实战

2023-02-27
中国香港
本文字数：4909 字
阅读完需：约 16 分钟

本文分享自华为云社区《ATK Loss论文复现与代码实战》，作者：李长安。

损失是一种非常通用的聚合损失，其可以和很多现有的定义在单个样本上的损失结合起来，如 logistic 损失，hinge 损失，平方损失（L2），绝对值损失（L1）等等。通过引入自由度 k，损失可以更好的拟合数据的不同分布。当数据存在多分布或类别分布不均衡的时候，最小化平均损失会牺牲掉小类样本以达到在整体样本集上的损失最小；当数据存在噪音或外点的时候，最大损失对噪音非常的敏感，学习到的分类边界跟 Bayes 最优边界相差很大；当采取损失最为聚合损失的时候（如 k=10），可以更好的保护小类样本，并且其相对于最大损失而言对噪音更加鲁棒。所以我们可以推测：最优的 k 即不是 k = 1（对应最大损失）也不是 k = n（对应平均损失），而是在[1, n]之间存在一个比较合理的 k 的取值区间。

论文地址

上图结合仿真数据显示了最小化平均损失和最小化最大损失分别得到的分类结果。可以看出，当数据分布不均衡或是某类数据存在典型分布和非典型分布的时候，最小化平均损失会忽略小类分布的数据而得到次优的结果；而最大损失对样本噪音和外点（outliers）非常的敏感，即使数据中仅存在一个外点也可能导致模型学到非常糟糕的分类边界；相比于最大损失损失，第 k 大损失对噪音更加鲁棒，但其在 k > 1 时非凸非连续，优化非常困难。

由于真实数据集非常复杂，可能存在多分布性、不平衡性以及噪音等等，为了更好的拟合数据的不同分布，我们提出了平均 Top-K 损失作为一种新的聚合损失。

本项目最初的思路来自于八月份参加比赛的时候。由于数据集复杂，所以就在想一些难例挖掘的方法。看看这个方法能否带来一个更好的模型效果。该方法的主要思想是使用数值较大的排在前面的梯度进行反向传播，可以认为是一种在线难例挖掘方法，该方法使模型讲注意力放在较难学习的样本上，以此让模型产生更好的效果。代码如下所示。

class topk_crossEntrophy(nn.Layer):
    def __init__(self, top_k=0.6):        super(topk_crossEntrophy, self).__init__()        self.loss = nn.NLLLoss()        self.top_k = top_k        self.softmax = nn.LogSoftmax()        return        def forward(self, inputs, target):        softmax_result = self.softmax(inputs)                loss1 = paddle.zeros([1])        for idx, row in enumerate(softmax_result):            gt = target[idx]            pred = paddle.unsqueeze(row, 0)            cost = self.loss(pred, gt)            loss1 = paddle.concat((loss1, cost), 0)               loss1 = loss1[1:]        if self.top_k == 1:            valid_loss1 = loss1        index = paddle.topk(loss1, int(self.top_k * len(loss1)))        valid_loss1 = loss1[index[1]]                return paddle.mean(valid_loss1)

复制代码

topk_loss 的主要思想
topk_loss 的核心思想，即通过控制损失函数的梯度反传，使模型对 Loss 值较大的样本更加关注。该函数即为 CrossEntropyLoss 函数的具体实现，只不过是在计算 nllloss 的时候取了前 70%的梯度，
数学逻辑：挖掘反向传播前 70% 梯度。

代码实战

此部分使用比赛中的数据集，并带领大家使用 Top-k Loss 完成模型训练。在本例中使用前 70%的 Loss。

!cd 'data/data107306' && unzip -q img.zip

复制代码

# 导入所需要的库from sklearn.utils import shuffleimport osimport pandas as pdimport numpy as npfrom PIL import Image
import paddleimport paddle.nn as nnfrom paddle.io import Datasetimport paddle.vision.transforms as Timport paddle.nn.functional as Ffrom paddle.metric import Accuracy
import warningswarnings.filterwarnings("ignore")
# 读取数据train_images = pd.read_csv('data/data107306/img/df_all.csv')
train_images = shuffle(train_images)# 划分训练集和校验集all_size = len(train_images)train_size = int(all_size * 0.9)train_image_list = train_images[:train_size]val_image_list = train_images[train_size:]
train_image_path_list = train_image_list['image'].valueslabel_list = train_image_list['label'].valuestrain_label_list = paddle.to_tensor(label_list, dtype='int64')
val_image_path_list = val_image_list['image'].valuesval_label_list1 = val_image_list['label'].valuesval_label_list = paddle.to_tensor(val_label_list1, dtype='int64')
# 定义数据预处理data_transforms = T.Compose([    T.Resize(size=(448, 448)),
    T.Transpose(),    # HWC -> CHW    T.Normalize(
        mean = [0, 0, 0],        std = [255, 255, 255],        to_rgb=True)    ])

复制代码

# 构建Datasetclass MyDataset(paddle.io.Dataset):    """    步骤一：继承paddle.io.Dataset类    """    def __init__(self, train_img_list, val_img_list,train_label_list,val_label_list, mode='train'):        """        步骤二：实现构造函数，定义数据读取方式，划分训练和测试数据集        """        super(MyDataset, self).__init__()        self.img = []        self.label = []        self.valimg = []        self.vallabel = []        # 借助pandas读csv的库        self.train_images = train_img_list        self.test_images = val_img_list        self.train_label = train_label_list        self.test_label = val_label_list        # self.mode = mode        if mode == 'train':            # 读train_images的数据            for img,la in zip(self.train_images, self.train_label):                self.img.append('data/data107306/img/imgV/'+img)                self.label.append(la)        else :            # 读test_images的数据            for img,la in zip(self.test_images, self.test_label):                self.img.append('data/data107306/img/imgV/'+img)                self.label.append(la)
    def load_img(self, image_path):        # 实际使用时使用Pillow相关库进行图片读取即可，这里我们对数据先做个模拟        image = Image.open(image_path).convert('RGB')        image = np.array(image).astype('float32')        return image
    def __getitem__(self, index):        """        步骤三：实现__getitem__方法，定义指定index时如何获取数据，并返回单条数据（训练数据，对应的标签）        """        # if self.mode == 'train':
        image = self.load_img(self.img[index])        label = self.label[index]
        return data_transforms(image), label
    def __len__(self):        """        步骤四：实现__len__方法，返回数据集总数目        """        return len(self.img)

复制代码

#train_loadertrain_dataset = MyDataset(train_img_list=train_image_path_list, val_img_list=val_image_path_list, train_label_list=train_label_list, val_label_list=val_label_list, mode='train')train_loader = paddle.io.DataLoader(train_dataset, places=paddle.CPUPlace(), batch_size=4, shuffle=True, num_workers=0)
#val_loaderval_dataset = MyDataset(train_img_list=train_image_path_list, val_img_list=val_image_path_list, train_label_list=train_label_list, val_label_list=val_label_list, mode='test')val_loader = paddle.io.DataLoader(val_dataset, places=paddle.CPUPlace(), batch_size=4, shuffle=True, num_workers=0)

复制代码

from res2net import Res2Net50_vd_26w_4s
# 模型封装model_re2 = Res2Net50_vd_26w_4s(class_dim=4)

复制代码

import paddle.nn.functional as Fimport paddlemodelre2_state_dict = paddle.load("Res2Net50_vd_26w_4s_pretrained.pdparams")
model_re2.set_state_dict(modelre2_state_dict, use_structured_name=True)

复制代码

model_re2.train()epochs = 2
optim1 = paddle.optimizer.Adam(learning_rate=3e-4, parameters=model_re2.parameters())

复制代码

class topk_crossEntrophy(nn.Layer):
    def __init__(self, top_k=0.7):        super(topk_crossEntrophy, self).__init__()        self.loss = nn.NLLLoss()        self.top_k = top_k        self.softmax = nn.LogSoftmax()        return        def forward(self, inputs, target):        softmax_result = self.softmax(inputs)                loss1 = paddle.zeros([1])        for idx, row in enumerate(softmax_result):            gt = target[idx]            pred = paddle.unsqueeze(row, 0)            cost = self.loss(pred, gt)            loss1 = paddle.concat((loss1, cost), 0)               loss1 = loss1[1:]        if self.top_k == 1:            valid_loss1 = loss1        # print(len(loss1))        index = paddle.topk(loss1, int(self.top_k * len(loss1)))        valid_loss1 = loss1[index[1]]                return paddle.mean(valid_loss1)
topk_loss = topk_crossEntrophy()

复制代码

from numpy import *# 用Adam作为优化函数for epoch in range(epochs):
    loss1_train = []    loss2_train = []    loss_train = []
    acc1_train = []    acc2_train = []    acc_train = []    for batch_id, data in enumerate(train_loader()):
        x_data = data[0]        y_data = data[1]        y_data1 = paddle.topk(y_data, 1)[1]        predicts1 = model_re2(x_data)
        loss1 = topk_loss(predicts1, y_data1)
        # 计算损失        acc1 = paddle.metric.accuracy(predicts1, y_data)
        loss1.backward()
        if batch_id % 1 == 0:            print("epoch: {}, batch_id: {}, loss1 is: {}, acc1 is: {}".format(epoch, batch_id, loss1.numpy(), acc1.numpy()))        optim1.step()        optim1.clear_grad()    loss1_eval = []    loss2_eval = []    loss_eval = []    acc1_eval = []    acc2_eval = []    acc_eval = []    for batch_id, data in enumerate(val_loader()):        x_data = data[0]        y_data = data[1]               y_data1 = paddle.topk(y_data, 1)[1]
        predicts1 = model_re2(x_data)
        loss1 = topk_loss(predicts1, y_data1)        loss1_eval.append(loss1.numpy())
        # 计算acc        acc1 = paddle.metric.accuracy(predicts1, y_data)        acc1_eval.append(acc1)
        if batch_id % 100 == 0:            print('************Eval Begin!!***************')            print("epoch: {}, batch_id: {}, loss1 is: {}, acc1 is: {}".format(epoch, batch_id, loss1.numpy(), acc1.numpy()))               print('************Eval End!!***************')

复制代码

总结

在该工作中，分析了平均损失和最大损失等聚合损失的优缺点，并提出了平均 Top-K 损失（损失）作为一种新的聚合损失，其包含了平均损失和最大损失并能够更好的拟合不同的数据分布，特别是在多分布数据和不平衡数据中。损失降低正确分类样本带来的损失，使得模型学习的过程中可以更好的专注于解决复杂样本，并由此提供了一种保护小类数据的机制。损失仍然是原始损失的凸函数，具有很好的可优化性质。我们还分析了损失的理论性质，包括 classification calibration 等。
Top-k loss 的参数设置为 1 时，此损失函数将变 cross_entropy 损失，对其进行测试，结果与原始 cross_entropy()完全一样。但是我在实际的使用中，使用此损失函数却没使模型取得一个更好的结果。需要做进一步的实验。

点击关注，第一时间了解华为云新鲜技术~

发布于: 刚刚阅读数: 5

原文链接:【http://xie.infoq.cn/article/fbef0c856bb8a2c472a00e2bc】。文章转载请联系作者。