Посібник із написання власних зворотних викликів TensorFlow/Keras

Вступ

Припустімо, ви хочете, щоб ваша модель Keras мала певну поведінку під час навчання, оцінювання чи прогнозування. Наприклад, ви можете зберігати свою модель у кожній епосі навчання. Одним із способів зробити це є використання зворотних викликів.

Загалом, зворотні виклики – це функції, які викликаються, коли відбувається якась подія, і передаються як аргументи іншим функціям. У випадку з Keras вони є інструментом для налаштування поведінки вашої моделі – під час навчання, оцінки чи висновку. Деякі програми включають журналювання, збереження моделі, ранню зупинку або зміну швидкості навчання. Це робиться шляхом передачі списку зворотних викликів як аргументів keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

Деякі поширені випадки використання зворотних викликів – це зміна швидкості навчання, ведення журналу, моніторинг і рання зупинка навчання. Keras має низку вбудованих зворотних викликів, деталізовано
в документації.

Однак для деяких більш специфічних програм може знадобитися спеціальний зворотний виклик. Наприклад, впровадження розігріву швидкості навчання з косинусним розпадом після періоду утримування наразі не є вбудованим, але широко використовується та приймається як планувальник.

Клас зворотного виклику та його методи

Keras має певний клас зворотного виклику, keras.callbacks.Callback, з методами, які можна викликати під час навчання, тестування та логічного висновку на глобальному, пакетному або епохальному рівні. Щоб створювати власні зворотні виклики, нам потрібно створити підклас і перевизначити ці методи.

Команда keras.callbacks.Callback клас має три види методів:

• глобальні методи: викликаються на початку або в кінці fit(), evaluate() та predict().
• методи на рівні партії: викликаються на початку або в кінці обробки партії.
• методи епохального рівня: викликаються на початку або в кінці навчальної партії.

Примітка: Кожен метод має доступ до викликаного dict logs. Ключі та значення logs є контекстними – вони залежать від події, яка викликає метод. Крім того, ми маємо доступ до моделі всередині кожного методу через self.model атрибут.

Давайте розглянемо три приклади спеціальних зворотних викликів – один для навчання, один для оцінки та один для прогнозування. Кожна з них на кожному етапі друкуватиме, що робить наша модель і до яких журналів ми маємо доступ. Це корисно для розуміння того, що можна робити з власними зворотними викликами на кожному етапі.

Почнемо з визначення моделі іграшки:

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

Зворотний дзвінок спеціального навчання

Наш перший зворотній дзвінок має бути викликаний під час навчання. Розділимо підклас Callback клас:

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

Якщо будь-який із цих методів не перевизначено, поведінка за замовчуванням продовжуватиметься, як і раніше. У нашому прикладі ми просто роздруковуємо доступні журнали та рівень, на якому застосовано зворотний виклик, із належним відступом.

Давайте подивимося на результати:

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}

Зауважте, що на кожному кроці ми можемо стежити за тим, що робить модель і до яких показників ми маємо доступ. Наприкінці кожної партії та епохи ми маємо доступ до функції втрат у вибірці та показників нашої моделі.

Зворотний виклик індивідуальної оцінки

Тепер давайте подзвонимо Model.evaluate() метод. Ми бачимо, що наприкінці партії ми маємо доступ до функції втрат і показників на той час, а наприкінці оцінки ми маємо доступ до загальних втрат і показників:

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}

Зворотний дзвінок спеціального передбачення

Нарешті, давайте подзвонимо Model.predict() метод. Зауважте, що наприкінці кожної партії ми маємо доступ до прогнозованих результатів нашої моделі:

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

За допомогою них ви можете налаштувати поведінку, налаштувати моніторинг або іншим чином змінити процеси навчання, оцінки чи висновку. Альтернативою підкассу є використання LambdaCallback.

Використання LambaCallback

Одним із вбудованих зворотних викликів у Keras є LambdaCallback клас. Цей зворотній виклик приймає функцію, яка визначає, як він поводиться і що він робить! У певному сенсі це дозволяє вам використовувати будь-яку довільну функцію як зворотний виклик, таким чином дозволяючи вам створювати спеціальні зворотні виклики.

Клас має додаткові параметри:
-on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

Кожен параметр приймається функція яка викликається у відповідній події моделі. Як приклад, давайте зробимо зворотній дзвінок, щоб надіслати електронний лист, коли модель закінчить навчання:

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

Щоб зробити наш власний зворотний виклик за допомогою LambdaCallback, нам просто потрібно реалізувати функцію, яку ми хочемо викликати, обгорнути її як a lambda і передати її в
LambdaCallback клас як параметр.

Зворотний дзвінок для візуалізації моделі навчання

У цьому розділі ми наведемо приклад власного зворотного виклику, який робить анімацію покращення продуктивності нашої моделі під час навчання. Для цього ми зберігаємо значення журналів у кінці кожної партії. Потім, наприкінці навчального циклу, ми створюємо анімацію за допомогою matplotlib.

Щоб покращити візуалізацію, втрати та показники будуть нанесені в логарифмічному масштабі:

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

Ми будемо використовувати ту саму модель, що й раніше, але з більшою кількістю навчальних зразків:

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

Наш результат — це анімація показників і функції втрат, коли вони змінюються в процесі навчання:

Висновок

У цьому посібнику ми розглянули реалізацію власних зворотних викликів у Keras.
Існує два варіанти реалізації спеціальних зворотних викликів – через підкласи keras.callbacks.Callback або за допомогою keras.callbacks.LambdaCallback клас.

Ми бачили один практичний приклад використання LambdaCallbackдля надсилання електронного листа в кінці навчального циклу та один приклад підкласу Callback клас, який створює анімацію навчального циклу.

Незважаючи на те, що Keras має багато вбудованих зворотних викликів, знання того, як реалізувати власний зворотний виклик, може бути корисним для більш конкретних програм.