[핸즈온 머신러닝] 10장. 케라스를 사용한 인공 신경망 소개 (2)

본 글은 [핸즈온 머신러닝 2판 (Hands-On Machine Learning with Scikit-Learn, keras & TensorFlow)] - (박해선 옮김) 책을 개인공부하기 위해 요약, 정리한 내용입니다.
전체 코드는 https://github.com/ingu627/handson-ml2에 기재했습니다.(원본 코드 fork)
뿐만 아니라 책에서 설명이 부족하거나 이해가 안 되는 것들은 외부 자료들을 토대로 메꿔 놓았습니다.
오타나 오류는 알려주시길 바라며, 도움이 되길 바랍니다.

10.2 케라스로 다층 퍼셉트론 구현하기

• 케라스는 모든 종류의 신경망을 손쉽게 만들고 훈련, 평가, 실행할 수 있는 고수준 딥러닝 API이다.
• 텐서플로와 케라스 다음으로 가장 인기 있는 딥러닝 라이브러리는 페이스북 파이토치 (PyTorch)이다.

10.2.1 텐서플로 2 설치

• https://ingu627.github.io/tips/install_cuda/에서 자세히 다룬바 있다.
• UPDATE : https://ingu627.github.io/tips/install_cuda2/ (Tensorflow 2.8 버전 설치)

10.2.2 시퀀셜 API를 사용하여 이미지 분류기 만들기

# 케라스를 사용하여 데이터셋 적재하기
import keras
import tensorflow as tf

fashion_mnist = tf.keras.datasets.fashion_mnist
(X_train_full, y_train_full), (X_test, y_test) = fashion_mnist.load_data()

print(X_train_full.shape) # (60000, 28, 28)
print(X_train_full.dtype) # uint8

# 경사 하강법으로 신경망을 훈련하기 때문에 입력 특성의 스케일을 조정해야 한다. 255(최대 범위)로 나누어 0~1 사이로 조정
X_valid, X_train = X_train_full[:5000] / 255.0, X_train_full[5000:] / 255.0
y_valid, y_train = y_train_full[:5000], y_train_full[5000:]
X_test = X_test / 255.0

# 클래스 이름의 리스트 만들기
class_names = ["T-shirt/top", "Trouser", "Pullover", "Dress", "Coat", "Sandal", "Shirt", "Sneaker", "Bag", "Ankle boot"]

시퀀셜 API를 사용하여 모델 만들기

from tensorflow.keras.layers import Flatten, Dense
from tensorflow.keras.models import Sequential

model = Sequential() # 순서대로 연결된 층을 일렬로 쌓아서 구성한다.
model.add(Flatten(input_shape=[28,28])) # 모델의 첫 번째 층이므로 input_shpae를 지정해야 한다. 이 층은 어떤 모델 파라미터도 가지지 않고 간단한 전처리를 수행할 뿐이다. 여기에는 배치 크기를 제외하고 샘플의 크기만 써야 한다. 
model.add(Dense(300, activation='relu')) # 뉴런 300개를 가진 Dense 은닉층을 추가한다. 이 층은 ReLU 활성화 함수를 사용한다. Dense 층마다 각자 가중치 행렬을 관리한다. 이 행렬에는 층의 뉴런과 입력 사이의 모든 연결 가중치가 포함된다. 또한 (뉴런마다 하나씩 있는) 편향도 벡터로 관리한다. 
model.add(Dense(300, activation='relu'))
model.add(Dense(10, activation='softmax')) # (클래스마다 하나씩) 뉴런 10개를 가진 Dense 출력층을 추가한다. 배타적인 클래스이므로 소프트맥스 활성화 함수를 사용한다.

• Flatten() : 입력 이미지를 1D 배열로 변환

• summary() : 모델에 있는 모든 층을 출력한다. (각 층의 이름 + 출력 크기 + 파라미터 개수)

model.summary()

Model: "sequential"
# _________________________________________________________________
# Layer (type)                 Output Shape              Param #   
# =================================================================
# flatten (Flatten)            (None, 784)               0         
# _________________________________________________________________
# dense (Dense)                (None, 300)               235500    
# _________________________________________________________________
# dense_1 (Dense)              (None, 300)               90300     
# _________________________________________________________________
# dense_2 (Dense)              (None, 10)                3010      
# =================================================================
# Total params: 328,810
# Trainable params: 328,810
# Non-trainable params: 0
# _________________________________________________________________

• 첫 번째 은닉층은 784 x 300 개의 연결 가중치와 300개의 편향을 가진다. 이를 더하면 235,500개의 파라미터를 가진다.
  • 이런 모델은 훈련 데이터를 학습하기 충분한 유연성을 가진다.
• Dense 층은 대칭성을 깨뜨리기 위해 연결 가중치를 무작위로 초기화한다. 편향은 0으로 초기화한다.
  • 가중치 행렬의 크기는 입력의 크기에 달려 있다. 이 때문에 Sequential 모델에 첫 번째 층을 추가할 때 input_shape 매개변수를 지정한 것이다.

모델 컴파일

• 모델을 만들고 나서 compile() 메서드를 호출하여 사용할 손실 함수와 옵티마이저를 지정해야 한다.

model.compile(loss="sparse_categorical_crossentropy",
optimizer='sgd',
metrics=['accuracy']) 

• 클래스가 배타적이므로 sparsed_categorical_crossentropy 사용.
  • 만약 샘플마다 클래스별 타깃 확률을 가지고 있다면 대신 categorical_crossentropy손실을 사용해야 한다.
  • 이진 분류나 다중 레이블 이진 분류를 수행한다면 “sigmoid” 함수를 사용하고 binary_Crossentropy 손실을 사용한다.
• sgd = stochastic gradient descent (default = lr=0.01)
  • 옵티마이저에 sgd를 지정하면 역전파 알고리즘을 수행하는 것이다.

모델 훈련과 평가

history = model.fit(X_train, y_train, epochs=30,
                    validation_data = (X_valid, y_valid))

# Epoch 1/30
# 1719/1719 [==============================] - 12s 6ms/step - loss: 1.0080 - accuracy: 0.6764 - val_loss: 0.5268 - val_accuracy: 0.8168
# Epoch 2/30
# 1719/1719 [==============================] - 10s 6ms/step - loss: 0.5108 - accuracy: 0.8239 - val_loss: 0.4558 - val_accuracy: 0.8476
# Epoch 3/30
# 1719/1719 [==============================] - 11s 6ms/step - loss: 0.4533 - accuracy: 0.8430 - val_loss: 0.4098 - val_accuracy: 0.8634
# Epoch 4/30
# 1719/1719 [==============================] - 10s 6ms/step - loss: 0.4222 - accuracy: 0.8512 - val_loss: 0.4230 - val_accuracy: 0.8574
# Epoch 5/30
# 1719/1719 [==============================] - 10s 6ms/step - loss: 0.4025 - accuracy: 0.8583 - val_loss: 0.4019 - val_accuracy: 0.8598
# Epoch 6/30
# 1719/1719 [==============================] - 10s 6ms/step - loss: 0.3807 - accuracy: 0.8658 - val_loss: 0.3893 - val_accuracy: 0.8618
# Epoch 7/30
# 1719/1719 [==============================] - 11s 6ms/step - loss: 0.3699 - accuracy: 0.8711 - val_loss: 0.3613 - val_accuracy: 0.8748
# ...
# Epoch 29/30
# 1719/1719 [==============================] - 10s 6ms/step - loss: 0.2322 - accuracy: 0.9151 - val_loss: 0.3025 - val_accuracy: 0.8916
# Epoch 30/30
# 1719/1719 [==============================] - 10s 6ms/step - loss: 0.2277 - accuracy: 0.9185 - val_loss: 0.3012 - val_accuracy: 0.8912

• 모델을 훈련하려면 fit() 메서드를 호출한다.
• 입력 특성과 타깃 클래스, 훈련할 에포크 횟수를 전달한다. 검증 세트도 전달한다. (optional)
  • X_train : 입력 특성
  • y_train : 타깃 클래스
  • epochs : 훈련할 에포크 횟수
• 케라스는 에포크가 끝날 때마다 검증 세트를 사용해 손실과 추가적인 측정 지표를 계산한다.
  • 이 지표는 모델이 얼마나 잘 수행되는지 확인하는 데 유용하다.
• 케라스는 (진행 표시줄과 함께) 처리한 샘플 개수와 샘플마다 걸린 평균 훈련 시간, 훈련 세트와 검증 세트에 대한 손실과 정확도를 출력한다.

• 모델 성능 테스트
  • 훈련된 모델의 evaluate 메서드 활용
  • 손실과 정확도 계산해줌

model.evalute(X_test, y_test)

• 어떤 클래스는 많이 등장하고 다른 클래스는 조금 등장하여 훈련 세트가 편중되어 있다면 fit() 메서드를 호출할 때 class_weight 매개변수를 지정하는 것이 좋다.
  • 적게 등장하는 클래스는 높은 가중치를 부여하고 많이 등장하는 클래스는 낮은 가중치를 부여한다.
  • 케라스가 손실을 계산할 때 이 가중치를 사용한다.
• 샘플별로 가중치를 부여하고 싶다면 sample_weight 매개변수를 지정한다.
  • 각 샘플의 손실에 가중치를 곱한 후 전체 샘플 개수로 나누어 손실을 계산한다.
• fit() 메서드가 반환하는 History 객체에는 에포크가 끝날 때마다 훈련 세트와 (있다면) 검증 세트에 대한 손실과 측정한 지표를 담은 딕셔너리 (history.history)가 있다.

# 학습 곡선
import pandas as pd
import matplotlib.pyplot as plt

pd.DataFrame(history.history).plot(figsize=(8,5))
plt.grid(True)
plt.gca().set_ylim(0,1)
plt.show()

• 훈련하는 동안 훈련 정확도와 검증 정확도가 꾸준히 상승하고 훈련 손실과 검증 손실은 감소한다.
• 검증 손실은 에포크가 끝난 후에 계산되고 훈련 손실은 에포크가 진행되는 동안 계산된다.
• 케라스에서는 fit() 메서드를 다시 호출하면 중지되었던 곳에서부터 훈련을 이어갈 수 있다.
• 모델 성능에 만족스럽지 않으면 처음부터 되돌아가서 하이퍼파라미터를 튜닝해야 한다.
  • 여전히 성능이 높지 않으면 층 개수, 층에 있는 뉴런 개수, 은닉층이 사용하는 활성화 함수와 같은 모델의 하이퍼파리미터를 튜닝해 본다.

모델을 사용해 예측 만들기

• predict() 메서드를 사용해 새로운 샘플에 대해 예측을 만들 수 있다.
• predict() : 각 클래스에 속할 확률 계산
• predict_class() : 가장 높은 확률의 클래스 지정

10.2.3 시퀀셜 API를 사용하여 회귀용 다층 퍼셉트론 만들기

from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

housing = fetch_california_housing()

X_train_full, X_test, y_train_full, y_test = train_test_split(
    housing.data, housing.target)
X_train, X_valid, y_train, y_valid = train_test_split(
    X_train_full, y_train_full
)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_valid = scaler.transform(X_valid)
X_test = scaler.transform(X_test)

• 이 데이터셋에는 잡음이 많기 때문에 과대적합을 막는 용도로 뉴런 수가 적은 은닉층 하나만 사용한다.

from tensorflow.keras.layers import Dense
from tensorflow.keras.models import Sequential

model = Sequential([
    Dense(30, activation='relu', input_shape=X_train.shape[1:]),
    Dense(1) # 하나의 값을 예측하기 때문에 1 (활성화 함수 X)
])
model.compile(loss='mse', optimizer='sgd')
history = model.fit(X_train, y_train, epochs=2,
                    validation_data=(X_valid, y_valid))
mse_test = model.evaluate(X_test, y_test)
X_new = X_test[:3]
y_pred = model.predict(X_new)

• 시퀀셜 API를 사용해 회귀용 MLP를 구축, 훈련, 평가, 예측하는 방법은 분류에서 했던 것과 비슷하지만
  1. 출력층이 활성화 함수가 없는 하나의 뉴런(하나의 값을 예측하기 때문)을 가진다는 것과
  2. 손실 함수로 평균 제곱 오차를 사용한다는 점이 주된 차이점이다.

케라스 Sequential 클래스 활용의 장단점

• 사용하기 매우 쉬우며 성능 우수하다.
• 입출력이 여러 개이거나 더 복잡한 네트워크를 구성하기 어려움
• Sequential 클래스 대신에 함수형 API, 하위클래스(subclassing) API 등을 사용하여 보다 복잡하며, 보다 강력한 딥러닝 모델을 구축 가능하다.

10.2.4 함수형 API를 사용해 복잡한 모델 만들기

• 와이드 & 딥 (Wide & Deep) 신경망
  • 신경망이 (깊게 쌓은 층을 사용한) 복잡한 패턴과 (짧은 경로를 사용한) 간단한 규칙을 모두 학습할 수 있다.

이미지 출처 ¹

from tensorflow.keras.layers import Input, Dense, Concatenate
from tensorflow.keras.models import Model

input_ = Input(shape=X_train.shape[1:]) # 이 객체는 shape과 dtype을 포함하여 모델의 입력을 정의한다.
hidden1 = Dense(30, activation='relu')(input_) # 30개의 뉴런과 ReLU 활성화 함수를 가진 Dense 층을 만든다. 이 층은 만들어지자마자 입력과 함께 함수처럼 호출된다.
# 이 메서드에는 build() 메서드를 호출하여 층의 가중치를 생성한다.
hidden2 = Dense(30, activation='relu')(hidden1) 
concat = Concatenate()([input_, hidden2]) # Concatenate 층을 만들고 또 다시 함수처럼 호출하여 두 번째 은닉층의 출력과 입력을 연결한다.
output = Dense(1)(concat) # 하나의 뉴런과 활성화 함수가 없는 출력층을 만들고 Concatenate 층이 만든 결과를 사용해 호출한다.
model = Model(inputs=[input_], outputs=[output]) # 사용할 입력과 출력을 지정하여 케라스 Model을 만든다. 

이미지 출처 ²

from tensorflow.keras.layers import Input, Dense, concatenate
from tensorflow.keras.models import Model

input_A = Input(shape=[5], name="wide_input") # 5개의 특성을 입력받을 수 있다.
input_B = Input(shape=[6], name="deep_input") # 6개의 특성을 입력받을 수 있다.
hidden1 = Dense(30, activation="relu")(input_B)
hidden2 = Dense(30, activation="relu")(hidden1)
concat = concatenate([input_A, hidden2])
output = Dense(1, name='output')(concat)
model = Model(inputs=[input_A, input_B], outputs=[output])

• 모델 컴파일은 이전과 동일하지만 fit() 메서드를 호출할 때 하나의 입력 행렬 X_train을 전달하는 것이 아니라 입력마다 하나씩 행렬의 튜플 (X_train_A, X_train_B)을 전달해야 한다.

from tensorflow.keras.optimizers import SGD

model.compile(loss='mse', optimizer=SGD(lr=1e-3))

X_train_A, X_train_B = X_train[:, :5], X_train[:, 2:] # input_A 입력값 : 0~4번 인덱스 특성, input_B 입력값 : 2~7번 인덱스 특성
X_valid_A, X_valid_B = X_valid[:, :5], X_valid[:, 2:]
X_test_A, X_test_B = X_test_A[:3], X_test_B[:3]
X_new_A, X_new_B = X_test_A[:3], X_test_B[:3]

history = model.fit((X_train_A, X_train_B), y_train, epochs=20,
                    validation_data=((X_valid_A, X_valid_B), y_valid))
mse_test = model.evaluate((X_test_A, X_test_B), y_test)
y_pred = model.predict((X_new_A, X_new_B))

• 다중 출력이 필요할 때
• 보조 출력을 사용해 하위 네트워크가 나머지 네트워크에 의존하지 않고 그 자체로 유용한 것을 학습하는지 확인할 수 있다.

이미지 출처 ³

from tensorflow.keras.layers import Input, Dense, concatenate
from tensorflow.keras.models import Model

input_A = Input(shape=[5], name="wide_input")
input_B = Input(shape=[6], name="deep_input")
hidden1 = Dense(30, activation="relu")(input_B)
hidden2 = Dense(30, activation="relu")(hidden1)
concat = concatenate([input_A, hidden2])
output = Dense(1, name="main_output")(concat) # 주 출력
aux_output = Dense(1, name="aux_output")(hidden2) # 보조 출력
model = Model(inputs=[input_A, input_B],
              outputs=[output, aux_output])

• 각 출력은 자신만의 손실 함수가 필요하다.
• 모델을 컴파일할 때 손실의 리스트를 전달해야 한다.
• 케라스는 나열된 손실을 모두 더하여 최종 손실을 구해 훈련에 사용한다.
• 출력 별로 손실가중치를 지정하여 출력별 중요도 지정 가능

model.compile(loss=['mse', 'mse'], loss_weights=[0.9, 0.1], optimizer='sgd')

• 모델을 훈련할 때 각 출력에 대한 레이블을 제공해야 한다.
• 여기에서는 주 출력과 보조 출력이 같은 것을 예측해야 하므로 동일한 레이블을 사용한다.

history = model.fit(
    [X_train_A, X_train_B], [y_train, y_train], epochs=20,
    validation_data=([X_valid_A, X_valid_B], [y_valid, y_valid])
)

• 모델을 평가하면 케라스는 개별 손실과 함께 총 손실을 반환한다.

total_loss, main_loss, aux_loss = model.evaluate(
    [X_test_A, X_test_B], [y_test, y_test]
)

• predict() 메서드는 각 출력에 대한 예측을 반환한다.

y_pred_main, y_pred_aux = model.predict([X_new_A, X_new_B])

10.2.5 서브클래싱 API로 동적 모델 만들기

• 시퀀셜 API와 함수형 API는 모두 선언적 (declarative)이다. (=정적)
  • 사용할 층과 연결 방식을 먼저 정의하고 모델에 데이터를 주입하여 훈련이나 추론을 시작할 수 있다.
  • 장점
    1. 모델을 저장하거나 복사, 공유하기 쉽다.
    2. 모델의 구조를 출력하거나 분석하기 좋다.
    3. 프레임워크가 크기를 짐작하고 타입을 학인하여 에러를 일찍 발견할 수 있다.
    4. 전체 모델이 층으로 구성된 정적 그래프이므로 디버깅하기 쉽다.
• 하지만 어떤 모델은 반복문을 포함하고 다양한 크기를 다루어야 하며 조건문을 가지는 등 여러 가지 동적인 구조를 필요로 한다.
  • 서브클래싱 API를 쓴다!!

• Model 클래스를 상속한 다음 다음 생성자 안에서 필요한 층을 만든다.
• 초기설정 메서드(__init__())를 이용하여 은닉층과 출력층 설정
• call() 메서드 안에 수행하려는 연산을 기술한다.

import tensorflow as tf

class WideAndDeepModel(tf.keras.Model):
    def __init__(self, units=30, activation='relu', **kwargs):
        super().__init__(**kwargs) # 표준 매개변수 처리
        self.hidden1 = tf.keras.layers.Dense(units, activation=activation)
        self.hidden2 = tf.keras.layers.Dense(units, activation=activation)
        self.main_output = tf.keras.layers.Dense(1)
        self.aux_output = tf.keras.layers.Dense(1)
    
    def call(self, inputs):
        input_A, input_B = inputs
        hidden1 = self.hidden1(input_B)
        hidden2 = self.hidden2(hidden1)
        concat = tf.keras.layers.concatenate([input_A, hidden2])
        main_output = self.main_output(concat)
        aux_output = self.aux_output(hidden2)
        return main_output, aux_output

model = WideAndDeepModel()

• 위 코드는 Input 클래스의 객체를 만들 필요가 없다. 대신 call() 메서드의 input 매개변수를 사용한다.
• 단점
  1. 모델 구조가 call() 메서드 안에 숨겨져 있기 때문에 케라스가 쉽게 이를 분석할 수 없다. (모델을 저장하거나 복사 X)
  2. summary() 메서드를 호출하면 층의 목록만 나열되고 층 간의 연결 정보를 얻을 수 없다.
  3. 케라스가 타입과 크기를 미리 확인할 수 없다.

10.2.6 모델 저장과 복원

• 모델 저장 형태 & 로드

model = Sequential([...])
model.compile([...])
model.fit([...])
model.save('my_keras_model.h5') # 저장

model = load_model('my_keras_model.h5') # 로드

save_weights()와 load_weights() 메서드를 사용하여 모델 파라미터를 저장하고 복원할 수 있다. 그 외에는 모두 수동으로 저장하고 복원해야 한다.

10.2.7. 콜백 사용하기

• fit() 메서드의 callbakcs 매개변수를 사용하여 케라스가 훈련의 시작이나 끝에 호출할 객체 리스트를 지정할 수 있다. 또는 에포크의 시작이나 끝, 각 배치 처리 전후에 호출할 수도 있다.
• ModelCheckpoint는 훈련하는 동안 일정한 간격으로 모델의 체크포인트를 저장한다.

import tensorflow as tf
from tensorflow.keras.callbacks import ModelCheckpoint

checkpoint_cb = ModelCheckpoint('my_keras_model.h5')
history = model.fit(X_train, y_train, epochs=10, callbacks=[checkpoint_cb])

• ModelCheckpoint에서 save_best_only=True로 지정하면 최상의 검증 세트 점수에서만 모델을 저장한다.

import tensorflow as tf
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras.models import load_model

checkpoint_cb = ModelCheckpoint('my_keras_model.h5',
                                save_best_only=True)
history = model.fit(X_train, y_train, epochs=10, 
                    validation_data=(X_valid, y_valid),
                    callbacks=[checkpoint_cb])
model = load_model("my_keras_model.h5")

• EarlyStopping 콜백은 일정 에포크 동안 검증 세트에 대한 점수가 향상되지 않으면 훈련을 멈춘다.
• 모델이 향상되지 않으면 훈련이 자동으로 중지되기 때문에 에포크의 숫자를 크게 지정해도 된다.
• restore_best_weights=True : 최상의 모델 복원 기능 설정
  • 훈련이 끝난 후 최적 가중치 바로 복원

from tensorflow.keras.callbacks import EarlyStopping

early_stopping_cb = EarlyStopping(patience=10,
                                  restore_best_weights=True)
history = model.fit(X_train, y_train, epochs=100,
                    validation_data=(X_valid, y_valid),
                    callbacks=[checkpoint_cb, early_stopping_cb])

10.3.1 은닉층 개수

• 심층신경망은 복잡한 함수를 모델링하는 데 얕은 신경망보다 훨씬 적은 수의 뉴런을 사용하므로 동일한 양의 훈련 데이터에서 더 높은 성능을 낼 수 있다.
• 하나 또는 두 개의 은닉층만으로도 많은 문제를 꽤 잘 해결할 수 있다.

10.3.2 은닉층의 뉴런 개수

• 네트워크가 과대적합이 시작되기 전까지 점진적으로 뉴런 수를 늘릴 수 있다.
• 과대적합되지 않도록 조기 종료나 규제 기법을 사용

References

1. handson-ml2-10-slides ↩

2. handson-ml2-10-slides ↩

3. handson-ml2-10-slides ↩