<jupyter_start><jupyter_text>특정 시점에 학습률을 조정하는 커스텀 케라스 콜백<jupyter_code>from tensorflow.keras.callbacks import Callback
from tensorflow.keras import backend as K

# Callback을 상속받아 Custom Callback을 정의합니다.
class CustomLearningLateCallback(Callback):
    def __init__(self):
        pass
    
    # 0.1배 만큼 학습률을 감소시킵니다.
    def down_lr(self, current_lr):
        return current_lr * 0.1
    
    # 기점 예시입니다.
    # 이 예제에서는 사용하지 않습니다.
    def on_train_begin(self, logs = None):
        pass
    
    def on_train_end(self, logs = None):
        pass
    
    def on_train_batch_begin(self, batch, logs = None):
        pass
    
    def on_train_batch_end(self, batch, logs = None):
        pass
    
    def on_epoch_begin(self, epoch, logs = None):
        current_lr = self.model.optimizer.lr
        
        if(epoch > 1):
            # 5, 8, 10번째마다 학습률을 감소시킬 것입니다.
            if((epoch == 4) or (epoch == 7) or (epoch == 9)):
                current_lr = self.down_lr(current_lr)
                
                # 감소된 학습률을 현재 모델 옵티마이저의 학습률로 설정합니다.
                K.set_value(self.model.optimizer.lr, current_lr)
                print('\nEpoch %03d: learning rate change! %s.' % (epoch + 1, current_lr.numpy()))
                
    def on_epoch_end(self, epoch, logs = None):
        pass<jupyter_output><empty_output><jupyter_text>커스텀 케라스 콜백을 사용하여 모델 학습시키기<jupyter_code>from tensorflow.keras.datasets import mnist

(x_train, y_train), (x_test, y_test) = mnist.load_data(path='mnist.npz')

from sklearn.model_selection import train_test_split

x_train, x_val, y_train, y_val = train_test_split(x_train, y_train, 
                                                  test_size = 0.3, random_state = 777)

x_train = (x_train.reshape(-1, 28, 28, 1)) / 255
x_val = (x_val.reshape(-1, 28, 28, 1)) / 255
x_test = (x_test.reshape(-1, 28, 28, 1)) / 255

from tensorflow.keras.utils import to_categorical

y_train = to_categorical(y_train)
y_val = to_categorical(y_val)
y_test = to_categorical(y_test)
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Conv2D, MaxPooling2D, GlobalAveragePooling2D, Dense
from tensorflow.keras.layers import Input

inputs = Input(shape = (28, 28, 1))

x = Conv2D(32, (3, 3), activation = 'relu')(inputs)
x = Conv2D(32, (3, 3), activation = 'relu')(x)
x = MaxPooling2D(strides = 2)(x)

x = GlobalAveragePooling2D()(x)
x = Dense(10, activation = 'softmax')(x)

model = Model(inputs = inputs, outputs = x)

# 정의한 손실 함수를 사용합니다.
model.compile(optimizer = 'adam', 
              loss = 'categorical_crossentropy',
              metrics = ['acc'])

model.fit(x_train, y_train,
         batch_size = 32,
         validation_data = (x_val, y_val),
         epochs = 10,
         callbacks = [CustomLearningLateCallback()])<jupyter_output><empty_output><jupyter_text>(책 내용 X) CosineAnnealing Learning Rate + 커스텀 케라스 콜백과 사용 방법은 동일합니다.<jupyter_code>class CosineAnnealingLearningRateSchedule(Callback):
    def __init__(self, n_epochs, init_lr, T_mult = 1, eta_min = 0,restart_decay = 0, verbose = 0):
        self.T_max = n_epochs
        self.T_mult = T_mult
        self.cycle_cnt = 0
        self.restart_decay = restart_decay
        self.init_lr = init_lr
        self.eta_min = eta_min
        self.lrates = list()
  # caculate learning rate for an epoch

    def cosine_annealing(self, epoch):
        lr = self.eta_min + (self.init_lr - self.eta_min) * (1 + math.cos(math.pi * (epoch / self.T_max))) / 2
        if(epoch == self.T_max):
            self.cycle_cnt += 1
            self.T_max = self.T_mult * self.T_max

        if(self.restart_decay >0):
            self.init_lr *= self.restart_decay
            print('change init learning rate {}'.format(self.init_lr))

    return lr
  # calculate and set learning rate at the start of the epoch

    def on_epoch_begin(self, epoch, logs = None):
        lr = self.cosine_annealing(epoch)
        print('\nEpoch %05d: CosineAnnealingScheduler setting learng rate to %s.' % (epoch + 1, lr))
        # set learning rate
        backend.set_value(self.model.optimizer.lr, lr)
        # log value
        self.lrates.append(lr)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup