전형적인 훈련 루프

1. 현재 배치 데이터에 대한 손실 값 계산: 그레이디언트 테이프 정방향 패스 실행
2. 모델 가중치에 대한 손실의 그레이디언트 계산
3. 현재 배치데이터에 모델 가중치 업데이트
   
   

훈련과 추론

• Dropout 및 BatchNormalization 레이어와 같이 일부 케라스 층의 훈련과 추론에서 동작이 다른 경우( model에서 그 기능을 사용하는 경우 )

  • __call__()에서 training 매개변수(boolean)를 제공

  • 정방향 패스에서 케라스 모델을 호출 시, training = True 지정

    • prediction = model(inputs, training=True)

  • 모델 가중치 그레이디언트 추출시,

    • tape.gradients(loss, model.trainable_weights) 사용

  • 층과 모델에는 두 종류의 가중치 존재
    - 훈련가능한 가중치
    - 훈련되지 않는 가중치 : 해당층 정방향 패스 동안 업데이트./ BatchNormalization층은 특성정규화(feature normalization)를 실시간으로 근사하므로 훈련되지 않는 가중치 필요.
    (ex: 얼마나 많은 배치처리를 하였는가 카운트하기 위한 층이 필요하면 이 정보를 훈련되지 않는 가중치에 저장하고 배치마다 값 증가 시킬 수 있음)
    
    
    

사용자 정의 훈련

측정지표의 저수준 사용법

• 측정지표 API (keras.metrics.Metric)의 update_state()를 호출하여 각 배치의 타깃과 예측값을 받음

• result()를 호출하여 현재 지표값을 얻음

• 참고: 딥러닝_사용자 정의 지표, 콜백

🔸 예시

metric = keras.metrics.SparseCategoricalAccuracy()
targets = [0, 1, 2]
predictions = [[1, 0, 0], [0, 1, 0], [0, 0, 1]]
metric.update_state(targets, predictions)
current_result = metric.result()
print(f"result: {current_result:.2f}")

result: 1.00

🔸 모델의 스칼라(손실) 값의 평균 추적:

• keras.metrics.Mean

values = [0, 1, 2, 3, 4]
mean_tracker = keras.metrics.Mean() 

for value in values:
    mean_tracker.update_state(value) 
print(f"Mean of values: {mean_tracker.result():.2f}")

Mean of values: 2.00

🔸 현재 결과를 재설정 하고 싶을 때:

• metric.reset_state()
  

단계별 훈련 루프: 훈련 스텝(fit())

🔸 훈련스텝 함수

model = get_mnist_model()
 
loss_fn = keras.losses.SparseCategoricalCrossentropy() # loss
optimizer = keras.optimizers.RMSprop()                 # optimizer
metrics = [keras.metrics.SparseCategoricalAccuracy()]  # metric

loss_tracking_metric = keras.metrics.Mean()            # 손실평균값 추적


def train_step(inputs, targets):
    with tf.GradientTape() as tape:
        predictions = model(inputs, training=True)
        loss = loss_fn(targets, predictions)                           # 정방향 패스
    
    gradients = tape.gradient(loss, model.trainable_weights)
    optimizer.apply_gradients(zip(gradients, model.trainable_weights)) # 역방향 패스
    
    #---------------------- 측정지표 계산 ---------------------#
    logs = {}
    for metric in metrics: 
        metric.update_state(targets, predictions)
        logs[metric.name] = metric.result()
    
    # ------------------ 손실평균값 계산 -----------------------#
    loss_tracking_metric.update_state(loss) 
    logs["loss"] = loss_tracking_metric.result() 
    
    return logs

🔸 매 에포크 시작 전 지표의 상태 재설정

def reset_metrics():
    for metric in metrics:
        metric.reset_state()
    loss_tracking_metric.reset_state()

🔸 훈련 루프

training_dataset = tf.data.Dataset.from_tensor_slices(
                   (train_images, train_labels))
training_dataset = training_dataset.batch(32)
epochs = 3 

for epoch in range(epochs):
    reset_metrics()
    for inputs_batch, targets_batch in training_dataset:
        logs = train_step(inputs_batch, targets_batch)
    print(f"Results at the end of epoch {epoch}")
    for key, value in logs.items():
        print(f"...{key}: {value:.4f}")

단계별 평가 루프: 평가 스텝(evaluate())

def test_step(inputs, targets):
    predictions = model(inputs, training=False)
    loss = loss_fn(targets, predictions)
  
    logs = {}
    for metric in metrics:
        metric.update_state(targets, predictions)
        logs["val_" + metric.name] = metric.result()
    loss_tracking_metric.update_state(loss)
    logs["val_loss"] = loss_tracking_metric.result()
    return logs
  
val_dataset = tf.data.Dataset.from_tensor_slices((val_images, val_labels))
val_dataset = val_dataset.batch(32)
reset_metrics() 
for inputs_batch, targets_batch in val_dataset:
    logs = test_step(inputs_batch, targets_batch) 
print("Evaluation results:") 
for key, value in logs.items(): 
    print(f"...{key}: {value:.4f}")

🔸 @tf.function

• 코드의 맨 앞라인에 @tf.function(데코레이터)을 추가하면 성능이 높아지게 된다.
• 버그 추적을 쉽게하기 위해선 이 데코레이션을 쓰지말고 즉시 실행 모드(원래의 코드)를 쓰는 게 좋다.
• 코드가 제대로 작동하는게 확실하고 성능을 높이고 싶을 때 @tf.function을 추가

사용자 정의 훈련 스텝 구현

loss_fn = keras.losses.SparseCategoricalCrossentropy()
loss_tracker = keras.metrics.Mean(name="loss")
 
class CustomModel(keras.Model):
    def train_step(self, data):
        inputs, targets = data
        with tf.GradientTape() as tape:
            predictions = self(inputs, training=True)
            loss = loss_fn(targets, predictions)
        gradients = tape.gradient(loss, self.trainable_weights)
        self.optimizer.apply_gradients(zip(gradients, self.trainable_weights))
  
        loss_tracker.update_state(loss)
        return {"loss": loss_tracker.result()}
 
    @property 
    def metrics(self):
        return [loss_tracker]