昇思MindSpore應用實踐

本系列文章主要用于記錄昇思25天學習打卡營的學習心得。

基于MindSpore的ResNet50圖像分類

1、ResNet50 簡介

ResNet-50是一種深度殘差網絡（Residual Network），是ResNet系列中的一種經典模型。它由微軟研究院的Kaiming He等人于2015年提出，被廣泛應用于計算機視覺任務，如圖像分類、目標檢測和圖像分割等。

殘差網絡結構主要由兩種：
一種是Building Block，適用于較淺的ResNet網絡，如ResNet18和ResNet34；
另一種是Bottleneck，適用于層數較深的ResNet網絡，如ResNet50、ResNet101和ResNet152。

其整體網絡結構如下圖所示：

ResNet主要解決的問題是：通過Kaiming He等人提出的基于殘差連接的訓練方式大大改善了網絡深度增加時的梯度消失和梯度爆炸問題。

當有這條跳躍連接線（殘差連接，Residual Connections）時，即使網絡層次很深導致梯度消失時，在網絡上堆疊這樣的結構（f(x)=0，y=g(x)=relu(x)=x），我什么也學不到，但至少能把原來的樣子恒等映射給下一層網絡，相當于在淺層網絡上堆疊了“復制層”，這樣至少不會比淺層網絡差。

當然，萬一我“不小心”學到了什么，那就賺大了，由于網絡中要用到很多次恒等映射，所以網絡有效學習到東西的概率很大。這就是ResNet的靈魂汁子，澆給~

ResNet50在多個常用的數據集上（如ImageNet）都有預訓練的模型可供下載和使用。這種預訓練模型含有大量且多樣化的圖像特征，能夠為圖像分類這樣需要重點關注圖像特征的視覺任務提供強有力的技術支撐。

2、數據集預處理及可視化

本案例基于MindSpore對CIFAR-10數據集進行圖形分類。

import mindspore as ms
import mindspore.dataset as ds
import mindspore.dataset.vision as vision
import mindspore.dataset.transforms as transforms
from mindspore import dtype as mstypedata_dir = "./datasets-cifar10-bin/cifar-10-batches-bin"  # 數據集根目錄
batch_size = 256  # 批量大小
image_size = 32  # 訓練圖像空間大小
workers = 4  # 并行線程個數
num_classes = 10  # 分類數量def create_dataset_cifar10(dataset_dir, usage, resize, batch_size, workers):data_set = ds.Cifar10Dataset(dataset_dir=dataset_dir,usage=usage,num_parallel_workers=workers,shuffle=True)trans = []if usage == "train":trans += [vision.RandomCrop((32, 32), (4, 4, 4, 4)),vision.RandomHorizontalFlip(prob=0.5)]trans += [vision.Resize(resize),vision.Rescale(1.0 / 255.0, 0.0),vision.Normalize([0.4914, 0.4822, 0.4465], [0.2023, 0.1994, 0.2010]),vision.HWC2CHW()]target_trans = transforms.TypeCast(mstype.int32)# 數據映射操作data_set = data_set.map(operations=trans,input_columns='image',num_parallel_workers=workers)data_set = data_set.map(operations=target_trans,input_columns='label',num_parallel_workers=workers)# 批量操作data_set = data_set.batch(batch_size)return data_set# 獲取處理后的訓練與測試數據集dataset_train = create_dataset_cifar10(dataset_dir=data_dir,usage="train",resize=image_size,batch_size=batch_size,workers=workers)
step_size_train = dataset_train.get_dataset_size()dataset_val = create_dataset_cifar10(dataset_dir=data_dir,usage="test",resize=image_size,batch_size=batch_size,workers=workers)
step_size_val = dataset_val.get_dataset_size()

數據可視化：

import matplotlib.pyplot as plt
import numpy as npdata_iter = next(dataset_train.create_dict_iterator())images = data_iter["image"].asnumpy()
labels = data_iter["label"].asnumpy()
print(f"Image shape: {images.shape}, Label shape: {labels.shape}")# 訓練數據集中，前六張圖片所對應的標簽
print(f"Labels: {labels[:6]}")classes = []with open(data_dir + "/batches.meta.txt", "r") as f:for line in f:line = line.rstrip()if line:classes.append(line)# 訓練數據集的前六張圖片
plt.figure()
for i in range(6):plt.subplot(2, 3, i + 1)image_trans = np.transpose(images[i], (1, 2, 0))mean = np.array([0.4914, 0.4822, 0.4465])std = np.array([0.2023, 0.1994, 0.2010])image_trans = std * image_trans + meanimage_trans = np.clip(image_trans, 0, 1)plt.title(f"{classes[labels[i]]}")plt.imshow(image_trans)plt.axis("off")
plt.show()

3、構建網絡

殘差網絡結構(Residual Network)是ResNet網絡的主要亮點，ResNet使用殘差網絡結構后可有效地減輕退化問題，實現更深的網絡結構設計，提高網絡的訓練精度。本節首先講述如何構建殘差網絡結構，然后通過堆疊殘差網絡來構建ResNet50網絡。

構建 Building Block

Building Block結構包含一個殘差支路和short-cut支路，比傳統的卷積結構多了一個short-cut支路，用于傳遞低層的信息使得網絡能夠訓練地很深。

Building Block結構如下圖所示，主分支有兩層卷積網絡結構：

• 主分支第一層網絡以輸入channel為64為例，首先通過一個$3\times 3$的卷積層，然后通過Batch Normalization層，最后通過Relu激活函數層，減小過擬合，梯度消失/爆炸的可能性，輸出channel為64；
• 主分支第二層網絡的輸入channel為64，首先通過一個$3\times 3$的卷積層，然后通過Batch Normalization層，輸出channel為64。

最后將主分支輸出的特征矩陣與shortcuts輸出的特征矩陣相加，通過Relu激活函數即為Building Block最后的輸出。

主分支與shortcuts輸出的特征矩陣相加時，需要保證主分支與shortcuts輸出的特征矩陣shape相同。如果主分支與shortcuts輸出的特征矩陣shape不相同，如輸出channel是輸入channel的一倍時，shortcuts上需要使用數量與輸出channel相等，大小為$1\times 1$的卷積核進行卷積操作；若輸出的圖像較輸入圖像縮小一倍，則要設置shortcuts中卷積操作中的stride為2，主分支第一層卷積操作的stride也需設置為2。

如下代碼定義ResidualBlockBase類實現Building Block結構:

from typing import Type, Union, List, Optional
import mindspore.nn as nn
from mindspore.common.initializer import Normal# 初始化卷積層與BatchNorm的參數
weight_init = Normal(mean=0, sigma=0.02)
gamma_init = Normal(mean=1, sigma=0.02)class ResidualBlockBase(nn.Cell):expansion: int = 1  # 最后一個卷積核數量與第一個卷積核數量相等def __init__(self, in_channel: int, out_channel: int,stride: int = 1, norm: Optional[nn.Cell] = None,down_sample: Optional[nn.Cell] = None) -> None:super(ResidualBlockBase, self).__init__()if not norm:self.norm = nn.BatchNorm2d(out_channel)else:self.norm = normself.conv1 = nn.Conv2d(in_channel, out_channel,kernel_size=3, stride=stride,weight_init=weight_init)self.conv2 = nn.Conv2d(in_channel, out_channel,kernel_size=3, weight_init=weight_init)self.relu = nn.ReLU()self.down_sample = down_sampledef construct(self, x):"""ResidualBlockBase construct."""identity = x  # 建立一個shortcuts分支out = self.conv1(x)    # 主分支第一層：3*3卷積層out = self.norm(out)out = self.relu(out)out = self.conv2(out)  # 主分支第二層：3*3卷積層out = self.norm(out)if self.down_sample is not None:identity = self.down_sample(x)out += identity        # 殘差連接，輸出為主分支與shortcuts之和out = self.relu(out)return out

構建 Bottleneck Block

Bottleneck結構圖如下圖所示，在輸入相同的情況下Bottleneck結構相對Building Block結構的參數數量更少，更適合層數較深的網絡，ResNet50使用的殘差結構就是Bottleneck。該結構的主分支有三層卷積結構，分別為$1\times 1$的卷積層、$3\times 3$卷積層和$1\times 1$的卷積層，其中$1\times 1$的卷積層分別起降維和升維的作用。

• 主分支第一層網絡以輸入channel為256為例，首先通過數量為64，大小為$1\times 1$的卷積核進行降維，然后通過Batch Normalization層，最后通過Relu激活函數層，其輸出channel為64；
• 主分支第二層網絡通過數量為64，大小為$3\times 3$的卷積核提取特征，然后通過Batch Normalization層，最后通過Relu激活函數層，其輸出channel為64；
• 主分支第三層通過數量為256，大小$1\times 1$的卷積核進行升維，然后通過Batch Normalization層，其輸出channel為256。

最后將主分支輸出的特征矩陣與shortcuts輸出的特征矩陣相加，通過Relu激活函數即為Bottleneck最后的輸出。

主分支與shortcuts輸出的特征矩陣相加時，需要保證主分支與shortcuts輸出的特征矩陣shape相同。如果主分支與shortcuts輸出的特征矩陣shape不相同，如輸出channel是輸入channel的一倍時，shortcuts上需要使用數量與輸出channel相等，大小為$1\times 1$的卷積核進行卷積操作；若輸出的圖像較輸入圖像縮小一倍，則要設置shortcuts中卷積操作中的stride為2，主分支第二層卷積操作的stride也需設置為2。

如下代碼定義ResidualBlock類實現Bottleneck結構。

class ResidualBlock(nn.Cell):expansion = 4  # 最后一個卷積核的數量是第一個卷積核數量的4倍def __init__(self, in_channel: int, out_channel: int,stride: int = 1, down_sample: Optional[nn.Cell] = None) -> None:super(ResidualBlock, self).__init__()self.conv1 = nn.Conv2d(in_channel, out_channel,kernel_size=1, weight_init=weight_init)self.norm1 = nn.BatchNorm2d(out_channel)self.conv2 = nn.Conv2d(out_channel, out_channel,kernel_size=3, stride=stride,weight_init=weight_init)self.norm2 = nn.BatchNorm2d(out_channel)self.conv3 = nn.Conv2d(out_channel, out_channel * self.expansion,kernel_size=1, weight_init=weight_init)self.norm3 = nn.BatchNorm2d(out_channel * self.expansion)self.relu = nn.ReLU()self.down_sample = down_sampledef construct(self, x):identity = x  # shortscuts分支out = self.conv1(x)    # 主分支第一層：1*1卷積層out = self.norm1(out)out = self.relu(out)out = self.conv2(out)  # 主分支第二層：3*3卷積層out = self.norm2(out)out = self.relu(out)out = self.conv3(out)  # 主分支第三層：1*1卷積層out = self.norm3(out)if self.down_sample is not None:identity = self.down_sample(x)out += identity  	   # 輸出為主分支與shortcuts之和out = self.relu(out)return out

構建 ResNet50 網絡

ResNet 網絡層結構如下圖所示，以輸入彩色圖像$224\times 224$為例：
1、首先通過數量64，卷積核大小為$7\times 7$，stride為2的卷積層conv1，該層輸出圖片大小為$112\times 112$，輸出channel為64；
2、然后通過一個$3\times 3$的最大下采樣池化層，該層輸出圖片大小為$56\times 56$，輸出channel為64；
3、再堆疊4個殘差網絡塊（conv2_x、conv3_x、conv4_x和conv5_x），此時輸出圖片大小為$7\times 7$，輸出channel為2048；
4、最后通過一個平均池化層、全連接層和softmax，得到分類概率。

對于每個殘差網絡塊，以ResNet50網絡中的conv2_x為例，其由3個Bottleneck結構堆疊而成，每個Bottleneck輸入的channel為64，輸出channel為256。

如下示例定義make_layer實現殘差塊的構建，其參數如下所示:

• last_out_channel：上一個殘差網絡輸出的通道數。
• block：殘差網絡的類別，分別為ResidualBlockBase和ResidualBlock。
• channel：殘差網絡輸入的通道數。
• block_nums：殘差網絡塊堆疊的個數。
• stride：卷積移動的步幅。

def make_layer(last_out_channel, block: Type[Union[ResidualBlockBase, ResidualBlock]],channel: int, block_nums: int, stride: int = 1):down_sample = None  # shortcuts分支if stride != 1 or last_out_channel != channel * block.expansion:down_sample = nn.SequentialCell([nn.Conv2d(last_out_channel, channel * block.expansion,kernel_size=1, stride=stride, weight_init=weight_init),nn.BatchNorm2d(channel * block.expansion, gamma_init=gamma_init)])layers = []layers.append(block(last_out_channel, channel, stride=stride, down_sample=down_sample))in_channel = channel * block.expansion# 堆疊殘差網絡for _ in range(1, block_nums):layers.append(block(in_channel, channel))return nn.SequentialCell(layers)

ResNet50網絡共有5個卷積結構，一個平均池化層，一個全連接層，以CIFAR-10數據集為例：

• conv1：輸入圖片大小為$32\times 32$，輸入channel為3。首先經過一個卷積核數量為64，卷積核大小為$7\times 7$，stride為2的卷積層；然后通過一個Batch Normalization層；最后通過Reul激活函數。該層輸出feature map大小為$16\times 16$，輸出channel為64。
• conv2_x：輸入feature map大小為$16\times 16$，輸入channel為64。首先經過一個卷積核大小為$3\times 3$，stride為2的最大下采樣池化操作；然后堆疊3個$[1\times 1，64；3\times 3，64；1\times 1，256]$結構的Bottleneck。該層輸出feature map大小為$8\times 8$，輸出channel為256。
• conv3_x：輸入feature map大小為$8\times 8$，輸入channel為256。該層堆疊4個[1×1，128；3×3，128；1×1，512]結構的Bottleneck。該層輸出feature map大小為$4\times 4$，輸出channel為512。
• conv4_x：輸入feature map大小為$4\times 4$，輸入channel為512。該層堆疊6個[1×1，256；3×3，256；1×1，1024]結構的Bottleneck。該層輸出feature map大小為$2\times 2$，輸出channel為1024。
• conv5_x：輸入feature map大小為$2\times 2$，輸入channel為1024。該層堆疊3個[1×1，512；3×3，512；1×1，2048]結構的Bottleneck。該層輸出feature map大小為$1\times 1$，輸出channel為2048。
• average pool & fc：輸入channel為2048，輸出channel為分類的類別數。

如下示例代碼實現ResNet50模型的構建，通過用調函數resnet50即可構建ResNet50模型，函數resnet50參數如下：

• num_classes：分類的類別數，默認類別數為1000，本案例的分類數為10。
• pretrained：下載對應的訓練模型，并加載預訓練模型中的參數到網絡中。

from mindspore import load_checkpoint, load_param_into_netclass ResNet(nn.Cell):def __init__(self, block: Type[Union[ResidualBlockBase, ResidualBlock]],layer_nums: List[int], num_classes: int, input_channel: int) -> None:super(ResNet, self).__init__()self.relu = nn.ReLU()# 第一個卷積層，輸入channel為3（彩色圖像），輸出channel為64self.conv1 = nn.Conv2d(3, 64, kernel_size=7, stride=2, weight_init=weight_init)self.norm = nn.BatchNorm2d(64)# 最大池化層，縮小圖片的尺寸self.max_pool = nn.MaxPool2d(kernel_size=3, stride=2, pad_mode='same')# 各個殘差網絡結構塊定義self.layer1 = make_layer(64, block, 64, layer_nums[0])self.layer2 = make_layer(64 * block.expansion, block, 128, layer_nums[1], stride=2)self.layer3 = make_layer(128 * block.expansion, block, 256, layer_nums[2], stride=2)self.layer4 = make_layer(256 * block.expansion, block, 512, layer_nums[3], stride=2)# 平均池化層self.avg_pool = nn.AvgPool2d()# flattern層self.flatten = nn.Flatten()# 全連接層self.fc = nn.Dense(in_channels=input_channel, out_channels=num_classes)def construct(self, x):x = self.conv1(x)x = self.norm(x)x = self.relu(x)x = self.max_pool(x)x = self.layer1(x)x = self.layer2(x)x = self.layer3(x)x = self.layer4(x)x = self.avg_pool(x)x = self.flatten(x)x = self.fc(x)return x

def _resnet(model_url: str, block: Type[Union[ResidualBlockBase, ResidualBlock]],layers: List[int], num_classes: int, pretrained: bool, pretrained_ckpt: str,input_channel: int):model = ResNet(block, layers, num_classes, input_channel)if pretrained:# 加載預訓練模型download(url=model_url, path=pretrained_ckpt, replace=True)param_dict = load_checkpoint(pretrained_ckpt)load_param_into_net(model, param_dict)return modeldef resnet50(num_classes: int = 1000, pretrained: bool = False):"""ResNet50模型"""resnet50_url = "https://mindspore-website.obs.cn-north-4.myhuaweicloud.com/notebook/models/application/resnet50_224_new.ckpt"resnet50_ckpt = "./LoadPretrainedModel/resnet50_224_new.ckpt"return _resnet(resnet50_url, ResidualBlock, [3, 4, 6, 3], num_classes,pretrained, resnet50_ckpt, 2048)# 定義ResNet50網絡
network = resnet50(pretrained=True)# 全連接層輸入層的大小
in_channel = network.fc.in_channels
fc = nn.Dense(in_channels=in_channel, out_channels=10)
# 重置全連接層
network.fc = fc

4、模型訓練

針對CIFAR-10中的10種目標訓練5個epoch（按MindSpore官方文檔中預設的80個Epoch訓練出來的效果會好得多）

# 設置學習率
num_epochs = 5
lr = nn.cosine_decay_lr(min_lr=0.00001, max_lr=0.001, total_step=step_size_train * num_epochs,step_per_epoch=step_size_train, decay_epoch=num_epochs)
# 定義優化器和損失函數
opt = nn.Momentum(params=network.trainable_params(), learning_rate=lr, momentum=0.9)
loss_fn = nn.SoftmaxCrossEntropyWithLogits(sparse=True, reduction='mean')def forward_fn(inputs, targets):logits = network(inputs)loss = loss_fn(logits, targets)return lossgrad_fn = ms.value_and_grad(forward_fn, None, opt.parameters)def train_step(inputs, targets):loss, grads = grad_fn(inputs, targets)opt(grads)return loss

import os# 創建迭代器
data_loader_train = dataset_train.create_tuple_iterator(num_epochs=num_epochs)
data_loader_val = dataset_val.create_tuple_iterator(num_epochs=num_epochs)# 最佳模型存儲路徑
best_acc = 0
best_ckpt_dir = "./BestCheckpoint"
best_ckpt_path = "./BestCheckpoint/resnet50-best.ckpt"if not os.path.exists(best_ckpt_dir):os.mkdir(best_ckpt_dir)

import mindspore.ops as opsdef train(data_loader, epoch):"""模型訓練"""losses = []network.set_train(True)for i, (images, labels) in enumerate(data_loader):loss = train_step(images, labels)if i % 100 == 0 or i == step_size_train - 1:print('Epoch: [%3d/%3d], Steps: [%3d/%3d], Train Loss: [%5.3f]' %(epoch + 1, num_epochs, i + 1, step_size_train, loss))losses.append(loss)return sum(losses) / len(losses)def evaluate(data_loader):"""模型驗證"""network.set_train(False)correct_num = 0.0  # 預測正確個數total_num = 0.0  # 預測總數for images, labels in data_loader:logits = network(images)pred = logits.argmax(axis=1)  # 預測結果correct = ops.equal(pred, labels).reshape((-1, ))correct_num += correct.sum().asnumpy()total_num += correct.shape[0]acc = correct_num / total_num  # 準確率return acc# 開始循環訓練
print("Start Training Loop ...")for epoch in range(num_epochs):curr_loss = train(data_loader_train, epoch)curr_acc = evaluate(data_loader_val)print("-" * 50)print("Epoch: [%3d/%3d], Average Train Loss: [%5.3f], Accuracy: [%5.3f]" % (epoch+1, num_epochs, curr_loss, curr_acc))print("-" * 50)# 保存當前預測準確率最高的模型if curr_acc > best_acc:best_acc = curr_accms.save_checkpoint(network, best_ckpt_path)print("=" * 80)
print(f"End of validation the best Accuracy is: {best_acc: 5.3f}, "f"save the best ckpt file in {best_ckpt_path}", flush=True)

效果如下：

5、圖像分類模型推理

import matplotlib.pyplot as pltdef visualize_model(best_ckpt_path, dataset_val):num_class = 10  # 對狼和狗圖像進行二分類net = resnet50(num_class)# 加載模型參數param_dict = ms.load_checkpoint(best_ckpt_path)ms.load_param_into_net(net, param_dict)# 加載驗證集的數據進行驗證data = next(dataset_val.create_dict_iterator())images = data["image"]labels = data["label"]# 預測圖像類別output = net(data['image'])pred = np.argmax(output.asnumpy(), axis=1)# 圖像分類classes = []with open(data_dir + "/batches.meta.txt", "r") as f:for line in f:line = line.rstrip()if line:classes.append(line)# 顯示圖像及圖像的預測值plt.figure()for i in range(6):plt.subplot(2, 3, i + 1)# 若預測正確，顯示為藍色；若預測錯誤，顯示為紅色color = 'blue' if pred[i] == labels.asnumpy()[i] else 'red'plt.title('predict:{}'.format(classes[pred[i]]), color=color)picture_show = np.transpose(images.asnumpy()[i], (1, 2, 0))mean = np.array([0.4914, 0.4822, 0.4465])std = np.array([0.2023, 0.1994, 0.2010])picture_show = std * picture_show + meanpicture_show = np.clip(picture_show, 0, 1)plt.imshow(picture_show)plt.axis('off')plt.show()# 使用測試數據集進行驗證
visualize_model(best_ckpt_path=best_ckpt_path, dataset_val=dataset_val)

上述訓練的5個Epoch分類效果如下：

CIFAR-10訓練集里有粉色的青蛙嘛？

Reference

[1] 昇思大模型平臺
[2] 昇思官方文檔-ResNet50圖像分類
[3] 深度學習（五）：pytorch遷移學習之resnet50
[4] Resnet-50網絡結構詳解