Guide d'écriture de rappels TensorFlow/Keras personnalisés

Introduction

Supposons que vous souhaitiez que votre modèle Keras ait un comportement spécifique lors de l'entraînement, de l'évaluation ou de la prédiction. Par exemple, vous souhaiterez peut-être enregistrer votre modèle à chaque époque d'entraînement. Une façon de faire est d'utiliser les rappels.

En général, les rappels sont des fonctions appelées lorsqu'un événement se produit et sont transmises en tant qu'arguments à d'autres fonctions. Dans le cas de Keras, il s'agit d'un outil permettant de personnaliser le comportement de votre modèle, que ce soit pendant la formation, l'évaluation ou l'inférence. Certaines applications sont la journalisation, la persistance du modèle, l'arrêt précoce ou la modification du taux d'apprentissage. Cela se fait en passant une liste de rappels comme arguments pour keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Certains cas d'utilisation courants des rappels sont la modification du taux d'apprentissage, la journalisation, la surveillance et l'arrêt précoce de la formation. Keras a un certain nombre de rappels intégrés, détaillés
dans la documentation.

Cependant, certaines applications plus spécifiques peuvent nécessiter un rappel personnalisé. Par exemple, mise en œuvre de l'échauffement du taux d'apprentissage avec une décroissance cosinus après une période de maintien n'est actuellement pas intégré, mais il est largement utilisé et adopté comme planificateur.

Classe de rappel et ses méthodes

Keras a une classe de rappel spécifique, keras.callbacks.Callback, avec des méthodes qui peuvent être appelées pendant la formation, les tests et l'inférence au niveau global, par lots ou d'époque. Afin de créer des rappels personnalisés, nous devons créer une sous-classe et remplacer ces méthodes.

Les keras.callbacks.Callback classe a trois types de méthodes :

• méthodes globales : appelées au début ou à la fin de fit(), evaluate() ainsi que predict().
• méthodes de niveau batch : appelées au début ou à la fin du traitement d'un batch.
• méthodes au niveau de l'époque : appelées au début ou à la fin d'un lot d'apprentissage.

Remarque: Chaque méthode a accès à un dict appelé logs. Les clés et les valeurs de logs sont contextuels - ils dépendent de l'événement qui appelle la méthode. De plus, nous avons accès au modèle à l'intérieur de chaque méthode à travers le self.model attribuer.

Examinons trois exemples de rappels personnalisés : un pour la formation, un pour l'évaluation et un pour la prédiction. Chacun imprimera à chaque étape ce que fait notre modèle et à quels journaux nous avons accès. Ceci est utile pour comprendre ce qu'il est possible de faire avec des rappels personnalisés à chaque étape.

Commençons par définir un modèle de jouet :

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:])

Rappel de formation personnalisé

Notre premier rappel est d'être appelé pendant la formation. Sous-classons le Callback classe:

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 l'une de ces méthodes n'est pas remplacée, le comportement par défaut continuera comme avant. Dans notre exemple, nous imprimons simplement les journaux disponibles et le niveau auquel le rappel est appliqué, avec une indentation appropriée.

Jetons un coup d'œil aux sorties :

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}

Notez que nous pouvons suivre à chaque étape ce que fait le modèle, et à quelles métriques nous avons accès. À la fin de chaque lot et époque, nous avons accès à la fonction de perte dans l'échantillon et aux métriques de notre modèle.

Rappel d'évaluation personnalisé

Maintenant, appelons le Model.evaluate() méthode. Nous pouvons voir qu'à la fin d'un lot, nous avons accès à la fonction de perte et aux métriques du moment, et à la fin de l'évaluation, nous avons accès à la perte globale et aux métriques :

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}

Rappel de prédiction personnalisé

Enfin, appelons le Model.predict() méthode. Notez qu'à la fin de chaque lot, nous avons accès aux sorties prédites de notre modèle :

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: {}

Avec ceux-ci, vous pouvez personnaliser le comportement, mettre en place une surveillance ou modifier autrement les processus de formation, d'évaluation ou d'inférence. Une alternative à la sous-classification consiste à utiliser le LambdaCallback.

Utilisation de LambaCallback

L'un des rappels intégrés dans Keras est le LambdaCallback classer. Ce callback accepte une fonction qui définit comment il se comporte et ce qu'il fait ! En un sens, cela vous permet d'utiliser n'importe quelle fonction arbitraire comme rappel, vous permettant ainsi de créer des rappels personnalisés.

La classe a les paramètres facultatifs :
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Chaque paramètre accepte une fonction qui est appelé dans l'événement de modèle respectif. Par exemple, effectuons un rappel pour envoyer un e-mail lorsque le modèle a terminé l'entraînement :

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],
)

Pour effectuer notre rappel personnalisé en utilisant LambdaCallback, il nous suffit d'implémenter la fonction que nous voulons appeler, de l'envelopper dans un lambda fonction et transmettez-la à la
LambdaCallback classe en paramètre.

Un rappel pour visualiser la formation du modèle

Dans cette section, nous donnerons un exemple de rappel personnalisé qui fait une animation de l'amélioration des performances de notre modèle pendant la formation. Pour ce faire, nous stockons les valeurs des logs à la fin de chaque batch. Puis, à la fin de la boucle de formation, nous créons une animation en utilisant matplotlib.

Afin d'améliorer la visualisation, la perte et les métriques seront tracées à l'échelle logarithmique :

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()

Nous utiliserons le même modèle qu'auparavant, mais avec plus d'échantillons d'apprentissage :

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()],
)

Notre résultat est une animation des métriques et de la fonction de perte à mesure qu'elles changent au cours du processus de formation :

Conclusion

Dans ce guide, nous avons examiné la mise en œuvre de rappels personnalisés dans Keras.
Il existe deux options pour implémenter des rappels personnalisés - en sous-classant le keras.callbacks.Callback classe, ou en utilisant la keras.callbacks.LambdaCallback classe.

Nous avons vu un exemple pratique utilisant LambdaCallbackpour envoyer un e-mail à la fin de la boucle de formation, et un exemple sous-classant le Callback classe qui crée une animation de la boucle d'entraînement.

Bien que Keras ait de nombreux rappels intégrés, savoir comment implémenter un rappel personnalisé peut être utile pour des applications plus spécifiques.