[TensorFlow] Callback 사용하기 - 커스텀 콜백 / 모델 학습 / 평가

🙋‍♀️ Callback은 모델 학습 중 호출할 수 있는 기능을 의미한다.

🙋‍♀️ tf.keras.callbacks에는 함수가 구현되어 있으며, 직접 필요한 기능을 구현해 사용하는 것도 가능하다.

---

Callback Classes

기본 Callback 관련 클래스
Callback	베이스 클래스로부터 새로운 콜백을 만들 수 있는 클래스
LamdaCallback	간단한 커스텀 콜백을 on-the-fly로 생성할 수 있는 콜백
CallbackList	콜백의 리스트를 추출할 수 있는 컨테이너
Logging 관련 클래스
BaseLogger	에포크별로 metric의 평균을 축적하는 콜백
CSVLogger	결과를 CSV 파일로 스트리밍하는 콜백
ProgbarLogger	metric을 stdout으로 프린트하는 콜백
RemoteMonitor	이벤트를 서버로 스트리밍하는 콜백
TensorBoard	TensorBoard에서 시각화가 가능하도록 함
History	History object로 이벤트를 기록하는 콜백
학습 관련 클래스
EarlyStopping	metric에 개선이 없을 때, 학습을 중단할 수 있는 콜백
LearningRateScheduler	학습율 스케쥴링에 사용
ReduceLROnPlateau	metric에 개선이 없을 때 learning rate를 줄이는 콜백
TerminateOnNaN	NaN loss가 발생하면 학습을 중단하는 콜백
ModelCheckpoint	Keras 모델, 혹은 모델 가중치를 일정 빈도에 따라 저장하는 콜백

---

콜백 함수 작성하기

tf.keras.callbacks.Callback을 상속하여 사용자 정의 콜백을 정의할 수 있다.

다음과 같이 정해진 시점마다 사용자가 정의한 기능을 수행하도록 함수를 구현할 수 있다.

 

   > fit / evaluate / predict가 시작하거나 끝날 때

      - on_train_begin / on_train_end

      - on_test_begin / on_test_end

      - on_predict_begin / on_predict_end

 

   > 각 에포크가 시작하거나 끝날 때

      - on_epoch_begin / on_epoch_end

 

   > 각 훈련 배치가 시작하거나 끝날 때

      - on_train_batch_begin / on_train_batch_end

 

   > 각 평가(테스트) 배치가 시작하거나 끝날 때

      - on_test_batch_begin / on_test_batch_end

 

   > 각 추론(예측) 배치가 시작하거나 끝날 때

      - on_predict_batch_begin / on_predict_batch_end

---

self.model 속성 사용하기

콜백 매서드 중 하나가 호출될 때 현재 훈련/평가/추론 라운드와 연결된 모델 (self.model)에 접근할 수 있다.

콜백에서 self.model로 수행할 수 있는 연산 :

   - 훈련을 즉시 중단 ( self.model.stop_training = True를 설정)

   - self.model.optimizer.learning_rate와 같은 옵티마이저의 하이퍼파라미터 변경 (self.model.optimizer 사용)

   - 주기적으로 모델 저장

   - 각 에포크가 끝날 때 몇 가지 테스트 샘플에 대해 model.predict()의 출력을 기록

   - 각 에포크가 끝날 때 중간 특성의 시각화를 추출하여 시간이 지남에 따라 학습하는 내용 모니터링

 

Ex1) 손실함수에 따라 모델 학습 중지하기

   - self.model.stop_training 속성 설정

   - 이 기능은 tf.keras.callbacks.EarlyStopping 에서 더 일반적인 기능을 사용할 수 있음

 

import numpy as np

class EarlyStoppingAtMinLoss(keras.callbacks.Callback):
    """Stop training when the loss is at its min, i.e. the loss stops decreasing.

  Arguments:
      patience: Number of epochs to wait after min has been hit. After this
      number of no improvement, training stops.
  """

    def __init__(self, patience=0): # 몇 번 loss 증가를 용인할지를 인풋으로 받음
        super(EarlyStoppingAtMinLoss, self).__init__()
        self.patience = patience
        # best_weights to store the weights at which the minimum loss occurs.
        self.best_weights = None

    def on_train_begin(self, logs=None): # 학습 시작시 -> EarlyStopping에 필요한 변수들 초기화
        # The number of epoch it has waited when loss is no longer minimum.
        self.wait = 0
        # The epoch the training stops at.
        self.stopped_epoch = 0
        # Initialize the best as infinity.
        self.best = np.Inf

    def on_epoch_end(self, epoch, logs=None): # 에포크가 끝날 때마다 loss를 체크하고, best weight 업데이트
        current = logs.get("loss")
        if np.less(current, self.best):
            self.best = current
            self.wait = 0
            # Record the best weights if current results is better (less).
            self.best_weights = self.model.get_weights() # model에 접근해 weight를 가지고 올 수 있음
            model.save('best_model') # 모델 저장하기
        else:
            self.wait += 1
            if self.wait >= self.patience:
                self.stopped_epoch = epoch
                self.model.stop_training = True
                print("Restoring model weights from the end of the best epoch.")
                self.model.set_weights(self.best_weights)

    def on_train_end(self, logs=None):
        if self.stopped_epoch > 0:
            print("Epoch %05d: early stopping" % (self.stopped_epoch + 1))

 

 

model.fit 에서 위의 콜백을 적용

model = get_model()
early_stopping_custom_callback = EarlyStoppingAtMinLoss()
model.fit(
    x_train,
    y_train,
    callbacks=[early_stopping_custom_callback],
)

👉 early_stopping_custom_callback.best 에 minimum loss가 저장되어 있다.

👉 early_stopping_custom_callback.best_weights에 best 모델의 파라미터가 저장되어 있다.

 

Ex2) 학습율 스케줄링하기

콜백 기능을 사용하여 optimizer의 학습율을 동적으로 변경할 수 있다.

이 콜백은 callbacks.LearningRateScheduler에서 일반적인 기능을 사용할 수 있다.

 

아래 코드는 에포크마다 지정된 learning rate를 사용하도록 커스텀 콜백을 작성한 예시이다.

 

class CustomLearningRateScheduler(keras.callbacks.Callback):
    """Learning rate scheduler which sets the learning rate according to schedule.

  Arguments:
      schedule: 
          인풋 - 현재 epoch(정수, 0부터 인덱싱 시작) & 현재 학습율
          새로운 learning rate(float)를 계산해 아웃풋으로 리턴하는 함수
  """

    def __init__(self, schedule):
        super(CustomLearningRateScheduler, self).__init__()
        self.schedule = schedule

    def on_epoch_begin(self, epoch, logs=None):
        if not hasattr(self.model.optimizer, "lr"):
            raise ValueError('Optimizer must have a "lr" attribute.')
        # 모델 옵티마이저에서 현재 learning rate를 가지고 옴
        lr = float(tf.keras.backend.get_value(self.model.optimizer.learning_rate))

        # schedule 함수를 호출하여 스케줄링된 learning rate를 받아옴
        scheduled_lr = self.schedule(epoch, lr)

        # 이번에 계산된 학습률을 optimizer에 설정해 줌
        tf.keras.backend.set_value(self.model.optimizer.lr, scheduled_lr)
        print("\nEpoch %05d: Learning rate is %6.4f." % (epoch, scheduled_lr))


LR_SCHEDULE = [
    # (epoch to start, learning rate) tuples
    (3, 0.05),
    (6, 0.01),
    (9, 0.005),
    (12, 0.001),
]


def lr_schedule(epoch, lr):
    """Helper function to retrieve the scheduled learning rate based on epoch."""
    if epoch < LR_SCHEDULE[0][0] or epoch > LR_SCHEDULE[-1][0]:
        return lr
    for i in range(len(LR_SCHEDULE)):
        if epoch == LR_SCHEDULE[i][0]:
            return LR_SCHEDULE[i][1]
    return lr

 

 

 

---

참고 자료

[TensorFlow API] 커스텀 콜백 작성하기

---