Panduan untuk Menulis Callback TensorFlow/Keras Kustom

Pengantar

Misalkan Anda ingin model Keras Anda memiliki perilaku tertentu selama pelatihan, evaluasi, atau prediksi. Misalnya, Anda mungkin ingin menyimpan model Anda di setiap periode pelatihan. Salah satu cara untuk melakukan ini adalah menggunakan Callback.

Secara umum, Callback adalah fungsi yang dipanggil ketika beberapa peristiwa terjadi, dan diteruskan sebagai argumen ke fungsi lain. Dalam kasus Keras, mereka adalah alat untuk menyesuaikan perilaku model Anda – baik itu selama pelatihan, evaluasi, atau inferensi. Beberapa aplikasi melakukan logging, persistensi model, penghentian lebih awal atau mengubah kecepatan pembelajaran. Ini dilakukan dengan melewatkan daftar Callback sebagai argumen untuk keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Beberapa kasus penggunaan umum untuk panggilan balik adalah memodifikasi kecepatan pembelajaran, pencatatan log, pemantauan, dan penghentian awal pelatihan. Keras memiliki sejumlah panggilan balik bawaan, terperinci
dalam dokumentasi.

Namun, beberapa aplikasi yang lebih spesifik mungkin memerlukan panggilan balik khusus. Contohnya, menerapkan pemanasan Tingkat Pembelajaran dengan Peluruhan Cosinus setelah periode penahanan saat ini tidak built-in, tetapi banyak digunakan dan diadopsi sebagai penjadwal.

Kelas Panggilan Balik dan Metodenya

Keras memiliki kelas panggilan balik tertentu, keras.callbacks.Callback, dengan metode yang dapat dipanggil selama pelatihan, pengujian, dan inferensi pada tingkat global, batch, atau epoch. Untuk buat panggilan balik khusus, kita perlu membuat subkelas dan mengganti metode ini.

Grafik keras.callbacks.Callback kelas memiliki tiga jenis metode:

• metode global: dipanggil di awal atau di akhir fit(), evaluate() dan predict().
• metode tingkat batch: dipanggil di awal atau di akhir pemrosesan batch.
• metode epoch-level: dipanggil di awal atau di akhir batch pelatihan.

Catatan: Setiap metode memiliki akses ke dict yang disebut logs. Kunci dan nilai dari logs bersifat kontekstual – bergantung pada peristiwa yang memanggil metode. Selain itu, kami memiliki akses ke model di dalam setiap metode melalui self.model atribut.

Mari kita lihat tiga contoh panggilan balik khusus – satu untuk pelatihan, satu untuk evaluasi, dan satu untuk prediksi. Masing-masing akan mencetak pada setiap tahap apa yang dilakukan model kita dan log mana yang dapat kita akses. Ini berguna untuk memahami apa yang mungkin dilakukan dengan panggilan balik khusus di setiap tahap.

Mari kita mulai dengan mendefinisikan model mainan:

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

Panggilan Balik Pelatihan Kustom

Panggilan balik pertama kami akan dipanggil selama pelatihan. Mari kita subkelaskan Callback kelas:

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

Jika salah satu dari metode ini tidak diganti – perilaku default akan berlanjut seperti sebelumnya. Dalam contoh kami - kami hanya mencetak log yang tersedia dan tingkat di mana panggilan balik diterapkan, dengan lekukan yang tepat.

Mari kita lihat outputnya:

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}

Perhatikan bahwa kita dapat mengikuti di setiap langkah apa yang dilakukan model, dan metrik mana yang dapat kita akses. Di akhir setiap batch dan epoch, kami memiliki akses ke fungsi kehilangan sampel dan metrik model kami.

Panggilan Balik Evaluasi Kustom

Sekarang, mari kita panggil Model.evaluate() metode. Kami dapat melihat bahwa pada akhir batch kami memiliki akses ke fungsi kerugian dan metrik pada saat itu, dan pada akhir evaluasi kami memiliki akses ke kerugian dan metrik keseluruhan:

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}

Panggilan Balik Prediksi Kustom

Akhirnya, mari kita panggil Model.predict() metode. Perhatikan bahwa pada akhir setiap batch kami memiliki akses ke output prediksi model kami:

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

Dengan ini – Anda dapat menyesuaikan perilaku, mengatur pemantauan atau mengubah proses pelatihan, evaluasi, atau inferensi. Sebuah alternatif untuk sublcassing adalah dengan menggunakan LambdaCallback.

Menggunakan LambaCallback

Salah satu panggilan balik bawaan di Keras adalah LambdaCallback kelas. Panggilan balik ini menerima fungsi yang mendefinisikan bagaimana perilakunya dan apa fungsinya! Dalam arti tertentu, ini memungkinkan Anda untuk menggunakan fungsi arbitrer apa pun sebagai panggilan balik, sehingga memungkinkan Anda membuat panggilan balik khusus.

Kelas memiliki parameter opsional:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Setiap parameter menerima sebuah fungsi yang disebut dalam acara model masing-masing. Sebagai contoh, mari kita membuat panggilan balik untuk mengirim email ketika model selesai pelatihan:

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

Untuk membuat panggilan balik khusus kami menggunakan LambdaCallback, kita hanya perlu mengimplementasikan fungsi yang ingin kita panggil, bungkus sebagai lambda berfungsi dan meneruskannya ke
LambdaCallback kelas sebagai parameter.

Panggilan Balik untuk Memvisualisasikan Pelatihan Model

Di bagian ini, kami akan memberikan contoh panggilan balik khusus yang membuat animasi kinerja model kami meningkat selama pelatihan. Untuk melakukan ini, kami menyimpan nilai log di akhir setiap batch. Kemudian, di akhir loop pelatihan, kami membuat animasi menggunakan matplotlib.

Untuk meningkatkan visualisasi, kerugian dan metrik akan diplot dalam skala log:

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

Kami akan menggunakan model yang sama seperti sebelumnya, tetapi dengan lebih banyak sampel pelatihan:

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

Output kami adalah animasi metrik dan fungsi kerugian saat mereka berubah melalui proses pelatihan:

Kesimpulan

Dalam panduan ini, kita telah melihat implementasi callback kustom di Keras.
Ada dua opsi untuk menerapkan panggilan balik khusus – melalui subklasifikasi keras.callbacks.Callback kelas, atau dengan menggunakan keras.callbacks.LambdaCallback kelas.

Kami telah melihat satu contoh praktis menggunakan LambdaCallbackuntuk mengirim email di akhir loop pelatihan, dan salah satu contoh subclassing the Callback class yang membuat animasi dari loop pelatihan.

Meskipun Keras memiliki banyak panggilan balik bawaan, mengetahui cara menerapkan panggilan balik khusus dapat berguna untuk aplikasi yang lebih spesifik.