Przewodnik po pisaniu niestandardowych wywołań zwrotnych TensorFlow/Keras

Wprowadzenie

Załóżmy, że chcesz, aby Twój model Keras wykazywał określone zachowanie podczas uczenia, oceny lub przewidywania. Na przykład możesz chcieć zapisać swój model w każdej epoce treningowej. Jednym ze sposobów na to jest użycie wywołań zwrotnych.

Ogólnie rzecz biorąc, wywołania zwrotne to funkcje, które są wywoływane po wystąpieniu jakiegoś zdarzenia i są przekazywane jako argumenty do innych funkcji. W przypadku Keras są narzędziem do dostosowywania zachowania Twojego modelu – czy to podczas uczenia, oceny czy wnioskowania. Niektóre aplikacje to logowanie, trwałość modelu, wczesne zatrzymywanie lub zmiana szybkości uczenia. Odbywa się to poprzez przekazanie listy Callbacków jako argumentów dla keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Niektóre typowe przypadki użycia wywołań zwrotnych to modyfikowanie szybkości uczenia się, rejestrowanie, monitorowanie i wczesne zatrzymywanie uczenia. Keras ma wiele wbudowanych funkcji zwrotnych, szczegółowo opisanych
w dokumentacji.

Jednak niektóre bardziej specyficzne aplikacje mogą wymagać niestandardowego wywołania zwrotnego. Na przykład, wdrożenie rozgrzewki współczynnika uczenia się z rozpadem kosinusowym po okresie utrzymywania nie jest obecnie wbudowany, ale jest szeroko stosowany i adoptowany jako harmonogram.

Klasa wywołania zwrotnego i jej metody

Keras ma specyficzną klasę wywołań zwrotnych, keras.callbacks.Callback, z metodami, które można wywoływać podczas uczenia, testowania i wnioskowania na poziomie globalnym, wsadowym lub epokowym. W celu tworzyć niestandardowe wywołania zwrotne, musimy utworzyć podklasę i nadpisać te metody.

Połączenia keras.callbacks.Callback klasa ma trzy rodzaje metod:

• metody globalne: wywoływane na początku lub na końcu fit(), evaluate() i predict().
• metody na poziomie wsadu: wywoływane na początku lub na końcu przetwarzania wsadu.
• metody na poziomie epoki: wywoływane na początku lub na końcu partii treningowej.

Uwaga: Każda metoda ma dostęp do dyktatu o nazwie logs. Klucze i wartości logs są kontekstowe – zależą od zdarzenia, które wywołuje metodę. Ponadto mamy dostęp do modelu wewnątrz każdej metody poprzez self.model atrybutów.

Przyjrzyjmy się trzem przykładom niestandardowych wywołań zwrotnych — jeden do szkolenia, jeden do oceny i jeden do przewidywania. Każdy wydrukuje na każdym etapie, co robi nasz model i do jakich logów mamy dostęp. Jest to pomocne w zrozumieniu, co można zrobić z niestandardowymi wywołaniami zwrotnymi na każdym etapie.

Zacznijmy od zdefiniowania modelu zabawki:

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

Niestandardowe oddzwonienie do szkolenia

Nasz pierwszy callback ma zostać wywołany podczas szkolenia. Zróbmy podklasy Callback klasa:

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

Jeśli którakolwiek z tych metod nie zostanie zastąpiona — zachowanie domyślne będzie kontynuowane tak, jak wcześniej. W naszym przykładzie – po prostu wypisujemy dostępne logi i poziom, na którym stosuje się callback, z odpowiednim wcięciem.

Przyjrzyjmy się wynikom:

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}

Pamiętaj, że na każdym kroku możemy śledzić, co robi model i do jakich metryk mamy dostęp. Na końcu każdej partii i epoki mamy dostęp do funkcji straty w próbce i metryk naszego modelu.

Niestandardowe oddzwonienie do oceny

Teraz zadzwońmy do Model.evaluate() metoda. Widzimy, że na końcu partii mamy dostęp do funkcji straty i metryk w danym momencie, a na końcu oceny mamy dostęp do całkowitej straty i metryk:

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}

Oddzwonienie do niestandardowej prognozy

Na koniec nazwijmy Model.predict() metoda. Zauważ, że na końcu każdej partii mamy dostęp do przewidywanych wyników naszego modelu:

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

Dzięki nim możesz dostosować zachowanie, skonfigurować monitorowanie lub w inny sposób zmienić procesy szkolenia, oceny lub wnioskowania. Alternatywą dla sublcastingu jest użycie LambdaCallback.

Korzystanie z LambaCallback

Jednym z wbudowanych wywołań zwrotnych w Keras jest LambdaCallback klasa. To wywołanie zwrotne akceptuje funkcję, która definiuje, jak się zachowuje i co robi! W pewnym sensie pozwala na użycie dowolnej funkcji jako wywołania zwrotnego, co pozwala na tworzenie niestandardowych wywołań zwrotnych.

Klasa posiada opcjonalne parametry:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Każdy parametr akceptuje funkcja który jest wywoływany w odpowiednim zdarzeniu modelowym. Jako przykład wykonajmy wywołanie zwrotne, aby wysłać wiadomość e-mail, gdy model zakończy trenowanie:

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

Aby wykonać nasze niestandardowe wywołanie zwrotne za pomocą LambdaCallback, wystarczy zaimplementować funkcję, którą chcemy wywołać, owinąć ją jako a lambda funkcji i przekaż ją do
LambdaCallback klasa jako parametr.

Oddzwanianie do wizualizacji szkolenia modeli

W tej sekcji podamy przykład niestandardowego wywołania zwrotnego, które powoduje animację poprawy wydajności naszego modelu podczas uczenia. W tym celu przechowujemy wartości dzienników na końcu każdej partii. Następnie na końcu pętli treningowej tworzymy animację za pomocą matplotlib.

Aby ulepszyć wizualizację, straty i metryki zostaną wykreślone w skali logarytmicznej:

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

Użyjemy tego samego modelu co poprzednio, ale z większą liczbą próbek szkoleniowych:

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

Naszym wynikiem jest animacja metryk i funkcji straty, gdy zmieniają się one w procesie uczenia:

Wnioski

W tym przewodniku przyjrzeliśmy się implementacji niestandardowych wywołań zwrotnych w Keras.
Istnieją dwie opcje implementacji niestandardowych wywołań zwrotnych – poprzez podklasy keras.callbacks.Callback klasy lub za pomocą keras.callbacks.LambdaCallback class.

Widzieliśmy jeden praktyczny przykład za pomocą LambdaCallbackza wysłanie e-maila na koniec pętli treningowej i jeden przykład podklasy Callback klasa, która tworzy animację pętli treningowej.

Althoug Keras ma wiele wbudowanych wywołań zwrotnych, wiedza o tym, jak zaimplementować niestandardowe wywołania zwrotne, może być przydatna w przypadku bardziej specyficznych aplikacji.