Introducción

Suponga que desea que su modelo de Keras tenga un comportamiento específico durante el entrenamiento, la evaluación o la predicción. Por ejemplo, es posible que desee guardar su modelo en cada época de entrenamiento. Una forma de hacer esto es usando Callbacks.

En general, las devoluciones de llamada son funciones que se llaman cuando ocurre algún evento y se pasan como argumentos a otras funciones. En el caso de Keras, son una herramienta para personalizar el comportamiento de su modelo, ya sea durante el entrenamiento, la evaluación o la inferencia. Algunas aplicaciones son el registro, la persistencia del modelo, la detención anticipada o el cambio de la tasa de aprendizaje. Esto se hace pasando una lista de devoluciones de llamada como argumentos para keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Algunos casos de uso comunes para las devoluciones de llamada son la modificación de la tasa de aprendizaje, el registro, la supervisión y la detención anticipada del entrenamiento. Keras tiene una serie de devoluciones de llamadas integradas, detalladas
en la documentación.

Sin embargo, algunas aplicaciones más específicas pueden requerir una devolución de llamada personalizada. Por ejemplo, implementando el calentamiento de la tasa de aprendizaje con una caída de coseno después de un período de espera actualmente no está integrado, pero es ampliamente utilizado y adoptado como programador.

Clase de devolución de llamada y sus métodos

Keras tiene una clase de devolución de llamada específica, keras.callbacks.Callback, con métodos que se pueden llamar durante el entrenamiento, la prueba y la inferencia a nivel global, por lotes o de época. Con el fin de crear devoluciones de llamadas personalizadas, necesitamos crear una subclase y anular estos métodos.

El keras.callbacks.Callback La clase tiene tres tipos de métodos:

• métodos globales: llamados al principio o al final de fit(), evaluate() y predict().
• métodos a nivel de lote: llamados al principio o al final del procesamiento de un lote.
• Métodos de nivel de época: llamados al principio o al final de un lote de entrenamiento.

Nota: Cada método tiene acceso a un dict llamado logs. Las claves y valores de logs son contextuales: dependen del evento que llama al método. Además, tenemos acceso al modelo dentro de cada método a través del self.model atributo.

Echemos un vistazo a tres ejemplos de devoluciones de llamada personalizadas: uno para entrenamiento, uno para evaluación y otro para predicción. Cada uno imprimirá en cada etapa qué está haciendo nuestro modelo y a qué registros tenemos acceso. Esto es útil para comprender lo que se puede hacer con las devoluciones de llamadas personalizadas en cada etapa.

Comencemos definiendo un modelo de juguete:

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

model = keras.Sequential()
model.add(keras.layers.Dense(10, input_dim = 1, activation='relu'))
model.add(keras.layers.Dense(10, activation='relu'))
model.add(keras.layers.Dense(1))
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=0.1),
    loss = "mean_squared_error",
    metrics = ["mean_absolute_error"]
)

x = np.random.uniform(low = 0, high = 10, size = 1000)
y = x**2
x_train, x_test = (x[:900],x[900:])
y_train, y_test = (y[:900],y[900:])

Devolución de llamada de capacitación personalizada

Nuestra primera devolución de llamada debe ser llamada durante el entrenamiento. Vamos a subclasificar el Callback clase:

class TrainingCallback(keras.callbacks.Callback):
    def __init__(self):
        self.tabulation = {"train":"", 'batch': " "*8, 'epoch':" "*4}
    def on_train_begin(self, logs=None):
        tab = self.tabulation['train']
        print(f"{tab}Training!")
        print(f"{tab}available logs: {logs}")

    def on_train_batch_begin(self, batch, logs=None):
        tab = self.tabulation['batch']
        print(f"{tab}Batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_train_batch_end(self, batch, logs=None):
        tab = self.tabulation['batch']
        print(f"{tab}End of Batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_epoch_begin(self, epoch, logs=None):
        tab = self.tabulation['epoch']
        print(f"{tab}Epoch {epoch} of training")
        print(f"{tab}available logs: {logs}")

    def on_epoch_end(self, epoch, logs=None):
        tab = self.tabulation['epoch']
        print(f"{tab}End of Epoch {epoch} of training")
        print(f"{tab}available logs: {logs}")

    def on_train_end(self, logs=None):
        tab = self.tabulation['train']
        print(f"{tab}Finishing training!")
        print(f"{tab}available logs: {logs}")

Si alguno de estos métodos no se anula, el comportamiento predeterminado continuará como antes. En nuestro ejemplo, simplemente imprimimos los registros disponibles y el nivel en el que se aplica la devolución de llamada, con la sangría adecuada.

Echemos un vistazo a las salidas:

model.fit(
    x_train,
    y_train,
    batch_size=500,
    epochs=2,
    verbose=0,
    callbacks=[TrainingCallback()],
)

Training!
available logs: {}
    Epoch 0 of training
    available logs: {}
        Batch 0
        available logs: {}
        End of Batch 0
        available logs: {'loss': 2172.373291015625, 'mean_absolute_error': 34.79669952392578}
        Batch 1
        available logs: {}
        End of Batch 1
        available logs: {'loss': 2030.1309814453125, 'mean_absolute_error': 33.30256271362305}
    End of Epoch 0 of training
    available logs: {'loss': 2030.1309814453125, 'mean_absolute_error': 33.30256271362305}
    Epoch 1 of training
    available logs: {}
        Batch 0
        available logs: {}
        End of Batch 0
        available logs: {'loss': 1746.2772216796875, 'mean_absolute_error': 30.268001556396484}
        Batch 1
        available logs: {}
        End of Batch 1
        available logs: {'loss': 1467.36376953125, 'mean_absolute_error': 27.10252571105957}
    End of Epoch 1 of training
    available logs: {'loss': 1467.36376953125, 'mean_absolute_error': 27.10252571105957}
Finishing training!
available logs: {'loss': 1467.36376953125, 'mean_absolute_error': 27.10252571105957}

Tenga en cuenta que podemos seguir en cada paso lo que está haciendo el modelo y a qué métricas tenemos acceso. Al final de cada lote y época, tenemos acceso a la función de pérdida en la muestra y las métricas de nuestro modelo.

Devolución de llamada de evaluación personalizada

Ahora, llamemos al Model.evaluate() método. Podemos ver que al final de un lote tenemos acceso a la función de pérdida y las métricas en ese momento, y al final de la evaluación tenemos acceso a la pérdida general y las métricas:

class TestingCallback(keras.callbacks.Callback):
    def __init__(self):
          self.tabulation = {"test":"", 'batch': " "*8}
      
    def on_test_begin(self, logs=None):
        tab = self.tabulation['test']
        print(f'{tab}Evaluating!')
        print(f'{tab}available logs: {logs}')

    def on_test_end(self, logs=None):
        tab = self.tabulation['test']
        print(f'{tab}Finishing evaluation!')
        print(f'{tab}available logs: {logs}')

    def on_test_batch_begin(self, batch, logs=None):
        tab = self.tabulation['batch']
        print(f"{tab}Batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_test_batch_end(self, batch, logs=None):
        tab = self.tabulation['batch']
        print(f"{tab}End of batch {batch}")
        print(f"{tab}available logs: {logs}")

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

Evaluating!
available logs: {}
        Batch 0
        available logs: {}
        End of batch 0
        available logs: {'loss': 382.2723083496094, 'mean_absolute_error': 14.069927215576172}
Finishing evaluation!
available logs: {'loss': 382.2723083496094, 'mean_absolute_error': 14.069927215576172}

Devolución de llamada de predicción personalizada

Finalmente, llamemos al Model.predict() método. Observe que al final de cada lote tenemos acceso a los resultados previstos de nuestro modelo:

class PredictionCallback(keras.callbacks.Callback):
    def __init__(self):
        self.tabulation = {"prediction":"", 'batch': " "*8}

    def on_predict_begin(self, logs=None):
        tab = self.tabulation['prediction']
        print(f"{tab}Predicting!")
        print(f"{tab}available logs: {logs}")

    def on_predict_end(self, logs=None):
        tab = self.tabulation['prediction']
        print(f"{tab}End of Prediction!")
        print(f"{tab}available logs: {logs}")

    def on_predict_batch_begin(self, batch, logs=None):
        tab = self.tabulation['batch']
        print(f"{tab}batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_predict_batch_end(self, batch, logs=None):
        tab = self.tabulation['batch']
        print(f"{tab}End of batch {batch}")
        print(f"{tab}available logs:n {logs}")

res = model.predict(x_test[:10],
                    verbose = 0, 
                    callbacks=[PredictionCallback()])

Predicting!
available logs: {}
        batch 0
        available logs: {}
        End of batch 0
        available logs:
 {'outputs': array([[ 7.743822],
       [27.748264],
       [33.082104],
       [26.530678],
       [27.939169],
       [18.414223],
       [42.610645],
       [36.69335 ],
       [13.096557],
       [37.120853]], dtype=float32)}
End of Prediction!
available logs: {}

Con estos, puede personalizar el comportamiento, configurar el monitoreo o alterar los procesos de capacitación, evaluación o inferencia. Una alternativa a la subcassing es usar el LambdaCallback.

Usando LambaCallback

Una de las devoluciones de llamadas integradas en Keras es la LambdaCallback clase. ¡Esta devolución de llamada acepta una función que define cómo se comporta y qué hace! En cierto sentido, le permite usar cualquier función arbitraria como devolución de llamada, lo que le permite crear devoluciones de llamada personalizadas.

La clase tiene los parámetros opcionales:
–on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Cada parámetro acepta Una función que se llama en el evento del modelo respectivo. Como ejemplo, hagamos una devolución de llamada para enviar un correo electrónico cuando el modelo termine de entrenarse:

import smtplib
from email.message import EmailMessage

def send_email(logs): 
    msg = EmailMessage()
    content = f"""The model has finished training."""
    for key, value in logs.items():
      content = content + f"n{key}:{value:.2f}"
    msg.set_content(content)
    msg['Subject'] = f'Training report'
    msg['From'] = '[email protected]'
    msg['To'] = 'receiver-email'

    s = smtplib.SMTP('smtp.gmail.com', 587)
    s.starttls()
    s.login("[email protected]", "your-gmail-app-password")
    s.send_message(msg)
    s.quit()

lambda_send_email = lambda logs : send_email(logs)

email_callback = keras.callbacks.LambdaCallback(on_train_end = lambda_send_email)

model.fit(
    x_train,
    y_train,
    batch_size=100,
    epochs=1,
    verbose=0,
    callbacks=[email_callback],
)

Para hacer nuestra devolución de llamada personalizada usando LambdaCallback, solo necesitamos implementar la función a la que queremos que se llame, envolverla como un lambda función y pásela al
LambdaCallback class como parámetro.

Una devolución de llamada para visualizar el entrenamiento del modelo

En esta sección, daremos un ejemplo de una devolución de llamada personalizada que hace que una animación del rendimiento de nuestro modelo mejore durante el entrenamiento. Para hacer esto, almacenamos los valores de los registros al final de cada lote. Luego, al final del ciclo de entrenamiento, creamos una animación usando matplotlib.

Para mejorar la visualización, la pérdida y las métricas se trazarán en escala logarítmica:

import matplotlib.pyplot as plt
import numpy as np
from matplotlib.animation import FuncAnimation
from IPython import display

class TrainingAnimationCallback(keras.callbacks.Callback):
    def __init__(self, duration = 40, fps = 1000/25):
        self.duration = duration
        self.fps = fps
        self.logs_history = []

    def set_plot(self):   
        self.figure = plt.figure()
        
        plt.xticks(
            range(0,self.params['steps']*self.params['epochs'], self.params['steps']),
            range(0,self.params['epochs']))
        plt.xlabel('Epoch')
        plt.ylabel('Loss & Metrics ($Log_{10}$ scale)')

        self.plot = {}
        for metric in self.model.metrics_names:
          self.plot[metric], = plt.plot([],[], label = metric)
          
        max_y = [max(log.values()) for log in self.logs_history]
        
        self.title = plt.title(f'batches:0')
        plt.xlim(0,len(self.logs_history)) 
        plt.ylim(0,max(max_y))

           
        plt.legend(loc='upper right')
  
    def animation_function(self,frame):
        batch = frame % self.params['steps']
        self.title.set_text(f'batch:{batch}')
        x = list(range(frame))
        
        for metric in self.model.metrics_names:
            y = [log[metric] for log in self.logs_history[:frame]]
            self.plot[metric].set_data(x,y)
        
    def on_train_batch_end(self, batch, logs=None):
        logarithm_transform = lambda item: (item[0], np.log(item[1]))
        logs = dict(map(logarithm_transform,logs.items()))
        self.logs_history.append(logs)
       
    def on_train_end(self, logs=None):
        self.set_plot()
        num_frames = int(self.duration*self.fps)
        num_batches = self.params['steps']*self.params['epochs']
        selected_batches = range(0, num_batches , num_batches//num_frames )
        interval = 1000*(1/self.fps)
        anim_created = FuncAnimation(self.figure, 
                                     self.animation_function,
                                     frames=selected_batches,
                                     interval=interval)
        video = anim_created.to_html5_video()
        
        html = display.HTML(video)
        display.display(html)
        plt.close()

Usaremos el mismo modelo que antes, pero con más muestras de entrenamiento:

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

model = keras.Sequential()
model.add(keras.layers.Dense(10, input_dim = 1, activation='relu'))
model.add(keras.layers.Dense(10, activation='relu'))
model.add(keras.layers.Dense(1))
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=0.1),
    loss = "mean_squared_error",
    metrics = ["mean_absolute_error"]
)

def create_sample(sample_size, train_test_proportion = 0.9):
    x = np.random.uniform(low = 0, high = 10, size = sample_size)
    y = x**2
    train_test_split = int(sample_size*train_test_proportion)
    x_train, x_test = (x[:train_test_split],x[train_test_split:])
    y_train, y_test = (y[:train_test_split],y[train_test_split:])
    return (x_train,x_test,y_train,y_test)

x_train,x_test,y_train,y_test = create_sample(35200)


model.fit(
    x_train,
    y_train,
    batch_size=32,
    epochs=2,
    verbose=0,
    callbacks=[TrainingAnimationCallback()],
)

Nuestro resultado es una animación de las métricas y la función de pérdida a medida que cambian durante el proceso de entrenamiento:

Conclusión

En esta guía, echamos un vistazo a la implementación de devoluciones de llamadas personalizadas en Keras.
Hay dos opciones para implementar devoluciones de llamada personalizadas: a través de la subclasificación del keras.callbacks.Callback clase, o usando el keras.callbacks.LambdaCallback clase.

Hemos visto un ejemplo práctico usando LambdaCallbackpara enviar un correo electrónico al final del ciclo de entrenamiento, y un ejemplo subclasificando el Callback clase que crea una animación del ciclo de entrenamiento.

Aunque Keras tiene muchas devoluciones de llamada integradas, saber cómo implementar una devolución de llamada personalizada puede ser útil para aplicaciones más específicas.