Ch7

7.1 다양한 워크플로

• 케라스 API 설계는 복잡성의 단계적 공개(progressive disclosure) 원칙을 따름
• 시작은 쉽게 하고, 필요할 때 단계마다 점진적으로 학습하여 아주 복잡한 경우를 처리할 수 있음
• 간단한 문제는 누구나 쉽게 처리할 수 있어야 하지만 어떤 고급 워크플로도 가능해야 함
• 원하는 작업이 얼마나 복잡하고 드문 경우인지에 상관없이 이를 달성하기 위한 명백한 방법이 있어야 함
• 이런 방법은 간단한 워크플로에서 배운 것을 기반으로 함
• 이는 초보자에서 전문가로 성장하면서 동일한 도구를 그대로 사용할 수 있다는 의미
• 다만 사용 방법만 달라질 뿐
• 케라스를 사용하는 '올바른' 한 가지 방법이 있는 것이 아님
• 케라스는 매우 간단한 것부터 매우 유연한 것까지 다양한 워크플로를 제공
• 케라스 모델을 만드는 방법은 여러 가지고, 모델을 훈련하는 방법도 여러 가지
• 이를 통해 다양한 요구 사항을 충족시킬 수 있음
• 이런 모든 워크플로는 동일하게 Layer와 Model 같은 API를 기반으로 하기 때문에 한 워크플로의 구성 요소를 다른 워크플로에서 사용할 수 있음
• 즉, 워크플로 간에 서로 호출될 수 있음

7.2 케라스 모델을 만드는 여러 방법

• 케라스에서 모델을 만드는 API는 세 가지
  • Sequential 모델이 가장 시작하기 쉬운 API
    • 기본적으로 하나의 파이썬 리스트
    • 단순히 층을 쌓을 수만 있음
  • 함수형 API(Functional API)는 그래프 같은 모델 구조를 주로 다룸
    • 이 API는 사용성과 유연성 사이의 적절한 중간 지점에 해당
    • 가장 널리 사용되는 모델 구축 API
  • Model 서브클래싱(subclassing)은 모든 것을 밑바닥부터 직접 만들 수 있는 저수준 방법
    • 모든 상세한 내용을 완전히 제어하고 싶은 경우에 적합
    • 여러 가지 케라스 내장 기능을 사용하지 못하기 때문에 실수가 발생할 위험이 많음

모델 구출 복잡성의 단계적 공개

Sequential 모델

• Sequential 클래스

from tensorflow import keras
from tensorflow.keras import layers

model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(10, activation="softmax")
])

• 동일한 모델을 add() 메서드를 통해 점진적으로 만들 수 있음
• 이 메서드는 파이썬 리스트의 append() 메서드와 비슷함
• 점진적으로 Sequential 모델 만들기

model = keras.Sequential()
model.add(layers.Dense(64, activation="relu"))
model.add(layers.Dense(10, activation="softmax"))

• 층은 처음 호출될 때 만들어진다는 것을 보았음(즉, 가중치를 만듦)
• 층의 가중치 크기가 입력 크기에 따라 달라지기 때문임
• 즉, 입력 크기를 알기 전까지 가중치를 만들 수 없음
• 앞의 Sequential 모델은 어떤 가중치도 가지고 있지 않음
• 가중치를 생성하려면 어떤 데이터로 호출하거나 입력 크기를 지정하여 build() 메서드를 호출
• build() 메서드가 호출 전의 모델은 가중치가 없다

model.weights # 이때는 아직 모델의 build()메서드가 호출되지 않았다.

• 가중치를 만들기 위해 모델을 호출

model.build(input_shape=(None, 3)) # 모델의 build() 메서드 호출: 이제 모델의 크기가 (3,)인 샘플을 기대한다. 입력 크기의 None은 어떤 배치 크기도 가능하다는 의미
model.weights

• build() 메서드가 호출된 후 디버깅에 유용한 summary() 메서드를 사용하여 모델 구조를 출력할 수 있음
• summary() 메소드

model.summary()

summary() 메서드

• 여기에서 보듯이 이 모델의 이름은 'sequential_1'
• 케라스에서는 모델과 층을 포함해서 모든 것에 이름을 지정할 수 있음
• name 매개변수로 모델과 층에 이름 지정하기

model = keras.Sequential(name="my_example_model")
model.add(layers.Dense(64, activation="relu", name="my_first_layer"))
model.add(layers.Dense(10, activation="softmax", name="my_last_layer"))
model.build((None, 3))
model.summary()

name 매개변수로 모델과 층에 이름 지정하기

• Sequential 모델을 점진적으로 만들 때 층을 추가하고 난 후 summary()메서드를 호출하여 현재 모델을 확인할 수 있으면 유용
• 모델의 build() 메서드를 호출하기 전까지는 summary() 메서드를 호출할 수 없음!
• Sequential 모델의 가중치를 바로 생성하는 방법이 있음
• 모델의 입력 크기를 미리 지정하면 됨
• 이를 위해 Input 클래스를 사용
• 모델의 입력 크기를 미리 지정하기

model = keras.Sequential()
model.add(keras.Input(shape=(3,)))
model.add(layers.Dense(64, activation="relu"))

• 이제 summary() 메서드를 사용하여 층을 추가함에 따라 모델의 출력 크기 변화를 확인할 수 있음

함수형 API

• Sequential 모델은 사용하기 쉽지만 적용할 수 있는 곳이 극히 제한적
• 하나의 입력과 하나의 출력을 가지며 순서대로 층을 쌓은 모델만 표현할 수 있음
• 실제로 다중 입력(예를 들어 이미지와 이미지의 메타데이터), 다중 출력(예를 들어 데이터에 대해 여러 가지 항목을 예측하는 모델) 또는 비선형적인 구조를 가진 모델을 자주 만날 수 있음
• 이런 경우에는 함수형 API를 사용한 모델을 만듦
• 실전에서 이런 케라스 모델을 가장 흔하게 만날 수 있음
• 이 방식은 레고 블록을 가지고 노는 것처럼 재미있고 강력함

간단한 예제

• 이전 절에서 했던 것처럼 2개의 층을 쌓아 보자
• 이 모델을 함수형 API 버전으로 만들면 다음과 같음
• 2개의 Dense 층을 가진 간단한 함수형 모델

inputs = keras.Input(shape=(3,), name="my_input")
features = layers.Dense(64, activation="relu")(inputs)
outputs = layers.Dense(10, activation="softmax")(features)
model = keras.Model(inputs=inputs, outputs=outputs)

• 단계별로 살펴보면 다음과 같음
• Input 클래스 객체를 정의하는 것으로 시작(다른 것과 마찬가지로 입력에도 이름을 지정할 수 있음)

inputs = keras.Input(shape=(3,), name="my_input")

• inputs 객체는 모델이 처리할 데이터의 크기와 dtype에 대한 정보를 가지고 있음

• 이런 객체를 심볼릭 텐서(symbolic tensor)라고 부름
• 실제 데이터를 가지고 있지 않지만 사용할 때 모델이 보게 될 데이터 텐서의 사양이 인코딩되어 있음
• 즉, 미래의 데이터 텐서를 나타냄
• 그 다음 층을 만들고 전의 입력으로 호출

features = layers.Dense(64, activation="relu")(inputs)

• 모든 케라스 층은 실제 데이터 텐서나 심볼릭 텐서로 호출할 수 있음
• 후자의 경우 크기와 dtype 정보가 업데이트된 새로운 심볼릭 텐서를 반환

• 최종 출력을 얻은 후 입력과 출력을 Model 클래스에 전달하여 모델 객체를 생성

outputs = layers.Dense(10, activation="softmax")(features)
model = keras.Model(inputs=inputs, outputs=outputs)

• 다음은 이 모델의 summary() 메서드 호출 결과

다중 입력, 다중 출력 모델

• 간단한 모델과 달리 배부분 딥러닝 모델은 리스트와 같은 형태가 아니라 그래프를 닮았음
• 예를 들어 입력이 여러 개이거나 출력이 여러 개
• 이런 종류의 모델에서 함수형 API가 진짜 빛을 발함
• 고객 이슈 티켓에 우선순위를 지정하고 적절한 부서로 전달하는 시스템을 만든다고 해 보자
• 이 모델은 3개의 입력을 사용
  • 이슈 티켓의 제목(텍스트 입력)
  • 이슈 티켓의 텍스트 본문(텍스트 입력)
  • 사용자가 추가한 태그(범주형 입력으로 여기에서는 원-핫 인코딩되었다고 가정)
• 텍스트 입력을 크기가 vocabulary_size인 0과 1로 이루어진 배열로 인코딩할 수 있음
• 이 모델은 출력도 2개
  • 이슈 티켓의 우선순위 점수로 0과 1 사이의 스칼라(시그모이드 출력)
  • 이슈 티켓을 처리해야 할 부서(전체 부서 집합에 대한 소프트맥스 출력)
• 다중 입력, 다중 출력 함수형 모델

vocabulary_size = 10000
num_tags = 100
num_departments = 4

# 모델의 입력을 정의한다.
title = keras.Input(shape=(vocabulary_size,), name="title")
text_body = keras.Input(shape=(vocabulary_size,), name="text_body")
tags = keras.Input(shape=(num_tags,), name="tags")

# 입력 특성을 하나의 텐서 features로 연결한다.
features = layers.Concatenate()([title, text_body, tags])

# 중간층을 적용하여 입력 특성을 더 풍부한 표현을 재결합시킨다.
features = layers.Dense(64, activation="relu")(features)

# 모델의 출력을 정의한다.
priority = layers.Dense(1, activation="sigmoid", name="priority")(features)
department = layers.Dense(
    num_departments, activation="softmax", name="department")(features)

# 입력과 출력을 지정하여 모델을 만든다.
model = keras.Model(inputs=[title, text_body, tags], outputs=[priority, department])

• 함수형 API는 간단하고 레고 블록 같지만 층으로 구성된 어떤 그래프도 정의할 수 있는 매우 유연한 방법

다중 입력, 다중 출력 모델 훈련하기

• Sequential 모델을 훈련하는 것과 거의 같은 방법으로 이 모델을 훈련할 수 있음
• 입력과 출력 데이터의 리스트로 fit() 메서드를 호출하면 됨
• 데이터의 리스트는 Model 클래스에 전달한 순서와 같아야 함
• 입력과 타깃 배열 리스트를 전달하여 모델 훈련하기

import numpy as np

num_samples = 1280

# 더미(dummy) 입력 데이터
title_data = np.random.randint(0, 2, size=(num_samples, vocabulary_size))
text_body_data = np.random.randint(0, 2, size=(num_samples, vocabulary_size))
tags_data = np.random.randint(0, 2, size=(num_samples, num_tags))

# 더미 타깃 데이터
priority_data = np.random.random(size=(num_samples, 1))
department_data = np.random.randint(0, 2, size=(num_samples, num_departments))

model.compile(optimizer="rmsprop",
              loss=["mean_squared_error", "categorical_crossentropy"],
              metrics=[["mean_absolute_error"], ["accuracy"]])
model.fit([title_data, text_body_data, tags_data],
          [priority_data, department_data],
          epochs=1)
model.evaluate([title_data, text_body_data, tags_data],
               [priority_data, department_data])
priority_preds, department_preds = model.predict([title_data, text_body_data, tags_data])

• 입력과 타깃 배열을 딕셔너리로 전달하여 모델 훈련하기

model.compile(optimizer="rmsprop",
              loss={"priority": "mean_squared_error", "department": "categorical_crossentropy"},
              metrics={"priority": ["mean_absolute_error"], "department": ["accuracy"]})
model.fit({"title": title_data, "text_body": text_body_data, "tags": tags_data},
          {"priority": priority_data, "department": department_data},
          epochs=1)
model.evaluate({"title": title_data, "text_body": text_body_data, "tags": tags_data},
               {"priority": priority_data, "department": department_data})
priority_preds, department_preds = model.predict(
    {"title": title_data, "text_body": text_body_data, "tags": tags_data})

함수형 API의 장점: 층 연결 구조 활용하기

• 함수형 모델은 명시적인 그래프 데이터 구조
• 층이 어떻게 연결되어 있는지 조사하고 이전 그래프 노드(node)(층의 출력)를 새 모델의 일부로 재사용할 수 있음
• 대부분의 연구자들이 심층 신경망에 대해 생각할 때 사용하는 '멘탈 모델(mental model)'인 층 그래프(graph of layers)에도 잘 맞음
• 이를 통해 모델 시각화와 특성 추출이라는 두 가지 중요한 기능이 가능
• 방금 정의한 모델의 연결 구조(모델의 토폴로지(topology))를 시각화해 보자
• plot_model() 함수를 사용하여 함수형 모델을 그래프로 그릴 수 있음

keras.utils.olot_model(model, "ticket_classifier.png")

plot_model()로 생성한 이슈 티켓 분류 모델 그림

• 모델에 있는 각 층의 입출력 크기를 추가하면 디버깅에 도움이 될 수 있다.

keras.utils.plot_model(
    model, "ticket_classifier_with_shape_info.png", show_shapes=True)

크기 정보가 추가된 모델 그림

• 텐서 크기에 None은 배치 크기를 나타냄
• 즉, 이 모델은 어떤 크기의 배치에서도 사용 가능
• 층 연결 구조를 참조하여 그래프에 있는 개별 노드를 조사하고 재사용(층 호출)할 수 있음
• model.layers 속성은 모델에 있는 모든 층의 리스트를 가지고 있음
• 각 층에 대해 layer.input과 layer.output을 출력해 볼 수 있음

• 이를 통해 특성 추출(feature extraction)을 수행하여 다른 모델에서 중간 특성을 재사용하는 모델을 만들 수 있음
• 이전 모델에 또 다른 출력을 추가한다고 가정해 보자
• 이슈 티켓이 해결되는 데 걸리는 시간, 즉 일종의 난이도를 추정하려고 함
• 이를 위해 'quick', 'medium', 'difficult' 3개의 범주에 대한 분류 층을 추가
• 모델을 처음부터 다시 만들고 재훈련할 필요가 없음
• 다음 코드와 같이 중간층을 참조할 수 있기 때문에 이전 모델의 중간 특성에서 수 있음

features = model.layers[4].output
difficulty = layers.Dense(3, activation="softmax", name="difficulty")(features)

new_model = keras.Model(
    inputs=[title, text_body, tags],
    outputs=[priority, department, difficulty])

새로운 모델의 그래프

Model 서브클래싱

• 마지막으로 알아야 할 모델 구축 패턴은 가장 고급 방법인 Model 서브클래싱
• Model 클래스를 상속하는 것도 매우 비슷함
  • __init__() 메서드에서 모델이 사용할 층을 정의
  • call() 메서드에서 앞서 만든 층을 사용하여 모델의 정방향 패스를 정의
  • 서브클래스의 객체를 만들고 데이터와 함께 호출하여 가중치를 만듦

이전 예제를 서브클래싱 모델로 다시 만들기

• Model 클래스를 상속하여 고객 이슈 티켓 관리 모델을 다시 구현해 보자

class CustomerTicketModel(keras.Model):

    def __init__(self, num_departments):
        super().__init__() # 꼭 부모 클래스의 생성자를 호출해야 한다.

        # 생성자에서 층을 정의
        self.concat_layer = layers.Concatenate()
        self.mixing_layer = layers.Dense(64, activation="relu")
        self.priority_scorer = layers.Dense(1, activation="sigmoid")
        self.department_classifier = layers.Dense(
            num_departments, activation="softmax")

    def call(self, inputs): # call() 메서드에서 정방향 패스를 정의
        title = inputs["title"]
        text_body = inputs["text_body"]
        tags = inputs["tags"]

        features = self.concat_layer([title, text_body, tags])
        features = self.mixing_layer(features)
        priority = self.priority_scorer(features)
        department = self.department_classifier(features)
        return priority, department

• 모델을 정의하고 나면 이 클래스의 객체를 만들 수 있음
• Layer 클래스와 마찬가지로 어떤 데이터로 처음 호출할 때 가중치를 만듦

model = CustomerTicketModel(num_departments=4)

priority, department = model(
    {"title": title_data, "text_body": text_body_data, "tags": tags_data})

• 지금까지는 모든 것이 이미 보았던 Layer 클래스 상속과 매우 비슷함
• Layer클래스 상속과 Model 클래스 상속의 차이점은 다음과 같다.
• '층'은 모델을 만드는 데 사용하는 구성 요소고 '모델'은 실제로 훈련하고 추론에 사용하는 최상위 객체
• 간단히 말해서 Model클래스는 fit(), evaluate(), predict() 메서드를 가지고 있음
• Layer 클래스에는 이런 메서드가 없음
• 그 외에는 두 클래스가 거의 동일
• Sequential이나 함수형 모델과 마찬가지로 Model을 상속하여 만든 모델을 컴파일하고 훈련할 수 있음

model.compile(optimizer="rmsprop",
              # 손실과 측정 지표로 전달하는 값은 call() 메서드가 반환하는 것과 정확히 일치해야 한다.(여기서는 2개의 원소를 가진 리스트)
              loss=["mean_squared_error", "categorical_crossentropy"],
              metrics=[["mean_absolute_error"], ["accuracy"]])

# 입력 데이터의 구조는 call() 메서드가 기대하는 것과 정확히 일치해야 한다.(여기에서는 title, text_body, tags 키를 가진 딕셔너리)
model.fit({"title": title_data,
           "text_body": text_body_data,
           "tags": tags_data},

          # 타깃 데이터의 구조는 call() 메서드가 반환하는 것과 정확히 일치해야 한다. (여기에서는 2개의 원소를 가진 리스트)
          [priority_data, department_data],
          epochs=1)
model.evaluate({"title": title_data,
                "text_body": text_body_data,
                "tags": tags_data},
               [priority_data, department_data])
priority_preds, department_preds = model.predict({"title": title_data,
                                                  "text_body": text_body_data,
                                                  "tags": tags_data})

• Model 서브클래싱 워크플로는 모델을 만드는 가장 유연한 방법 (사용자가 원하는대로 정의 가능)
• 층의 유향 비순환 그래프(directed ascyclic graph)(순환이 없는)로 표현할 수 없는 모델을 만들 수 있음
• 예를 들어 call() 메서드가 for루프 안에서 층을 사용하거나 재귀적으로 호출하는 모델
• 모든 것이 가능
• 다만 책임은 작성자에게 있음

주의: 서브클래싱된 모델이 지원하지 않는 것

• 이런 자유에는 대가가 따름
• 서브클래싱 모델에서는 모델 로직을 많이 책임져야 하며 잠재적인 오류 가능성이 훨씬 큼
• 결과적으로 더 많은 디버깅 작업을 해야 함
• 레고 블록을 맞추는 것이 아니라 새로운 파이썬 객체를 개발하고 있기 때문임
• 함수형과 서브클래싱 모델은 태생적으로 크게 다름
• 함수형 모델은 명시적인 데이터 구조인 층의 그래프이므로 출력하고 조사하고 수정할 수 있음
• 서브클래싱 모델은 한 덩어리의 바이트코드(bytecode)
• 원시 코드가 담긴 call() 메서드를 가진 파이썬 클래스
• 이것이 (원하는 어떤 기능도 코드로 작성할 수 있어) 서브클래싱 워크플로의 유연성의 원천이지만 새로운 제약 사항이 발생
• 예를 들어 층이 서로 연결되는 방식이 call() 메서드 안에 감추어지기 때문에 이 정보를 활용할 수 없음
• summary() 메서드가 층의 연결 구조를 출력할 수 없고 plot_model()함수로 모델의 구조를 그래프로 그릴 수 없음
• 비슷하게 서브클래싱 모델은 그래프가 없기 때문에 특성 추출을 위해 층 그래프의 노드를 참조할 수 없음
• 이 모델의 객체를 생성하고 나면 정방향 패스는 완전히 블랙박스(진행상황을 볼 수 없는)가 됨

여러 방식을 혼합하여 사용하기

• 중요한 것은 Sequential 모델, 함수형 API, Model 서브클래싱 패턴 중 하나를 선택한다고 다른 패턴의 사용을 제한하지 않는다는 점
• 케라스 API로 만든 모델을 Sequential 모델, 함수형 모델 또는 밑바닥부터 만든 서브클래싱 모델인지에 상관없이 부드럽게 서로 상호 운영할 수 있음
• 각각의 방법은 모두 동일한 워크플로 스펙트럼(범주)의 일부분
• 서브클래싱한 모델을 포함하는 함수형 모델 만들기

class Classifier(keras.Model):

    def __init__(self, num_classes=2):
        super().__init__()
        if num_classes == 2:
            num_units = 1
            activation = "sigmoid"
        else:
            num_units = num_classes
            activation = "softmax"
        self.dense = layers.Dense(num_units, activation=activation)

    def call(self, inputs):
        return self.dense(inputs)

inputs = keras.Input(shape=(3,))
features = layers.Dense(64, activation="relu")(inputs)
outputs = Classifier(num_classes=10)(features)
model = keras.Model(inputs=inputs, outputs=outputs)

• 함수형 모델을 포함하는 서브클래싱 모델 만들기

inputs = keras.Input(shape=(64,))
outputs = layers.Dense(1, activation="sigmoid")(inputs)
binary_classifier = keras.Model(inputs=inputs, outputs=outputs)

class MyModel(keras.Model):

    def __init__(self, num_classes=2):
        super().__init__()
        self.dense = layers.Dense(64, activation="relu")
        self.classifier = binary_classifier

    def call(self, inputs):
        features = self.dense(inputs)
        return self.classifier(features)

model = MyModel()

작업에 적합한 도구 사용하기

• 캐라스 모델 구축을 위해 가장 단순한 워크플로인 Sequential 모델부터 가장 고급인 Model 서브클래싱까지 다양한 워크플로를 학습함
• 각 방법은 장단점이 있으므로 현재 작업에 가장 잘 맞는 것을 선택
• 일반적으로 함수형 API가 쉬운 사용성과 유연성 사이에 적절한 절충점
• 층 연결 구조를 활용하여 모델 출력이나 특성 추출과 같은 용도에 잘 맞음
• 함수형 API를 사용할 수 있다면, 즉 모델을 층의 유향 비순환 그래프로 표현할 수 있다면 Model 서브클래싱보다 이 방식을 사용할 것을 권장
• 일반적으로 서브클래싱 층을 포함한 함수형 모델을 사용하면 함수현 API의 장점을 유지하면서 높은 개발 유연성을 제공할 수 있음

7.3 내장된 훈련 루프와 평가 루프 사용하기

• 아주 쉬운 것부터 매우 유연한 것까지 한 번에 한 단계씩 전체 워크플로에 접근할 수 있는 복잡성의 단계적 공개 원칙은 모델 훈련에도 적용
• 케라스는 모델 훈련을 위해 다양한 워크플로를 제공
• 간단하게 데이터로 fit() 메서드를 호출하거나 밑바닥부터 새로운 훈련 알고리즘을 작성할 수 있는 고급 방법도 제공
• 이미 compile(), fit(), evaluate(), predict() 워크플로를 잘 알고 있음
• 표준 워크플로: compile(), fit(), evaluate(), predict()

from tensorflow.keras.datasets import mnist

def get_mnist_model(): # 모델은 만든다. (나중에 재사용할 수 있도록 별도의 함수로 만든다.)
    inputs = keras.Input(shape=(28 * 28,))
    features = layers.Dense(512, activation="relu")(inputs)
    features = layers.Dropout(0.5)(features)
    outputs = layers.Dense(10, activation="softmax")(features)
    model = keras.Model(inputs, outputs)
    return model

(images, labels), (test_images, test_labels) = mnist.load_data() # 데이터를 로드하고 검증을 위해 일부를 떼어 놓는다.
images = images.reshape((60000, 28 * 28)).astype("float32") / 255
test_images = test_images.reshape((10000, 28 * 28)).astype("float32") / 255
train_images, val_images = images[10000:], images[:10000]
train_labels, val_labels = labels[10000:], labels[:10000]

model = get_mnist_model()

# 옵티마이저, 최소화할 손실 함수, 모니터링할 지표를 지정하여 모델을 컴파일
model.compile(optimizer="rmsprop",
              loss="sparse_categorical_crossentropy",
              metrics=["accuracy"])

# fit() 메서드를 사용해서 모델을 훈련. 본 적 없는 데이터에 대한 성능을 모니터링하기 위해 검증 데이터를 함께 제공
model.fit(train_images, train_labels,
          epochs=3,
          validation_data=(val_images, val_labels))

# evaluate() 메서드를 사용해서 새로운 데이터에 대한 손실과 측정 지표를 계산
test_metrics = model.evaluate(test_images, test_labels)

# predict() 메서드를 사용하여 새로운 데이터에 대한 분류 확률을 계산
predictions = model.predict(test_images)

• 이 간단한 워크플로를 커스터마이징할 수 있는 몇 가지 방법이 있음
  • 사용자 정의 측정 지표를 전달
  • fit() 메서드에 콜백(callback)을 전달하여 훈련하는 동안 특정 시점에 수행될 행동을 예약

사용자 정의 지표 만들기

• 지표(metric)는 모델의 성능을 측정하는 열쇠
• 특히 훈련 데이터 성능과 테스트 데이터 성능 차이를 측정하는 것이 중요(과대적합과 과소적합의 절충점인 최적접합을 찾기 위해)
• 분류와 회귀에 일반적으로 사용되는 지표는 keras.metrics 모듈에 이미 포함되어 있음
• 대부분의 경우 여기에 포함된 지표를 사용할 것
• 일반적이지 않은 작업을 한다면 사용자 정의 지표를 만들 수 있어야 함
• 케라스 지표는 keras.metrics.Metric 클래스에 상속한 클래스
• 층과 마찬가지로 지표는 텐서플로 변수에 내부 상태를 저장
• 층과 다른 점은 이 변수가 역전파로 업데이트되지 않는다는 것
• 상태 업데이트 로직은 update_state() 메서드 안에 직접 작성
• Metric 클래스를 상속하여 사용자 정의 지표 구현하기

# tensorflow 라이버르리 import
import tensorflow as tf

# Metric 클래스를 상속한다. RootMeanSquareError라는 클래스를 정의한다. 모델의 예측으로부터 rmse지표를 계산하는 사용자 정의 지표 클래스이다.
class RootMeanSquaredError(keras.metrics.Metric): 

    # 생성자에서 상태 변수를 정의한다. 층과 마찬가지로 add_weight() 메서드를 사용한다.
    def __init__(self, name="rmse", **kwargs): 
        # 부모 생성자를 호출해 기본 생성자를 초기화한다.
        super().__init__(name=name, **kwargs)
        # 이 줄은 add_weight() 메서드를 사용하여 상태 변수 mse_sum을 생성합니다. 모델 평가 중에 평균 제곱 오차의 누적 합계를 저장합니다.
        self.mse_sum = self.add_weight(name="mse_sum", initializer="zeros")
        # 이 줄은 add_weight()를 사용하여 또 다른 상태 변수 total_samples를 생성합니다. 모델 평가 중에 처리된 총 샘플 수를 저장합니다.
        self.total_samples = self.add_weight(
            name="total_samples", initializer="zeros", dtype="int32")
    # 이 메서드는 각 데이터 배치에 대한 실제 레이블(y_true) 및 모델 예측(y_pred)을 기반으로 상태 변수(mse_sum 및 total_samples) 업데이트를 담당합니다.
    def update_state(self, y_true, y_pred, sample_weight=None):
        # 이 라인은 평균 제곱 오차를 계산하기 위해 y_true를 원-핫 인코딩 표현으로 변환합니다.
        y_true = tf.one_hot(y_true, depth=tf.shape(y_pred)[1])
        # 이 라인은 예측 값에서 실제 레이블을 빼고 결과를 제곱하여 배치의 평균 제곱 오차(MSE)를 계산합니다. 그런 다음 제곱 차이를 합산합니다.
        mse = tf.reduce_sum(tf.square(y_true - y_pred))
        # 이 줄은 현재 배치의 MSE를 추가하여 mse_sum 상태 변수를 업데이트합니다.
        self.mse_sum.assign_add(mse)

        # 이 줄은 현재 배치(num_samples)의 샘플 수를 계산하고 num_samples를 추가하여 total_samples 상태 변수를 업데이트합니다.
        num_samples = tf.shape(y_pred)[0]
        self.total_samples.assign_add(num_samples)

    # 이 메서드는 누적된 mse_sum 및 total_samples를 기반으로 최종 RMSE 메트릭 값을 계산합니다.
    def result(self):
        return tf.sqrt(self.mse_sum / tf.cast(self.total_samples, tf.float32))

    # 이 메서드는 상태 변수(mse_sum 및 total_samples)를 초기 값으로 재설정하여 이전 평가에서 누적된 데이터를 지웁니다.
    def reset_state(self):
        self.mse_sum.assign(0.)
        self.total_samples.assign(0)

사용자 정의 지표 테스트

콜백 사용하기

• 대규모 데이터셋에서 model.fit() 메서드를 사용하여 수십 번의 에포크를 실행하는 것은 종이 비행기를 날리는 것과 조금 비슷함
• 일단 손을 떠나면 종이 비행기 경로와 착륙 지점을 제어할 방법이 없음
• 나쁜 결과를 피하려면 (그래서 종이 비행기를 낭비하지 않으려면) 종이 비행기 대신 다른 것을 사용하는 것이 좋음
• 드론은 주변 환경을 감지한 데이터를 조작부에 전달하여 현재 상태를 바탕으로 자동으로 운전
• 케라스 콜백(callback)은 model.fit() 호출을 종이 비행기에서 스스로 판단하고 동적으로 결정하는 똑똑한 자동 드론으로 바꾸어 줄 것
• 콜백은 fit() 메서드 호출 시 모델에 전달되는 객체(특정 메서드를 구현한 클래스 객체)
• 훈련하는 동안 모델은 여러 지점에서 콜백을 호출
• 콜백은 모델의 상태와 성능에 대한 모든 정보에 접근하고 훈련 중지, 모델 저장, 가중치 적재 또는 모델 상태 변경 등을 처리할 수 있음
• 다음은 콜백을 사용하는 몇 가지 사례
  • 모델 체크포인트(checkpoint) 저장: 훈련하는 동안 어떤 지점에서 모델의 현재 가중치를 저장
  • 조기 종료(early stopping): 검증 손실이 더 이상 향상되지 않을 때 훈련을 중지(물론 훈련하는 동안 얻은 가장 좋은 모델을 저장)
  • 훈련하는 동안 하이퍼파라미터 값을 동적으로 조정: 옵티마이저의 학습률 같은 경우
  • 훈련과 검증 지표를 로그에 기록하거나 모델이 학습한 표현이 업데이트될 때마다 시각화: 앞서 보았던 fit() 메서드의 진행 표시줄이 하나의 콜백
• keras.callbacks 모듈에는 여러 가지 내장 콜백이 포함되어 있음(다음은 전체 리스트가 아님)

ModelCheckpoint와 EarlyStopping 콜백

• 모델을 훈련할 때는 미리 예상할 수 없는 것이 많음
• 특히 최적의 검증 손실을 얻기 위해 얼마나 많은 에포크가 필요한지 알지 못함
• 지금까지 예제는 적절한 훈련 에포크를 알아내기 위해 첫 번째 실행에서 과대적합이 시작될 때까지 충분한 에포크로 훈련했음
• 그런 다음 최적의 에포크 횟수로 처음부터 다시 훈련을 시작
• 당연히 이런 방식은 낭비가 많음
• 더 좋은 처리 방법은 검증 손실이 더 이상 향상되지 않을 때 훈련을 멈추는 것
• EarlyStopping 콜백을 사용하여 이를 구현할 수 있음
• EarlyStopiing 콜백은 정해진 에포크 동안 모니터링 지표가 향상되지 않을 때 훈련을 중지
• 예를 들어 과대적합이 시작되자마자 훈련을 중지할 수 있음
• 에포크 횟수를 줄여 다시 모델을 훈련할 필요가 없음
• 일반적으로 이 콜백은 훈련하는 동안 모델을 계속 저장해 주는 ModelCheckpoint 콜백과 함께 사용
• (선택적으로 지금까지 가장 좋은 모델만 저장할 수 있고 즉, 에포크 끝에서 최고의 성능을 낸 모델)
• fit() 메서드에서 callbacks 매개변수 사용하기

# fit() 메서드의 callbacks 매개변수를 사용하여 콜백의 리스트를 모델로 전달한다. 몇 개의 콜백이라도 전달할 수 있다.
callbacks_list = [
    # 성능 향상이 멈추면 훈련을 중지시킨다.
    keras.callbacks.EarlyStopping(
        # 모델의 검증 정확도를 모니터링한다.
        monitor="val_accuracy",
        # 두 번의 에포크 동안 정확도가 향상되지 않으면 훈련을 중지
        patience=2,
    ),
    # 매 에포크 끝에서 현재 가중치를 저장
    keras.callbacks.ModelCheckpoint(
        # 모델 파일의 저장 경로
        filepath="checkpoint_path.keras",

        # 이 두 매개변수는 val_loss가 좋아지지 않으면 모델 파일을 덮어쓰지 않는다는 뜻이다. 훈련하는 동안 가장 좋은 모델이 저장된다.
        monitor="val_loss",
        save_best_only=True,
    )
]
model = get_mnist_model()
model.compile(optimizer="rmsprop",
              loss="sparse_categorical_crossentropy",
              metrics=["accuracy"]) # 정확도를 모니터링하므로 모델 지표에 포함되어야 한다.

# 콜백이 검증 손실과 검증 정확도를 모니터링하기 때문에 fit() 메서드를 호출할 때 validation_data 매개변수로 검증 데이터를 전달해야 한다. 학습 후 항상 모델을 수동으로 저장할 수 있다.
model.fit(train_images, train_labels,
          epochs=10,
          callbacks=callbacks_list,
          validation_data=(val_images, val_labels))

• 저장된 모델은 다음과 같이 로드할 수 있음

model = keras.models.load_model("checkpoint_path.keras")

사용자 정의 콜백 만들기

• 내장 콜백에서 제공하지 않는 특정 행동이 훈련 도중 필요하면 자신만의 콜백을 만들 수 있음
• 콜백은 keras.callbacks.Callback 클래스를 상속받아 구현
• 그 다음 이름에서 알 수 있듯이 훈련하는 동안 여러 지점에서 호출될 다음과 같은 메서드를 구현

• 이 메서드들은 모두 logs 매개변수와 함께 호출
• 이 매개변수 값은 이전 배치, 에포크 또는 훈련 실행에 대한 정보(훈련과 검증 지표 등)가 담긴 딕셔너리
• 또한, on_epoch_*와 on_batch_* 메서드는 에포크 인덱스나 배치 인덱스를 첫 번째 매개변수(정수)로 받음
• Callback 클래스를 상속하여 사용자 정의 콜백 만들기

from matplotlib import pyplot as plt

# keras.callbacks.Callback을 상속받는 사용자 정의 클래스 LossHistory를 정의합니다. 이 클래스는 모델 훈련 중 손실을 추적하고 시각화하는 데 사용
class LossHistory(keras.callbacks.Callback):
    # 이 메서드는 훈련 과정이 시작될 때 호출되는 콜백입니다. 이는 per_batch_losses라는 인스턴스 변수를 빈 리스트로 초기화합니다. 이 리스트는 훈련 중 각 배치의 손실을 저장
    def on_train_begin(self, logs):
        self.per_batch_losses = []

    # 이 메서드는 각 훈련 배치가 끝날 때 호출되는 콜백입니다. 현재 배치의 손실을 per_batch_losses 리스트에 추가
    def on_batch_end(self, batch, logs):
        self.per_batch_losses.append(logs.get("loss"))

    # 이 메서드는 각 훈련 에포크가 끝날 때 호출되는 콜백입니다. 에포크 동안 각 배치의 훈련 손실을 그래프로 만들어 이미지로 저장
    def on_epoch_end(self, epoch, logs):
        plt.clf()
        plt.plot(range(len(self.per_batch_losses)), self.per_batch_losses,
                 label="Training loss for each batch")
        plt.xlabel(f"Batch (epoch {epoch})")
        plt.ylabel("Loss")
        plt.legend()
        plt.savefig(f"plot_at_epoch_{epoch}")
        self.per_batch_losses = []

손실 그래프를 저장하는 사용자 정의 콜백의 출력 결과

텐서보드를 사용한 모니터링 시각화

• 좋은 연구를 하거나 좋은 모델을 개발하려면 실험하는 모델 내부에서 어떤 일이 일어나는지 자주 그리고 많은 피드백을 받아야 함
• 모델이 얼마나 잘 작동하는지 가능한 많은 정보를 얻는 것이 실험을 하는 목적
• 발전은 루프처럼 반복되는 과정을 통해 일어남
• 한 아이디어가 떠오르면 이 아이디어를 검증할 실험을 계획
• 실험을 수행하고 생성된 정보를 가공
• 이 정보는 다음 아이디어에 영감을 줌
• 이 루프를 더 많이 실행할수록 아이디어는 더 정제되고 강력해질 것
• 이때 케라스는 가능한 최단 시간에 아이디어를 실험으로 구현하도록 도와줌
• 고속 GPU를 통해 가능한 빠르게 실험의 결과를 얻도록 도와줄 것
• 실험 결과를 처리하기 위해 텐서보드(TensorBoard)가 필요함

발전의 반복 루프

• 텐서보드는 로컬에서 실행할 수 있는 브라우저 기반 애플리케이션
• 훈련하는 동안 모델 안에서 일어나는 모든 것을 모니터링하기 위한 가장 좋은 방법
• 텐서보드를 사용하여 다음과 같은 일을 수행할 수 있음
  • 훈련하는 동안 측정 지표를 시각적으로 모니터링
  • 모델 구조를 시각화
  • 활성화 출력과 그레이디언트의 히스토그램을 그림 임베딩을 3D로 표현
• 모델의 최종 손실 외에 더 많은 정보를 모니터링하면 모델 작동에 대한 명확한 그림을 그릴 수 있음
• 결국 모델을 더 빠르게 개선할 수 있음
• 케라스 모델의 fit() 메서드와 함께 텐서보드를 사용하는 가장 쉬운 방법은 keras.callbacks.TensorBoard 콜백

model = get_mnist_model()
model.compile(optimizer="rmsprop",
              loss="sparse_categorical_crossentropy",
              metrics=["accuracy"])

tensorboard = keras.callbacks.TensorBoard(
    log_dir="./tb_logs",
)
model.fit(train_images, train_labels,
          epochs=10,
          validation_data=(val_images, val_labels),
          callbacks=[tensorboard])

• 모델이 실행되고 나면 지정한 위치에 로그를 기록할 것
• 로컬 컴퓨터에서 파이썬 스크립트를 실행한다면 다음 명령으로 로컬에서 텐서보드를 실행할 수 있음
• (pip 명령으로 텐서플로를 설치했다면 tensorboard 명령을 사용할 수 있고 만약 이 실행 파일이 없다면 pip install tensorboard와 같이 수동으로 텐서보드를 설치할 수 있음)

tensorboard --logdir /full_path_to_your_log_dir

• 이 명령을 실행하고 화면에 출력된 URL에 접속하면 텐서보드 인터페이스를 볼 수 있음
• 코랩 노트북을 사용한다면 다음 명령을 사용하여 텐서보드 인스턴스를 노트북 일부로 실행할 수 있음

%load_ext tensorboard
%tensorboard --logdir ./tb_logs

7.4 사용자 정의 훈련, 평가 루프 만들기

• fit() 워크플로는 쉬운 사용성과 유연성 사이에서 알맞은 군형을 유지
• 대부분의 경우 이를 사용할 것
• 사용자 정의 지표, 사용자 정의 손실, 사용자 정의 콜백이 있음에도 딥러닝 연구자들이 원하는 모든 것을 지원하지는 못함
• 무엇보다도 내장 fit() 워크플로는 지도 학습(supervised learning)에만 초점이 맞추어져 있음
• 지도 학습은 입력 데이터와 이에 연관된 타깃(레이블 또는 어노테이션이라고도 부름)을 가지며 모델의 예측과 이 타깃의 함수로 손실을 계산
• 모든 형태의 머신 러닝이 카테고리에 속하는 것은 아님
• 생성 학습(generative learning), 자기지도 학습(selfsupervised learning)(타깃을 입력에서 얻는다.), 강화 학습(reinforcement learning)(강아지를 훈련하는 것처럼 간혈적인 '보상(reward)'으로 학습)등 명시적인 타깃이 없는 경우가 있음
• 일반적인 지도 학습을 하는 경우에도 연구자로서 저수준의 유연성이 필요한 새로운 기능을 추가하고 싶을 수 있음
• 내장 fit() 메서드로 충분하지 않은 상황이면 자신만의 훈련 로직을 직접 작성
• 다시 정리하면 전형적인 훈련 루프틑 다음과 같은 내용이 포함

1. 현재 배치 데이터에 대한 손실 값을 얻기 위해 그레이디언트 테이프 안에서 정방향 패스를 실행(모델의 출력을 계산)
2. 모델 가중치에 대한 손실의 그레이디언트를 계산
3. 현재 배치 데이터에 대한 손실 값을 낮추는 방향으로 모델 가중치를 업데이트

• 이런 단계가 필요한 만큼 많은 배치에 대해 반복
• 이것이 fit() 메서드가 처리하는 일

훈련 vs 추론

• 지금까지 본 저수준 훈련 루프 예제에서는 predictions = model(inputs)를 통해 단계 1(정방향 계산)을 수행해고 gradients = tape.gradient(loss, model.weights)를 통해 단계 2(그레이디언트 테이프로 계산한 그레이디언트를 추출)를 수행
• 일반적으로 두 가지 중요한 세부 사항을 고려해야 함
• Dropout 층과 같은 일부 케라스 층은 훈련(training)과 (예측을 만들기 위해 모델을 사용하는) 추론(inference)에서 동작이 다름
• 이런 층은 call() 메서드에 training 불리언(boolean) 매개변수를 제공
• dropout(inputs, training=True)와 같이 호출하면 이전 층의 활성화 출력 값을 일부 랜덤하게 제외
• dropout(inputs, training=False)와 같이 호출하면 아무런 일을 수행하지 않음
• 함수형 모델과 Sequential 모델도 call() 메서드에서 training 매개변수를 제공
• 정방향 패스에서 캐라스 모델을 호출할 때는 training=True로 지정해야 하는 것을 기억하자!
• 정방향 패스는 predictions = model(inputs, training=True)가 됨
• 모델 가중치 그레이디언트를 추론할 때 tape.gradients(loss, model.weights)가 아니라 tape.gradients(loss, model.trainable_weights)를 사용해야 함
• 사실 층과 모델에는 두 종류의 가중치가 있음
  • 훈련 가능한 가중치: Dense 층의 커널과 편향처럼 모델의 손실을 최소화하기 위해 역전파로 업데이트
  • 훈련되지 않는 가중치: 해당 층의 정방향 패스 동안 업데이트. 예를 들어 얼마나 많은 배치를 처리했는지 카운트하는 사용자 정의 층이 필요하다면 이 정보를 훈련되지 않는 가중치에 저장하고 배치마다 값을 1씩 증가시킴
• 케라스에 내장된 층 중에는 훈련되지 않는 가중치를 가진 층은 9장에서 소개할 Batch Normalization뿐
• BatchNormalization 층은 처리하는 데이터의 평균과 표준 편차에 대한 정보를 추적하여 (6장에서 배운 개념인) 특성 정규화(feature normalization)를 실시간으로 근사하기 위해 훈련되지 않은 가중치가 필요함
• 이 두 가지를 고려하면 지도 학습을 위한 훈련 스탭은 다음과 같이 작성할 수 있음

def train_step(inputs, targets):
    with tf.GradientTape() as tape:
        predictions = model(inputs, training=True)

    loss = loss_fn(targets, predictions)
gradients = tape.gradients(loss, model.trainable_weights)
optimizer.apply_gradients(zip(model.trainable_weights, gradients))

측정 지표의 저수준 사용법

• 저수준 훈련 루프에서 (사용자 정의 지표든 내장 지표든) 케라스 지표를 사용하게 될 것
• 앞에서 측정 지표 API에 대해 배웠음
• 단순히 각 배치의 타깃과 예측에 대해 update_state(y_true, y_pred)를 호출하면 됨
• 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"결과: {current_result:.2f}")

• 모델의 손실처럼 스칼라 값의 평균을 추적해야 할 수도 있음
• 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_tracker.result():.2f}")

• (훈련 에포크나 평가를 시작할 때처럼) 현재 결과를 재설정하고 싶을 때 metric.reset_state()를 사용하는 것을 잊지 말자

완전한 훈련과 평가 루프

• 정방향 패스, 역방향 패스, 지표 추적을 fit()과 유사한 훈련 스탭 함수로 연결해 보겠음
• 단계별 훈련 루프 작성하기: 훈련 스탭 함수

model = get_mnist_model()

# 손실 함수를 준비
loss_fn = keras.losses.SparseCategoricalCrossentropy()

# 옵티마이저를 준비
optimizer = keras.optimizers.RMSprop()

# 모니터링할 지표 리스트를 준비
metrics = [keras.metrics.SparseCategoricalAccuracy()]

# 손실 평균을 추적할 평균 지표를 준비
loss_tracking_metric = keras.metrics.Mean()

def train_step(inputs, targets):
    # 정방향 패스를 실행. training = True를 전달
    with tf.GradientTape() as tape:
        predictions = model(inputs, training=True)
        loss = loss_fn(targets, predictions)

    # 역방향 패스를 실행한다. model.trainable_weights를 사용
    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()

• 이제 완전한 훈련 루프를 구성할 수 있음
• tf.data.Dataset객체를 사용하여 넘파이 데이터를 크기가 32인 배치로 데이터를 순회하는 반복자로 바꿈
• 단계별 훈련 루프 작성하기: 훈련 루프 자체

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"{epoch}번째 에포크 결과")
    for key, value in logs.items():
        print(f"...{key}: {value:.4f}")

• 다음은 평가 루프
• 간단한 for 루프로 하나의 배치 데이터를 처리하는 test_step() 함수를 반복하여 호출
• test_step() 함수는 train_step() 함수 로직의 일부분을 사용
• 모델의 가중치를 업데이트하는 코드가 빠져 있음
• 단계별 평가 루프 작성하기

def test_step(inputs, targets):
    predictions = model(inputs, training=False) # 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("평가 결과:")
for key, value in logs.items():
    print(f"...{key}: {value:.4f}")

• 방금 fit() 메서드와 evaluate() 메서드를 구현!
• 사실 완전히 똑같지는 않음
• fit()과 evaluate() 메서드는 대규모 분산 계산과 같은 더 많은 기능을 제공
• 이를 위해서는 조금 더 많은 작업이 필요함
• 또한, 몇 가지 핵심적인 성능 최적화가 포함되어 있음

tf.function으로 성능 높이기

• 기본적으로 동일한 로직을 구현했지만 아마 직접 작성한 루프가 케라스에 내장된 fit()과 evaluate() 메서드보다 훨씬 느리게 실행됨
• 기본적으로 텐서플로 코드는 넘파이나 일반적인 파이썬 코드와 비슷하게 즉시(eagerly) 라인 단위로 실행되기 때문임
• 즉시 실행(eager execution)은 코드 디버깅을 쉽게 만들어 줌
• 성능 측면에서는 최적이 아님
• 텐서플로 코드는 계산 그래프(computation graph)로 컴파일하는 것이 더 성능이 좋음
• 여기에서는 라인 단위로 해석되는 코드에서는 할 수 없는 전역적인 최적화가 가능
• 이렇게 만드는 문법은 매우 간단함
• 다음과 같이 실행하기 전에 컴파일하고 싶은 함수에 @tf.function
• 평가 스텝 함수에 @tf.function 데코레이터 추가하기

@tf.function # 이 라인만 추가되었다.
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("평가 결과:")
for key, value in logs.items():
    print(f"...{key}: {value:.4f}")

• 코랩 CPU에서 이 평가 루프를 실행하는데 2.8초에서 0.6초로 줄었음
• 훨씬 빨라졌음!
• 코드를 디버깅할 때 @tf.function 데코레이터를 쓰지 말고 즉시 실행 모드를 사용하는 것이 좋음
• 이 방식이 버그를 추적하기 더 쉬움
• 코드가 제대로 작동하고 성능을 높이고 싶을 때 훈련 스탬과 평가 스탭에 @tf.function 데코레이터를 추가

fit() 메서드를 사용자 정의 루프로 활용하기

• 이전 절에서 밑바닥부터 완전한 사용자 정의 훈련 루프를 만들었음
• 이런 방식이 가장 높은 유연성을 제공하지만 많은 코드를 작성해야 하고 콜백이나 분산 훈련 지원 같은 fit() 메서드가 제공하는 많은 편리한 기능을 사용할 수 없음
• 사용자 정의 훈련 알고리즘이 필요하지만 케라스에 내장된 훈련 로직의 기능을 활용하고 싶다면 어떨까?
• 사실 fit() 메서드와 밑바닥부터 작성한 훈련 루프 사이의 중간 지점이 있음
• 사용자 정의 훈련 스탭 함수를 제공하고 나머지 처리는 프레임워크에 위임할 수 있음
• 이렇게 하려면 Model 클래스의 train_step() 메서드를 오버라이딩(overriding)
• 이 함수는 fit() 메서드가 배치 데이터마다 호출하는 메서드
• 그 다음 이전처럼 fit() 메서드를 호출하면 자신만의 학습 알고리즘을 실행시킬 수 있음
• 간단한 예는 다음과 같음
  • keras.Model을 상속한 새로운 클래스를 만듦
  • train_step(self, data)메서드를 오버라이드. 이 메서드의 내용은 이전 절에서 만든 것과 동일. (손실을 포함하여) 측정 지표 이름과 현재 값이 매핑된 딕셔너리를 반환
  • 모델의 Metric 객체들을 반환하는 metrics 속성을 구현. 이를 활용하여 매 에포크 시작이나 evaluate()를 호출할 때 모델이 지표 객체들의 reset_state() 메서드를 자동으로 호출할 수 있음. 수동으로 지표를 재설정할 필요가 없음
• fit()이 사용할 사용자 정의 훈련 스탭 구현하기

loss_fn = keras.losses.SparseCategoricalCrossentropy()
loss_tracker = keras.metrics.Mean(name="loss") # 이 객체는 훈련과 평균 과정에서 배치 손실의 평균을 추적한다.

class CustomModel(keras.Model):
    def train_step(self, data): # train_step 메서드를 오버라이딩한다.
        inputs, targets = data
        with tf.GradientTape() as tape:
            predictions = self(inputs, training=True) # 모델이 클래스 자체이므로 model(inputs, training=True)대신에 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) # 손실의 평균을 추적하는 loss_tracker를 업데이트한다.
        return {"loss": loss_tracker.result()} # 평균 손실을 구한다.

    @property
    # 에포크마다 재설정할 지표는 여기에 나열해야 한다.
    def metrics(self):
        return [loss_tracker]

• 이제 사용자 정의 모델의 객체를 만들고 컴파일하고 (손실은 모델 밖에서 이미 정의했기 때문에 옵티마이저만 전달), 보통 때처럼 fit() 메서드로 훈련할 수 있음

inputs = keras.Input(shape=(28 * 28,))
features = layers.Dense(512, activation="relu")(inputs)
features = layers.Dropout(0.5)(features)
outputs = layers.Dense(10, activation="softmax")(features)
model = CustomModel(inputs, outputs)

model.compile(optimizer=keras.optimizers.RMSprop())
model.fit(train_images, train_labels, epochs=3)

• 몇 가지 주의할 점이 있음
  • 이 패턴 때문에 함수형 API로 모델을 만드는 데 문제가 되지 않음. Sequential 모델, 함수형 모델, 서브클래싱 모델을 만드는지에 상관없이 이 방식을 하용할 수 있음
  • 프레임워크가 알아서 처리하기 때문에 train_step 메서드를 오버라이딩할 때 @tf.function 데코레이터를 사용할 필요가 없음
  • 이제 compile() 메서드를 통해 지표와 손실을 설정하면 어떨까?
  • compile() 메서드를 호출한 후 다음을 참조할 수 있음
    • self.compiled_loss: compile() 메서드에 전달할 손실 함수
    • self.compiled_metrics: compile() 메서드에 전달된 지표 목록이 포함되어 있는 객체. self.compiled_metrics.update_state()를 호출하여 모든 지표를 동시에 업데이트 할 수 있음
    • self.metrics: compile() 메서드에 전달한 실제 지표의 목록. 앞서 loss_tracking_metric으로 수동으로 했던 것과 비슷하게 손실을 추적하는 지표도 포함
• 다음과 같이 사용 가능

class CustomModel(keras.Model):
    def train_step(self, data):
        inputs, targets = data
        with tf.GradientTape() as tape:
            predictions = self(inputs, training=True)
            loss = self.compiled_loss(targets, predictions) # self.compiled_loss를 사용해서 손실을 계산
        gradients = tape.gradient(loss, self.trainable_weights)
        self.optimizer.apply_gradients(zip(gradients, self.trainable_weights))
        self.compiled_metrics.update_state(targets, predictions) # self.compiled_metrics로 모델의 지표를 업데이트
        return {m.name: m.result() for m in self.metrics} # 측정 지표 이름과 현재 값을 매핑한 딕셔너리를 반환한다.

• 테스트

inputs = keras.Input(shape=(28 * 28,))
features = layers.Dense(512, activation="relu")(inputs)
features = layers.Dropout(0.5)(features)
outputs = layers.Dense(10, activation="softmax")(features)
model = CustomModel(inputs, outputs)

model.compile(optimizer=keras.optimizers.RMSprop(),
              loss=keras.losses.SparseCategoricalCrossentropy(),
              metrics=[keras.metrics.SparseCategoricalAccuracy()])
model.fit(train_images, train_labels, epochs=3)