2023/02/23 ~ 2023/03/08

[njs03332]

• 1: 12.3.4
• 2: 12.3.5
• 3: 12.3.6
• 3/8 수요일 9:30pm

[njs03332]

12.3.6 사용자 정의 모델

• 사용자 정의 모델 클래스 만들기

  • keras.Model 클래스를 상속하여 생성자에서 층과 변수를 만들고 call() 메서드에서 모델이 해야할 작업 정의

• 사용자 정의 모델 예시 (스킵 연결이 있는 사용자 정의 잔차블록 (ResidualBlock) 층을 가진 모델)

• 동일한 블록을 여러 개 만들어야 하므로 ResidualBlock 층을 먼저 만듦

  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 속성을 감지하고 필요한 변수를 자동으로 이 층의 변수 리스트에 추가함

• 서브클래싱 API를 사용해 모델 정의

  class ResidualRegressor(tf.keras.Model):
  def __init__(self, output_dim, **kwargs):
      super().__init__(**kwargs)
      self.hidden1 = tf.keras.layers.Dense(30, activation='elu',
                                           kernel_initializer='he_normal')
      self.block1 = ResidualBlock(2, 30)
      self.block2 = ResidualBlock(2, 30)
      self.out = tf.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() 메서드에서 사용
  • 다른 일반 모델 처럼 사용 가능
  • save(), load_model() 메서드를 사용하고 싶다면 ResidualBlock 클래스와 ResidualRegressor 클래스에서 모두 get_config() 메서드를 구현해야 함
  • save_weights(), load_weights() 메서드를 사용해 가중치 저장 & 로드할 수 있음

• Model 클래스는 Layer 클래스의 서브클래스이지만 추가 기능이 있음

  • compile(), fit(), evaluate(), predict(), get_layers(), save() 등

[danbi5228]

12.3.5 사용자 정의 층

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

# case1. 가중치가 필요없는 가장 간단한 사용자 정의 층 만들기
exponential_layer = keras.layers.Lambda(lambda x: tf.exp(x))

# case2. 상태가 있는, 가중치를 가지는 사용자 정의 층 만들기: keras.layser.Layer 상속 필요
class MyDense(keras.layers.Layer):

    # 모든 하이퍼파라미터를 매개변수를 받음 (units, activation)
    # 부모 생성자 kwargs; input_shape, trainable, name 과 같은 기본 매개변수 처리
    def __init__(self, units, activation=None, **kwargs):
        super().__init__(**kwargs)
        self.units = units
        self.acitivation = keras.activations.get(activation) # 적절한 활성화 함수로 바꾸는데 이용

    # weight 초기화; 가중치마다 add_weight 메서드를 호출하여 층의 변수 생성
    def build(self, batch_input_shape):
        self.kernel = self.add_weight(
            name='kernel', shape=[batch_input_shape[-1], self.units],
            initailizer='glorot_normal')
        self.bias = self.add_weight(
            name='bias', shape=[self.units], initializer="zeros"
        )
        super().build(batch_input_shape) # 이 명령어를 통해 층이 만들어 졌다는 것을 keras가 인식함(마지막에 호출 필수)

    # 층에서 필요한 연산 수행
    # 이 경우 행렬 곱셈 후 편향을 더한 뒤 활성화 함수 적용 - 이 값이 층의 출력
    def call(self, X):
        return self.activation(X @ self.kernel + self.bias)

    # 이 층의 출력 크기를 반환
    # 예시의 경우 마지막 차원(뉴런 개수0을 제외하고 입력과 크기가 같음.
    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, "unit":self.units, "activation": keras.activations.serialize(self.activation)}

# case 3. 여러가지 입력을 받는 층을 만들고 싶을 때 + 여러 출력을 가진 층을 만들고 싶을 때
# 두개의 입력 층과 세개의 출력 층을 만드는 경우
# 이 경우, 함수형 API와 서브클래싱API에만 사용할 수 있음
# 하나의 입력과 하나의 출력 층만 사용하는 시퀀셜 API에는 사용할 수 없음
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] 

# case4. 훈련과 테스트에서 다르게 동작하는 층이 필요한 경우
# 훈련 과정에서 가우시안 잡음을 추가해준 형태 - 테스트 시에는 아무것도 하지 않음
# keras.layers.GussianNoise 층과 동작이 동일
class MyGaussianNoise(keras.layers.Layer):
    def __init__(self, stddev, **kwargs):
        super().__init__(**kwargs)
        self.stddev = stddev

    def call(self, X, training=None):
        if training:
            noise = tf.random.normal(tf.shape(X), stddev=self.stddev)
            return X+noise
        else:
            return X

    def compute_output_shape(self, batch_input_shape):
        return batch_input_shape

[givitallugot]

12.3.4 사용자 정의 지표

• 손실: (ex 크로스 엔트로피) 미분 가능해야 하고 그레이디언트가 모든 곳에서 0이 아니어야 함
• 지표: 미분이 가능하지 않거나 모든 곳에서 그레이디언트가 0이어도 괜찮음
• 그러나 사용자 지표 함수를 만드는 것은 사용자 손실 함수를 만드는 것과 동일
• 훈련 동안 케라스는 지표를 계산하고 에포크가 시작할 때부터 평균을 기록
• 그러나 그냥 평균을 내면 안됨 (첫 번째 배치에서 5개 중에 4개 맞고, 두 번째 배치에서 3개 중에 0개 맞았다면? 80%와 0%의 평균 40%가 아닌, (4+0)/(5+3) = 50%여야 함
• keras.metrics.Precision: 진짜 양성 개수와 거짓 양성 개수를 기록하고 필요할 때 정밀도를 계산하는 객체
• 지금까지 진행한 배치의 정밀도, 처음에는 80%, 그 다음은 50%
• 이렇게 점진적으로 업데이트 되는 지표를 스트리밍 지표라고 함

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>
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>

• result() 메서드를 호출하여 현재 지푯값을 얻을 수 있음
• variables 속성을 사용하여 변수를 확인할 수도 있음

precision.result()
# <tf.Tensor: shape=(), dtype=float32, numpy=0.5>
precision.variables
# [<tf.Variable 'true_positives:0' shape=(1,) dtype=float32, numpy=array([4.], dtype=float32)>,
#  <tf.Variable 'false_positives:0' shape=(1,) dtype=float32, numpy=array([4.], dtype=float32)>]
precision.reset_states()

• 이러한 스트리밍 지표를 만들고 싶다면 keras.metrics.Metric 클래스를 상속
• 다음은 전체 후버 손실과 지금까지 처리한 샘플 수를 기록한 클래스

  class HuberMetric(keras.metrics.Metric):
  def __init__(self, threshold=1.0, **kwargs):
      super().__init__(**kwargs) # 기본 매개변수 처리 (예를 들면, dtype)
      self.threshold = threshold
      self.huber_fn = create_huber(threshold)
      self.total = self.add_weight("total", initializer="zeros")
      self.count = self.add_weight("count", initializer="zeros")
  def update_state(self, y_true, y_pred, sample_weight=None):
      metric = self.huber_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}