昇思25天學習打卡營第19天 | ResNet50遷移學習再續

訓練模型部分代碼解析

構建Resnet50網絡

兩行初始化代碼

weight_init = Normal(mean=0, sigma=0.02)

這行代碼定義了一個初始化器weight_init，它將使用均值為0，標準差為0.02的正態分布來初始化網絡中的權重。這種初始化策略有助于在網絡的初始階段避免梯度消失或爆炸的問題，因為權重被初始化為接近于0但又不至于太小的值。

gamma_init = Normal(mean=1, sigma=0.02)

這行代碼定義了一個初始化器gamma_init，通常用于初始化批量歸一化（Batch Normalization）層中的尺度參數（gamma）。批量歸一化是一種常用的技術，用于加速神經網絡的訓練過程，并通過提高穩定性來改善性能。在這種情況下，gamma參數的初始化通常設置為接近1的值，因為它們在批量歸一化層中起到縮放的作用，初始化為1意味著在初始階段不會改變激活的尺度。

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

這段代碼定義了一個名為ResidualBlockBase的類，它是一個殘差塊（Residual Block）的基礎實現。殘差塊是殘差網絡的基本構建塊，它允許網絡在增加層數的同時，仍然能夠通過反向傳播有效地訓練。

函數解析：

__init__:
- 作用：這是一個構造函數，用于初始化殘差塊。它定義了殘差塊的基本結構和操作，包括卷積層、批量歸一化層和ReLU激活函數。
- 參數：
  - in_channel：輸入通道數。
  - out_channel：輸出通道數。
  - stride：卷積層的步幅。
  - norm：可選的規范化層，默認為None，此時會使用批量歸一化。
  - down_sample：可選的下采樣層，用于匹配輸入和輸出的維度，默認為None。
- 代碼邏輯：
  - 調用父類的構造函數：super(ResidualBlockBase, self).__init__()。
  - 判斷是否提供了自定義的規范化層，如果沒有，則使用批量歸一化：nn.BatchNorm2d(out_channel)。
  - 定義兩個卷積層conv1和conv2，其中conv1可能包含步幅stride用于降采樣。
  - 定義ReLU激活函數relu。
  - 保存可選的下采樣層down_sample。
construct:
- 作用：這是執行前向傳播的函數，用于定義輸入數據通過殘差塊的流程。
- 參數：
  - x：輸入數據，通常是特征圖。
- 代碼邏輯：
  - 保存輸入數據x作為身份映射（shortcuts分支）。
  - 通過第一個卷積層conv1處理輸入數據，然后進行規范化、ReLU激活。
  - 通過第二個卷積層conv2處理激活后的數據，然后進行規范化。
  - 判斷是否存在下采樣層down_sample，如果存在，則對身份映射進行下采樣以匹配輸出的維度。
  - 將主分支的輸出out與身份映射identity相加，實現跳躍連接（skip connection）。
  - 對相加的結果應用ReLU激活函數。
  - 返回最終的輸出out。
    殘差塊通過跳躍連接允許梯度直接流過身份映射，這有助于解決深度網絡中的梯度消失問題，并允許網絡學習更復雜的函數。

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

---------------------------------------------------------------------------NameError                                 Traceback (most recent call last)Cell In[3], line 1
----> 1 class ResidualBlockBase(nn.Cell):2     expansion: int = 1  # 最后一個卷積核數量與第一個卷積核數量相等4     def __init__(self, in_channel: int, out_channel: int,5                  stride: int = 1, norm: Optional[nn.Cell] = None,6                  down_sample: Optional[nn.Cell] = None) -> None:NameError: name 'nn' is not defined

這段代碼定義了一個名為ResidualBlock的類，它也是殘差網絡（ResNet）中的一個殘差塊，但是與之前的基礎版殘差塊相比，這個類實現了更復雜的殘差學習結構，這個殘差塊包含了三個卷積層，其中第一個和第三個卷積層是1x1的，用于降維和升維，第二個卷積層是3x3的，用于提取特征。

函數解析

__init__:
- 作用：這是一個構造函數，用于初始化殘差塊。它定義了殘差塊的基本結構和操作，包括三個卷積層、三個批量歸一化層和ReLU激活函數。
- 參數：
  - in_channel：輸入通道數。
  - out_channel：輸出通道數，這里的輸出通道數是指第一個卷積層之后的通道數，最終的輸出通道數將是這個數的4倍，由expansion變量控制。
  - stride：第二個卷積層的步幅，用于降采樣。
  - down_sample：可選的下采樣層，用于匹配輸入和輸出的維度，默認為None。
- 代碼邏輯：
  - 調用父類的構造函數：super(ResidualBlock, self).__init__()。
  - 定義三個卷積層conv1、conv2和conv3，其中conv1和conv3是1x1卷積，conv2是3x3卷積，conv2可能包含步幅stride用于降采樣。
  - 定義三個對應的批量歸一化層norm1、norm2和norm3。
  - 定義ReLU激活函數relu。
  - 保存可選的下采樣層down_sample。
construct:
- 作用：這是執行前向傳播的函數，用于定義輸入數據通過殘差塊的流程。
- 參數：
  - x：輸入數據，通常是特征圖。
- 代碼邏輯：
  - 保存輸入數據x作為身份映射（shortcuts分支）。
  - 通過第一個卷積層conv1處理輸入數據，然后進行規范化、ReLU激活。
  - 通過第二個卷積層conv2處理激活后的數據，然后進行規范化、ReLU激活。
  - 通過第三個卷積層conv3處理激活后的數據，然后進行規范化。
  - 判斷是否存在下采樣層down_sample，如果存在，則對身份映射進行下采樣以匹配輸出的維度。
  - 將主分支的輸出out與身份映射identity相加，實現跳躍連接（skip connection）。
  - 對相加的結果應用ReLU激活函數。
  - 返回最終的輸出out。
    這個殘差塊通過跳躍連接允許梯度直接流過身份映射，這有助于解決深度網絡中的梯度消失問題，并允許網絡學習更復雜的函數。這種結構在許多圖像識別任務中表現出色，并且是現代深度學習架構的重要組成部分。

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)

這段代碼定義了一個名為make_layer的函數，它的作用是創建并堆疊多個殘差塊（Residual Blocks），形成殘差網絡（ResNet）中的一個層。這個函數可以根據給定的參數生成不同數量的殘差塊，并且可以處理不同層的輸入和輸出通道數不匹配的情況。

函數解析

函數參數:
- last_out_channel：前一個層的輸出通道數。
- block：殘差塊類型，可以是ResidualBlockBase或ResidualBlock。
- channel：當前層的輸入通道數（對于第一個殘差塊）。
- block_nums：要堆疊的殘差塊數量。
- stride：第一個殘差塊的步幅，用于降采樣。
函數邏輯:
- 首先定義一個down_sample變量，用于可能需要的下采樣操作。
- 檢查是否需要下采樣，即如果stride不等于1或者last_out_channel不等于channel * block.expansion，則創建一個下采樣層，包括一個1x1的卷積層和一個批量歸一化層。
- 初始化一個layers列表，并將第一個殘差塊添加到列表中，這個殘差塊使用down_sample作為其下采樣層。
- 然后使用一個循環，將剩余的殘差塊添加到layers列表中，每個殘差塊的輸入通道數是前一個殘差塊的輸出通道數。
- 最后，使用nn.SequentialCell(layers)將堆疊的殘差塊封裝為一個順序層（Sequential Layer），并返回這個層。
  這個函數是構建殘差網絡的關鍵部分，因為它允許開發者通過簡單地指定殘差塊的數量和配置來創建復雜的網絡結構。通過這種方式，可以很容易地增加網絡的深度，而不會遇到梯度消失問題，因為殘差塊的結構允許梯度直接流過身份映射。

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 xdef _resnet(model_url: str, block: Type[Union[ResidualBlockBase, ResidualBlock]],layers: List[int], num_classes: int, pretrained: bool, pretrianed_ckpt: str,input_channel: int):model = ResNet(block, layers, num_classes, input_channel)if pretrained:# 加載預訓練模型download(url=model_url, path=pretrianed_ckpt, replace=True)param_dict = load_checkpoint(pretrianed_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)

函數部分解析：

__init__:
- 作用：這是一個構造函數，用于初始化ResNet模型。它定義了模型的所有層和操作。
- 參數：
  - block：殘差塊類型，可以是ResidualBlockBase或ResidualBlock。
  - layer_nums：一個列表，指定每個殘差網絡結構塊中殘差塊的數量。
  - num_classes：輸出類別數，用于全連接層的輸出尺寸。
  - input_channel：全連接層的輸入通道數。
- 代碼邏輯：
  - 調用父類的構造函數：super(ResNet, self).__init__()。
  - 定義ReLU激活函數relu。
  - 定義第一個卷積層conv1，用于處理輸入圖像。
  - 定義批量歸一化層norm。
  - 定義最大池化層max_pool，用于縮小圖像尺寸。
  - 定義四個殘差網絡結構塊layer1到layer4，每個塊由make_layer函數創建。
  - 定義平均池化層avg_pool。
  - 定義Flatten層flatten，用于將特征圖展平為一維向量。
  - 定義全連接層fc，用于最終分類。
construct:
- 作用：這是執行前向傳播的函數，用于定義輸入數據通過整個ResNet模型的流程。
- 參數：
  - x：輸入數據，通常是特征圖。
- 代碼邏輯：
  - 通過第一個卷積層conv1處理輸入數據，然后進行批量歸一化和ReLU激活。
  - 通過最大池化層max_pool處理激活后的數據。
  - 通過四個殘差網絡結構塊layer1到layer4處理數據。
  - 通過平均池化層avg_pool處理數據。
  - 通過Flatten層flatten將特征圖展平為一維向量。
  - 通過全連接層fc進行最終的分類。
  - 返回最終的輸出。
_resnet:
- 作用：這是一個輔助函數，用于創建并初始化ResNet模型，可以選擇是否加載預訓練的權重。
- 參數：
  - model_url：預訓練模型的URL。
  - block：殘差塊類型。
  - layers：殘差塊的數量列表。
  - num_classes：輸出類別數。
  - pretrained：一個布爾值，指示是否加載預訓練模型。
  - pretrianed_ckpt：預訓練模型的本地路徑。
  - input_channel：全連接層的輸入通道數。
- 代碼邏輯：
  - 創建ResNet模型model。
  - 如果pretrained為True，則從model_url下載預訓練模型并將其加載到model中。
resnet50:
- 作用：這是一個工廠函數，用于創建ResNet50模型。
- 參數：
  - num_classes：輸出類別數，默認為1000。
  - pretrained：一個布爾值，指示是否加載預訓練模型，默認為False。
- 代碼邏輯：
  - 定義ResNet50模型的URL和本地檢查點路徑。
  - 調用_resnet函數創建并返回ResNet50模型，可以選擇是否加載預訓練的權重。

固定特征進行訓練

這部分代碼附有注釋，看代碼即可

import mindspore as ms
import matplotlib.pyplot as plt
import os
import timenet_work = resnet50(pretrained=True)# 全連接層輸入層的大小
in_channels = net_work.fc.in_channels
# 輸出通道數大小為狼狗分類數2
head = nn.Dense(in_channels, 2)
# 重置全連接層
net_work.fc = head# 平均池化層kernel size為7
avg_pool = nn.AvgPool2d(kernel_size=7)
# 重置平均池化層
net_work.avg_pool = avg_pool# 凍結除最后一層外的所有參數
for param in net_work.get_parameters():if param.name not in ["fc.weight", "fc.bias"]:param.requires_grad = False# 定義優化器和損失函數
opt = nn.Momentum(params=net_work.trainable_params(), learning_rate=lr, momentum=0.5)
loss_fn = nn.SoftmaxCrossEntropyWithLogits(sparse=True, reduction='mean')def forward_fn(inputs, targets):logits = net_work(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# 實例化模型
model1 = train.Model(net_work, loss_fn, opt, metrics={"Accuracy": train.Accuracy()})

訓練和評估

訓練這塊之前有做過類似操作，不過多贅述

import mindspore as ms
import matplotlib.pyplot as plt
import os
import time
dataset_train = create_dataset_canidae(data_path_train, "train")
step_size_train = dataset_train.get_dataset_size()dataset_val = create_dataset_canidae(data_path_val, "val")
step_size_val = dataset_val.get_dataset_size()num_epochs = 5# 創建迭代器
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_ckpt_dir = "./BestCheckpoint"
best_ckpt_path = "./BestCheckpoint/resnet50-best-freezing-param.ckpt"

import mindspore as ms
import matplotlib.pyplot as plt
import os
import time
# 開始循環訓練
print("Start Training Loop ...")best_acc = 0for epoch in range(num_epochs):losses = []net_work.set_train()epoch_start = time.time()# 為每輪訓練讀入數據for i, (images, labels) in enumerate(data_loader_train):labels = labels.astype(ms.int32)loss = train_step(images, labels)losses.append(loss)# 每個epoch結束后，驗證準確率acc = model1.eval(dataset_val)['Accuracy']epoch_end = time.time()epoch_seconds = (epoch_end - epoch_start) * 1000step_seconds = epoch_seconds/step_size_trainprint("-" * 20)print("Epoch: [%3d/%3d], Average Train Loss: [%5.3f], Accuracy: [%5.3f]" % (epoch+1, num_epochs, sum(losses)/len(losses), acc))print("epoch time: %5.3f ms, per step time: %5.3f ms" % (epoch_seconds, step_seconds))if acc > best_acc:best_acc = accif not os.path.exists(best_ckpt_dir):os.mkdir(best_ckpt_dir)ms.save_checkpoint(net_work, 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)

可視化模型預測

看代碼注釋即可

import matplotlib.pyplot as plt
import mindspore as msdef visualize_model(best_ckpt_path, val_ds):net = resnet50()# 全連接層輸入層的大小in_channels = net.fc.in_channels# 輸出通道數大小為狼狗分類數2head = nn.Dense(in_channels, 2)# 重置全連接層net.fc = head# 平均池化層kernel size為7avg_pool = nn.AvgPool2d(kernel_size=7)# 重置平均池化層net.avg_pool = avg_pool# 加載模型參數param_dict = ms.load_checkpoint(best_ckpt_path)ms.load_param_into_net(net, param_dict)model = train.Model(net)# 加載驗證集的數據進行驗證data = next(val_ds.create_dict_iterator())images = data["image"].asnumpy()labels = data["label"].asnumpy()class_name = {0: "dogs", 1: "wolves"}# 預測圖像類別output = model.predict(ms.Tensor(data['image']))pred = np.argmax(output.asnumpy(), axis=1)# 顯示圖像及圖像的預測值plt.figure(figsize=(5, 5))for i in range(4):plt.subplot(2, 2, i + 1)# 若預測正確，顯示為藍色；若預測錯誤，顯示為紅色color = 'blue' if pred[i] == labels[i] else 'red'plt.title('predict:{}'.format(class_name[pred[i]]), color=color)picture_show = np.transpose(images[i], (1, 2, 0))mean = np.array([0.485, 0.456, 0.406])std = np.array([0.229, 0.224, 0.225])picture_show = std * picture_show + meanpicture_show = np.clip(picture_show, 0, 1)plt.imshow(picture_show)plt.axis('off')plt.show()