custom_callback.ipynb

Kernel: Python 3

Copyright 2020 The TensorFlow Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Escrevendo seus próprios callbacks

Veja em TensorFlow.org

Executar no Google Colab

Ver fonte em GitHub

Baixar notebook

Introdução

Um callback é uma ferramenta poderosa para personalizar o comportamento de um modelo Keras durante o treinamento, avaliação ou inferência. Exemplos incluem tf.keras.callbacks.TensorBoard para visualizar o progresso e os resultados do treinamento com o TensorBoard ou tf.keras.callbacks.ModelCheckpoint para salvar seu modelo periodicamente durante o treinamento.

Neste guia, você aprenderá o que é um callback Keras, o que ele pode fazer e como você pode criar o seu próprio. Fornecemos algumas demonstrações de aplicativos de callback simples para você começar.

Configuração

import tensorflow as tf
from tensorflow import keras

Visão geral dos callbacks do Keras

Todos os callbacks são subclasses da classe keras.callbacks.Callback e substituem um conjunto de métodos chamados em diferentes estágios de treinamento, teste e previsão. Callbacks são úteis para se obter uma visão dos estados internos e estatísticas do modelo durante o treinamento.

Você pode passar uma lista de callbacks (como o argumento de palavra-chave callbacks) para os seguintes métodos de modelo:

• keras.Model.fit()

• keras.Model.evaluate()

• keras.Model.predict()

Uma visão geral dos métodos de callback

Métodos globais

on_(train|test|predict)_begin(self, logs=None)

Chamado no início de fit/evaluate/predict .

on_(train|test|predict)_end(self, logs=None)

Chamado no final de fit/evaluate/predict .

Métodos em nível de lote para treinamento/teste/previsão

on_(train|test|predict)_batch_begin(self, batch, logs=None)

Chamado imediatamente antes de processar um lote durante o treinamento/teste/previsão.

on_(train|test|predict)_batch_end(self, batch, logs=None)

Chamado ao final do treinamento/teste/previsão de um lote. Nesse método, logs é um dict que contém os resultados das métricas.

Métodos de nível de época (somente treinamento)

on_epoch_begin(self, epoch, logs=None)

Chamado no início de uma época durante o treinamento.

on_epoch_end(self, epoch, logs=None)

Chamado ao final de uma época durante o treinamento.

Um exemplo básico

Vamos dar uma olhada num exemplo concreto. Para começar, vamos importar o tensorflow e definir um modelo Keras Sequential simples:

# Define the Keras model to add callbacks to
def get_model():
    model = keras.Sequential()
    model.add(keras.layers.Dense(1, input_dim=784))
    model.compile(
        optimizer=keras.optimizers.RMSprop(learning_rate=0.1),
        loss="mean_squared_error",
        metrics=["mean_absolute_error"],
    )
    return model

Em seguida, carregue os dados MNIST para treinamento e teste da API de datasets do Keras:

# Load example MNIST data and pre-process it
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()
x_train = x_train.reshape(-1, 784).astype("float32") / 255.0
x_test = x_test.reshape(-1, 784).astype("float32") / 255.0

# Limit the data to 1000 samples
x_train = x_train[:1000]
y_train = y_train[:1000]
x_test = x_test[:1000]
y_test = y_test[:1000]

Agora, defina um callback personalizado simples que registre:

• Quando o fit/evaluate/predict começa e termina

• Quando cada época começa e termina

• Quando cada lote de treinamento começa e termina

• Quando cada lote de avaliação (teste) começa e termina

• Quando cada lote de inferência (previsão) começa e termina

class CustomCallback(keras.callbacks.Callback):
    def on_train_begin(self, logs=None):
        keys = list(logs.keys())
        print("Starting training; got log keys: {}".format(keys))

    def on_train_end(self, logs=None):
        keys = list(logs.keys())
        print("Stop training; got log keys: {}".format(keys))

    def on_epoch_begin(self, epoch, logs=None):
        keys = list(logs.keys())
        print("Start epoch {} of training; got log keys: {}".format(epoch, keys))

    def on_epoch_end(self, epoch, logs=None):
        keys = list(logs.keys())
        print("End epoch {} of training; got log keys: {}".format(epoch, keys))

    def on_test_begin(self, logs=None):
        keys = list(logs.keys())
        print("Start testing; got log keys: {}".format(keys))

    def on_test_end(self, logs=None):
        keys = list(logs.keys())
        print("Stop testing; got log keys: {}".format(keys))

    def on_predict_begin(self, logs=None):
        keys = list(logs.keys())
        print("Start predicting; got log keys: {}".format(keys))

    def on_predict_end(self, logs=None):
        keys = list(logs.keys())
        print("Stop predicting; got log keys: {}".format(keys))

    def on_train_batch_begin(self, batch, logs=None):
        keys = list(logs.keys())
        print("...Training: start of batch {}; got log keys: {}".format(batch, keys))

    def on_train_batch_end(self, batch, logs=None):
        keys = list(logs.keys())
        print("...Training: end of batch {}; got log keys: {}".format(batch, keys))

    def on_test_batch_begin(self, batch, logs=None):
        keys = list(logs.keys())
        print("...Evaluating: start of batch {}; got log keys: {}".format(batch, keys))

    def on_test_batch_end(self, batch, logs=None):
        keys = list(logs.keys())
        print("...Evaluating: end of batch {}; got log keys: {}".format(batch, keys))

    def on_predict_batch_begin(self, batch, logs=None):
        keys = list(logs.keys())
        print("...Predicting: start of batch {}; got log keys: {}".format(batch, keys))

    def on_predict_batch_end(self, batch, logs=None):
        keys = list(logs.keys())
        print("...Predicting: end of batch {}; got log keys: {}".format(batch, keys))

Vamos experimentar:

model = get_model()
model.fit(
    x_train,
    y_train,
    batch_size=128,
    epochs=1,
    verbose=0,
    validation_split=0.5,
    callbacks=[CustomCallback()],
)

res = model.evaluate(
    x_test, y_test, batch_size=128, verbose=0, callbacks=[CustomCallback()]
)

res = model.predict(x_test, batch_size=128, callbacks=[CustomCallback()])

Uso do dict logs

O dict logs contém o valor da perda e todas as métricas ao final de um lote ou época. O exemplo inclui a perda e o erro absoluto médio.

class LossAndErrorPrintingCallback(keras.callbacks.Callback):
    def on_train_batch_end(self, batch, logs=None):
        print(
            "Up to batch {}, the average loss is {:7.2f}.".format(batch, logs["loss"])
        )

    def on_test_batch_end(self, batch, logs=None):
        print(
            "Up to batch {}, the average loss is {:7.2f}.".format(batch, logs["loss"])
        )

    def on_epoch_end(self, epoch, logs=None):
        print(
            "The average loss for epoch {} is {:7.2f} "
            "and mean absolute error is {:7.2f}.".format(
                epoch, logs["loss"], logs["mean_absolute_error"]
            )
        )


model = get_model()
model.fit(
    x_train,
    y_train,
    batch_size=128,
    epochs=2,
    verbose=0,
    callbacks=[LossAndErrorPrintingCallback()],
)

res = model.evaluate(
    x_test,
    y_test,
    batch_size=128,
    verbose=0,
    callbacks=[LossAndErrorPrintingCallback()],
)

Uso do atributo self.model

Além de receber informações de log quando um de seus métodos é chamado, os callbacks têm acesso ao modelo associado à rodada atual de treinamento/avaliação/inferência: self.model.

Eis algumas coisas que você pode fazer com self.model em um callback:

• Definir self.model.stop_training = True para interromper imediatamente o treinamento.

• Modificar hiperparâmetros do otimizador (disponível como self.model.optimizer), tais como self.model.optimizer.learning_rate.

• Salvar o modelo em intervalos periódicos.

• Registrar a saída de model.predict() em algumas amostras de teste ao final de cada época, para usar como uma verificação de sanidade durante o treinamento.

• Extrair visualizações de recursos intermediários ao final de cada época, para monitorar o que o modelo está aprendendo ao longo do tempo.

• etc.

Vamos ver isto em ação em alguns exemplos.

Exemplos de aplicativos de callback do Keras

Parada antecipada com perda mínima

Este primeiro exemplo mostra a criação de um Callback que interrompe o treinamento quando a perda mínima é atingida, configurando o atributo self.model.stop_training (boolean). Opcionalmente, você pode fornecer um argumento patience para especificar quantas épocas devemos esperar antes de parar depois de atingir um mínimo local.

O tf.keras.callbacks.EarlyStopping fornece uma implementação mais completa e geral.

import numpy as np


class EarlyStoppingAtMinLoss(keras.callbacks.Callback):
    """Stop training when the loss is at its min, i.e. the loss stops decreasing.

  Arguments:
      patience: Number of epochs to wait after min has been hit. After this
      number of no improvement, training stops.
  """

    def __init__(self, patience=0):
        super(EarlyStoppingAtMinLoss, self).__init__()
        self.patience = patience
        # best_weights to store the weights at which the minimum loss occurs.
        self.best_weights = None

    def on_train_begin(self, logs=None):
        # The number of epoch it has waited when loss is no longer minimum.
        self.wait = 0
        # The epoch the training stops at.
        self.stopped_epoch = 0
        # Initialize the best as infinity.
        self.best = np.Inf

    def on_epoch_end(self, epoch, logs=None):
        current = logs.get("loss")
        if np.less(current, self.best):
            self.best = current
            self.wait = 0
            # Record the best weights if current results is better (less).
            self.best_weights = self.model.get_weights()
        else:
            self.wait += 1
            if self.wait >= self.patience:
                self.stopped_epoch = epoch
                self.model.stop_training = True
                print("Restoring model weights from the end of the best epoch.")
                self.model.set_weights(self.best_weights)

    def on_train_end(self, logs=None):
        if self.stopped_epoch > 0:
            print("Epoch %05d: early stopping" % (self.stopped_epoch + 1))


model = get_model()
model.fit(
    x_train,
    y_train,
    batch_size=64,
    steps_per_epoch=5,
    epochs=30,
    verbose=0,
    callbacks=[LossAndErrorPrintingCallback(), EarlyStoppingAtMinLoss()],
)

Programação da taxa de aprendizado

Neste exemplo, mostramos como um Callback personalizado pode ser usado para alterar dinamicamente a taxa de aprendizado do otimizador durante o treinamento.

Veja callbacks.LearningRateScheduler para implementações mais gerais.

class CustomLearningRateScheduler(keras.callbacks.Callback):
    """Learning rate scheduler which sets the learning rate according to schedule.

  Arguments:
      schedule: a function that takes an epoch index
          (integer, indexed from 0) and current learning rate
          as inputs and returns a new learning rate as output (float).
  """

    def __init__(self, schedule):
        super(CustomLearningRateScheduler, self).__init__()
        self.schedule = schedule

    def on_epoch_begin(self, epoch, logs=None):
        if not hasattr(self.model.optimizer, "lr"):
            raise ValueError('Optimizer must have a "lr" attribute.')
        # Get the current learning rate from model's optimizer.
        lr = float(tf.keras.backend.get_value(self.model.optimizer.learning_rate))
        # Call schedule function to get the scheduled learning rate.
        scheduled_lr = self.schedule(epoch, lr)
        # Set the value back to the optimizer before this epoch starts
        tf.keras.backend.set_value(self.model.optimizer.lr, scheduled_lr)
        print("\nEpoch %05d: Learning rate is %6.4f." % (epoch, scheduled_lr))


LR_SCHEDULE = [
    # (epoch to start, learning rate) tuples
    (3, 0.05),
    (6, 0.01),
    (9, 0.005),
    (12, 0.001),
]


def lr_schedule(epoch, lr):
    """Helper function to retrieve the scheduled learning rate based on epoch."""
    if epoch < LR_SCHEDULE[0][0] or epoch > LR_SCHEDULE[-1][0]:
        return lr
    for i in range(len(LR_SCHEDULE)):
        if epoch == LR_SCHEDULE[i][0]:
            return LR_SCHEDULE[i][1]
    return lr


model = get_model()
model.fit(
    x_train,
    y_train,
    batch_size=64,
    steps_per_epoch=5,
    epochs=15,
    verbose=0,
    callbacks=[
        LossAndErrorPrintingCallback(),
        CustomLearningRateScheduler(lr_schedule),
    ],
)

Callbacks Keras integrados

Não deixe de dar uma olhada nos callbacks Keras existentes lendo a documentação da API. As aplicações incluem fazer log em CSV, salvar o modelo, visualizar métricas no TensorBoard e muito mais!