사용자 정의 손실 함수

• tensorflow를 사용하면서 손실 함수를 사용자가 따로 정의하여 만들 수도 있다.
• 아래의 경우 Huber loss의 경우다.
• 아래 코드를 보면 단순 연산들도 tensorflow의 연산으로 사용한 것을 알 수 있다.
• tensorflow 그래프의 장점을 살릴려면 이처럼 tensorflow 연산만으로 만드는 것이 좋다.
• 또한 성능을 위해서는 아래처럼 벡터화 하여 구현해야 한다.

import tensorflow as tf

def huber(y_true, y_pred):
  error = y_true - y_pred
  se = tf.abs(error) < 1
  sq_loss = tf.square(error) / 2
  linear_loss = tf.abs(error) - 0.5
  return tf.where(se,sq_loss,linear_loss)

#모델 컴파일
#model.compile(loss=huber, optimizer="nadam")
#model.fit(...)

사용자 정의 요소 모델 저장 & 로드

• 모델 저장의 경우 , keras가 함수 이름을 저장하기 때문에 문제 없이 저장된다.
• 모델 로드의 경우 , 이름과 객체를 매핑 해야 한다.
• 아래 예시의 경우 모델을 저장할때 threshold 값은 저장되지 않는다.
• 따라서 모델을 로드 할때 threshold 값을 지정해야 한다.

def create_huber(threshold=1.0):
  def huber(y_true, y_pred):
    error = y_true - y_pred
    se = tf.abs(error) < threshold
    sq_loss = tf.square(error) / 2
    linear_loss = threshold * tf.abs(error) - threshold**2 / 2
    return tf.where(se,sq_loss,linear_loss)
  return huber
#model.compile(loss=create_huber(2.0), optimizer="nadam")

#model = keras.models.load_model("models.h5",
#       custom_objects={"huber": create_huber(2.0)})


#이러한 문제는 keras.losses.Loss 클래스를 상속하여
#get_config() 메소드를 구현해서 해결 할 수 있다.
#아래 코드를 확인하자.

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
    se = tf.abs(error) < self.threshold
    sq_loss = tf.square(error) / 2
    linear_loss = self.threshold * tf.abs(error) - self.threshold**2 / 2
    return tf.where(se,sq_loss,linear_loss)
  def get_config(self):
    base_config = super().get_config()
    return {**base_config, "threshold" : self.threshold}

• keras는 현재 현재 layer , 모델 , 콜백 , 규제를 상속하는 방법만 정의하고 있다.
• 손실 , 지표 , 초기화 , 제한 등 상속을 통해 다른 요소들을 만들면 기존의 keras 구현과 호환되지 않을 수도 있다.
• 일단 위의 코드를 살펴보면 먼저 생성자는 **kwargs로 받은 매개변수 값을 부모클래스의 생성자에게 전달한다.
• call() 메소드는 label과 prediction을 받고 huber loss 계산을 진행해 값을 도출한다.
• get_config() 메소드는 하이퍼 파라미터 이름과 같이 매핑된 딕셔너리를 반환한다.
• 먼저 부모 클래스의 get_config() 메소드를 호출하고 반환된 딕셔너리에 새로운 하이퍼 파라미터를 추가한다.
• 모델 저장시 get_config()로 반환된 설정들이 HDF5 파일에 JSON 형태도 저장된다.

• 모델을 로드하면 HuberLoss 클래스의 from_config() [이는 keras.losses.Loss에 구현되어 있다] 클래스 메소드를 호출해 생성자에게 config 매개변수를 전달해 주고 이를 통해 클래스의 인스턴스를 만든다.
  
  
  ### 파이썬 문법 : {x}

• * : 언패킹 연산자 / ** : 패킹 연산자

• 일반적으로 객체를 언패킹 하여 원소를 추가하고 다시 리스트나 딕셔너리를 만드는 경우 / 매개변수가 여러개인 함수에 한번에 값을 전달하고 싶은 경우에 사용한다.
• ex) param = {"a" : 1 , "b" : 2 } 라면 함수 func(**param) 을 하면 func(a=1 , b=2) 와 같아지는 것이다.

#모델을 컴파일 한다면 다음과 같이 할 수 있을 것이다.

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

• 모델을 저장 할 때 threshold 값도 같이 저장된다.
• 로드시에는 클래스 이름과 클래스 자체를 매핑 해야 한다.

#model = keras.models.load_model("models_5", 
#         custom_objects={HuberLoss : HuberLoss})

사용자 정의 커스터마이징

• 손실 , 규제 , 초기화 , 제한 , 지표 , activation function 등등 을 keras와 유사하게 커스터마이징 할 수 있다.
• 다음은 사용자 정의로 손실 , 규제 , 제한 등을 만든 코드다.
• 아래 layer에 적용된 경우를 보면 ,

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

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

def my_l1_regular(weights):
  return tf.reduce_sum(tf.abs(0.01 * weights))

def my_positive_weights(weights):
  return tf.where(weights < 0. , tf.zeros_like(weights), weights)

#위의 사용자 정의로 만든 함수들은 layer에서 일반적으로 사용하듯이 사용할 수 있다.

#layer = keras.layers.Dense(30, activation=my_softplus,
#           kernel_initializer = my_glorot_initial,
#           kernel_regularizer = my_l1_regular,
#           kernel_constraint = my_positive_weights )

사용자 정의 지표

• 배치별로 양성 음성을 나누어 정밀도를 확인한다고 생각해보자.
• 첫번째 배치에서는 5개중 4개가 맞앗고 , 두번째 배치에서는 3개중 0개가 맞았다.
• 이런 경우 80%와 0% 의 평균인 40%가 평균적인 정밀도라고 이야기 하면 안된다.
• 4 / ( 3 + 5 ) 이므로 50% 가 맞는 정밀도다.
• 이처럼 진짜 양성 , 가짜 양성 개수들을 저장하는 것처럼 정밀도를 계산할 수 있는 객체가 필요하다.
• 이것이 바로 keras.metrics.Precision 이다.

pre = keras.metrics.Precision()
pre([0,1,1,1],[1,1,1,0])
pre([0,0,0],[1,1,1])

#Precision 객체 안에 배치의 레이블과 실제 예측이 들어간다.
#이때 샘플 가중치를 매개변수로 전달 할 수 있다.
#2번째 줄에서 첫배치를 , 3번째 줄에서 두번째 배치를 처리해
#정밀도를 알려준다.

#정밀도는 이처럼 배치를 거치면서 점점 업데이트 된다.
#이러한 지표를 스트리밍 지표 라고 한다.

pre.result()
#result() 메소드로 현재 지표값을 얻을 수 있다.

pre.variables
#variables 속성으로 변수도 확인 가능하다.

pre.reset_states()
#reset_states 메소드로 변수 초기화도 가능하다.


#위와 같은 스트리밍 지표를 만들고 싶다면 , 
#keras.metrics.Metric 클래스를 상속해 만들 수 있다.

<tf.Tensor: shape=(), dtype=float32, numpy=0.33333334>

사용자 정의 층

• 일반적인 layer가 아닌 특이한 layer를 만들어야 할 경우가 있을 수 있다.
• keras.layers.Flatten 이나 keras.layers.ReLU 같은 층들은 weight가 없다.
• 가중치가 필요없는 층을 만들기 위해서는 함수를 만든 후 keras.layers.Lambda층으로 감싸는 방법으로 만들 수 있다.
• 아래는 입력에 지수 함수를 적용하는 층의 예시이다.

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

• 이러한 방식으로 만들어진 층들은 시퀀셜 , 함수형 , 서브 클래싱 API에서 일반적인 층들 처럼 사용될 수 있다.
• 가중치가 없는 층은 위처럼 만들 수 있지만 , 가중치가 있는 층들은 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)

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

• 위 코드에서 생성자는 모든 하이퍼 파라미터들을 매개 변수로 받는다.
• 이후 부모 생성자를 호출하면서 kwargs를 전달한다.
• input_shape이나 trainable, name 같은 기본적인 매개변수들을 처리할 수 있다.
• 이후 하이퍼파라미터들을 속성으로 전달하고 activation function 매개 변수를 get 함수를 통해 적절한 activation fucntion으로 바꾼다.
• build() 메소드는 add_weight()를 호출해 층의 변수를 만드는 것이다.
• build()는 처음 층이 만들어 질 때 호출되며 , 가중치를 만들 떄 크기가 필요한 경우도 있다.
• 이 크기는 입력의 마지막 차원 크기에 해당된다.
• call() 메소드는 층에 필요한 연산을 수행한다. 위 코드는 X와 행렬곱을 수행한다.
• compute_output_shape()은 층의 출력 크기를 반환한다.
• 크기는 tf.TensorShape 객체이고 as_list()로 파이썬 리스트로 바꿀 수 있다.

• get_config는 activation fucntion의 전체 설정을 저장한다.
  
  
  

• 만약 여러 입력을 받는 층을 만들고 싶다면 , call() 쪽에서 매개변수를 튜플로 입력 받아야 할 것이다.

• Train과 Test에서 다르게 동작하는 층이 필요하면 , call() 메소드에 training 매개변수를 추가해 if training == True: 면 이런식으로 작동하고 , False 면 이런식으로 작동하고 나누어서 만들 수 있다.

사용자 정의 모델

• 기존에 존재하지 않는 특별한 모델을 만들기 위해서는 어떠한 구조의 모델이던지 모두 만들 수 있어야 한다.
• 사용자 정의 모댈의 경우 keras.Model 클래스를 상속해서 만들 수 있다.
• def __init__ 생성자 에서 모델 layer의 구조를 만들고 , call() 메소드에서 이를 사용한다.
• 만약 save()를 통해 저장한 모델을 불러오고 싶다면 , 사용하는 모델이나 층에 get_config() 메소드를 구현해놓아야 한다.
• 또한 가중치 저장을 위한 save_weights() , 불러오기를 위한 load_weights()를 모두 사용해야 한다.
• keras.Model을 상속받았으므로 compile() , fit() , evaluate(), predict() 등등 모델과 관련된 다양한 메소드를 사용할 수 있다.

모델 구성요소로 만드는 손실과 지표

• 직접 만든 모델의 layer안의 weight나 activation fucntion 같은 구성요소를 가지고 loss를 만들거나 지표를 만들어야 할 수도 있다.
• 모델을 만들고 모델 내부 call() 메소드에 self.add_loss() 메소드를 통해 계산된 모델의 손실을 loss에 추가한다.
• 이와 비슷하게 임의의 계산을 수행하는 사용자 정의 지표를 만들 수 있다.

자동 미분으로 경사 계산하기

• 차원이 늘어날 때마다 또 차수가 늘어날 떄마다 , 항이 늘때 마다 도함수를 구하기 위해 미분하는 것은 점점 복잡해진다.
• 손으로 도함수를 구하는 방법 이외에 좀 더 편한 방법은 파라미터가 바뀔 때 마다 함수의 출력이 얼마나 변하는지 측정하여 도함수의 근삿값을 계산하는 것이다.
• 아래의 예시 코드를 살펴보자.

#수동 미분

w1 , w2 =  5 , 3
eps = 1e-6

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

#f 함수는 3w1^2 + 2w1w2 이고 이 함수를 미분 즉 도함수를 구한다고 가정
#w1 에 대한 도함수는 6*w1 + 2*w2
#w2 에 대한 도함수는 2*w1

#(5,3)을 대입했을 떄 경사 벡터는 (36,10)

print("w1에 대한 경사 값")
print((f(w1+eps,w2) - f(w1,w2)) / eps)
print("w2에 대한 경사 값")
print((f(w1, w2+eps) - f(w1,w2)) / eps) 

#비슷하게 나오는 것을 알 수 있다.
#이는 f(x+a) - f(x) / (x+a) - x 같은 식으로 a를 작게 만들어서
#limit로 미분하는 방식과 같다.

w1에 대한 경사 값
36.000003007075065
w2에 대한 경사 값
10.000000003174137

#자동 미분

#위의 수동 미분은 미분시 함수 f()를 계속 호출하므로 대규모 Network에서는 사용하기 어렵다.
#텐서플로우 에서는 다음과 같이 자동미분을 구현해 쉽게 미분이 가능하다.

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

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

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

• 위의 자동 미분 코드의 경우 변수 w1 , w2를 정의한 후 tf.GradientTape 블럭을 만들어 이 변수들과 관련된 연산을 모두 저장한다.
• 이후 이 테이프에 두 변수에 대한 z의 경사를 요구한다.
• tf.GradientTape() 블록안에 최소한의 코드만 담는 것이 메모리를 줄이는 최고의 방법이다.
• 다른 방법은 with tape.stop_recordind() 블록을 만들어 계산 기록을 저장하지 않을 수도 있다.
• gradient() 메소드가 호출된 이후 자동으로 tape가 사라진다. 따라서 gradient() 메소드를 두번 호출하면 예외로 실행 에러가 발생한다.
• 만약 2번 이상 사용해야 한다면 지속 가능한 tape를 만들고 계산이 끝난 tape은 del로 삭제 후 사용해야 한다.
• 아래 코드는 이와 관련된 코드다.

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

gt1 = tape.gradient(z, w1)
print(gt1)
gt2 = tape.gradient(z, w2)
print(gt2)
del tape

tf.Tensor(36.0, shape=(), dtype=float32)
tf.Tensor(10.0, shape=(), dtype=float32)

#기본적으로 tape은 변수가 포함된 연산만 기록한다.
#아래는 Variable을 constant 즉 상수로 두고 한 예시이다.

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

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

#확인 경과 [None, None] 이 출력됨

[None, None]

#변수가 아닌 것들을 가지고 하려면 다음과 같이 하면 된다.

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

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

#예를 들어 입력이 작은 경우 변동 폭이 큰 activation function에
#대하여 규제 손실을 구현하는 경우 입력은 변수가 아니므로 이런식으로
#구현해야 할 것이다.

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

#모델을 만들다 보면 경사 역전파를 막아야 하는 경우도 있다.
#이때는 tf.stop_gradient() 함수를 사용한다.
#이 함수는 정방향 계산시 일반적인 식으로 작동하고 , 역전파 시에는
#작동하지 않는다.

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

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

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

#30과 None이 나오는 것을 볼 수 있다.

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

• 가끔씩 자동 미분을 사용할 때 수치적인 문제가 발생할 수 있다.
• 자동 미분을 통해 함수의 경사를 계산하는 것이 수치적으로 불안정 하기 때문이다.
• 부동 소수점 정밀도 오류 때문에 자동 미분이 무한 나누기 무한을 계산한다.
• 이런 경우 도함수를 해석적으로 구해 사용하는 방법이 존재한다.
• 또한 tf.where()를 사용해 값이 클 때 입력을 그대로 반환하는 방법도 있다.
  
  
• 이외에도 극도의 유연성을 추구한다면 fit() 메소드 대신 사용자 정의 fit() 반복 훈련 메소드를 만들어 쓸 수도 있다.
• 하지만 굳이 이런 경우는 등장하지도 않는다. 특히 팀단위로 모델을 구성할 때는 이러한 일은 자주 발생하지 않는다.
• 또한 사용자 정의 반복 훈련은 주의해야할 부분이 많고 그런 부분들에서 지뢰를 밟지 않도록 해야 한다.
• 실수가 나기 쉬운 부분이 정말 많고 코드도 매우 길어지기 때문이다.