[HandsOn]12. 텐서플로를 사용한 사용자 정의 모델과 훈련 - 연습문제

https://github.com/rickiepark/handson-ml2/blob/master/12_custom_models_and_training_with_tensorflow.ipynb

GitHub - rickiepark/handson-ml2: 핸즈온 머신러닝 2/E의 주피터 노트북

답안은 여기를 참고하였다!

정말 어렵지만 내맘대로 커스텀한 모델을 짜기 위해선 필수적이니 꼭 알아두도록 하자. 

절대 첫 번째에 익숙해지지 않으니 보고 또 볼 것!

 

12. 사용자 정의 모델과 훈련 알고리즘

1. 사용자 정의 손실 함수

후버 손실(keras.losses.Huber)을 없다고 생각해보고 짜도록 하자.

이 때 모델을 저장하고 로드하는 것까지 생각해야 한다.

 

후버 손실에서 threshold를 매개변수로 받아야 한다면? ( 1.0 이하면 이 함수를 쓰고, 이상이면 이 함수를 써라 이런식으로 커스텀을 하고 싶다면) 

1. keras.losses.Loss 클래스를 상속

2. get_config() 메서드를 구현

하여 해결할 수 있다. ( 이해가 안되므로 외우자 그냥 )

 

import tensorflow as tf
import keras

class HuberLoss(keras.losses.Loss):
    def __init__(self,threshold=1.0,**kwargs):
        self.threshold=threshold
        super().__init__(**kwargs)
    def call(self, y_true, y_pred):
        error=y_true-y_pred
        is_small_error=tf.abs(error) < self.threshold
        squared_loss=tf.square(error)/2
        linear_loss=self.threshold*tf.abs(error)-self.threshold**2/2
        return tf.where(is_small_error,squared_loss,linear_loss)
        def get_config(self):
            base_config=super().get_config()
            return {**base_config,"threshold":self.threshold}

생성자에서 기본적인 하이퍼 파라미터를 상속을 받고, threshold도 지정을 해 줌.

call 메서드에서 손실 계산하여 반환.

get_config 메서드는 모르겠당.. 그냥 매개변수 받을 때 갖다 붙이는듯... 

 

이렇게 해 놓으면 모델을 컴파일할때 이 클래스의 인스턴스를 사용할 수 있다. 

모델을 저장할 때 threshold도 같이 저장되고, 로드할 때 클래스 이름과 클래스를 매핑해 주면 된다. 

 

2. 사용자 정의 손실 지표

비슷함.. (지금까지 해 왔던 것처럼)지표를 정의만 하면 케라스가 자동으로 함수 호출하고 에포크동안 평균 기록해줌. 

클래스를 일일히 다 정의해주는 거는

1. 매개변수를 저장할 때!

2. 배치가 계속될 때마다 전체 배치를 포함하는 '정밀도' 같은 '스트리밍 지표'를 구현할 때!

 

3. 사용자 정의 층

텐서플로에 없는 특이한 층을 가진 네트워크를 만들어야 할 때가 있음!

가중치를 가진 층을 만들려면 keras.layers.Layer를 상속해야 한다. 

class MyDense(keras.layers.Layer):
    def __init__(self, units, activation=None, **kwargs):
        super().__init__(**kwargs)
        self.units = units
        self.activation = keras.activations.get(activation)

    def build(self, batch_input_shape):
        self.kernel = self.add_weight(
            name="kernel", shape=[batch_input_shape[-1], self.units],
            initializer="glorot_normal")
        self.bias = self.add_weight(
            name="bias", shape=[self.units], initializer="zeros")
        super().build(batch_input_shape) # must be at the end

    def call(self, X):
        return self.activation(X @ self.kernel + self.bias)

    def compute_output_shape(self, batch_input_shape):
        return tf.TensorShape(batch_input_shape.as_list()[:-1] + [self.units])

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "units": self.units,
                "activation": keras.activations.serialize(self.activation)}

1. 생성자는 모든 하이퍼 파라미터를 매개변수로 받음. 부모 생성자 호출하고, 하이퍼파라미터를 속성으로 저장.

2. build() 메서드의 역할은 가중치를 만드는 것.

   "kernel": 연결 가중치(입력의 마지막 차원 크기를 받음) ,

   "bias" : 편향, 0으로 초기화

3. compute_output_shape(): 이 층의 출력 크기를 반환 

4. get_config(): 앞서 보았던 것과 같음. keras.activations.serialize()를 사용하여 활성화 함수의 전체 설정 저장

model = keras.models.Sequential([
    MyDense(30, activation="relu", input_shape=input_shape),
    MyDense(1)
])

 

4. 사용자 정의 모델

 

한 마디로, keras.Model 클래스를 상속하여 생성자에서 층과 변수를 만들고 모델이 해야 할 작업을 call()메서드에 구현한다. 

이런 모델이 있다고 가정: 스킵 연결이 있는 ResidualBlock 층을 가진 예제 모델

먼저 Residual Block 층을 만들자.

 

class ResidualBlock(keras.layers.Layer):
    def __init__(self, n_layers, n_neurons, **kwargs):
        super().__init__(**kwargs)
        self.hidden = [keras.layers.Dense(n_neurons, activation="elu",
                                          kernel_initializer="he_normal")
                       for _ in range(n_layers)]

    def call(self, inputs):
        Z = inputs
        for layer in self.hidden:
            Z = layer(Z)
        return inputs + Z

층 안에 층이 포함되어 있음 !

hidden 에 층을 레이어 갯수만큼 지정해줌.

그 다음에 call()메서드에서 Z를 hidden 에 통과시키고, input 이랑 더해준다. 

 

ResidualBlock 층 완성!

 

해당 층을 이용해서 모델을 만든다. 

class ResidualRegressor(keras.models.Model):
    def __init__(self, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.hidden1 = keras.layers.Dense(30, activation="elu",
                                          kernel_initializer="he_normal")
        self.block1 = ResidualBlock(2, 30)
        self.block2 = ResidualBlock(2, 30)
        self.out = keras.layers.Dense(output_dim)

    def call(self, inputs):
        Z = self.hidden1(inputs)
        for _ in range(1 + 3):
            Z = self.block1(Z)
        Z = self.block2(Z)
        return self.out(Z)

모델을 상속받고, 생성자에 필요한 층을 정의하고( 내맘대로 적고있음 )

call() 메서드에 모델의 형태를 정의한다. 

모델 완성!

 

5.  자동 미분을 사용하여 그레이디언트 계산하기

간단한 함수로 예를 살펴보자.

def f(w1,w2):
    return 3*w1**2 + 2*w1*w2

자동 미분을 사용해 텐서플로에서 쉽게 그레이디언트를 계산해보자. 

w1,w2=tf.Variable(5.),tf.variable(3.)
with tf.GradientTape() as tape:
    z=f(w1,w2)

gradients=tape.gradient(z,[w1,w2])

gradients

Out[172]:

[<tf.Tensor: shape=(), dtype=float32, numpy=36.0>,
 <tf.Tensor: shape=(), dtype=float32, numpy=10.0>]

포인트 (5,3)에서의 그레이디언트 벡터가 (36,10) 으로 계산되었다!

 

6.  사용자 정의 훈련 반복

만약에 모델을 피팅할때, fit()의 유연성이 충분하지 않다면? (예를 들어 fit()메서드에서는 compile할 때 하나의 옵티마이저만 지원하는데, 두 개의 다른 옵티마이저를 레이어마다 쓰고 싶다면)

 

훈련 반복을 직접 구현해야 함! 중요한데 넘나어려운것...

 

간단한 모델을 만들어보자. 훈련 반복을 직접 다루기 때문에 컴파일할 필요는 없음.

그냥 간단한 시퀀셜 모델을 만들었다.

l2_reg = keras.regularizers.l2(0.05)
model = keras.models.Sequential([
    keras.layers.Dense(30, activation="elu", kernel_initializer="he_normal",
                       kernel_regularizer=l2_reg),
    keras.layers.Dense(1, kernel_regularizer=l2_reg)
])

훈련 세트에서 샘플 배치를 랜덤하게 추출하는 함수를 만들자.

def random_batch(X, y, batch_size=32):
    idx = np.random.randint(len(X), size=batch_size)
    return X[idx], y[idx]

훈련 상태를 출력하는 함수도 만든다. (사실 내가만든건 아니고 책에 적혀있는거임)

간편하게 tqdm 라이브러리를 쓸 수도 있음.

def print_status_bar(iteration, total, loss, metrics=None):
    metrics = " - ".join(["{}: {:.4f}".format(m.name, m.result())
                         for m in [loss] + (metrics or [])])
    end = "" if iteration < total else "\n"
    print("\r{}/{} - ".format(iteration, total) + metrics,
          end=end)

 

이제 몇 개의 하이퍼파라미터를 정의하고 옵티마이저, 손실 함수, 지표를 선택해야 함.

n_epochs = 5
batch_size = 32
n_steps = len(X_train) // batch_size
optimizer = keras.optimizers.Nadam(learning_rate=0.01)
loss_fn = keras.losses.mean_squared_error
mean_loss = keras.metrics.Mean()
metrics = [keras.metrics.MeanAbsoluteError()]

 

사용자 정의 훈련 반복문: print status bar를 사용하지 않고 tqdm을 사용하였다.

try:
    from tqdm.notebook import trange
    from collections import OrderedDict
    with trange(1, n_epochs + 1, desc="All epochs") as epochs:
        for epoch in epochs:
            with trange(1, n_steps + 1, desc="Epoch {}/{}".format(epoch, n_epochs)) as steps:
                for step in steps:
                    X_batch, y_batch = random_batch(X_train_scaled, y_train)
                    with tf.GradientTape() as tape:
                        y_pred = model(X_batch)
                        main_loss = tf.reduce_mean(loss_fn(y_batch, y_pred))
                        loss = tf.add_n([main_loss] + model.losses)
                    gradients = tape.gradient(loss, model.trainable_variables)
                    optimizer.apply_gradients(zip(gradients, model.trainable_variables))
#                    for variable in model.variables:
#                        if variable.constraint is not None:
#                            variable.assign(variable.constraint(variable))   
모델에 가중치 제한을 추가한 항.근데 잘 모르겠어서 주석처리함
                    status = OrderedDict()
                    mean_loss(loss)
                    status["loss"] = mean_loss.result().numpy()
                    for metric in metrics:
                        metric(y_batch, y_pred)
                        status[metric.name] = metric.result().numpy()
                    steps.set_postfix(status)
            for metric in [mean_loss] + metrics:
                metric.reset_states()
except ImportError as ex:
    print("To run this cell, please install tqdm, ipywidgets and restart Jupyter")

넘나 어려워보인다. 천천히 이해해보자. 

한 에포크 안에서 배치가 돌고 있다.

랜덤 배치 함수를 이용해서 X 배치와 y 배치를 지정하고, 

 그레이디언트 테이프를 이용해서 배치마다 나온 loss를 "훈련 가능한 변수" 로 미분한다. 

 이 값들을 optimizer에 보냄. 

 

status라는 dictionary를 만들고, 상태들을 여기에 저장.

한 에포크가 다 돌면 metric 을 리셋.

 

너무 어렵다...