回調函數是一組在訓練的特定階段被調用的函數集，你可以使用回調函數來觀察訓練過程中網絡內部的狀態和統計信息。通過傳遞回調函數列表到模型的.fit()中，即可在給定的訓練階段調用該函數集中的函數。

【Tips】雖然我們稱之為回調“函數”，但事實上Keras的回調函數是一個類，回調函數只是習慣性稱呼

Callback

例子No.1：官網示例?
保存訓練過程中最好的模型

from __future__ import print_function
import numpy as np
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers.core import Dense, Activation
from keras.optimizers import SGD
from keras.utils import np_utils
from keras.callbacks import ModelCheckpoint
from keras.callbacks import TensorBoard
import matplotlib.pyplot as plt
?
np.random.seed(1671) ?# for reproducibility
?
# network and training
NB_EPOCH = 20
BATCH_SIZE = 128
VERBOSE = 1
NB_CLASSES = 10 ? # number of outputs = number of digits
OPTIMIZER = SGD() # optimizer, explained later in this chapter
N_HIDDEN = 128
VALIDATION_SPLIT=0.2 # how much TRAIN is reserved for VALIDATION
?
# data: shuffled and split between train and test sets
(X_train, y_train), (X_test, y_test) = mnist.load_data()
?
#X_train is 60000 rows of 28x28 values --> reshaped in 60000 x 784
RESHAPED = 784
#
X_train = X_train.reshape(60000, RESHAPED)
X_test = X_test.reshape(10000, RESHAPED)
X_train = X_train.astype('float32')
X_test = X_test.astype('float32')
?
# normalize
X_train /= 255
X_test /= 255
print(X_train.shape[0], 'train samples')
print(X_test.shape[0], 'test samples')
?
# convert class vectors to binary class matrices
Y_train = np_utils.to_categorical(y_train, NB_CLASSES)
Y_test = np_utils.to_categorical(y_test, NB_CLASSES)
?
# M_HIDDEN hidden layers
# 10 outputs
# final stage is softmax
?
model = Sequential()
model.add(Dense(N_HIDDEN, input_shape=(RESHAPED,)))
model.add(Activation('relu'))
model.add(Dense(N_HIDDEN))
model.add(Activation('relu'))
model.add(Dense(NB_CLASSES))
model.add(Activation('softmax'))
model.summary()
?
#model.load_weights("weight_bst.hdf5")
?
model.compile(loss='categorical_crossentropy',
? ? ? ? ? ? ? optimizer=OPTIMIZER,
? ? ? ? ? ? ? metrics=['accuracy'])
?
#保存.fit()過程中最好的模型
checkpoint=ModelCheckpoint("/weights.hdf5",verbose=1,save_best_only=True)
?
history = model.fit(X_train, Y_train,
? ? ? ? ? ? ? ? ? ? batch_size=BATCH_SIZE, epochs=5,
? ? ? ? ? ? ? ? ? ? verbose=VERBOSE, validation_split=VALIDATION_SPLIT,
? ? ? ? ? ? ? ? ? ? callbacks=[checkpoint])
?

例子No.2：保存每一個改進CheckPoint模型
應用Checkpoint時，應在每次訓練中觀察到改進時輸出模型權重。

Checkpoint設置成當驗證數據集的分類精度提高時保存網絡權重(monitor=’val_acc’ and mode=’max’)。權重存儲在一個包含評價的文件中(weights-improvement – { val_acc = .2f } .hdf5)。

from __future__ import print_function
import numpy as np
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers.core import Dense, Activation
from keras.optimizers import SGD
from keras.utils import np_utils
from keras.callbacks import ModelCheckpoint
from keras.callbacks import TensorBoard
import matplotlib.pyplot as plt
?
np.random.seed(1671) ?# for reproducibility
?
# network and training
NB_EPOCH = 20
BATCH_SIZE = 128
VERBOSE = 1
NB_CLASSES = 10 ? # number of outputs = number of digits
OPTIMIZER = SGD() # optimizer, explained later in this chapter
N_HIDDEN = 128
VALIDATION_SPLIT=0.2 # how much TRAIN is reserved for VALIDATION
?
# data: shuffled and split between train and test sets
(X_train, y_train), (X_test, y_test) = mnist.load_data()
?
#X_train is 60000 rows of 28x28 values --> reshaped in 60000 x 784
RESHAPED = 784
#
X_train = X_train.reshape(60000, RESHAPED)
X_test = X_test.reshape(10000, RESHAPED)
X_train = X_train.astype('float32')
X_test = X_test.astype('float32')
?
# normalize
X_train /= 255
X_test /= 255
print(X_train.shape[0], 'train samples')
print(X_test.shape[0], 'test samples')
?
# convert class vectors to binary class matrices
Y_train = np_utils.to_categorical(y_train, NB_CLASSES)
Y_test = np_utils.to_categorical(y_test, NB_CLASSES)
?
# M_HIDDEN hidden layers
# 10 outputs
# final stage is softmax
?
model = Sequential()
model.add(Dense(N_HIDDEN, input_shape=(RESHAPED,)))
model.add(Activation('relu'))
model.add(Dense(N_HIDDEN))
model.add(Activation('relu'))
model.add(Dense(NB_CLASSES))
model.add(Activation('softmax'))
model.summary()
?
#model.load_weights("weight_bst.hdf5")
?
model.compile(loss='categorical_crossentropy',
? ? ? ? ? ? ? optimizer=OPTIMIZER,
? ? ? ? ? ? ? metrics=['accuracy'])
?
#保存.fit()訓練過程中，精確度提高的模型
filepath="weights-improvement-{epoch:02d}-{val_acc:.2f}.hdf5"
checkpoint=ModelCheckpoint(filepath,monitor='val_acc',verbose=1,save_best_only=True,mode='max')
?
history = model.fit(X_train, Y_train,
? ? ? ? ? ? ? ? ? ? batch_size=BATCH_SIZE, epochs=5,
? ? ? ? ? ? ? ? ? ? verbose=VERBOSE, validation_split=VALIDATION_SPLIT,
? ? ? ? ? ? ? ? ? ? callbacks=[checkpoint])
程序運行結果：?

在epoch=5次訓練中，val_acc均提高，所以每次都保存了一個新的模型

?

例子No.3：保存最佳的Checkpoint模型
如果驗證精度提高的話，一個更簡單的Checkpoint策略是將模型權重保存到相同的文件中。

這可以使用上述相同的代碼輕松完成，并將輸出文件名更改為固定（不包括評價或次數的信息）。

在這種情況下，只有當驗證數據集上的模型的分類精度提高到到目前為止最好的時候，才會將模型權重寫入文件“weights.best.hdf5”。

?

from __future__ import print_function
import numpy as np
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers.core import Dense, Activation
from keras.optimizers import SGD
from keras.utils import np_utils
from keras.callbacks import ModelCheckpoint
from keras.callbacks import TensorBoard
import matplotlib.pyplot as plt
?
np.random.seed(1671) ?# for reproducibility
?
# network and training
NB_EPOCH = 20
BATCH_SIZE = 128
VERBOSE = 1
NB_CLASSES = 10 ? # number of outputs = number of digits
OPTIMIZER = SGD() # optimizer, explained later in this chapter
N_HIDDEN = 128
VALIDATION_SPLIT=0.2 # how much TRAIN is reserved for VALIDATION
?
# data: shuffled and split between train and test sets
(X_train, y_train), (X_test, y_test) = mnist.load_data()
?
#X_train is 60000 rows of 28x28 values --> reshaped in 60000 x 784
RESHAPED = 784
#
X_train = X_train.reshape(60000, RESHAPED)
X_test = X_test.reshape(10000, RESHAPED)
X_train = X_train.astype('float32')
X_test = X_test.astype('float32')
?
# normalize
X_train /= 255
X_test /= 255
print(X_train.shape[0], 'train samples')
print(X_test.shape[0], 'test samples')
?
# convert class vectors to binary class matrices
Y_train = np_utils.to_categorical(y_train, NB_CLASSES)
Y_test = np_utils.to_categorical(y_test, NB_CLASSES)
?
# M_HIDDEN hidden layers
# 10 outputs
# final stage is softmax
?
model = Sequential()
model.add(Dense(N_HIDDEN, input_shape=(RESHAPED,)))
model.add(Activation('relu'))
model.add(Dense(N_HIDDEN))
model.add(Activation('relu'))
model.add(Dense(NB_CLASSES))
model.add(Activation('softmax'))
model.summary()
?
#model.load_weights("weight_bst.hdf5")
?
model.compile(loss='categorical_crossentropy',
? ? ? ? ? ? ? optimizer=OPTIMIZER,
? ? ? ? ? ? ? metrics=['accuracy'])
?
#保存訓練過程中最好的模型
filepath="weight_bst.hdf5"
checkpoint=ModelCheckpoint(filepath,monitor='val_acc',verbose=1,save_best_only=True,mode='max')
?
history = model.fit(X_train, Y_train,
? ? ? ? ? ? ? ? ? ? batch_size=BATCH_SIZE, epochs=5,
? ? ? ? ? ? ? ? ? ? verbose=VERBOSE, validation_split=VALIDATION_SPLIT,
? ? ? ? ? ? ? ? ? ? callbacks=[checkpoint)])
'''
score=model.evaluate(X_test,Y_test,verbose=VERBOSE)
print(score)
'''
程序運行結果：

這是一個在你的實驗中需要經常用到的方便的Checkpoint策略。它將確保你的最佳模型被保存，以便稍后使用。它避免了輸入代碼來手動跟蹤，并在訓練時序列化最佳模型。

?

例子No.4:加載之前保存過的Checkpoint模型
Checkpoint只包括模型權重。它假定你了解網絡結構。這也可以序列化成JSON或YAML格式。

在下面的示例中，模型結構是已知的，并且最好的權重從先前的實驗中加載，然后存儲在weights.best.hdf5文件的工作目錄中。

那么將該模型用于對整個數據集進行預測。

from __future__ import print_function
import numpy as np
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers.core import Dense, Activation
from keras.optimizers import SGD
from keras.utils import np_utils
from keras.callbacks import ModelCheckpoint
from keras.callbacks import TensorBoard
import matplotlib.pyplot as plt
?
np.random.seed(1671) ?# for reproducibility
?
# network and training
NB_EPOCH = 20
BATCH_SIZE = 128
VERBOSE = 1
NB_CLASSES = 10 ? # number of outputs = number of digits
OPTIMIZER = SGD() # optimizer, explained later in this chapter
N_HIDDEN = 128
VALIDATION_SPLIT=0.2 # how much TRAIN is reserved for VALIDATION
?
# data: shuffled and split between train and test sets
(X_train, y_train), (X_test, y_test) = mnist.load_data()
?
#X_train is 60000 rows of 28x28 values --> reshaped in 60000 x 784
RESHAPED = 784
#
X_train = X_train.reshape(60000, RESHAPED)
X_test = X_test.reshape(10000, RESHAPED)
X_train = X_train.astype('float32')
X_test = X_test.astype('float32')
?
# normalize
X_train /= 255
X_test /= 255
print(X_train.shape[0], 'train samples')
print(X_test.shape[0], 'test samples')
?
# convert class vectors to binary class matrices
Y_train = np_utils.to_categorical(y_train, NB_CLASSES)
Y_test = np_utils.to_categorical(y_test, NB_CLASSES)
?
# M_HIDDEN hidden layers
# 10 outputs
# final stage is softmax
?
model = Sequential()
model.add(Dense(N_HIDDEN, input_shape=(RESHAPED,)))
model.add(Activation('relu'))
model.add(Dense(N_HIDDEN))
model.add(Activation('relu'))
model.add(Dense(NB_CLASSES))
model.add(Activation('softmax'))
model.summary()
?
#加載模型
model.load_weights("weight_bst.hdf5")
?
model.compile(loss='categorical_crossentropy',
? ? ? ? ? ? ? optimizer=OPTIMIZER,
? ? ? ? ? ? ? metrics=['accuracy'])
?
?
score=model.evaluate(X_test,Y_test,verbose=VERBOSE)
print(score)
?

?例子：LearningRateScheduler調整學習率
from keras.callbacks import LearningRateScheduler
def scheduler(epoch):
? ? if epoch % 100 == 0 and epoch != 0:
? ? ? ? lr = K.get_value(model.optimizer.lr)
? ? ? ? K.set_value(model.optimizer.lr, lr * 0.1)
? ? ? ? print("lr changed to {}".format(lr * 0.1))
? ? return K.get_value(model.optimizer.lr)
?
reduce_lr = LearningRateScheduler(scheduler)
model.fit(train_x, train_y, batch_size=32, epochs=5, callbacks=[reduce_lr])
?

例子：ReduceLROnPlateau自動減少學習率?
from keras.callbacks import ReduceLROnPlateau
reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2,patience=5, min_lr=0.001)
model.fit(X_train, Y_train, callbacks=[reduce_lr])
?

例子：CSVLoggerb保存訓練結果?
csv_logger = CSVLogger('training.log')
model.fit(X_train, Y_train, callbacks=[csv_logger])
?

例子：輕量級自定義回調函數LambdaCallback
# Print the batch number at the beginning of every batch.
batch_print_callback = LambdaCallback(
? ? on_batch_begin=lambda batch,logs: print(batch))
?
# Plot the loss after every epoch.
import numpy as np
import matplotlib.pyplot as plt
plot_loss_callback = LambdaCallback(
? ? on_epoch_end=lambda epoch, logs: plt.plot(np.arange(epoch),
? ? ? ? ? ? ? ? ? ? ? logs['loss']))
?
# Terminate some processes after having finished model training.
processes = ...
cleanup_callback = LambdaCallback(
? ? on_train_end=lambda logs: [
? ? p.terminate() for p in processes if p.is_alive()])
?
model.fit(...,
? ? ? callbacks=[batch_print_callback,
? ? ? ? ?plot_loss_callback,
? ? ? ? ?cleanup_callback])
?

?例子：查看TensorBoard監控
from __future__ import print_function
import numpy as np
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers.core import Dense, Activation
from keras.optimizers import SGD
from keras.utils import np_utils
from keras.callbacks import ModelCheckpoint
from keras.callbacks import TensorBoard
import matplotlib.pyplot as plt
?
np.random.seed(1671) ?# for reproducibility
?
# network and training
NB_EPOCH = 20
BATCH_SIZE = 128
VERBOSE = 1
NB_CLASSES = 10 ? # number of outputs = number of digits
OPTIMIZER = SGD() # optimizer, explained later in this chapter
N_HIDDEN = 128
VALIDATION_SPLIT=0.2 # how much TRAIN is reserved for VALIDATION
?
# data: shuffled and split between train and test sets
(X_train, y_train), (X_test, y_test) = mnist.load_data()
?
#X_train is 60000 rows of 28x28 values --> reshaped in 60000 x 784
RESHAPED = 784
#
X_train = X_train.reshape(60000, RESHAPED)
X_test = X_test.reshape(10000, RESHAPED)
X_train = X_train.astype('float32')
X_test = X_test.astype('float32')
?
# normalize
X_train /= 255
X_test /= 255
print(X_train.shape[0], 'train samples')
print(X_test.shape[0], 'test samples')
?
# convert class vectors to binary class matrices
Y_train = np_utils.to_categorical(y_train, NB_CLASSES)
Y_test = np_utils.to_categorical(y_test, NB_CLASSES)
?
# M_HIDDEN hidden layers
# 10 outputs
# final stage is softmax
?
model = Sequential()
model.add(Dense(N_HIDDEN, input_shape=(RESHAPED,)))
model.add(Activation('relu'))
model.add(Dense(N_HIDDEN))
model.add(Activation('relu'))
model.add(Dense(NB_CLASSES))
model.add(Activation('softmax'))
model.summary()
?
?
model.compile(loss='categorical_crossentropy',
? ? ? ? ? ? ? optimizer=OPTIMIZER,
? ? ? ? ? ? ? metrics=['accuracy'])
?
?
history = model.fit(X_train, Y_train,
? ? ? ? ? ? ? ? ? ? batch_size=BATCH_SIZE, epochs=5,
? ? ? ? ? ? ? ? ? ? verbose=VERBOSE, validation_split=VALIDATION_SPLIT,
? ? ? ? ? ? ? ? ? ? callbacks=[TensorBoard(log_dir="./log")])
?
?
score=model.evaluate(X_test,Y_test,verbose=VERBOSE)
print(score)
?在命令行輸入命令：

運行結果：

?

?

編寫自己的回調函數
我們可以通過繼承keras.callbacks.Callback編寫自己的回調函數，回調函數通過類成員self.model訪問訪問，該成員是模型的一個引用。

這里是一個簡單的保存每個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'))
?

例子：記錄損失函數的歷史數據
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'))
?
model = Sequential()
model.add(Dense(10, input_dim=784, kernel_initializer='uniform'))
model.add(Activation('softmax'))
model.compile(loss='categorical_crossentropy', optimizer='rmsprop')
?
history = LossHistory()
model.fit(X_train, Y_train, batch_size=128, epochs=20, verbose=0, callbacks=[history])
?
print history.losses
# outputs
'''
[0.66047596406559383, 0.3547245744908703, ..., 0.25953155204159617, 0.25901699725311789]
參考：https://blog.csdn.net/weixin_38517705/article/details/82971147
?

總結

以上是生活随笔為你收集整理的Keras中Callback函数的使用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。