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

12.3.1 사용자 정의 손실 함수

Huber 손실

출처 : http://doc.mindscale.kr/km/data_mining/dm02.html

• 훈련 데이터셋에 outlier가 존재한다고 가정해보자.
• mse를 사용하면 outlier로 인해 손실함수의 크기가 너무 커지기 때문에 적절치 않다.
• mae를 사용하면 outlier에 대한 규제가 너무 적기 때문에 또한 적절치 않을 것이다.
• 이럴 때 Huber 손실 함수를 사용한다.
• 위 식에서 $e$는 true y와 pred y의 차이인 error를 뜻한다.
• error가 $\delta$ 보다 작거나 같으면, 손실함수로 $frac{1}{2}e^2$를 적용한다.
• 다른 경우엔 $\delta(|e| - frac{1}{2} \delta)$를 적용한다.
• Huber 손실은 공식 keras api에서는 제공하지 않고 tf.keras의 keras.losses.Huber 클래스를 사용하면 된다.
• 그럼에도 불구하고 직접 사용자 정의 함수를 생성해보면 다음과 같다.

def create_huber(threshold=0.1):
  def huber_fn(y_true, y_pred):
    error=y_true - y_pred
    is_small_error=tf.abs(error)<threshold
    squared_loss=tf.square(error)/2
    linear_loss=threshold*tf.abs(error) - threshold**2/2
    return tf.where(is_small_error, squared_loss, linear_loss)
  return huber_fn

• 사용자 정의 손실함수 사용시 mode.compile()의 loss parameter에 함수명을 적으면 된다. ex) `model.compile(loss=create_huber(2.0), optimizer='nadam')

12.3.2 사용자 정의 요소를 가진 모델을 저장하고 로드하기

model = keras.models.load_model('my_model_with_a_custom_loss_threshold_2.h5', 
                                custom_objects={'huber_fn' : create_huber(2.0)})

• 만약 모델에 threshold를 2로 사용하였다면(기본 매개변수 값을 사용하지 않았다면), 모델을 저장할 때 threshold값은 저장되지 않기 때문에 load시 인자로 다시 넣어줘야 한다.
• 사용자 정의 객체를 포함한 모델을 로드할 때는 위와 같이 이름과 객체를 매핑시켜줘야 한다.

keras.losses.Loss 상속해서 만들기

클래스 생성

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
    square_loss=tf.square(error)
    linear_loss=self.threshold*tf.abs(error)-self.threshold**2/2
    return tf.where(is_small_error, square_loss, linear_loss)
  def get_config(self):
    base_config=super().get_config() #model 구성 반환
    return {**base_config, 'threshold' : self.threshold} #여러개의 model 구성(**base_config)과 설정해 놓은 threshold반환

모델 컴파일

model.compile(loss=HuberLoss(2.), optimizer='nadam')

모델 로드

model=keras.models.load\_model('my\_model\_with\_a\_custom\_loss\_class.h5', custom\_objects={'HuberLoss' : HuberLoss})

• 위와 같이 keras.losses.Lose를 상속하여 사용자 정의 클래스를 만들 수 있다.
• 상속을 통해 사용자 정의 클래스를 만들면, 모델 저장시 케라스가 손실 객체의 get_config()메서드를 통해 반환된 설정을 JSON 형태로 저장한다.
• 그리고 모델을 로드하면 HuberLoss 클래스의 from_config() 클래스 메서드(Loss에 구현되어 있음)를 호출한다. 생성자에게 ** config 매개변수를 전달해 클래스의 인스턴스를 만든다.

12.3.3 활성화 함수, 초기화, 규제, 제한을 커스터마이징 하기

• loss, regularizer, constraint ,initializer, activation function, layer, model등의 요소는 커스터마이징이 가능하다.

def my_softplus(z):
  return tf.math.log(tf.exp(z) + 1.0)

def my_glorot_initializer(shape, dtype=tf.float32):
  stddev = tf.sqrt(2. / (shape[0] + shape[1]))
  return tf.random.normal(shape, stddev=stddev, dtype=dtype)

def my_l1_regularizer(weights): #
  return tf.reduce_sum(tf.abs(0.01 + weights))

def my_positive_weights(weights): #tf.nn.relu() 혹은 keras.constraints.nonneg()와 동일
  return tf.where(weights < 0. , tf.zeros_like(weights), weights)

• 이렇게 만들어진 함수들은 케라스에서 제공하는 함수와 같이 모델 생성에 사용 가능하다.

layer=keras.layers.Dense(30, activation=my_softplus,
                         kernel_initializer=my_glorot_initializer,
                         kernel_regularizer=my_l1_regularizer,
                         kernel_constraint=my_positive_weights)

• 앞서 사용자 정의 함수를 통해 저장해야 될 파라미터까지 모델로서 저장하는 방법(ex - threshold)을 알아보았다.
• 같은 방법으로 부여한 하이퍼 파라미터 값을 저장하고 싶다면 사용자 정의 함수를 만들어야 한다.

class MyL1Regularizer(keras.regularizers.Regularizer):
  def __init__(self, factor):
    self.factor=factor
  def __call__(self, weights):
    return tf.reduce_sum(tf.abs(self.factor * weights))
  def get_config(self):
    return {'factor' : self.factor}

• 손실, 층, 모델의 경우 call()메소드를 구현해야 한다.
• 규제, 초기화, 제한의 경우 __ call__() 메소드를 구현해야 한다.

12.3.4 사용자 정의 지표

손실과 지표의 차이점

• 손실(loss)과 지표(metric)는 개념적으로는 같다.

출처 : http://doc.mindscale.kr/km/data_mining/dm02.html

• 손실 중 대표적으로 후버 손실을 예로 들면, 후버 손실에서 e는 error로 true y와 pred y의 차이를 뜻한다. 결국 true y와 pred y의 차이를 계산하되, 약간의 규제를 가하는 식이라는 것을 알 수 있다.

출처 : https://dailyheumsi.tistory.com/167

• 지표 중 대표적으로 mse를 예로 들면, mse 또한 true y와 pred y의 차이를 계산하는 식이라는 것을 알 수 있다.
• 결국 손실과 지표는 둘 다 예측 값과 실제 값을 가지고 오차를 계산하는 식인 것이다.
• 하지만, 손실은 weight을 update하기 위해 gradient descent algorithm에서 사용되므로 미분 가능해야 하고, gradient가 모든 곳에서 0이 아니어야 하지만, 지표는 그러한 조건이 필요 없다는 점이 차이점이다.

스트리밍 지표(Streaming Metric)

• 배치가 거듭될 수록 udpate되는 지표를 스트리밍 지표 혹은 상태가 있는 지표(stateful metric)라고 한다.

precision=keras.metrics.Precision()
precision([0,1,1,1,0,1,0,1],[1,1,0,1,0,1,0,1]) 
#<tf.Tensor: shape=(), dtype=float32, numpy=0.8>

• 첫 번째 배치가 [0,1,1,1,0,1,0,1] 의 값을 예측했다고 생각하고, true값이 [1,1,0,1,0,1,0,1]일 때 precision은 0.8이다.

precision([0,1,0,0,1,0,1,1],[1,0,1,1,0,0,0,0])
#<tf.Tensor: shape=(), dtype=float32, numpy=0.5>

• 두 번째 배치가 [0,1,0,0,1,0,1,1] 라고 했을 떄 정밀도는 두 번째 정밀도가 아닌, 첫 번째 배치와 두 번째 배치의 정밀도를 모두 반영한 0.5가 된다.
• result() 메소드를 사용하면 전체 정밀도가 아닌, 현재 정밀도만을 확인할 수 있다.
• variables 속성을 사용하면 true positive와 false postivie에 대한 변수를 확인할 수도 있다.
• 이렇게 Precision()클래스는 스트리밍 지표로서 사용될 수 있다.

스트리밍 지표 사용자 정의 함수

• 스트리밍 지표를 만들고 싶다면 keras.metrics.Metric 클래스를 상속하면 된다.

class HuberMetric(keras.metrics.Metric):
  def __init__(self, threshold=1.0, **kwargs):
    super().__init__(**kwargs)
    self.threshold=threshold
    self.huber_fn=create_huber(threshold)
    self.total=self.add_weight('total', initializer='zeros') #total = 후버 손실의 합
    self.count=self.add_weight('count', initializer='zeros') #count = 지금까지 처리한 샘플 수
  def update_state(self, y_true, y_pred, sample_weight=None):
    metric=self.hubber_fn(y_true, y_pred)
    self.total.assign_add(tf.reduce_sum(metric))
    self.count.assign_add(tf.cast(tf.size(y_true), tf.float32))
  def result(self):
    return self.total / self.count
  def get_config(self):
    base_config=super().get_config()
    return {**base_config, 'threshold' : self.threshold}

12.3.5 사용자 정의 층

가중치가 있는 층

• 텐서플로우에 없는 층을 만들 대 사용자 정의 층을 생성한다.
• 만약 A, B, C, A, B, C, A, B, C로 이뤄진 층을 생성하고 싶다면, 사용자 정의 층을 정의해 D, D, D (D = A, B, C)로 정의하는 것이 가능해진다.
• keras.layers.Flatten이나 keras.layers.ReLU와 같은 층은 가중치가 없다.
• 이러한 가중치가 없는 층은 간단하게 keras.layers.Lambda층으로 감싸는 것으로 가능하다.

exponential_layer=keras.layers.Lambda(lambda x: tf.exp(x))

• 이렇게 입력에 지수함수를 적용해 출력으로 내보내는 층을 Lambda함수를 이용해 정의하는 것이 가능하다.

사용자 정의 Dense layer

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

  def build(self, batch_input_shape):
    #batch shape만큼 weights 생성
    self.kernel=self.add_weight(
        name='kernel', shape=[batch_input_shape[-1], self.units],
        initializer='glorot_normal')
    #units 수 만큼 bias 생성
    self.bias=self.add_weight(
        name='bias', shape=[self.units], initializer='zeros')
    #부모 클래스의 build() 메소드를 호출해줌으로서 self.build = True로 변함.
    #이에 따라 케라스가 layer가 생성되었다는 것을 인식함.
    super().build(batch_input_shape) 

  def call(self, X): #output
    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,
            #serialize를 사용해 전체 설정 저장
            'activation' : keras.activations.serialize(self.activation)} 

• 이렇게 정의한 Dense layer의 형태를 띠는 사용자 정의 층은 아래와 같이 모델의 층으로 사용할 수 있다.

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

다중 입력 model(Concatenate층)

class MyMultiLayer(keras.layers.Layer):
  def call(self.X):
    X1, X2 = X
    return [X1 + X2, X1 * X2, X1 / X2]

  def compute_output_shape(self, batch_input_shape):
    b1, b2=batch_input_shape
    return [b1,b1,b1]

input_A = keras.layers.Input(shape=X_train_scaled_A.shape[-1])
input_B = keras.layers.Input(shape=X_train_scaled_B.shape[-1])
hidden_A, hidden_B = MyMultiLayer()((input_A, input_B))
hidden_A = keras.layers.Dense(30, activation='selu')(hidden_A)
hidden_B = keras.layers.Dense(30, activation='selu')(hidden_B)
concat = keras.layers.Concatenate()((hidden_A, hidden_B))
output = keras.layers.Dense(1)(concat)
model = keras.models.Model(inputs=[input_A, input_B], outputs=[output])

• 다중 입력 모델을 위해서는 call()메서드에 tuple형식으로(입력들을 하나의 tuple로) 입력들을 모두 인자로서 전달해줘야 한다.
• 2개의 입력이 있다고 했을 때, X1, X2 = (input1, input2)를 통해 X1=input1, X2=input2로 나누어 준다.
• 또한 output layer도 출력마다 하나씩 배치 출력 크기의 리스트를 반환해야 한다.

훈련과 테스트에서 다르게 동작하는 층

class MyGaussianNoise(keras.layers.Layer):
  def __init__(self, stddev, **kwargs):
    super().__init__(**kwargs)
    self.stddev=stddev

  def call(self, X, training=None):
    if training: #training이 true면, noise를 더함
      noise=tf.random.normal(tf.shape(X), stddev=self.stddev)
      return X+noise
    else: #training이 false, 즉 test중이면 X를 그대로 return함
      return X

  def compute_output_shape(self, batch_input_shape):
    return batch_input_shape

model = keras.models.Sequential([
    AddGaussianNoise(stddev=1.0),
    keras.layers.Dense(30, activation="selu"),
    keras.layers.Dense(1)
])

• 훈련과 테스트에서 다르게 동작하는 Dropout, Batchnormalization등은 call 메소드 안에 training 매개변수를 추가해 훈련과 테스트 두 상황에 따라 다르게 적용할 수 있다.
• 위 예제는 training에서는 Gaussian noise를 더하여 학습하고 test에서는 noise 없이 예측하는 layer를 구현한 것이다.
• keras.layers.GaussianNoise 를 통해 손쉽게 사용하는 것도 가능하다.

12.3.6 사용자 정의 모델

• 사용자 정의 모델을 만드려면 keras.Model 클래스를 상속해 생성자에서 층과 변수를 만들고 모델이 해야 할 작업을 call()메소드에 구현하면 된다.
• 먼저 위 그림과 같은 모델을 구현해본다.
• 위 모델에서 ResidualBlock은 전(prior)층의 입력을 두 개의 Dense layer와 skip 연결 layer로 구성된 모델을 반복 통과시킨다.

#사용자 정의 층 생성
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: #layer개수만큼 반복
      Z=layer(Z) #self.hidden에 포함된 반복되는 layer에 inputs 통과
    return inputs+Z #Dense layer 통과 inputs와 기존 inputs를 더해서 return

#사용자 정의 모델 생성
class ResidualRegressor(keras.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) #neuron 30개의 hidden Dense layer 2개 통과 + inputs
    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): #ResidualBlock 반복 통과
      Z=self.block1(Z)
    Z=self.block2(Z)
    return self.out(Z)

model = ResidualRegressor(1)
model.compile(loss="mse", optimizer="nadam")
history = model.fit(X_train_scaled, y_train, epochs=5)
score = model.evaluate(X_test_scaled, y_test)
y_pred = model.predict(X_new_scaled)

• 만약 save()메소드를 이용해 모델을 저장하고 keras.models.load_model() 함수를 사용해 모델을 로드하고 싶다면, 두 클래스에 각각 get_config()메소드를 구현해 놓아야 한다.
• 이렇게 Model과 Layer 클래스를 상속해서 모델과 층을 생성할 수 있다.
• Model은 Layer클래스의 서브클래스이므로 여러가지 추가적인 기능들을 구현 가능하다. 예를 들면 compile(), fit(), evaluate(), predict() 등이 있다. 하지만, 층을 Model 클래스가 아닌 Layer클래스를 통해 상속받는 것은 모델안의 구성요소와 모델을 분리하기 위해서이다.

12.3.7 모델 구성 요소에 기반한 손실과 지표

• 앞서 정의한 사용자 손실과 지표는 모두 레이블(y true)과 예측(y pred)을 기반으로 한다.
• 은닉층의 가중치나 활성화 함수 등과 같이 모델의 구성 요소에 기반한 손실을 정의해야 할 때가 있다.

class ReconstructingRegressor(keras.Model):
  def __init__(self, output_dim, **kwargs):
    super().__init__(**kwargs)
    self.hidden=[keras.layersDense(30, activation='selu', 
                                   kernel_initializer='lecun_normal') for _ in range(5)]

  #모델 입력 재구성 layer
  def build(self, batch_input_shape):
    n_inputs=batch_input_shape[-1]
    self.reconstruct=keras.layers.Dense(n_inputs)
    super().build(batch_input_shape)

  def call(self, inputs):
    Z=inputs
    for layer in self.hidden:
      Z=layer(Z)
    reconstruction=self.reconstruct(Z)
    recon_loss=tf.reduce_mean(tf.sqaure(recontruction-inputs)) #재구성 손실
    self.add_loss(0.05*recon_loss) #손실 리스트에 (재구성손실*0.05)를 더함
    return self.out(Z)

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

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

w1, w2 = 5, 3
eps=1e-6
(f(w1+eps, w2) - f(w1, w2)) / eps #w1 미분 
(f(w1, w2+eps) - f(w1, w2)) / eps #w2 미분

• 도함수를 계산하는 것은 변화분을 측정하는 것으로 가능하다.
• 하지만 이 방법을 사용하려면 미분마다 함수 f()를 호출해야하기 때문에 대규모 신경망에서는 훈련 속도 측면에서 비효율적이다.
• 이를 해결하기 위해 tensorflow의 자동미분을 활용할 수 있다.

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

dz_dw1=tape.gradient(z, w1)
dz_dw2=tape.gradient(z, w2)
del tape

• 두 변수를 정의하고 gradiet tape 블럭에 이 변수와 관련된 모든 연산을 자동으로 기록한다.
• 마지막으로 변수 w1에 대한 z의 그레디언트와 변수 w2에 대한 z의 그레디언트를 요청한다.
• 이렇게 관련 연산만 기록해 놓으면 gradient()메소드를 호출하는 것 만으로 한 번에 미분이 가능하다는 것(기울기를 확인할 수 있다는 것)을 확인할 수 있다.
• gradient() 메소드가 호출된 이후에는 tape가 자동으로 지워진다. 때문에 메소드를 한 번 이상 호출해야 한다면 지속 가능한 테이프를 만들고 사용이 끝난 테이프는 del tape를 통해 삭제해줘야 한다.

c1, c2 = tf.constant(5.), tf.constant(3.)
with tf.GradientTape() as Tape:
    tape.watch(c1)
    tape.watch(c2)
    z=f(c1, c2)

gradients=tape.gradient(z, [c1, c2])

• 기본적으로 tape는 변수(tf.Variable)만 기록하기 때문에 상수인 tf.constant는 기록하지 못한다.
• 하지만, tape.watch()를 사용하면 텐서의 타입에 관계없이 강제로 기록하는 것이 가능해진다.
• 이러한 강제적인 기록은 입력이 작을 때 변동 폭이 큰 활성화 함수에 대한 규제 손실을 구현할 떄 유용하다. 손실은 활성화 함수의 gradient에 기반하고, 이 경우 입력은 tf.Variable이 아니기 때문에 강제적으로 기록하는 watch()메소드를 사용하는 것이 유용하게 쓰일 것이다.

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

with tf.GradientTape() as tape:
    z=f(w1, w2)

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

• 신경망에서 gradient가 역전파되지 않도록 막아야 할 때도 있다. 이 때는 위와 같이 tf.stop_gradient()를 사용한다.
• 정방향 계산 때는 입력을 반환하고 역방향 계산을 통해 gradient를 활용할 때는 gradient를 전파하지 않는다(상수처럼 동작한다).

def my_softplus(z):
    return tf.math.log(tf.exp(z)+1.0)

x=tf.Variable([100.])
with tf.GradientTape() as tape:
    z=my_softplus(x)

tape.gradient(z, [x])

• 위와 같이 gradient가 nan이 나올 경우가 있다. 자동 미분을 사용하여 gradient를 사용하는 것이 수치적으로 불안정하기 때문(부동소수점 정밀도 오류로 인해 자동 미분이 무한 나누기 무한 계산을 하게 됨)이다.

@tf.custon_gradient
def my_better_softplus(z):
    exp=tf.exp(z)
    def my_softplus_gradients(grad):
        return grad / (1 + 1 / exp)
    return tf.math.log(exp+1), my_softplus_gradients

• 위와 같이 일반 출력과 도함수를 계산하는 함수를 반환하면 해결 가능하다.

12.3.9 사용자 정의 훈련 반복

• fit() 메소드의 유연성은 크지 않다. 예를 들어 와이드 네트워크와 딥 네트워크 두 개에 다른 optimizer를 부여하고 싶을 경우 fit()메소드로는 구현이 불가능하다.
• 이러한 문제점을 해결하기 위해서는 훈련 과정을 직접 구현하는 수 밖에 없다.

모델 생성

l2_reg=keras.regularizers.l2(0.05)
model=keras.models.Sequenatial([
    keras.layers.Dense(30, activation='elu', kernel_initializer='he_normal',
                       kernel_initializer=l2_reg),
    keras.layers.Dense(1, kernel_initializer=l2_leg)                       

])

훈련 세트에서 샘플 배치를 랜덤 추출하는 함수 생성

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

• 이렇게 랜덤하게 샘플 배치를 추출하는 함수를 만들어도 되지만, 13장에서 사용하는 Data API가 더 활용성이 좋다.

현재 스텝 수, 전체 스텝 횟수, 에포크 시작부터 평균 손실 등의 훈련 상태를 출력하는 함수

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)

hyperparamter, optimizer, loss function, metrics 정하기

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

훈련 정의

for epoch in range(1, n_epochs+1):
    print("에포크 {}/{}".format(epoch, n_epochs)) #에포크 출력
    for step in range(1, n_steps + 1): 
        X_batch, y_batch = random_batch(X_train_scaled, y_train) #각 step마다 sample batch random 추출
           with tf.GradientTape() as tape():
            y_pred=model(X_batch, training=True) 
            main_loss=tf.reduce_mean(loss_fn(y_batch, y_pred))
            loss=tf.add_n([main_loss] + model.losses) #main loss에 규제 손실(le_reg) 더함
        gradients = tape.gradients(zip(gradients, model.trainable_variables)) #훈련 가능한 변수의 gradient계산
        optimizer.apply_gradients(zip(gradients, model.trainable_variables)) #위에서 계산한 gradients를 optimizer에 적용
        mean_loss(loss) #
        for metric in metrics:
            metric(y_batch, y_pred)
        print_status_bar(step * batch_size, len(y_train), mean_loss, metrics)
    print_status_bar(len(y_train), len(y_train), mean_loss, metrics)
    for metric in [mean_loss] + metrics:
        metric.reset_states()

• 위와 같은 사용자 정의 훈련은 Batch Nomalization이나 Dropout과 같이 훈련과 테스트 시에 다르게 동작하는 층은 다루지 못한다는 점이 단점이다.
• 만약 훈련과 테스트를 구분하고 싶다면 training=True로 모델을 호출하여 필요한 모든 층에 이 매개변수가 전파되도록 해야 한다.