カスタム TensorFlow/Keras コールバックの記述ガイド

タイムスタンプ：14年2022月6日30:XNUMX AM

概要

トレーニング、評価、または予測中に Keras モデルに特定の動作をさせたいとします。 たとえば、トレーニング エポックごとにモデルを保存したい場合があります。 これを行う XNUMX つの方法は、コールバックを使用することです。

一般に、コールバックは、何らかのイベントが発生したときに呼び出され、引数として他の関数に渡される関数です。 Keras の場合、それらはモデルの動作をカスタマイズするためのツールです – トレーニング中、評価中、または推論中です。 一部のアプリケーションは、ログ記録、モデルの永続性、学習率の早期停止または変更です。 これは、コールバックのリストを引数として渡すことによって行われます keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

コールバックの一般的なユース ケースには、学習率の変更、ログ記録、監視、およびトレーニングの早期停止があります。 Keras には多数の組み込みコールバックがあり、詳細を説明しています。
ドキュメントで.

ただし、一部のより具体的なアプリケーションでは、カスタム コールバックが必要になる場合があります。 例えば、 保持期間後にコサイン減衰を使用して学習率ウォームアップを実装する 現在は組み込まれていませんが、スケジューラとして広く使用され採用されています。

コールバック クラスとそのメソッド

Keras には特定のコールバック クラスがあり、 keras.callbacks.Callback、グローバル、バッチ、またはエポック レベルでのトレーニング、テスト、および推論中に呼び出すことができるメソッドを使用します。 そうするには カスタム コールバックを作成する、サブクラスを作成してこれらのメソッドをオーバーライドする必要があります。

　 keras.callbacks.Callback クラスには XNUMX 種類のメソッドがあります。

  • グローバル メソッド: の最初または最後に呼び出されます fit(), evaluate() & predict().
  • バッチレベルのメソッド: バッチ処理の開始時または終了時に呼び出されます。
  • エポックレベルのメソッド: トレーニング バッチの開始時または終了時に呼び出されます。

注： 各メソッドは、呼ばれる辞書にアクセスできます logs. のキーと値 logs コンテキストに依存します – メソッドを呼び出すイベントに依存します。 さらに、各メソッド内のモデルには、 self.model 属性。

トレーニング用、評価用、予測用の XNUMX つのカスタム コールバックの例を見てみましょう。 それぞれが各段階で、モデルが何をしているか、どのログにアクセスできるかを出力します。 これは、各段階でカスタム コールバックを使用して何ができるかを理解するのに役立ちます。

おもちゃのモデルを定義することから始めましょう。

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

ベストプラクティス、業界で認められた標準、および含まれているチートシートを含む、Gitを学習するための実践的で実用的なガイドを確認してください。 グーグルGitコマンドを停止し、実際に 学ぶ それ！

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 の組み込みコールバックの XNUMX つは、 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、呼び出したい関数を実装する必要があるだけで、それを 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()],
)

出力は、トレーニング プロセスを通じて変化するメトリクスと損失関数のアニメーションです。

お使いのブラウザはHTMLビデオをサポートしていません。

まとめ

このガイドでは、Keras でのカスタム コールバックの実装を見てきました。
カスタム コールバックを実装するには XNUMX つのオプションがあります。 keras.callbacks.Callback クラス、または keras.callbacks.LambdaCallback とに提供されます。

を使用した実用的な例を XNUMX つ見てきました。 LambdaCallbackトレーニング ループの最後に電子メールを送信するため、および XNUMX つの例をサブクラス化 Callback トレーニング ループのアニメーションを作成するクラス。

Keras には多くの組み込みコールバックがありますが、カスタム コールバックを実装する方法を知っていると、より具体的なアプリケーションに役立ちます。

タイムスタンプ：

より多くの スタックアバス