Útmutató az egyéni TensorFlow/Keras visszahívások írásához

Bevezetés

Tegyük fel, hogy azt szeretné, hogy Keras modellje bizonyos viselkedést tanúsítson a képzés, az értékelés vagy az előrejelzés során. Például érdemes lehet elmenteni a modellt minden képzési szakaszban. Ennek egyik módja a visszahívások használata.

A visszahívások általában olyan függvények, amelyek akkor hívódnak meg, amikor valamilyen esemény bekövetkezik, és argumentumként adják át más függvényeknek. A Keras esetében ezek egy olyan eszköz, amellyel személyre szabhatja modellje viselkedését – legyen az edzés, értékelés vagy következtetés során. Egyes alkalmazások naplóznak, modellek maradnak, korai leállítást végeznek vagy módosítják a tanulási sebességet. Ez úgy történik, hogy argumentumként adjuk át a visszahívások listáját keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

A visszahívások gyakori felhasználási esetei a tanulási sebesség módosítása, a naplózás, a figyelés és a képzés korai leállítása. A Keras számos beépített visszahívással rendelkezik, részletesen
a dokumentációban.

Néhány konkrétabb alkalmazás azonban egyéni visszahívást igényelhet. Például, a Tanulási sebesség bemelegítés végrehajtása koszinusz-csökkentéssel egy tartási időszak után jelenleg nincs beépítve, de széles körben használják és elfogadják ütemezőként.

Visszahívási osztály és módszerei

A Kerasnak van egy speciális visszahívási osztálya, keras.callbacks.Callback, a betanítás, tesztelés és következtetés során globális, kötegelt vagy korszakos szinten hívható módszerekkel. Azért, hogy egyéni visszahívások létrehozása, létre kell hoznunk egy alosztályt, és felül kell írni ezeket a metódusokat.

A keras.callbacks.Callback osztály háromféle metódussal rendelkezik:

• globális módszerek: az elején vagy a végén hívják fit(), evaluate() és a predict().
• kötegszintű módszerek: a köteg feldolgozásának elején vagy végén hívják meg.
• korszak szintű módszerek: a képzési köteg elején vagy végén hívják meg.

Jegyzet: Mindegyik metódus hozzáfér egy dict-hoz logs. A kulcsai és értékei logs kontextuálisak – a módszert meghívó eseménytől függenek. Sőt, az egyes metódusokon belül hozzáférünk a modellhez a self.model tulajdonság.

Nézzünk meg három egyedi visszahívási példát – egyet a képzéshez, egyet az értékeléshez és egyet az előrejelzéshez. Mindegyik minden szakaszban kinyomtatja, hogy mit csinál a modellünk, és mely naplókhoz férünk hozzá. Ez segít megérteni, mit lehet tenni az egyéni visszahívásokkal az egyes szakaszokban.

Kezdjük a játékmodell meghatározásával:

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

Egyéni képzési visszahívás

Az első visszahívásunkat edzés közben hívjuk. Alosztályozzuk a Callback osztály:

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

Ha a fenti módszerek egyike sem kerül felülírásra, az alapértelmezett viselkedés folytatódik, mint korábban. Példánkban – egyszerűen kinyomtatjuk a rendelkezésre álló naplókat és azt a szintet, amelyen a visszahívás alkalmazásra kerül, megfelelő behúzással.

Nézzük a kimeneteket:

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}

Vegye figyelembe, hogy minden lépésnél nyomon követhetjük, hogy mit csinál a modell, és mely mérőszámokhoz férünk hozzá. Minden köteg és korszak végén hozzáférhetünk a mintán belüli veszteségfüggvényhez és a modellünk metrikáihoz.

Egyéni értékelési visszahívás

Most pedig hívjuk a Model.evaluate() módszer. Láthatjuk, hogy egy köteg végén hozzáférünk a veszteségfüggvényhez és az akkori mérőszámokhoz, a kiértékelés végén pedig a teljes veszteséghez és mérőszámokhoz:

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}

Egyéni előrejelzés visszahívás

Végül hívjuk a Model.predict() módszer. Figyeljük meg, hogy minden köteg végén hozzáférünk modellünk előre jelzett kimeneteihez:

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

Ezekkel – testreszabhatja a viselkedést, beállíthatja a megfigyelést vagy más módon módosíthatja a képzési, értékelési vagy következtetési folyamatokat. A továbbadás alternatívája a LambdaCallback.

A LambaCallback használata

A Keras egyik beépített visszahívása a LambdaCallback osztály. Ez a visszahívás olyan függvényt fogad el, amely meghatározza, hogyan viselkedik és mit csinál! Bizonyos értelemben lehetővé teszi, hogy tetszőleges függvényt használjon visszahívásként, így egyéni visszahívásokat hozhat létre.

Az osztály választható paraméterekkel rendelkezik:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Minden paraméter elfogadja egy függvény amelyet az adott modelleseményben hívunk meg. Példaként hívjunk vissza e-mailt, amikor a modell befejezi a képzést:

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

Egyedi visszahívásunk elkészítéséhez a LambdaCallback, csak meg kell valósítanunk azt a függvényt, amelyet el akarunk hívni, csomagold be a lambda függvényt, és adja át a
LambdaCallback osztályt paraméterként.

Visszahívás a modellképzés megjelenítéséhez

Ebben a részben példát mutatunk be egy egyéni visszahívásra, amely animációt készít modellünk teljesítményének javulását az edzés során. Ennek érdekében minden köteg végén tároljuk a naplók értékeit. Ezután a képzési ciklus végén animációt készítünk a segítségével matplotlib.

A megjelenítés javítása érdekében a veszteség és a mérőszámok log skálán kerülnek ábrázolásra:

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

Ugyanazt a modellt fogjuk használni, mint korábban, de több képzési mintával:

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

Eredményünk a mérőszámok és a veszteségfüggvény animációja, ahogyan azok a képzési folyamat során változnak:

Következtetés

Ebben az útmutatóban áttekintettük az egyéni visszahívások megvalósítását a Kerasban.
Két lehetőség van az egyéni visszahívások megvalósítására – az alosztályozáson keresztül keras.callbacks.Callback osztályban, vagy a keras.callbacks.LambdaCallback osztály.

Láttunk egy gyakorlati példát a használatára LambdaCallbacke-mail küldésére a képzési ciklus végén, és egy példa alosztályozására a Callback osztály, amely animációt készít a képzési ciklusról.

Az Althoug Keras számos beépített visszahívással rendelkezik, így az egyéni visszahívás megvalósításának ismerete hasznos lehet konkrétabb alkalmazásoknál.