Categories
程式開發

TensorFlow 篇| TensorFlow 2.x 基於Keras 模型的本地訓練與評估


TensorFlow 篇| TensorFlow 2.x 基於Keras 模型的本地訓練與評估 1

「導語」模型的訓練與評估是整個機器學習任務流程的核心環節。只有掌握了正確的訓練與評估方法,並靈活使用,才能使我們更加快速地進行實驗分析與驗證,從而對模型有更加深刻的理解。

前言

在上一篇Keras 模型構建的文章中,我們介紹了在TensorFlow 2.x 版本中使用Keras 構建模型的三種方法,那麼本篇將在上一篇的基礎上著重介紹使用Keras 模型進行本地訓練、評估以及預測的流程和方法。 Keras 模型有兩種訓練評估的方式,一種方式是使用模型內置API ,如model.fit() , model.evaluate() 和model.predict() 等分別執行不同的操作;另一種方式是利用即時執行策略(eager execution) 以及GradientTape 對象自定義訓練和評估流程。對所有Keras 模型來說這兩種方式都是按照相同的原理來工作的,沒有本質上的區別。在一般情況下,我們更願意使用第一種訓練評估方式,因為它更為簡單,更易於使用,而在一些特殊的情況下,我們可能會考慮使用自定義的方式來完成訓練與評估。

內置API 進行訓練評估

端到端完整示例

下面介紹使用模型內置API 實現的一個端到端的訓練評估示例,可以認為要使用該模型去解決一個多分類問題。這裡使用了函數式API 來構建Keras 模型,當然也可以使用Sequential 方式以及子類化方式去定義模型。示例代碼如下所示:

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np

# Train and Test data from numpy array.
x_train, y_train = (
np.random.random((60000, 784)),
np.random.randint(10, size=(60000, 1)),
)
x_test, y_test = (
np.random.random((10000, 784)),
np.random.randint(10, size=(10000, 1)),
)

# Reserve 10,000 samples for validation.
x_val = x_train[-10000:]
y_val = y_train[-10000:]
x_train = x_train[:-10000]
y_train = y_train[:-10000]

# Model Create
inputs = keras.Input(shape=(784, ), name="digits")
x = layers.Dense(64, activation='relu', name="dense_1")(inputs)
x = layers.Dense(64, activation='relu', name="dense_2")(x)
outputs = layers.Dense(10, name="predictions")(x)
model = keras.Model(inputs=inputs, outputs=outputs)

# Model Compile.
model.compile(
# Optimizer
optimizer=keras.optimizers.RMSprop(),
# Loss function to minimize
loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
# List of metrics to monitor
metrics=['sparse_categorical_accuracy'],
)

# Model Training.
print('# Fit model on training data')
history = model.fit(
x_train,
y_train,
batch_size=64,
epochs=3,
# We pass some validation for monitoring validation loss and metrics
# at the end of each epoch
validation_data=(x_val, y_val),
)
print('nhistory dict:', history.history)

# Model Evaluate.
print('n# Evaluate on test data')
results = model.evaluate(x_test, y_test, batch_size=128)
print('test loss, test acc:', results)

# Generate predictions (probabilities -- the output of the last layer)
# Model Predict.
print('n# Generate predictions for 3 samples')
predictions = model.predict(x_test[:3])
print('predictions shape:', predictions.shape)

從代碼中可以看到,要完成模型的訓練與評估的整體流程,首先要構建好模型;然後要對模型進行編譯(compile),目的是指定模型訓練過程中需要用到的優化器(optimizer) ,損失函數(losses) 以及評估指標(metrics) ;接著開始進行模型的訓練與交叉驗證(fit),此步驟需要提前指定好訓練數據和驗證數據,並設置好一些參數如epochs 等才能繼續,交叉驗證操作會在每輪(epoch) 訓練結束後自動觸發;最後是模型評估(evaluate) 與預測(predict),我們會根據評估與預測結果來判斷模型的好壞。這樣一個完整的模型訓練與評估流程就完成了,下面來對示例裡的一些實現細節進行展開講解。

模型編譯(compile)

在模型訓練之前首先要進行模型編譯,因為只有知道了要優化什麼目標,如何進行優化以及要關注什麼指標,模型才能被正確的訓練與調整。 compile 方法包含三個主要參數,一個是待優化的損失(loss) ,它指明了要優化的目標,一個是優化器(optimizer),它指明了目標優化的方向,還有一個可選的指標項(metrics),它指明了訓練過程中要關注的模型指標。 Keras API 中已經包含了許多內置的損失函數,優化器以及指標,可以拿來即用,能夠滿足大多數的訓練需要。

損失函數類主要在tf.keras.losses 模塊下,其中包含了多種預定義的損失,比如我們常用的二分類損失BinaryCrossentropy ,多分類損失CategoricalCrossentropy 以及均方根損失MeanSquaredError 等。傳遞給compile 的參數既可以是一個字符串如binary_crossentropy 也可以是對應的losses 實例如tf.keras.losses.BinaryCrossentropy() ,當我們需要設置損失函數的一些參數時(比如上例中from_logits=True) ,則需要使用實例參數。

優化器類主要在tf.keras.optimizers 模塊下,一些常用的優化器如SGD , Adam 以及RMSprop 等均包含在內。同樣它也可以通過字符串或者實例的方式傳給compile 方法,一般我們需要設置的優化器參數主要為學習率(learning rate) ,其他的參數可以參考每個優化器的具體實現來動態設置,或者直接使用其默認值即可。

指標類主要在tf.keras.metrics 模塊下,二分類裡常用的AUC 指標以及lookalike 裡常用的召回率(Recall) 指標等均有包含。同理,它也可以以字符串或者實例的形式傳遞給compile 方法,注意compile 方法接收的是一個metric 列表,所以可以傳遞多個指標信息。

當然如果losses 模塊下的損失或metrics 模塊下的指標不滿足你的需求,也可以自定義它們的實現。

5.1. 對於自定義損失,有兩種方式,一種是定義一個損失函數,它接收兩個輸入參數y_true 和y_pred ,然後在函數內部計算損失並返回。代碼如下:

def basic_loss_function(y_true, y_pred):
return tf.math.reduce_mean(tf.abs(y_true - y_pred))

model.compile(optimizer=keras.optimizers.Adam(), loss=basic_loss_function)

5.2. 如果你需要的損失函數不僅僅包含上述兩個參數,則可以採用另外一種子類化的方式來實現。定義一個類繼承自tf.keras.losses.Loss 類,並實現其__init__(self) 和call(self, ytrue, ypred) 方法,這種實現方式與子類化層和模型比較相似。比如要實現一個加權的二分類交叉熵損失,其代碼如下:

class WeightedBinaryCrossEntropy(keras.losses.Loss):
"""
Args:
pos_weight: Scalar to affect the positive labels of the loss function.
weight: Scalar to affect the entirety of the loss function.
from_logits: Whether to compute loss from logits or the probability.
reduction: Type of tf.keras.losses.Reduction to apply to loss.
name: Name of the loss function.
"""
def __init__(self,
pos_weight,
weight,
from_logits=False,
reduction=keras.losses.Reduction.AUTO,
name="weighted_binary_crossentropy"):
super().__init__(reduction=reduction, name=name)
self.pos_weight = pos_weight
self.weight = weight
self.from_logits = from_logits

def call(self, y_true, y_pred):
ce = tf.losses.binary_crossentropy(
y_true,
y_pred,
from_logits=self.from_logits,
)[:, None]
ce = self.weight * (ce * (1 - y_true) + self.pos_weight * ce * y_true)
return ce

model.compile(
optimizer=keras.optimizers.Adam(),
loss=WeightedBinaryCrossEntropy(
pos_weight=0.5,
weight=2,
from_logits=True,
),
)

5.3. 對於自定義指標,也可以通過子類化的方式來實現,首先定義一個指標類繼承自tf.keras.metrics.Metric 類並實現其四個方法,分別是__init__(self) 方法,用來創建狀態變量, update_state(self, y_true, y_pred, sample_weight=None) 方法,用來更新狀態變量, result(self) 方法,用來返回狀態變量的最終結果, 以及reset_states(self) 方法,用來重新初始化狀態變量。比如要實現一個多分類中真正例(True Positives) 數量的統計指標,其代碼如下:

class CategoricalTruePositives(keras.metrics.Metric):
def __init__(self, name="categorical_true_positives", **kwargs):
super().__init__(name=name, **kwargs)
self.true_positives = self.add_weight(name="tp", initializer="zeros")

def update_state(self, y_true, y_pred, sample_weight=None):
y_pred = tf.reshape(tf.argmax(y_pred, axis=1), shape=(-1, 1))
values = tf.cast(y_true, 'int32') == tf.cast(y_pred, 'int32')
values = tf.cast(values, 'float32')
if sample_weight is not None:
sample_weight = tf.cast(sample_weight, 'float32')
values = tf.multiply(values, sample_weight)
self.true_positives.assign_add(tf.reduce_sum(values))

def result(self):
return self.true_positives

def reset_states(self):
# The state of the metric will be reset at the start of each epoch.
self.true_positives.assign(0.)

model.compile(
optimizer=keras.optimizers.RMSprop(learning_rate=1e-3),
loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=[CategoricalTruePositives()],
)

5.4. 對於一些在層(layers) 內部定義的損失,可以通過在自定義層的call 方法裡調用self.add_loss() 來實現,而且在模型訓練時,它會自動添加到整體的損失中,不用人為乾預。通過對比加入自定義損失前後模型訓練輸出的loss 值的變化來確認這部分損失是否被加入到了整體的損失中。還可以在build 模型後,打印model.losses 來查看該模型的所有損失。注意正則化損失是內置在Keras 的所有層中的,只需要在調用層時加入相應正則化參數即可,無需在call 方法中add_loss()。

5.5. 對於指標信息來說,可以在自定義層的call 方法裡調用self.add_metric() 來新增指標,同樣的,它也會自動出現在整體的指標中,無需人為乾預。

5.6. 函數式API 實現的模型,可以通過調用model.add_loss() 和model.add_metric() 來實現與自定義模型同樣的效果。示例代碼如下:

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

inputs = keras.Input(shape=(784, ), name="digits")
x1 = layers.Dense(64, activation='relu', name="dense_1")(inputs)
x2 = layers.Dense(64, activation='relu', name="dense_2")(x1)
outputs = layers.Dense(10, name="predictions")(x2)
model = keras.Model(inputs=inputs, outputs=outputs)

model.add_loss(tf.reduce_sum(x1) * 0.1)

model.add_metric(
keras.backend.std(x1),
name="std_of_activation",
aggregation='mean',
)

model.compile(
optimizer=keras.optimizers.RMSprop(1e-3),
loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
)
model.fit(x_train, y_train, batch_size=64, epochs=1)

如果要編譯的是多輸入多輸出模型,則可以為每一個輸出指定不同的損失函數以及不同的指標,後面會詳細介紹。

模型訓練與驗證(fit)

模型的訓練通過調用model.fit() 方法來實現, fit 方法包括訓練數據與驗證數據參數,它們可以是numpy 類型數據,也可以是tf.data 模塊下dataset 類型的數據。另外fit 方法還包括epochs , batch_size 以及stepsperepoch 等控制訓練流程的參數,並且還可以通過callbacks 參數控制模型在訓練過程中執行一些其它的操作,如Tensorboard 日誌記錄等。

模型的訓練和驗證數據可以是numpy 類型數據,最開始的端到端示例即是採用numpy 數組作為輸入。一般在數據量較小且內存能容下的情況下採用numpy 數據作為訓練和評估的數據輸入。

2.1. 對於numpy 類型數據來說,如果指定了epochs 參數,則訓練數據的總量為原始樣本數量* epochs。

2.2. 默認情況下一輪訓練(epoch) 所有的原始樣本都會被訓練一遍,下一輪訓練還會使用這些樣本數據進行訓練,每一輪執行的步數(steps) 為原始樣本數量/batch_size ,如果batch_size 不指定,默認為32 。交叉驗證在每一輪訓練結束後觸發,並且也會在所有驗證樣本上執行一遍,可以指定validationbatchsize 來控制驗證數據的batch 大小,如果不指定默認同batch_size。

2.3. 對於numpy 類型數據來說,如果設置了steps_per_epoch 參數,表示一輪要訓練指定的步數,下一輪會在上輪基礎上使用下一個batch 的數據繼續進行訓練,直到所有的epochs 結束或者訓練數據的總量被耗盡。要想訓練流程不因數據耗盡而結束,則需要保證數據的總量要大於stepsperepoch epochs batch_size。同理也可以設置validation_steps ,表示交叉驗證所需步數,此時要注意驗證集的數據總量要大於validationsteps * validationbatch_size。

2.4. fit 方法還提供了另外一個參數validation_split 來自動從訓練數據集中保留一定比例的數據作為驗證,該參數取值為0-1 之間,比如0.2 代表20% 的訓練集用來做驗證, fit方法會默認保留numpy 數組最後面20% 的樣本作為驗證集。

TensorFlow 2.0 之後,更為推薦的是使用tf.data 模塊下dataset 類型的數據作為訓練和驗證的數據輸入,它能以更加快速以及可擴展的方式加載和預處理數據。

3.1. 使用dataset 進行訓練的代碼如下:

train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
# Shuffle and slice the dataset.
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

# Prepare the validation dataset
val_dataset = tf.data.Dataset.from_tensor_slices((x_val, y_val))
val_dataset = val_dataset.batch(64)

# Now we get a test dataset.
test_dataset = tf.data.Dataset.from_tensor_slices((x_test, y_test))
test_dataset = test_dataset.batch(64)

# Since the dataset already takes care of batching,
# we don't pass a `batch_size` argument.
model.fit(train_dataset, epochs=3, validation_data=val_dataset)
result = model.evaluate(test_dataset)

3.2. dataset 一般是一個二元組,第一個元素為模型的輸入特徵,如果為多輸入就是多個特徵的字典(dict) 或元組(tuple),第二個元素是真實的數據標籤( label) ,即ground truth。

3.3. 使用from_tensor_slices 方法可以從nunpy 數組直接生成dataset 類型數據,是一種比較方便快捷的生成方式,一般在測試時使用。其它較為常用的生成方式,比如從TFRecord 文件或文本文件(TextLine) 中生成dataset ,可以參考tf.data 模塊下的相關類的具體實現。

3.4. dataset 可以調用內置方法提前對數據進行預處理,比如數據打亂(shuffle), batch 以及repeat 等操作。 shuffle 操作是為了減小模型過擬合的機率,它僅為小範圍打亂,需要藉助於一個緩存區,先將數據填滿,然後在每次訓練時從緩存區裡隨機抽取batch_size 條數據,產生的空缺用後面的數據填充,從而實現了局部打亂的效果。 batch 是對數據進行分批次,常用於控制和調節模型的訓練速度以及訓練效果,因為在dataset 中已經batch 過,所以fit 方法中的batch_size 就無需再提供了。 repeat 用來對數據進行複制,以解決數據量不足的問題,如果指定了其參數count,則表示整個數據集要復制count 次,不指定就會無限次復制,此時必須要設置stepsperepoch 參數,不然訓練無法終止。

3.5. 上述例子中, train dataset 的全部數據在每一輪都會被訓練到,因為一輪訓練結束後, dataset 會重置,然後被用來重新訓練。但是當指定了steps_per_epoch 之後, dataset 在每輪訓練後不會被重置,一直到所有epochs 結束或所有的訓練數據被消耗完之後終止,要想訓練正常結束,須保證提供的訓練數據總量要大於steps_per_epoch epochs batch_size。同理也可以指定validation_steps ,此時數據驗證會執行指定的步數,在下次驗證開始時, validation dataset 會被重置,以保證每次交叉驗證使用的都是相同的數據。 validation_split 參數不適用於dataset 類型數據,因為它需要知道每個數據樣本的索引,這在dataset API 下很難實現。

3.6. 當不指定steps_per_epoch 參數時, numpy 類型數據與dataset 類型數據的處理流程完全一致。但當指定之後,要注意它們之間在處理上的差異。對於numpy 類型數據而言,在處理時,它會被轉為dataset 類型數據,只不過這個dataset 被repeat 了epochs 次,而且每輪訓練結束後,這個dataset 不會被重置,會在上次的batch 之後繼續訓練。假設原始數據量為n ,指定stepsperepoch 參數之後,兩者的差異主要體現在真實的訓練數據量上, numpy 為n * epochs , dataset 為n。具體細節可以參考源碼實現。

3.7. dataset 還有map 與prefetch 方法比較實用。 map 方法接收一個函數作為參數,用來對dataset 中的每一條數據進行處理並返回一個新的dataset ,比如我們在使用TextLineDataset 讀取文本文件後生成了一個dataset ,而我們要抽取輸入數據中的某些列作為特徵(features),某些列作為標籤(labels),此時就會用到map 方法。 prefetch 方法預先從dataset 中準備好下次訓練所需的數據並放於內存中,這樣可以減少每輪訓練之間的延遲等待時間。

除了訓練數據和驗證數據外,還可以向fit 方法傳遞樣本權重(sample_weight) 以及類別權重(class_weight) 參數。這兩個參數通常被用於處理分類不平衡問題,通過給類別少的樣本賦予更高的權重,使得各個類別對整體損失的貢獻趨於一致。

4.1. 對於numpy 類型的輸入數據,可以使用上述兩個參數,以上面的多分類問題為例,如果要給分類5 一個更高的權重,可以使用如下代碼來實現:

import numpy as np

# Here's the same example using `class_weight`
class_weight = {0: 1., 1: 1., 2: 1., 3: 1., 4: 1.,
# Set weight "2" for class "5",
# making this class 2x more important
5: 2.,
6: 1., 7: 1., 8: 1., 9: 1.}
print('Fit with class weight')
model.fit(x_train, y_train, class_weight=class_weight, batch_size=64, epochs=4)

# Here's the same example using `sample_weight` instead:
sample_weight = np.ones(shape=(len(y_train), ))
sample_weight[y_train == 5] = 2.
print('nFit with sample weight')

model.fit(
x_train,
y_train,
sample_weight=sample_weight,
batch_size=64,
epochs=4,
)

4.2. 而對於dataset 類型的輸入數據來說,不能直接使用上述兩個參數,需要在構建dataset 時將sample_weight 加入其中,返回一個三元組的dataset ,格式為(input_batch, target_batch, sampleweightbatch) 。示例代碼如下所示:

sample_weight = np.ones(shape=(len(y_train), ))
sample_weight[y_train == 5] = 2.

# Create a Dataset that includes sample weights
# (3rd element in the return tuple).
train_dataset = tf.data.Dataset.from_tensor_slices((
x_train,
y_train,
sample_weight,
))

# Shuffle and slice the dataset.
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

model.fit(train_dataset, epochs=3)

在模型的訓練過程中有一些特殊時間點,比如在一個batch 結束或者一個epoch 結束時,一般都會做一些額外的處理操作來輔助我們進行訓練,上面介紹過的模型交叉驗證就是其中之一。還有一些其它的操作,比如當模型訓練停滯不前時(loss 值在某一值附近不斷波動),自動減小其學習速率(learning rate) 以使損失繼續下降,從而得到更好的收斂效果;在訓練過程中保存模型的權重信息,以備重啟模型時可以在已有權重的基礎上繼續訓練,從而減少訓練時間;還有在每輪的訓練結束後記錄模型的損失(loss) 和指標(metrics) 信息,以供Tensorboard 分析使用等等,這些操作都是模型訓練過程中不可或缺的部分。它們都可以通過回調函數(callbacks) 的方式來實現,這些回調函數都在tf.keras.callbacks 模塊下,可以將它們作為列表參數傳遞給fit 方法以達到不同的操作目的。

5.1. 下面以EarlyStopping 為例說明callbacks 的使用方式。本例中,當交叉驗證損失val_loss 至少在2 輪(epochs) 訓練中的減少值都低於1e-2 時,我們會提前停止訓練。其示例代碼如下所示:

callbacks = [
keras.callbacks.EarlyStopping(
# Stop training when `val_loss` is no longer improving
monitor="val_loss",
# "no longer improving" being defined as "no better than 1e-2 less"
min_delta=1e-2,
# "no longer improving" being further defined as "for at least 2 epochs"
patience=2,
verbose=1,
)
]

model.fit(
x_train,
y_train,
epochs=20,
batch_size=64,
callbacks=callbacks,
validation_split=0.2,
)

5.2. 一些比較常用的callbacks 需要了解並掌握, 如ModelCheckpoint 用來保存模型權重信息, TensorBoard 用來記錄一些指標信息, ReduceLROnPlateau 用來在模型停滯時減小學習率。更多的callbacks 函數可以參考tf.keras.callbacks 模塊下的實現。

5.3. 當然也可以自定義callbacks 類,該子類需要繼承自tf.keras.callbacks.Callback 類,並按需實現其內置的方法,比如如果需要在每個batch 訓練結束後記錄loss 的值,則可以使用如下代碼實現:

class LossHistory(keras.callbacks.Callback):
def on_train_begin(self, logs):
self.losses = []

def on_batch_end(self, batch, logs):
self.losses.append(logs.get('loss'))

5.4. 在TensorFlow 2.0 之前, ModelCheckpoint 內容和TensorBoard 內容是同時記錄的,保存在相同的文件夾下,而在2.0 之後的keras API 中它們可以通過不同的回調函數分開指定。記錄的日誌文件中,含有checkpoint 關鍵字的文件一般為檢查點文件,含有events.out.tfevents 關鍵字的文件一般為Tensorboard 相關文件。

多輸入輸出模型

TensorFlow 篇| TensorFlow 2.x 基於Keras 模型的本地訓練與評估 2

考慮如圖所示的多輸入多輸出模型,該模型包括兩個輸入和兩個輸出, score_output 輸出表示分值, class_output 輸出表示分類,其示例代碼如下:

from tensorflow import keras
from tensorflow.keras import layers

image_input = keras.Input(shape=(32, 32, 3), name="img_input")
timeseries_input = keras.Input(shape=(None, 10), name="ts_input")

x1 = layers.Conv2D(3, 3)(image_input)
x1 = layers.GlobalMaxPooling2D()(x1)

x2 = layers.Conv1D(3, 3)(timeseries_input)
x2 = layers.GlobalMaxPooling1D()(x2)

x = layers.concatenate([x1, x2])

score_output = layers.Dense(1, name="score_output")(x)
class_output = layers.Dense(5, name="class_output")(x)

model = keras.Model(
inputs=[image_input, timeseries_input],
outputs=[score_output, class_output],
)

在進行模型編譯時,如果只指定一個loss 明顯不能滿足不同輸出的損失計算方式,所以此時可以指定loss 為一個列表(list),其中每個元素分別對應於不同的輸出。示例代碼如下:

model.compile(
optimizer=keras.optimizers.RMSprop(1e-3),
loss=[
keras.losses.MeanSquaredError(),
keras.losses.CategoricalCrossentropy(from_logits=True)
],
loss_weights=[1, 1],
)

此時模型的優化目標為所有單個損失值的總和,如果想要為不同的損失指定不同的權重,可以設置loss_weights 參數,該參數接收一個標量係數列表(list),用以對模型不同輸出的損失值進行加權。如果僅為模型指定一個loss ,則該loss 會應用到每一個輸出,在模型的多個輸出損失計算方式相同時,可以採用這種方式。

同樣的對於模型的指標(metrics),也可以指定為多個,注意因為metrics 參數本身即為一個列表,所以為多個輸出指定metrics 應該使用二維列表。示例代碼如下:

model.compile(
optimizer=keras.optimizers.RMSprop(1e-3),
loss=[
keras.losses.MeanSquaredError(),
keras.losses.CategoricalCrossentropy(from_logits=True),
],
metrics=[
[
keras.metrics.MeanAbsolutePercentageError(),
keras.metrics.MeanAbsoluteError()
],
[keras.metrics.CategoricalAccuracy()],
],
)

對於有明確名稱的輸出,可以通過字典的方式來設置其loss 和metrics。示例代碼如下:

model.compile(
optimizer=keras.optimizers.RMSprop(1e-3),
loss={
'score_output': keras.losses.MeanSquaredError(),
'class_output': keras.losses.CategoricalCrossentropy(from_logits=True),
},
metrics={
'score_output': [
keras.metrics.MeanAbsolutePercentageError(),
keras.metrics.MeanAbsoluteError()
],
'class_output': [
keras.metrics.CategoricalAccuracy(),
]
},
)

對於僅被用來預測的輸出,也可以不指定其loss。示例代碼如下:

model.compile(
optimizer=keras.optimizers.RMSprop(1e-3),
loss=[
None,
keras.losses.CategoricalCrossentropy(from_logits=True),
],
)

# Or dict loss version
model.compile(
optimizer=keras.optimizers.RMSprop(1e-3),
loss={
'class_output': keras.losses.CategoricalCrossentropy(from_logits=True),
},
)

對於多輸入輸出模型的訓練來說,也可以採用和其compile 方法相同的方式來提供數據輸入,也就是說既可以使用列表的方式,也可以使用字典的方式來指定多個輸入。

6.1. numpy 類型數據示例代碼如下:

# Generate dummy Numpy data
img_data = np.random.random_sample(size=(100, 32, 32, 3))
ts_data = np.random.random_sample(size=(100, 20, 10))
score_targets = np.random.random_sample(size=(100, 1))
class_targets = np.random.random_sample(size=(100, 5))

# Fit on lists
model.fit(
x=[img_data, ts_data],
y=[score_targets, class_targets],
batch_size=32,
epochs=3,
)

# Alternatively, fit on dicts
model.fit(
x={
'img_input': img_data,
'ts_input': ts_data,
},
y={
'score_output': score_targets,
'class_output': class_targets,
},
batch_size=32,
epochs=3,
)

6.2. dataset 類型數據示例代碼如下:

# Generate dummy dataset data from numpy
train_dataset = tf.data.Dataset.from_tensor_slices((
(img_data, ts_data),
(score_targets, class_targets),
))

# Alternatively generate with dict
train_dataset = tf.data.Dataset.from_tensor_slices((
{
'img_input': img_data,
'ts_input': ts_data,
},
{
'score_output': score_targets,
'class_output': class_targets,
},
))
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

model.fit(train_dataset, epochs=3)

自定義訓練流程

如果你不想使用model 內置提供的fit 和evaluate 方法,而想使用低階API 自定義模型的訓練和評估的流程,則可以藉助於GradientTape 來實現。深度神經網絡在後向傳播過程中需要計算損失(loss) 關於權重矩陣的導數(也稱為梯度),以更新權重矩陣並獲得最優解,而GradientTape 能自動提供求導幫助,無需我們手動求導,它本質上是一個求導記錄器,能夠記錄前項傳播的過程,並據此計算導數。

模型的構建過程與之前相比沒有什麼不同,主要體現在訓練的部分,示例代碼如下:

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np

# Get the model.
inputs = keras.Input(shape=(784, ), name="digits")
x = layers.Dense(64, activation='relu', name="dense_1")(inputs)
x = layers.Dense(64, activation='relu', name="dense_2")(x)
outputs = layers.Dense(10, name="predictions")(x)
model = keras.Model(inputs=inputs, outputs=outputs)

# Instantiate an optimizer.
optimizer = keras.optimizers.SGD(learning_rate=1e-3)
# Instantiate a loss function.
loss_fn = keras.losses.SparseCategoricalCrossentropy(from_logits=True)

# Prepare the metrics.
train_acc_metric = keras.metrics.SparseCategoricalAccuracy()
val_acc_metric = keras.metrics.SparseCategoricalAccuracy()

# Prepare the training dataset.
batch_size = 64
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(batch_size)

# Prepare the validation dataset.
val_dataset = tf.data.Dataset.from_tensor_slices((x_val, y_val))
val_dataset = val_dataset.batch(64)

epochs = 3
for epoch in range(epochs):
print('Start of epoch %d' % (epoch, ))

# Iterate over the batches of the dataset.
for step, (x_batch_train, y_batch_train) in enumerate(train_dataset):

# Open a GradientTape to record the operations run
# during the forward pass, which enables autodifferentiation.
with tf.GradientTape() as tape:

# Run the forward pass of the layer.
# The operations that the layer applies
# to its inputs are going to be recorded
# on the GradientTape.
logits = model(x_batch_train,
training=True) # Logits for this minibatch

# Compute the loss value for this minibatch.
loss_value = loss_fn(y_batch_train, logits)

# Use the gradient tape to automatically retrieve
# the gradients of the trainable variables with respect to the loss.
grads = tape.gradient(loss_value, model.trainable_weights)

# Run one step of gradient descent by updating
# the value of the variables to minimize the loss.
optimizer.apply_gradients(zip(grads, model.trainable_weights))

# Update training metric.
train_acc_metric(y_batch_train, logits)

# Log every 200 batches.
if step % 200 == 0:
print('Training loss (for one batch) at step %s: %s' %
(step, float(loss_value)))
print('Seen so far: %s samples' % ((step + 1) * 64))

# Display metrics at the end of each epoch.
train_acc = train_acc_metric.result()
print('Training acc over epoch: %s' % (float(train_acc), ))
# Reset training metrics at the end of each epoch
train_acc_metric.reset_states()

# Run a validation loop at the end of each epoch.
for x_batch_val, y_batch_val in val_dataset:
val_logits = model(x_batch_val)
# Update val metrics
val_acc_metric(y_batch_val, val_logits)
val_acc = val_acc_metric.result()
val_acc_metric.reset_states()
print('Validation acc: %s' % (float(val_acc), ))

注意with tf.GradientTape() as tape 部分的實現,它記錄了前向傳播的過程,然後使用tape.gradient 方法計算出loss 關於模型所有權重矩陣(model.trainable_weights) 的導數(也稱作梯度),接著利用優化器(optimizer) 去更新所有的權重矩陣。

在上述訓練流程中,模型的訓練指標在每個batch 的訓練中進行更新操作(update_state()) ,在一個epoch 訓練結束後打印指標的結果(result()) ,然後重置該指標(reset_states( )) 並進行下一輪的指標記錄,交叉驗證的指標也是同樣的操作。

注意與使用模型內置API 進行訓練不同,在自定義訓練中,模型中定義的損失,比如正則化損失以及通過add_loss 添加的損失,是不會自動累加在loss_fn 之內的。如果要包含這部分損失,則需要修改自定義訓練的流程,通過調用model.losses 來將模型的全部損失加入到要優化的損失中去。示例代碼如下所示:

with tf.GradientTape() as tape:
logits = model(x_batch_train)
loss_value = loss_fn(y_batch_train, logits)

# Add extra losses created during this forward pass:
loss_value += sum(model.losses)
grads = tape.gradient(loss_value, model.trainable_weights)
optimizer.apply_gradients(zip(grads, model.trainable_weights))

參考資料

Keras 模型訓練與評估Keras 模型fit 方法tf.data.Dataset