Vodnik za pisanje povratnih klicev TensorFlow/Keras po meri

Predstavitev

Recimo, da želite, da ima vaš model Keras določeno obnašanje med usposabljanjem, ocenjevanjem ali napovedovanjem. Na primer, morda boste želeli shraniti svoj model v vsaki vadbeni dobi. Eden od načinov za to je uporaba povratnih klicev.

Na splošno so povratni klici funkcije, ki se kličejo, ko se zgodi nek dogodek, in se kot argumenti posredujejo drugim funkcijam. V primeru Kerasa so orodje za prilagoditev vedenja vašega modela – med usposabljanjem, vrednotenjem ali sklepanjem. Nekatere aplikacije so beleženje, obstojnost modela, zgodnja ustavitev ali spreminjanje stopnje učenja. To se naredi s posredovanjem seznama povratnih klicev kot argumentov za keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Nekateri običajni primeri uporabe povratnih klicev so spreminjanje stopnje učenja, beleženje, spremljanje in zgodnja ustavitev usposabljanja. Keras ima številne vgrajene povratne klice, podrobno
v dokumentaciji.

Vendar pa lahko nekatere bolj specifične aplikacije zahtevajo povratni klic po meri. Na primer, izvajanje ogrevanja stopnje učenja s kosinusnim upadanjem po obdobju zadrževanja trenutno ni vgrajen, vendar se pogosto uporablja in je sprejet kot razporejevalnik.

Razred povratnega klica in njegove metode

Keras ima poseben razred povratnega klica, keras.callbacks.Callback, z metodami, ki jih je mogoče priklicati med usposabljanjem, testiranjem in sklepanjem na globalni, paketni ali epohalni ravni. Da bi ustvarite povratne klice po meri, moramo ustvariti podrazred in preglasiti te metode.

O keras.callbacks.Callback ima tri vrste metod:

• globalne metode: kličejo se na začetku ali na koncu fit(), evaluate() in predict().
• metode na ravni serije: kličejo se na začetku ali na koncu obdelave serije.
• metode na ravni epohe: kličejo se na začetku ali na koncu učnega sklopa.

Opomba: Vsaka metoda ima dostop do imenovanega dict logs. Ključi in vrednote logs so kontekstualni – odvisni so od dogodka, ki kliče metodo. Poleg tega imamo dostop do modela znotraj vsake metode prek self.model atribut.

Oglejmo si tri primere povratnih klicev po meri – enega za usposabljanje, enega za ocenjevanje in enega za napovedovanje. Vsak bo na vsaki stopnji natisnil, kaj počne naš model in do katerih dnevnikov imamo dostop. To je koristno za razumevanje, kaj je mogoče narediti s povratnimi klici po meri na vsaki stopnji.

Začnimo z opredelitvijo modela igrače:

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

Povratni klic za usposabljanje po meri

Naš prvi povratni klic je treba poklicati med treningom. Razvrstimo v podrazred Callback razred:

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

Če kateri koli od teh načinov ni preglasen – se bo privzeto vedenje nadaljevalo kot prej. V našem primeru preprosto natisnemo razpoložljive dnevnike in raven, na kateri je uporabljen povratni klic, z ustreznim zamikom.

Oglejmo si rezultate:

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}

Upoštevajte, da lahko na vsakem koraku spremljamo, kaj počne model in do katerih meritev imamo dostop. Na koncu vsake serije in epohe imamo dostop do funkcije izgube v vzorcu in meritev našega modela.

Povratni klic ocene po meri

Zdaj pa pokličimo Model.evaluate() metoda. Vidimo lahko, da imamo na koncu paketa dostop do funkcije izgube in metrike v tem času, na koncu vrednotenja pa imamo dostop do celotne izgube in metrike:

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}

Povratni klic s predvidevanjem po meri

Končno pokličimo Model.predict() metoda. Upoštevajte, da imamo na koncu vsake serije dostop do predvidenih rezultatov našega modela:

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

S temi – lahko prilagodite vedenje, nastavite spremljanje ali kako drugače spremenite procese usposabljanja, vrednotenja ali sklepanja. Alternativa podohišju je uporaba LambdaCallback.

Uporaba LambaCallback

Eden od vgrajenih povratnih klicev v Keras je LambdaCallback razred. Ta povratni klic sprejme funkcijo, ki določa, kako se obnaša in kaj počne! V nekem smislu vam omogoča, da uporabite katero koli poljubno funkcijo kot povratni klic, kar vam omogoča ustvarjanje prilagojenih povratnih klicev.

Razred ima neobvezne parametre:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Vsak parameter sprejme funkcija ki se kliče v zadevnem dogodku modela. Kot primer naredimo povratni klic za pošiljanje e-pošte, ko model konča usposabljanje:

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

Za povratni klic po meri z uporabo LambdaCallback, samo implementirati moramo funkcijo, ki jo želimo poklicati, jo zaviti kot a lambda funkcijo in jo posredujte v
LambdaCallback razred kot parameter.

Povratni klic za vizualizacijo modela usposabljanja

V tem razdelku bomo podali primer povratnega klica po meri, ki naredi animacijo izboljšanja zmogljivosti našega modela med usposabljanjem. Da bi to naredili, shranimo vrednosti dnevnikov na koncu vsake serije. Nato na koncu vadbene zanke ustvarimo animacijo z uporabo matplotlib.

Za izboljšanje vizualizacije bodo izguba in meritve narisane v logaritmičnem merilu:

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

Uporabili bomo enak model kot prej, vendar z več vzorci usposabljanja:

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

Naš rezultat je animacija meritev in funkcije izgube, ko se spreminjajo skozi proces usposabljanja:

zaključek

V tem priročniku smo si ogledali izvedbo povratnih klicev po meri v Kerasu.
Obstajata dve možnosti za implementacijo povratnih klicev po meri – prek podrazreda keras.callbacks.Callback razreda ali z uporabo keras.callbacks.LambdaCallback razred.

Videli smo en praktični primer uporabe LambdaCallbackza pošiljanje e-pošte na koncu učne zanke in en primer podrazreda Callback razred, ki ustvari animacijo vadbene zanke.

Čeprav ima Keras veliko vgrajenih povratnih klicev, je lahko znanje, kako implementirati povratni klic po meri, koristno za bolj specifične aplikacije.