Guide till att skriva anpassade TensorFlow/Keras-återuppringningar

Beskrivning

Anta att du vill att din Keras-modell ska ha något specifikt beteende under träning, utvärdering eller förutsägelse. Till exempel kanske du vill spara din modell vid varje träningsepok. Ett sätt att göra detta är att använda återuppringningar.

I allmänhet är Callbacks funktioner som anropas när någon händelse inträffar, och som skickas som argument till andra funktioner. När det gäller Keras är de ett verktyg för att anpassa beteendet hos din modell – vare sig det är under träning, utvärdering eller slutledning. Vissa applikationer är loggning, modellbeständighet, tidig stopp eller ändring av inlärningshastigheten. Detta görs genom att skicka en lista med återuppringningar som argument för keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Några vanliga användningsfall för återuppringningar är modifiering av inlärningshastigheten, loggning, övervakning och tidigt avbrytande av utbildning. Keras har ett antal inbyggda återuppringningar, detaljerade
i dokumentationen.

Vissa mer specifika applikationer kan dock kräva en anpassad återuppringning. Till exempel, implementera Learning Rate-uppvärmning med en Cosine Decay efter en uppehållsperiod är för närvarande inte inbyggt, men används ofta och används som schemaläggare.

Callback Class och dess metoder

Keras har en specifik återuppringningsklass, keras.callbacks.Callback, med metoder som kan anropas under utbildning, testning och slutledning på global, batch- eller epoknivå. För att skapa anpassade återuppringningarmåste vi skapa en underklass och åsidosätta dessa metoder.

Smakämnen keras.callbacks.Callback klass har tre typer av metoder:

• globala metoder: anropas i början eller slutet av fit(), evaluate() och predict().
• Metoder på batchnivå: anropas i början eller i slutet av bearbetningen av en batch.
• Metoder på epoknivå: anropas i början eller i slutet av en träningssats.

Notera: Varje metod har tillgång till ett dict som kallas logs. Nycklar och värderingar av logs är kontextuella – de beror på händelsen som anropar metoden. Dessutom har vi tillgång till modellen inuti varje metod genom self.model attribut.

Låt oss ta en titt på tre anpassade återuppringningsexempel – ett för utbildning, ett för utvärdering och ett för förutsägelse. Var och en kommer i varje skede att skriva ut vad vår modell gör och vilka loggar vi har tillgång till. Detta är användbart för att förstå vad som är möjligt att göra med anpassade återuppringningar i varje steg.

Låt oss börja med att definiera en leksaksmodell:

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örsta återuppringning är att bli uppringd under träning. Låt oss underklassa Callback klass:

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

Om någon av dessa metoder inte åsidosätts – kommer standardbeteendet att fortsätta som tidigare. I vårt exempel skriver vi helt enkelt ut de tillgängliga loggarna och nivån på vilken återuppringningen tillämpas, med korrekt indrag.

Låt oss ta en titt på utgångarna:

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}

Observera att vi vid varje steg kan följa vad modellen gör, och till vilka mätvärden vi har tillgång till. I slutet av varje batch och epok har vi tillgång till förlustfunktionen i urvalet och mätvärdena för vår modell.

Anpassad utvärdering återuppringning

Låt oss nu ringa Model.evaluate() metod. Vi kan se att i slutet av en batch har vi tillgång till förlustfunktionen och mätvärdena vid tidpunkten, och i slutet av utvärderingen har vi tillgång till den övergripande förlusten och mätvärdena:

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}

Custom Prediction Callback

Slutligen, låt oss ringa Model.predict() metod. Observera att i slutet av varje batch har vi tillgång till de förutspådda utgångarna från vår modell:

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 dessa – kan du anpassa beteendet, ställa in övervakning eller på annat sätt ändra processerna för utbildning, utvärdering eller slutledning. Ett alternativ till sublcasing är att använda LambdaCallback.

Använder LambaCallback

En av de inbyggda återuppringningarna i Keras är LambdaCallback klass. Denna callback accepterar en funktion som definierar hur den beter sig och vad den gör! På sätt och vis låter det dig använda vilken godtycklig funktion som helst som en återuppringning, vilket gör att du kan skapa anpassade återuppringningar.

Klassen har de valfria parametrarna:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Varje parameter accepteras en funktion som kallas i respektive modellhändelse. Som ett exempel, låt oss ringa tillbaka för att skicka ett e-postmeddelande när modellen är klar med träningen:

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

För att göra vår anpassade återuppringning med hjälp av LambdaCallback, vi behöver bara implementera funktionen som vi vill ska kallas, linda in den som en lambda funktion och skicka den till
LambdaCallback klass som en parameter.

En återuppringning för visualisering av modellträning

I det här avsnittet kommer vi att ge ett exempel på en anpassad återuppringning som gör en animering av vår modells prestanda som förbättras under träning. För att göra detta lagrar vi loggarnas värden i slutet av varje batch. Sedan, i slutet av träningsslingan, skapar vi en animation med hjälp av matplotlib.

För att förbättra visualiseringen kommer förlusten och mätvärdena att plottas 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 kommer att använda samma modell som tidigare, men med fler träningsexempel:

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

Vår produktion är en animering av mätvärdena och förlustfunktionen när de förändras under utbildningsprocessen:

Slutsats

I den här guiden har vi tagit en titt på implementeringen av anpassade återuppringningar i Keras.
Det finns två alternativ för att implementera anpassade återuppringningar – genom att underklassa keras.callbacks.Callback klass, eller genom att använda keras.callbacks.LambdaCallback klass.

Vi har sett ett praktiskt exempel på att använda LambdaCallbackför att skicka ett e-postmeddelande i slutet av träningsslingan, och ett exempel på underklassning av Callback klass som skapar en animering av träningsslingan.

Även om Keras har många inbyggda återuppringningar, kan det vara användbart att veta hur man implementerar en anpassad återuppringning för mer specifika applikationer.