맞춤형 TensorFlow/Keras 콜백 작성 가이드

개요

학습, 평가 또는 예측 중에 Keras 모델에 특정 동작이 있기를 원한다고 가정합니다. 예를 들어, 모든 훈련 에포크에서 모델을 저장하고 싶을 수 있습니다. 이를 수행하는 한 가지 방법은 콜백을 사용하는 것입니다.

일반적으로 콜백은 어떤 이벤트가 발생할 때 호출되고 다른 함수에 인수로 전달되는 함수입니다. Keras의 경우 교육, 평가 또는 추론 중 모델의 동작을 사용자 지정하는 도구입니다. 일부 응용 프로그램은 로깅, 모델 지속성, 조기 중지 또는 학습률 변경입니다. 이것은 콜백 목록을 인수로 전달하여 수행됩니다. keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

콜백의 몇 가지 일반적인 사용 사례는 학습률 수정, 로깅, 모니터링 및 교육 조기 중지입니다. Keras에는 여러 내장 콜백이 있습니다.
문서에서.

그러나 일부 특정 응용 프로그램에는 사용자 지정 콜백이 필요할 수 있습니다. 예를 들어, 보유 기간 후 Cosine Decay로 학습률 워밍업 구현 현재 내장되어 있지는 않지만 널리 사용되고 스케줄러로 채택되고 있습니다.

콜백 클래스 및 해당 메서드

Keras에는 특정 콜백 클래스가 있습니다. keras.callbacks.Callback, 글로벌, 배치 또는 에포크 수준에서 훈련, 테스트 및 추론 중에 호출할 수 있는 메서드가 있습니다. 하기 위해 사용자 정의 콜백 생성, 우리는 하위 클래스를 만들고 이러한 메서드를 재정의해야 합니다.

XNUMXD덴탈의 keras.callbacks.Callback 클래스에는 세 가지 종류의 메서드가 있습니다.

• 전역 메서드: 시작 또는 끝에서 호출 fit(), evaluate() 와 predict().
• 배치 수준 메서드: 배치 처리의 시작 또는 끝에서 호출됩니다.
• epoch 수준 메서드: 훈련 배치의 시작 또는 끝에서 호출됩니다.

참고 : 각 메소드는 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.

Lamba콜백 사용

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, 호출하려는 함수를 구현하고 이를 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에는 많은 내장 콜백이 있지만 사용자 지정 콜백을 구현하는 방법을 알고 있으면 보다 구체적인 응용 프로그램에 유용할 수 있습니다.