Ghid pentru scrierea de apeluri personalizate TensorFlow/Keras

Introducere

Să presupunem că doriți ca modelul Keras să aibă un anumit comportament în timpul antrenamentului, evaluării sau predicției. De exemplu, ați putea dori să vă salvați modelul la fiecare epocă de antrenament. O modalitate de a face acest lucru este utilizarea apelurilor inverse.

În general, apelurile inverse sunt funcții care sunt apelate atunci când are loc un eveniment și sunt transmise ca argumente altor funcții. În cazul Keras, acestea sunt un instrument pentru a personaliza comportamentul modelului dumneavoastră – fie în timpul antrenamentului, al evaluării sau al inferenței. Unele aplicații sunt înregistrarea în jurnal, persistența modelului, oprirea timpurie sau modificarea ratei de învățare. Acest lucru se face prin trecerea unei liste de apeluri inverse ca argumente pentru keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Unele cazuri comune de utilizare pentru apeluri inverse sunt modificarea ratei de învățare, înregistrarea în jurnal, monitorizarea și oprirea timpurie a antrenamentului. Keras are o serie de apeluri încorporate, detaliate
în documentație.

Cu toate acestea, unele aplicații mai specifice ar putea necesita un apel invers personalizat. De exemplu, implementarea încălzirii ratei de învățare cu o declinare a cosinusului după o perioadă de reținere nu este în prezent încorporat, dar este utilizat pe scară largă și adoptat ca programator.

Clasa de apel invers și metodele sale

Keras are o anumită clasă de apel invers, keras.callbacks.Callback, cu metode care pot fi apelate în timpul antrenamentului, testării și inferenței la nivel global, de lot sau de epocă. Pentru a creați apeluri personalizate, trebuie să creăm o subclasă și să înlocuim aceste metode.

keras.callbacks.Callback clasa are trei tipuri de metode:

• metode globale: numite la începutul sau la sfârşitul fit(), evaluate() și predict().
• metode la nivel de lot: apelate la începutul sau la sfârșitul procesării unui lot.
• metode la nivel de epocă: apelate la începutul sau la sfârșitul unui lot de antrenament.

Notă: Fiecare metodă are acces la un dict numit logs. Cheile și valorile logs sunt contextuale – depind de evenimentul care apelează metoda. Mai mult, avem acces la modelul din interiorul fiecărei metode prin intermediul self.model atribut.

Să aruncăm o privire la trei exemple de apeluri personalizate – unul pentru instruire, unul pentru evaluare și unul pentru predicție. Fiecare va tipări în fiecare etapă ceea ce face modelul nostru și la ce jurnale avem acces. Acest lucru este util pentru a înțelege ce se poate face cu apelurile personalizate în fiecare etapă.

Să începem prin a defini un model de jucărie:

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

Inapoi personalizat de formare

Primul nostru apel invers va fi sunat în timpul antrenamentului. Să subclasăm Callback clasă:

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

Dacă oricare dintre aceste metode nu este suprascrisă – comportamentul implicit va continua ca înainte. În exemplul nostru – pur și simplu tipărim jurnalele disponibile și nivelul la care este aplicată apelul invers, cu indentarea corespunzătoare.

Să aruncăm o privire la ieșiri:

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}

Rețineți că putem urmări la fiecare pas ce face modelul și la ce metrici avem acces. La sfârșitul fiecărui lot și epocă, avem acces la funcția de pierdere în eșantion și la valorile modelului nostru.

Evaluare personalizată apel invers

Acum, să numim Model.evaluate() metodă. Putem vedea că la sfârșitul unui lot avem acces la funcția de pierdere și metricile la momentul respectiv, iar la sfârșitul evaluării avem acces la pierderea totală și metricile:

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}

Reapel cu predicție personalizată

În cele din urmă, să numim Model.predict() metodă. Observați că la sfârșitul fiecărui lot avem acces la rezultatele prezise ale modelului nostru:

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

Cu acestea – puteți personaliza comportamentul, configura monitorizarea sau modifica în alt mod procesele de instruire, evaluare sau inferență. O alternativă la sublcassing este folosirea LambdaCallback.

Folosind LambaCallback

Unul dintre apelurile încorporate în Keras este LambdaCallback clasă. Acest apel invers acceptă o funcție care definește cum se comportă și ce face! Într-un fel, vă permite să utilizați orice funcție arbitrară ca apel invers, permițându-vă astfel să creați apeluri personalizate.

Clasa are parametrii opționali:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Fiecare parametru acceptă o functie care este numit în evenimentul model respectiv. De exemplu, să facem un apel invers pentru a trimite un e-mail când modelul termină antrenamentul:

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

Pentru a face apelul nostru personalizat folosind LambdaCallback, trebuie doar să implementăm funcția pe care vrem să fie apelată, să o încapsulăm ca a lambda funcția și transmiteți-o către
LambdaCallback clasa ca parametru.

Un apel invers pentru vizualizarea antrenamentului de model

În această secțiune, vom oferi un exemplu de apel invers personalizat care face o animație a performanței modelului nostru să se îmbunătățească în timpul antrenamentului. Pentru a face acest lucru, stocăm valorile jurnalelor la sfârșitul fiecărui lot. Apoi, la sfârșitul buclei de antrenament, creăm o animație folosind matplotlib.

Pentru a îmbunătăți vizualizarea, pierderile și valorile vor fi reprezentate pe scară logică:

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

Vom folosi același model ca înainte, dar cu mai multe mostre de antrenament:

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

Rezultatul nostru este o animație a valorilor și a funcției de pierdere pe măsură ce acestea se modifică prin procesul de antrenament:

Concluzie

În acest ghid, am aruncat o privire asupra implementării apelurilor personalizate în Keras.
Există două opțiuni pentru implementarea apelurilor personalizate – prin subclasarea keras.callbacks.Callback clasa, sau prin utilizarea keras.callbacks.LambdaCallback clasă.

Am văzut un exemplu practic de utilizare LambdaCallbackpentru trimiterea unui e-mail la sfârșitul buclei de antrenament și un exemplu de subclasare a Callback clasă care creează o animație a buclei de antrenament.

Deși Keras are multe apeluri încorporate, a ști cum să implementați un apel invers personalizat poate fi util pentru aplicații mai specifice.