How can I create a custom callback in Keras?

user_10331791 0
Viewed 9367

I'm interested in creating a callback while fitting my keras model. More in detail I'd like to receive a message from a bot telegram with val_acc each time an epoch is over. I know you can add a callback_list as a parameter in classifier.fit() but many callbacks are prebuilt by keras and I don't know how to add a custom one.

Thank you!

keras tensorflow neural-network callback python
3 Answers

Here is an example of how I would add validation accuracy to a callback:

class AccuracyHistory(keras.callbacks.Callback):
    def on_train_begin(self, logs={}):
        self.acc = []

    def on_epoch_end(self, batch, logs={}):
        self.acc.append(logs.get('val_acc'))

history = AccuracyHistory()

model.fit(x, y,
          ...
          callbacks=[history])
user_22768310

As example, I provide my custom callback with F1 metric. It calculates F1 at the end of each epoch not batch-wise but for ALL the train data passed (and optionally also validation). It can be easily customized with every other metric

class F1History(tf.keras.callbacks.Callback):

    def __init__(self, train, validation=None):
        super(F1History, self).__init__()
        self.validation = validation
        self.train = train

    def on_epoch_end(self, epoch, logs={}):

        logs['F1_score_train'] = float('-inf')
        X_train, y_train = self.train[0], self.train[1]
        y_pred = (self.model.predict(X_train).ravel()>0.5)+0
        score = f1_score(y_train, y_pred)       

        if (self.validation):
            logs['F1_score_val'] = float('-inf')
            X_valid, y_valid = self.validation[0], self.validation[1]
            y_val_pred = (self.model.predict(X_valid).ravel()>0.5)+0
            val_score = f1_score(y_valid, y_val_pred)
            logs['F1_score_train'] = np.round(score, 5)
            logs['F1_score_val'] = np.round(val_score, 5)
        else:
            logs['F1_score_train'] = np.round(score, 5)

for fitting:

es = EarlyStopping(patience=3, verbose=1, min_delta=0.001, monitor='F1_score_val', mode='max', restore_best_weights=True)
model.fit(x_train,y_train, epochs=10, 
          callbacks=[F1History(train=(x_train,y_train),validation=(x_val,y_val)),es])
user_103750490

You can perform any action while the model is training. for this purpose keras has provided some methods as following:

on_train_begin, on_train_end, on_epoch_begin, on_epoch_end, on_test_begin,
on_test_end, on_predict_begin, on_predict_end, on_train_batch_begin, on_train_batch_end,
on_test_batch_begin, on_test_batch_end, on_predict_batch_begin,on_predict_batch_end

Example:

If you want to make predictions on the test data at the end of every epoch using the model which is being trained you can do it using the following code

class CustomCallback(keras.callbacks.Callback):
    def __init__(self, model, x_test, y_test):
        self.model = model
        self.x_test = x_test
        self.y_test = y_test

    def on_epoch_end(self, epoch, logs={}):
        y_pred = self.model.predict(self.x_test, self.y_test)
        print('y predicted: ', y_pred)

You need mention the callback during the model.fit

model.sequence()
# your model architecture
model.fit(x_train, y_train, epochs=10, 
          callbacks=[CustomCallback(model, x_test, y_test)])
user_110062840
Related