Guide til at skrive brugerdefinerede TensorFlow/Keras-tilbagekald

Introduktion

Antag, at du ønsker, at din Keras-model skal have en bestemt adfærd under træning, evaluering eller forudsigelse. For eksempel vil du måske gemme din model ved hver træningsepoke. En måde at gøre dette på er at bruge tilbagekald.

Generelt er tilbagekald funktioner, der kaldes, når en hændelse sker, og som sendes som argumenter til andre funktioner. I tilfældet med Keras er de et værktøj til at tilpasse din models adfærd – det være sig under træning, evaluering eller slutning. Nogle applikationer er logning, modelpersistens, tidlig stop eller ændring af indlæringshastigheden. Dette gøres ved at sende en liste over tilbagekald som argumenter for keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Nogle almindelige brugstilfælde for tilbagekald er ændring af indlæringshastigheden, logning, overvågning og tidlig stop af træning. Keras har en række indbyggede tilbagekald, detaljerede
i dokumentationen.

Nogle mere specifikke applikationer kan dog kræve et tilpasset tilbagekald. For eksempel, implementering af Learning Rate warmup med et Cosinus Decay efter en holdeperiode er i øjeblikket ikke indbygget, men er meget udbredt og vedtaget som en skemalægger.

Callback-klasse og dens metoder

Keras har en specifik tilbagekaldsklasse, keras.callbacks.Callback, med metoder, der kan kaldes under træning, test og inferens på globalt, batch- eller epokeniveau. For at oprette tilpassede tilbagekald, skal vi oprette en underklasse og tilsidesætte disse metoder.

keras.callbacks.Callback klasse har tre slags metoder:

• globale metoder: kaldet i begyndelsen eller i slutningen af fit(), evaluate() , predict().
• batch-niveau metoder: kaldet i begyndelsen eller slutningen af ​​behandlingen af ​​en batch.
• Metoder på epokeniveau: kaldet ved begyndelsen eller slutningen af ​​en træningsbatch.

Bemærk: Hver metode har adgang til en dict kaldet logs. Nøglerne og værdierne af logs er kontekstuelle – de afhænger af den begivenhed, der kalder metoden. Desuden har vi adgang til modellen inde i hver metode gennem self.model attribut.

Lad os tage et kig på tre eksempler på tilpassede tilbagekald – et til træning, et til evaluering og et til forudsigelse. Hver enkelt vil på hvert trin udskrive, hvad vores model laver, og hvilke logfiler vi har adgang til. Dette er nyttigt for at forstå, hvad der er muligt at gøre med tilpassede tilbagekald på hvert trin.

Lad os starte med at definere en legetøjsmodel:

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

Tilbagekald til tilpasset træning

Vores første tilbagekald er at blive ringet op under træning. Lad os underklassificere Callback klasse:

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}")

Hvis nogen af ​​disse metoder ikke tilsidesættes – vil standardadfærd fortsætte som før. I vores eksempel - vi udskriver blot de tilgængelige logfiler og det niveau, som tilbagekaldet anvendes på, med korrekt indrykning.

Lad os tage et kig på udgangene:

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}

Bemærk, at vi ved hvert trin kan følge med i, hvad modellen laver, og hvilke målinger vi har adgang til. I slutningen af ​​hver batch og epoke har vi adgang til in-sample tab-funktionen og metrikken for vores model.

Tilbagekald til tilpasset evaluering

Lad os nu ringe til Model.evaluate() metode. Vi kan se, at i slutningen af ​​en batch har vi adgang til tabsfunktionen og målene på det tidspunkt, og i slutningen af ​​evalueringen har vi adgang til de overordnede tab og målinger:

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}

Tilbagekald med tilpasset forudsigelse

Lad os endelig ringe til Model.predict() metode. Bemærk, at i slutningen af ​​hver batch har vi adgang til de forudsagte output fra vores model:

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

Med disse – kan du tilpasse adfærden, opsætte overvågning eller på anden måde ændre processerne for træning, evaluering eller slutning. Et alternativ til sublcassing er at bruge LambdaCallback.

Brug af LambaCallback

Et af de indbyggede tilbagekald i Keras er LambdaCallback klasse. Dette tilbagekald accepterer en funktion, som definerer, hvordan den opfører sig, og hvad den gør! På en måde giver det dig mulighed for at bruge enhver vilkårlig funktion som et tilbagekald, hvilket giver dig mulighed for at oprette tilpassede tilbagekald.

Klassen har de valgfrie parametre:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Hver parameter accepterer en funktion som kaldes i den respektive modelbegivenhed. Lad os som et eksempel ringe tilbage for at sende en e-mail, når modellen er færdig med at træne:

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

For at foretage vores tilpassede tilbagekald ved hjælp af LambdaCallback, vi skal bare implementere den funktion, som vi ønsker at blive kaldt, indpakke den som en lambda funktion og videregive den til
LambdaCallback klasse som parameter.

Et tilbagekald til visualisering af modeltræning

I dette afsnit giver vi et eksempel på et tilpasset tilbagekald, der laver en animation af vores models præstation, der forbedres under træning. For at gøre dette gemmer vi værdierne af logfilerne i slutningen af ​​hver batch. Derefter, i slutningen af ​​træningsløkken, laver vi en animation vha matplotlib.

For at forbedre visualiseringen vil tabet og metrikken blive plottet i log-skala:

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

Vi vil bruge samme model som før, men med flere træningseksempler:

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

Vores output er en animation af metrikken og tabsfunktionen, når de ændrer sig gennem træningsprocessen:

Konklusion

I denne vejledning har vi taget et kig på implementeringen af ​​tilpassede tilbagekald i Keras.
Der er to muligheder for at implementere tilpassede tilbagekald - gennem underklassificering af keras.callbacks.Callback klasse, eller ved at bruge keras.callbacks.LambdaCallback klasse.

Vi har set et praktisk eksempel på at bruge LambdaCallbackfor at sende en e-mail i slutningen af ​​træningssløjfen, og et eksempel på underklassificering af Callback klasse, der laver en animation af træningsløkken.

Selvom Keras har mange indbyggede tilbagekald, kan det være nyttigt for mere specifikke applikationer at vide, hvordan man implementerer et tilpasset tilbagekald.