Veiledning for å skrive tilpassede TensorFlow/Keras tilbakeringinger

Introduksjon

Anta at du vil at Keras-modellen din skal ha en viss oppførsel under trening, evaluering eller prediksjon. For eksempel kan det være lurt å lagre modellen din ved hver treningsepoke. En måte å gjøre dette på er å bruke tilbakeringinger.

Generelt er tilbakeringinger funksjoner som kalles når en hendelse inntreffer, og sendes som argumenter til andre funksjoner. Når det gjelder Keras, er de et verktøy for å tilpasse oppførselen til modellen din – det være seg under trening, evaluering eller slutning. Noen applikasjoner er logging, modellutholdenhet, tidlig stopp eller endring av læringshastigheten. Dette gjøres ved å sende en liste over tilbakeringinger som argumenter for keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Noen vanlige brukstilfeller for tilbakeringing er endring av læringshastigheten, logging, overvåking og tidlig stopp av trening. Keras har en rekke innebygde tilbakeringinger, detaljert
i dokumentasjonen.

Noen mer spesifikke applikasjoner kan imidlertid kreve en tilpasset tilbakeringing. For eksempel, implementere Learning Rate oppvarming med en Cosinus Decay etter en holdeperiode er for øyeblikket ikke innebygd, men er mye brukt og tatt i bruk som en planlegger.

Tilbakeringingsklasse og dens metoder

Keras har en spesifikk tilbakeringingsklasse, keras.callbacks.Callback, med metoder som kan kalles under trening, testing og inferens på globalt, batch- eller epokenivå. For å opprette tilpassede tilbakeringinger, må vi lage en underklasse og overstyre disse metodene.

De keras.callbacks.Callback klasse har tre typer metoder:

• globale metoder: kalt i begynnelsen eller slutten av fit(), evaluate() og predict().
• batch-nivå metoder: kalles i begynnelsen eller på slutten av behandlingen av en batch.
• Metoder på epokenivå: kalt ved begynnelsen eller slutten av en treningsgruppe.

OBS: Hver metode har tilgang til en dict kalt logs. Nøklene og verdiene til logs er kontekstuelle – de avhenger av hendelsen som kaller metoden. Dessuten har vi tilgang til modellen i hver metode gjennom self.model attributt.

La oss ta en titt på tre tilpassede tilbakeringingseksempler – ett for opplæring, ett for evaluering og ett for prediksjon. Hver og en vil skrive ut på hvert trinn hva modellen vår gjør og hvilke logger vi har tilgang til. Dette er nyttig for å forstå hva som er mulig å gjøre med tilpassede tilbakeringinger på hvert trinn.

La oss begynne med å definere en leketøysmodell:

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

Custom Training Callback

Vår første tilbakeringing er å bli oppringt under trening. La oss underklasser 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 noen av disse metodene ikke overstyres – vil standard oppførsel fortsette som før. I vårt eksempel - vi skriver ganske enkelt ut de tilgjengelige loggene og nivået som tilbakeringingen brukes på, med riktig innrykk.

La oss ta en titt på utgangene:

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 at vi ved hvert trinn kan følge med på hva modellen gjør, og hvilke beregninger vi har tilgang til. På slutten av hver batch og epoke har vi tilgang til tapsfunksjonen i prøven og beregningene til modellen vår.

Tilpasset evaluering tilbakeringing

La oss nå ringe Model.evaluate() metode. Vi kan se at på slutten av en batch har vi tilgang til tapsfunksjonen og beregningene på det tidspunktet, og på slutten av evalueringen har vi tilgang til det totale tapet og beregningene:

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}

Tilpasset tilbakeringing for prediksjon

Til slutt, la oss ringe Model.predict() metode. Legg merke til at på slutten av hver batch har vi tilgang til de forutsagte utgangene til modellen vår:

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 atferden, sette opp overvåking eller på annen måte endre prosessene for trening, evaluering eller slutning. Et alternativ til subcassing er å bruke LambdaCallback.

Bruker LambaCallback

En av de innebygde tilbakeringingene i Keras er LambdaCallback klasse. Denne tilbakeringingen aksepterer en funksjon som definerer hvordan den oppfører seg og hva den gjør! På en måte lar den deg bruke en hvilken som helst vilkårlig funksjon som en tilbakeringing, slik at du kan lage tilpassede tilbakeringinger.

Klassen har de valgfrie parameterne:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Hver parameter aksepterer en funksjon som kalles i den respektive modellhendelsen. Som et eksempel, la oss ringe tilbake for å sende en e-post når modellen er ferdig med treningen:

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 å gjøre vår tilpassede tilbakeringing ved hjelp av LambdaCallback, vi trenger bare å implementere funksjonen vi vil kalles, pakke den inn som en lambda funksjon og send den til
LambdaCallback klasse som en parameter.

En tilbakeringing for visualisering av modelltrening

I denne delen vil vi gi et eksempel på en tilpasset tilbakeringing som gjør en animasjon av modellens ytelse som forbedres under trening. For å gjøre dette lagrer vi verdiene til loggene på slutten av hver batch. Så, på slutten av treningsløkken, lager vi en animasjon ved hjelp av matplotlib.

For å forbedre visualiseringen vil tapet og beregningene plottes i loggskala:

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 bruke samme modell som før, men med flere treningsprøver:

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

Resultatet vårt er en animasjon av beregningene og tapsfunksjonen etter hvert som de endres gjennom treningsprosessen:

konklusjonen

I denne veiledningen har vi tatt en titt på implementeringen av tilpassede tilbakeringinger i Keras.
Det er to alternativer for å implementere tilpassede tilbakeringinger – gjennom underklassifisering av keras.callbacks.Callback klasse, eller ved å bruke keras.callbacks.LambdaCallback klasse.

Vi har sett ett praktisk eksempel på bruk LambdaCallbackfor å sende en e-post på slutten av treningssløyfen, og ett eksempel på underklassing av Callback klasse som lager en animasjon av treningsløkken.

Selv om Keras har mange innebygde tilbakeringinger, kan det være nyttig å vite hvordan man implementerer en tilpasset tilbakeringing for mer spesifikke applikasjoner.