Handleiding voor het schrijven van aangepaste TensorFlow/Keras-callbacks

Introductie

Stel dat u wilt dat uw Keras-model specifiek gedrag vertoont tijdens training, evaluatie of voorspelling. U wilt bijvoorbeeld uw model bij elke trainingsperiode opslaan. Een manier om dit te doen is het gebruik van terugbellen.

Over het algemeen zijn callbacks functies die worden aangeroepen wanneer een gebeurtenis plaatsvindt en die als argumenten worden doorgegeven aan andere functies. In het geval van Keras zijn ze een hulpmiddel om het gedrag van uw model aan te passen - of het nu tijdens training, evaluatie of gevolgtrekking is. Sommige toepassingen zijn logging, modelpersistentie, vroegtijdig stoppen of het leertempo wijzigen. Dit wordt gedaan door een lijst met terugbelverzoeken door te geven als argumenten voor: keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Enkele veelvoorkomende use-cases voor callbacks zijn het wijzigen van de leersnelheid, logging, monitoring en het vroegtijdig stoppen van training. Keras heeft een aantal ingebouwde callbacks, gedetailleerd
in de documentatie.

Voor sommige meer specifieke toepassingen is mogelijk echter een aangepaste callback vereist. Bijvoorbeeld, het implementeren van Learning Rate warming-up met een Cosinus Decay na een wachtperiode is momenteel niet ingebouwd, maar wordt veel gebruikt en aangenomen als planner.

Terugbelklasse en zijn methoden

Keras heeft een specifieke callback-klasse, keras.callbacks.Callback, met methoden die kunnen worden aangeroepen tijdens training, testen en gevolgtrekking op globaal, batch- of tijdperkniveau. Om zo te aangepaste callbacks maken, moeten we een subklasse maken en deze methoden overschrijven.

De keras.callbacks.Callback klasse heeft drie soorten methoden:

• globale methoden: aangeroepen aan het begin of aan het einde van fit(), evaluate() en predict().
• methoden op batchniveau: aangeroepen aan het begin of aan het einde van de verwerking van een batch.
• methoden op epoch-niveau: aangeroepen aan het begin of aan het einde van een trainingsbatch.

Opmerking: Elke methode heeft toegang tot een dict genaamd logs. De sleutels en waarden van logs zijn contextueel - ze zijn afhankelijk van de gebeurtenis die de methode aanroept. Bovendien hebben we toegang tot het model binnen elke methode via de self.model attribuut.

Laten we eens kijken naar drie voorbeelden van aangepaste callbacks: een voor training, een voor evaluatie en een voor voorspelling. Elk zal in elke fase afdrukken wat ons model doet en tot welke logboeken we toegang hebben. Dit is handig om te begrijpen wat er mogelijk is met aangepaste callbacks in elke fase.

Laten we beginnen met het definiëren van een speelgoedmodel:

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

Aangepaste training terugbellen

Onze eerste terugbelactie wordt tijdens de training gebeld. Laten we de . onderklassen 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}")

Als een van deze methoden niet wordt overschreven, blijft het standaardgedrag zoals voorheen. In ons voorbeeld printen we eenvoudig de beschikbare logs en het niveau waarop de callback wordt toegepast, met de juiste inspringing.

Laten we eens kijken naar de uitgangen:

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}

Merk op dat we bij elke stap kunnen volgen wat het model doet en tot welke statistieken we toegang hebben. Aan het einde van elke batch en elk tijdperk hebben we toegang tot de verliesfunctie in de steekproef en de statistieken van ons model.

Aangepaste evaluatie terugbellen

Laten we nu de bellen Model.evaluate() methode. We kunnen zien dat we aan het einde van een batch toegang hebben tot de verliesfunctie en de statistieken op dat moment, en aan het einde van de evaluatie hebben we toegang tot het totale verlies en de statistieken:

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}

Aangepaste voorspelling terugbellen

Laten we tot slot de . bellen Model.predict() methode. Merk op dat we aan het einde van elke batch toegang hebben tot de voorspelde uitvoer van ons 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: {}

Hiermee kunt u het gedrag aanpassen, monitoring instellen of anderszins de processen van training, evaluatie of gevolgtrekking wijzigen. Een alternatief voor subcasseren is het gebruik van de LambdaCallback.

LambaCallback gebruiken

Een van de ingebouwde callbacks in Keras is de LambdaCallback klas. Deze callback accepteert een functie die bepaalt hoe het zich gedraagt ​​en wat het doet! In zekere zin kunt u elke willekeurige functie als een callback gebruiken, waardoor u aangepaste callbacks kunt maken.

De klasse heeft de optionele parameters:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Elke parameter accepteert: een functie die in de betreffende modelgebeurtenis wordt aangeroepen. Laten we als voorbeeld terugbellen om een ​​e-mail te sturen wanneer het model klaar is met trainen:

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

Om onze aangepaste callback te maken met behulp van LambdaCallback, we hoeven alleen de functie die we willen aanroepen te implementeren, deze in te pakken als a lambda functie en geef het door aan de
LambdaCallback klasse als parameter.

Een terugroepactie voor het visualiseren van modeltraining

In dit gedeelte geven we een voorbeeld van een aangepaste callback waarmee een animatie van de prestaties van ons model tijdens de training wordt verbeterd. Om dit te doen, slaan we aan het einde van elke batch de waarden van de logs op. Vervolgens maken we aan het einde van de trainingslus een animatie met matplotlib.

Om de visualisatie te verbeteren, worden het verlies en de metrieken in logschaal uitgezet:

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

We gebruiken hetzelfde model als voorheen, maar met meer trainingsvoorbeelden:

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

Onze output is een animatie van de statistieken en de verliesfunctie terwijl ze veranderen tijdens het trainingsproces:

Conclusie

In deze handleiding hebben we de implementatie van aangepaste callbacks in Keras bekeken.
Er zijn twee opties voor het implementeren van aangepaste callbacks – door middel van subclassificatie van de keras.callbacks.Callback klasse, of met behulp van de keras.callbacks.LambdaCallback klasse.

We hebben een praktisch voorbeeld gezien met LambdaCallbackvoor het verzenden van een e-mail aan het einde van de trainingslus, en een voorbeeld van subclassificatie van de Callback klasse die een animatie van de trainingslus maakt.

Hoewel Keras veel ingebouwde callbacks heeft, kan het handig zijn om te weten hoe een aangepaste callback moet worden geïmplementeerd voor meer specifieke toepassingen.