小狐狸事務所: 機器學習筆記 : 強化式學習-打造最強通用演算法 (三D)

2022年5月3日星期二

機器學習筆記 : 強化式學習-打造最強通用演算法 (三D)

本篇繼續來整理我讀布留川英一寫的 "強化式學習-打造最強通用演算法 AlphaZero" 這本書的摘要筆記, 本系列之前的筆記參考 :

# 機器學習筆記 : 強化式學習-打造最強通用演算法 AlphaZero (一)

以下是我讀完第三章 "深度學習" 第四個演練範例 (殘差神經網路 ResNet) 的摘要筆記, 作者詳細解說了 ResNet 的實作方式, 這在深度學習書籍中是很少見的. 此例為了比較 ResNet 模型與前一篇使用的 CNN 模型的表現, 仍然使用 CIFAR-10 影像資料集. 以下演練不再檢視資料集, 請參考前一篇.

加深神經網路的層數理論上可以提取更複雜的特徵, 但層數到達某個程度時, 訓練的準確度會先上升再下降, 稱為退化問題 (degradation problem), 參考 ResNet 論文 :

# https://arxiv.org/abs/1512.03385

微軟研究院團隊於 2015 年提出第一個殘差神經網路 ResNet 並在當年 ILSVRC 視覺辨識大賽中於辨識, 偵測, 與圖像分割等項目奪得冠軍. ResNet 使用了 152 層的神經網路, 如此深的網路準確率卻能獲得顯著提升, 其獨到之處是使用所謂的殘差塊 (residual block) 結構來解決深度神經網路的退化問題, 使得深層網路的訓練可以獲得更好的結果, 參考 :

# ResNet - 2015年 ILSVRC 的赢家

基本的 ResNet 的結構如下圖所示 :

殘差網路的主角殘差塊事實上就是具有捷徑連接 (shortcut connection) 的卷積塊, 此連接的作用是將傳入卷積層的輸入跨層傳遞, 然後與經卷積運算之結果相加後再送給 ReLu 激活函數. 此處的相加是廣義而言, 但其實是相減, 所謂殘差 (residual) 就是卷積運算後的值減掉運算前 (即經捷徑傳遞者) 的值, 若神經網路在前一層就已經學得很好, 而在這一層卻甚麼都沒學到, 則相減之後殘差為 0, 此時這層的輸出就等於輸入 (捷徑), 故若這一層沒學到甚麼, 模型也不會退化, 亦即沒學到新的, 舊的也不會忘記, 這就是殘差學習.

殘差塊最後會經過批次正規化 (batch normalization) 處理, 讓輸出呈現 "平均值為 0, 標準差為 1" 的分布, 它可使深層神經網路容易被訓練, 也具有防止過度配適的作用, 正規化通常添加於卷積層與激活函數之間. 根據經驗, 正規化的效果比使用丟棄層要好, 但最好不要與丟棄層混合使用.

ResNet 結構的另一個特點是它採用 "全域平均池化 (global average maximum pooling)" 方法來處理與密集層介接時參數過多的問題. 傳統 CNN 網路經過卷積層萃取圖案特徵後需要經過展平層將陣列拉平為向量, 以便能介接到密集層, 但殘差網路層數很多, 若用展平層會有參數過多容易形成過度配適, 而 "全域平均池化" 就是用來取代展平層, 作法是將最後一層的每張特徵圖取平均值, 然後將這些平均特徵值組成的向量直接餵給密集層, 這樣就能有效減少特徵值數量. 參考 :

# Deep Residul Learning in Image Recogition

# Identity Mapping in Deep Residual Network

由於 ResNet 結構較複雜 (有分岔捷徑之非線性堆疊), 無法使用 Sequential() 以一個輸入接一個輸出的線性堆疊串接方式來建構, 必須改用 TensorFlow 的 Functional API 來實作非線性堆疊, 此方法可用來建構任何神經網路模型, 例如多輸出 (分類 + 迴歸) 與非線性堆疊或其組合. 線性堆疊通常使用 Sequential() 較直觀方便, 但也可以使用 Functional API 來搭建, 例如下面這個網路 :

model=Sequential() # 建構序列模型
model.add(Dense(256, activation='sigmoid', input_shape=(784,))) # 輸入層+密集層1
model.add(Dense(128, activation='sigmoid')) # 密集層 2

model.add(Dropout(0.5)) # 丟棄層
model.add(Dense(10, activation='softmax')) # 輸出層

如果用 Functional API 方式來建構是這樣 :

input=Input(shape=(784, )) # 輸入層
x=Dense(256, activation='sigmoid')(input) # 密集層 1
x=Dense(128, activation='sigmoid')(x) # 密集層 2

x=Dropout(rate=0.5)(x) # 丟棄層

x=Dense(10, activation='softmax')(x) # 輸出層

model=Model(inputs=input, outputs=x) # 建構模型

以下是演練部分 :

Matplotlib 繪圖內嵌 :
%matplotlib inline
匯入套件模組 :
from tensorflow.keras.datasets import cifar10
from tensorflow.keras.callbacks import LearningRateScheduler
from tensorflow.keras.layers import Activation, Add, BatchNormalization
from tensorflow.keras.layers import Conv2D, Dense
from tensorflow.keras.layers import GlobalAveragePooling2D, Input
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import SGD
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.regularizers import l2
from tensorflow.keras.utils import to_categorical
import numpy as np
import matplotlib.pyplot as plt

此處從 tf.keras.layers 匯入的模組與類別其實可以寫成一列, 因為太長會破壞網頁結構, 所以分成幾列匯入. 其中 Conv2D 用來建立卷積層, BatchNormalization 用來將特徵圖做批次正規化處理, GlobalAveragePooling 用來做全域平均池化以減少參數數量, 而 Add 就是殘差塊中用來相加的函數.
載入資料集 :
為了與前一個範例的 CNN 網路做比較, 此例仍使用 CIFAR-10 資料集 :
(train_images, train_labels), (test_images, test_labels)= cifar10.load_data()

CIFAR-10 資料集的檢視參考上一篇 CNN 網路, 此處不再重複 :

# 機器學習筆記 : 強化式學習-打造最強通用演算法 (三C)
資料預處理 :
由於 ResNet 有自己的批次正規化處理, 故此處僅需要對標籤做預處理, 也就是將原本 0~9 的分類答案傳給 to_categorical() 函數轉成 one-hot 編碼 :

train_labels=to_categorical(train_labels)
test_labels=to_categorical(test_labels)
print(train_images.shape)
print(train_labels.shape)
print(test_images.shape)
print(test_labels.shape)

結果如下 :

(50000, 32, 32, 3)
(50000, 10)
(10000, 32, 32, 3)
(10000, 10)

可見每一筆標籤都已變成 10 維向量了.

建構 ResNet 神經網路 :
(1). 建立卷積層 :
本例的 ResNet 網路是仿製原論文中的 54 層殘差網路, 殘差塊中的卷積層是用 Conv2D 類別來建構, 此類別的建構子常用參數如下 :

Conv2D() 常用參數	說明
filters	卷積核個數 (int)
kernel_size	卷積核尺寸 (int 或 tuple)
strides	步幅 (int 或 tuple)
padding	填補方式 (str), 預設 'valid' (無填補), 欲填補須設為 'same'
use_bias	是否啟用偏值向量 (bool), 預設 False (不使用)
kernel_initializer	卷積核權重之初始值 (str), 預設 'glorot_uniform'
kernel_regularizer	卷積核常規化 (Regularizer), 預設 None

因為要建構的卷積層很多, 故自訂一個函式來簡化程式碼 :

# 建構卷積層
def conv(filters, kernel_size, strides=1):
    return Conv2D(filters, kernel_size, strides=strides, padding='same', 
            use_bias=True, kernel_initializer='he_normal', 
            kernel_regularizer=l2(0.0001))

此卷積層函式會建立預設步幅為 1 且有填補之卷積層 (輸出尺寸不變), 不使用偏值向量, 卷積核初始化方法為 'he_normal' (常態分佈), 採用 L2 常規化.

(2). 建立殘差塊 :
殘差塊的結構有許多種設計, 此例使用下列 A, B 兩種殘差塊 :

可見這兩種殘差塊都是三組 BatchNormalization -> ReLu -> 卷積層結構的重複串接, 只是內部捷徑連接的方式不同而已, 右邊的殘差塊 B 從輸入端就抄捷徑了; 而左邊的殘差塊 A 從第一組的 ReLu 輸出才抄捷徑.

因為本例的 ResNet 要使用的殘差塊有 54 塊之多, 所以最好是將上面這兩種殘差塊結構分別寫成函式以簡化程式碼, 首先殘差塊 A 以函式 first_residual_block() 來實現 :

# 建構殘差塊 A
def first_residual_unit(filters, strides):
    def f(x):
        # 正規化 → ReLU 激活
        x=BatchNormalization()(x)
        x_b=Activation('relu')(x)
        # 卷積層 → 正規化 → ReLU 激活
        x=conv(filters // 4, 1, strides)(x_b)
        x=BatchNormalization()(x)
        x=Activation('relu')(x)        
        # 卷積層 → 正規化 → ReLU 激活
        x=conv(filters // 4, 3)(x)
        x=BatchNormalization()(x)
        x=Activation('relu')(x)
        # 卷積層
        x=conv(filters, 1)(x)
        # 調整捷徑之 shape 的尺寸
        x_b=conv(filters, 1, strides)(x_b)
        # Add
        return Add()([x, x_b])
    return f

注意這是一個巢狀函式, 它的傳回值是內部函式 f(x), 因為此例要建立兩類三組合計 54 個殘差塊, 每一組的參數不同, 所以要靠這個傳回的 f(x) 來建立. 殘差塊 A 有兩個參數 filters 與 strides, 分別用來設定卷積核的數目與步幅. 注意, 傳入的 fiters 是指殘差塊 A 中最大的卷積核數目, 用在第三個卷積層, 前面兩個卷積層之卷積核數目為 1/4, 故使用 //4 來獲得.

接著建構殘差塊 B, 此處以 residual_unit() 函式來實現 :

# 建構殘差塊 B
def residual_unit(filters):
    def f(x):
        x_b = x        
        # → 正規化 → ReLU 激活
        x=BatchNormalization()(x)
        x=Activation('relu')(x)        
        # 卷積層 → 正規化 → ReLU 激活
        x=conv(filters // 4, 1)(x)
        x=BatchNormalization()(x)
        x=Activation('relu')(x)        
        # 卷積層 → 正規化 → ReLU 激活
        x=conv(filters // 4, 3)(x)
        x=BatchNormalization()(x)
        x=Activation('relu')(x)        
        # 卷積層
        x=conv(filters, 1)(x)
        # Add
        return Add()([x, x_b])
    return f

此函式只有一個傳入參數, 即卷積核數目 filters, 因為殘差塊 B 裡面的三個卷積層步幅都是固定的, 不須設定.

本例的 ResNet 殘差神經網路是利用這兩種殘差塊去做組合建構的, 組合方式如下 :

整個殘差網路由三個組合構成, 每一個組合包含一個殘差塊 A 與 17 個殘差塊 B, 所以總共有 54 個殘差塊, 每組最大卷積核數目分別是 64, 128, 與 256. 建立殘差塊組合是以下列這個自訂函式 residual_block() 來實現 :

# 建構殘差塊 A×1 與 殘差塊 B×17
def residual_block(filters, strides, unit_size):
    def f(x):
        x=first_residual_unit(filters, strides)(x) # 建立殘差塊 A
        for i in range(unit_size-1):       
            x=residual_unit(filters)(x)  # 建立殘差塊 B
        return x
    return f

此處參數 unit_size 就是殘差塊 B 的數目, 所以呼叫時要傳入 18. 以上基本組件都準備好之後即可開始建構整個 ResNet 了 :

# 輸入層
input=Input(shape=(32,32, 3)) # CIFAR 圖片 32x32, RGB 三通道
# 卷積層
x=conv(16, 3)(input) # 16 個 3x3 卷積核
# 殘差塊 x 54
x=residual_block(64, 1, 18)(x)   # 第一組殘差塊
x=residual_block(128, 2, 18)(x)  # 第二組殘差塊
x=residual_block(256, 2, 18)(x)  # 第三組殘差塊
# 正規化 → ReLU 激活
x=BatchNormalization()(x)
x=Activation('relu')(x)
# 池化層
x=GlobalAveragePooling2D()(x)
# 密集層
output=Dense(10, activation='softmax', kernel_regularizer=l2(0.0001))(x)
# 建構模型
model=Model(inputs=input, outputs=output)

model.summary()  # 檢視網路摘要

這網路結構非常複雜, 以下僅列出摘要最後的參數統計 :

Total params: 1,740,042
Trainable params: 1,715,818
Non-trainable params: 24,224

總參數數目高達 174 萬!

編譯模型 :
model.compile(loss='categorical_crossentropy', optimizer=SGD(momentum=0.9),
metrics=['acc'])
與上一篇 CNN 網路不同之處為優化器採用指定動量的 SGD (根據 ResNet 論文).
訓練模型 :
在訓練模型之前還要做兩項準備, 一是資料擴增 (data augmentation) 與資料正規化 (normalization), 二是利用 callbacks 機制動態調整學習率. 所謂資料擴增是指利用小幅度修改現有樣本來產生更多新的樣本, 例如將圖片旋轉或位移等, 這可讓模型學習到更多樣化的資料, 避免過度配適, 這是利用 tf.keras 預處理模組中的 ImageDataGenerator 類別來做的 :

# 設定訓練集的 ImageDataGenerator
train_gen=ImageDataGenerator(
    featurewise_center=True, # 設定整體資料集之輸入平均值=0
    featurewise_std_normalization=True, # 以訓練集的標準差做正規化
    width_shift_range=0.125, # 隨機向左向右平移 12.5% 寬度以內之像素
    height_shift_range=0.125, # 隨機向上向下平移 12.5% 高度以內之像素
    horizontal_flip=True) # 隨機水平翻轉圖片
# 設定測試集的 ImageDataGenerator
test_gen=ImageDataGenerator(
    featurewise_center=True, # 設定整體資料集之輸入平均值=0
    featurewise_std_normalization=True) # 以測試集的標準差做正規化
for data in (train_gen, test_gen): # 對訓練集與測試集做正規化
    data.fit(train_images) # 以 train_images 之平均值與標準差做正規化

注意, 此處只對訓練集做資料擴增, 測試集不需要 (只做正規化).

第二項準備是要自訂一個可動態調整學習率的回呼函式, 然後利用 fit() 的 callbacks 參數讓模型能夠透過此回呼函式動態調整學習率, 這樣就能在距離正確答案較遠時使用較大的學習率, 而較近時改用較小的學習率, 以便能快速找到正確答案. 此處使用 tf.keras.callbacks 模組中的 LearningRateSchedular 類別來實現, 它可以在每個訓練週期 (epoch) 規劃不同的學習率 :

# 設定 LearningRateScheduler
def step_decay(epoch): # 以 epoch 為單位傳回新的學習率
    x=0.1  # 初始學習率
    if epoch >= 80: x=0.01  # 第 80 週期後學習率降為 0.01
    if epoch >= 120: x=0.001 # 第 120 週期後學習率降為 0.001
    return x
lr_decay=LearningRateScheduler(step_decay, verbose = 1) # 指定學習率函式

這樣就可以開始來訓練模型了, 但是因為 ResNet 網路複雜參數多, 計算量很大, 必須先更改 Colab 的執行環境設定為 GPU 才能提速 :

# 訓練模型
batch_size=128
history=model.fit(
    train_gen.flow(train_images, train_labels, batch_size=batch_size),
    epochs=120,
    steps_per_epoch=train_images.shape[0] // batch_size,
    validation_data=test_gen.flow(test_images, test_labels,
                   batch_size=batch_size),
    validation_steps=test_images.shape[0] // batch_size,
    callbacks=[lr_decay])

此處 batch_size 為在一個 epoch 中, 每次從訓練集中依序取出的圖片量 (128 張). 參數 steps_per_epoch 用來指定每個周期會做幾個批次, 這在一般的訓練中不需要特別指定, 因為訓練資料總筆數固定, 系統會自動計算, 以 CIFAR 為例, 一個 epoch 的批次數為 50000/128=390 個批次. 但此處因為使用了資料擴增, 這樣 ImageDataGenerator 會無限制地隨機動態產生擴充資料, 所以必須明確地指定每個周期的批次數, 否則模型會無限次地訓練下去.

(用 GPU 還是跑很久 ...)

一個 epoch 要花 5 分鐘 15 秒, 預估跑完整個 120 輪要花掉 630 分鐘, 也就是 10.5 小時, 這還在 Colab 免費帳戶 12 小時重置範圍內, 但必須 90 分鐘內重新整理網頁才能避免因為閒置而被重置, 參考 :

# 機器學習筆記 : 強化式學習-打造最強通用演算法 (二)

但跑了三次都不到半小時就停住了, 殘念 ~~~. 雖然一直跑不出訓練結果, 還是先將後續指令記錄下來, 等 Jetson Nano 環境布置好再試看看 :
儲存模型 :
model.save('resnet.h5')
繪製訓練圖 :
# 繪製圖形
plt.plot(history.history['acc'], label='acc')
plt.plot(history.history['val_acc'], label='val_acc')
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend(loc='best')
plt.show()
評估 :
# 評估
test_loss, test_acc=model.evaluate(test_gen.flow(test_images, test_labels))
print('loss: {:.3f}\nacc: {:.3f}'.format(test_loss, test_acc ))
預測 :
for i in range(10):
    plt.subplot(2, 5, i+1)
    plt.imshow(test_images[i])
plt.show()

test_predictions=model.predict(test_gen.flow(test_images[0:10], shuffle=False))
test_predictions=np.argmax(test_predictions, axis=1)
labels=['airplane', 'automobile', 'bird', 'cat', 'deer',
    'dog', 'frog', 'horse', 'ship', 'truck']
print('前 10 筆預測標籤:',[labels[n] for n in test_predictions])
test_ans=np.argmax(test_labels[:10], axis=1)
print('前 10 筆原始標籤:',[labels[n] for n in test_ans])