✱ 핸즈온 머신러닝 12장 텐서플로를 사용한 사용자 정의 모델과 훈련 ✱

12.1 넘파이처럼 텐서플로 사용하기

텐서플로는 강력한 수치 계산용 라이브러리로 대규모 머신러닝에 잘 맞도록 튜닝되어 있다.

텐서플로 API는 텐서를 순환시킨다. 텐서는 한 연산에서 다른 연산으로 흐르며 일반적으로 다차원 배열이다. 

사용자 정의 손실 함수, 사용자 정의 지표, 사용자 정의 층 등을 만들 때 텐서가 중요하다.

12.1.1 텐서와 연산

tf.constant() 함수로 텐서를 만들 수 있다.

import tensorflow as tf
t = tf.constant([[1., 2., 3.], [4., 5., 6.]]) # 행렬
t
# <tf.Tensor: shape=(2, 3), dtype=float32, numpy=
# array([[1., 2., 3.],
         [4., 5., 6.]]), dtype=float32)>

t.shape
# TensorShape([2, 3])
t.dtype
# tf.float32

tf.Tensor는 크기(shape)와 데이터 타입(dtype)을 가진다.

t[:, 1:]
# <tf.Tensor: shape=(2, 2), dtype=float32, numpy=
# array([[2., 3.],
#        [5., 6.]], dtype=float32)>
t[..., 1, tf.nexaxis]
# <tf.tensor: shape=(2, 1), dtype=float32, numpy=
# array([[2.].
         [5.]], dtype=float32)>

인덱스 참조도 넘파이와 비슷하게 작동한다.

t + 10
# <tf.Tensor: shape=(2, 3), dtype=float32, numpy=
# array([[11., 12., 13.],
         [14., 15., 16.]], dtype=float32)>
         
tf.square(t)
# <tf.Tensor: shape=(2, 3), dtype=float32, numpy=
# array([[ 1., 4., 9.],
#        [16., 25., 36.]], dtype=float32)>

t @ tf.transpose(t)
# <tf.Tensor: shape=(2, 2), dtype=float32, numpy=
# array([[14., 32.],
         [32., 77.]], dtype=float32)>

모든 종류의 텐서 연산이 가능하다.

t + 10은 tf.add(t, 10)을 호출하는 것과 같다. @ 연산은 tf.matmul() 함수를 호출하는 것과 같다.

tf.constant(42)
# <tf.Tensor: shape(), dtype=int32, numpy=42>

텐서는 스칼라값도 가질 수 있으며 크기는 비어 있다. 

12.1.2 텐서와 넘파이

텐서는 넘파이와 함께 사용하기 편하다. 넘파이 배열로 텐서를 만들 수 있고 그 반대도 가능하다.

넘파이 배열에 텐서플로 연산을 적용할 수 있고 텐서에 넘파이 연산을 적용할 수 있다. 

import numpy as np
a = np.array([2., 4., 5.])

tf.constant
# ‹tf. Tensor: id=111, shape=(3,), dtype=float64, numpy=array([2., 4., 5.])>

t.numpy()
# array([[1., 2., 3.1,
#        [4., 5., 6.]1, dtype=float32)

tf.square(a)
# ‹tf. Tensor: id=116, shape=(3,), dtype=float64, numpy=array([4., 16., 25.])>

np.square(t)
# array([[ 1., 4., 9.],
#        [16., 25., 36.]], dtype=float32)

12.1.3 타입 변환

텐서플로는 어떤 타입 변환도 자동으로 수행하지 않는다.

호환되지 않는 타입의 텐서로 연산을 실행하면 예외가 발생한다.

12.1.4 변수

tf.Tensor는 변경이 불가능한 객체이므로 텐서의 내용을 바꿀 수 없다.

따라서 일반적인 텐서로는 역전파로 변경되어야 하는 신경망의 가중치를 구현할 수 없다.

시간에 따라 변경되어야 하는 파라미터도 있다.

이것이 tf.Variable이 필요한 이유이다.

v = tf.Variable([[1., 2., 3.], [4., 5., 6.]])
v
# <tf.Variable 'Variable:0' shape=(2, 3) dtype=float32, numpy=
# array([[1., 2., 3.],
#        [4., 5., 6.]1, dtype=float32)>

v.assign(2 * v) # v는 이제 [[2., 4., 6.], [8., 10., 12.]]
v[0, 1]- assign(42) # v는 이제 [[2., 42., 6.], [8., 10., 12.]]
v[:, 2].assign([0., 1.] # v는 이제 [[2., 42., 0.], [8., 10., 1.]]
v.scatter_nd_update(indices=[[0, 0], [1, 2]], updates=[100., 200.]) # v는 이제 [[100., 42., 0.], [8., 10., 200.]]

tf.Variable은 tf.Tensor와 비슷하게 작동한다. 동일한 연산을 수행할 수 있고 넘파이와도 잘 호환된다.

assign() 메서드를 사용하면 변숫값을 바꿀 수 있다. 

assign_add()나 assign_sub() 메서드를 사용하면 주어진 값만큼 변수를 증가 혹은 감소시킬 수 있다.

또는 원소의 assign() 메서드나 scatter_update(), scatter_nd_update() 메서드를 사용하여 개별 원소를 수정할 수도 있다.

직접 수정은 안 된다.

---

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

12.2.1 사용자 정의 손실 함수

회귀 모델을 훈련하는 데 훈련 데이터에 잡음 데이터가 조금 있다면 이상치를 제거하거나 고쳐서 데이터셋을 수정할 수 있지만, 잡음 데이터는 여전히 남아 있을 것이다. 이럴 땐 후버 손실을 사용하면 좋다.

def hubor_fn(y_true, y_pred):
    error = y_true - y_pred
    is_small_error = tf.ads(error) < 1
    squared_loss = tf.square(error) / 2
    linear_loss = tf.ads(error) - 0.5
    return tf.where(is_small_error, squared_loss, linear_loss)

레이블과 모델의 예측을 매개변수로 받는 함수를 만들고 텐서플로 연산을 사용해 각 샘플의 손실을 모두 담은 텐서를 계산하면 된다.

model.compile(loss=huber_fn, optimizer="nadam")
model.fit(X_train, y_train, [...])

후버 손실을 사용해 케라스 모델위 컴파일 메서드를 호출하고 모델을 훈련할 수 있다.

 

훈련하는 동안 배치마다 케라스는 huber_fn() 함수를 호출하여 손실을 계산하고 후진 모드 자동 미분을 사용해 모델 파라미터에 대한 손실의 그레이디언트를 계산한다. 

그다음 경사 하강법 단계를 수행하고 에포크 시작부터 전체 손실을 기록하여 평균 손실을 출력한다.

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

사용자 정의 손실 함수를 사용하는 모델을 저장하는 데는 문제가 없지만, 모델을 로드할 때는 함수 이름과 실제 함수를 매핑한 딕셔너리를 전달해야 한다. 

사용자 정의 객체를 포함한 모델을 로드할 떄는 그 이름과 객체를 매핑해야 한다.

model = tf.keras.models.load_model("my_model_with_a_custom_loss",
                                   custom_objects={"huber_fn":huber_fn})

 

def create_huber(threshold=1.0):
    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_eror, squared_loss, linear_loss)
    return huber_fn

model.compile(loss=create_huber(2.0), optimizer="nadam")

오차를 작은 것으로 판단할 기준이 필요할 떄는 매개변수를 받을 수 있는 함수를 만들면 된다.

 

model = tf.keras.models.load_model(
    "my_model_with_a_custom_loss_threshold_2",
    custom_objects={"huber_fn": create_huber(2.0)]
)

모델을 저장할 때 이 threshold 값은 저장되지 않으므로 모델을 로드할 때 threshold 값을 지정해야 한다. 

tf.keras.losses.Loss 클래스를 상속하고 get_config() 메서드를 구현하여 해결할 수 있다.

class HuberLoss(tf.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)

생성자는 기본적인 하이퍼파라미터를 **kwargs로 받은 매개변수 값을 부모 클래스의 생성자에게 전달한다. 

call() 메서드는 레이블과 예측을 받고 모든 샘플의 손실을 계산하여 반환한다.

get_config() 메서드는 하이퍼파라미터 이름과 같이 매핑된 딕셔너리를 반환한다. 부모 클래스의 get_config() 메서드를 호출하고 그다음 반호나된 딕셔너리에 새로운 하이퍼파라미터를 추가한다.

 

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

모델을 컴파일할 때 이 클래스의 인스턴스를 사용할 수 있다.

model = tf.keras.models.load_model("my_model_with_a_custom_loss_class",
                                   custom_objects={"HuberLoss": HuberLoss})

모델을 저장할 때 임곗값도 함께 저장된다. 모델을 로드할 때 클래스 이름과 클래스 자체를 매핑해줘야 한다. 

모델을 저장할 때 케라스는 손실 객체의 get_config() 메서드를 호출하여 반환된 설정을 HDF5 파일에 JSON 형태로 저장한다.

모델을 로드하면 HuberLoss 클래스의 from_config() 클래스 메서드를 호출한다.

이 메서드는 기본 손실 클래스에 구현되어 있고 생성자에게 **config 매개변수를 전달해 클래스의 인스턴스를 만든다.

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

def my_softplus(z):
    return tf.math.log(1.0 + tf.exp(z))
    
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):
    return tf.where(weights < 0., tf.zeros_like(weights), weights)

매개변수는 사용자 정의하련느 함수의 종류에 따라 다르다.

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

이 활성화 함수는 Dense 층의 출력에 적용되고 다음 층에 그 결과가 전달된다.

층의 갖우치는 초기화 함수에서 반환된 값으로 초기화된다.

훈련 스텝마다 가중치가 규제 함수에 전달되어 규제 손실울 계산하고 전체 손실에 추가되어 훈련을 위한 최종 손실을 만들어낸다.

마지막으로 제한 함수가 훈련 스텝마다 호출되어 층의 가중치를 제한한 가중치 값으로 바뀐다.

 

함수가 모델과 함께 저장되야 할 하이퍼파라미터를 가지고 있다면 tf.keras.regularizers.Regularizer, tf.keras.constraints.Contraint, tf.keras.initializers.Initializer, tf.keras.layers.Layer와 같이 적절한 클래스를 상속한다. 

 

[factor 하이퍼파라미터를 저장하는 l1 규제를 위한 클래스의 예]

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

12.2.4 사용자 정의 지표

손실은 모델을 훈련하기 위해 경사 하강법에서 사용되므로 (적어도 평가할 지점에서는) 미분 가능해야 하고 그레이디언트가 모든 곳에서 0이 아니어야 한다.

지표는 모델을 평가할 때 사용되며 훨씬 이해되기 쉬워야 한다. 미분이 가능하지 않거나 모든 곳에서 그레이디언트가 0이어도 괜찮다.

 

대부분의 경우 사용자 지표 함수를 만드는 것은 사용자 손실 함수를 만드는 것과 동일하다.

 

하지만 진짜 양성 개수와 거짓 양성 개수를 기록하고 필요할 떄 정밀도를 계산할 수 있는 객체가 필요하다.

precision = tf.keras.metrics.Precision()
pricision([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>

이는 스트리밍 지표이다. 

precision.result()
# <tf.Tensor: shape(), dtype=float32, nupy=0.5>
precision.variables
# [<tf. Variable 'true_positives:0' [...], numpy=array([4.], dtype=float32)>,
# ‹tf.Variable 'false_positives:0' [...], numpy=array([4.], dtype=float32)>]
precision.reset_states() # 두 변수가 0.0으로 초기화된다.

result() 메서드를 호출하여 현재 지표 값을 얻을 수 있다.

variables 속성을 사용하여 진짜 양성과 거짓 양성을 기록한 변수를 확인할 수 있다.

reset_states() 메서드를 사용해 이 변수를 초기화할 수 있다.

 

사용자 정의 스트리밍 지표를 만들고 싶다면 tf.keras.metrics.Metric 클래스를 상속한다.

class HuberMetric(tf.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_weights("total", initializer="zeros")
        self.count = self.add_weights("count", initializer="zeros")
        
    def update_state(self, y_true, y_pred, sample_weight=None):
        sample_metrics = self.huber_fn(y_true, y_pred)
        self.total.assign_add(tf.reduce_sum(sample_matrics))
        self.count.assign_add(tf.cast(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)

생성자는 add_weight() 메서드를 사용해 여러 배치에 걸쳐 지표의 상태를 기록하기 위한 변수를 만든다.

이 예에서는 후버 손실의 합(total)과 지금까지 처리한 샘플 수(count)를 기록한다. 수동으로 변수를 만들 수 있다.

케라스는 속성으로 만들어진 모든 tf.Variable을 관리한다.

update_state() 메서드는 이 클래스를 함수처럼 사용할 때 호출된다.

배치의 레이블과 예측을 바탕으로 변수를 업데이트한다.

result() 메서드는 최종 결과를 계산하고 반환한다.  이 지표 클래스를 함수처럼 사용하면 먼저 update_state() 메서드가 호출되고 그다음 result() 메서드가 호출되어 출력이 반환된다.

get_config() 메서드를 구현하여 threshold 변수를 모델과 함께 저장한다.

 

지표를 간단한 함수로 정의하면 케라스가 배치마다 자동으로 이 함수를 호출하고 에포크 동안 평균을 기록한다.

HuberMetric 클래스를 정의하는 것의 유일한 이점은 threshold를 저장하는 것이다. 

12.2.5 사용자 정의 층

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

또는 동일한 층 블록이 여러 번 반복되는 네트워크를 만들 경우 각각의 블록을 하나의 층으로 다루는 게 편하다.

이런 경우 사용자 정의 층을 만든다. 

 

tf.keras.layers.Flatten이나 tf.keras.layers.ReLU와 같이 가중치가 없는 층이 있다.

가중치가 필요 없는 사용자 정의 층을 만드는 가장 간단한 방법은 파이썬 함수를 만든 후 tf.keras.layers.Lambda 층으로 감싸는 것이다. 

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

 

이 사용자 정의 층을 시퀀셜 API나 함수형 API, 서브클래싱 API에서 보통의 층과 동일하게 사용할 수 있다.

또는 활성화 함수로 사용하거나 activation=tf.exp와 같이 지정할 수도 있다.

지수 함수는 이따금 회귀 모델에서 예측값의 스케일이 매우 다를 때 출력 층에 사용된다.

 

상태가 있는 층(가중치를 가진 층)을 만드려면 tf.keras.layers.Layer를 상속해야 한다. 

class MyDense(tf.keras.layers.Layer):
     def __init__(self, units, activation=None, **kwargs):
         super().__init__(**kwargs)
         self.units = units
         self.activation = tf.keras.activations.get(activation)
         
     def build(self, batch_input_shape):
         se;f.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")
     def call(self, X):
         return self.activation(X @ self.kernel + self.bias)
     def get_config(self):
         base_config = super().get_config()
         return {**base__config, "units": self.units,
                 "activation": tf.keras.activations.serialize(self.activation)}

생성자는 모든 하이퍼파라미터를 매개변수로 받는다.

build() 메서드의 역할은 가중치마다 add_weight() 메서드를 호출하여 층의 변수를 만드는 것이다. 층이 처음 사용될 때 호출된다.  이 시점이 되면 케라스가 층의 입력 크기를 알고 있을 것이므로 build() 메서드의 입력으로 크기를 전달한다. 

call() 메서드는 이 층에 필요한 연산을 수행한다. 입력 X와 층의 커널을 행렬 곱셈하고 편향을 더한다. 결과에 활성화 함수를 적용한다. 이 값이 층의 출력이다.

get_config() 메서드는 앞의 코드와 같다. tf.keras.activations.serialize()를 사용하여 활성화 함수의 전체 설정을 저장한다. 

 

여러 가지 입력을 받는 층을 만들려면 call() 메서드에 모든 입력이 포함된 튜플을 매개변수 값으로 전달해야 한다.

여러 출력을 가진 층을 만드려면 call() 메서드가 출력의 리스트를 반환해야 한다.

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

함수형 API와 서브클래싱 API에만 사용할 수 있다. 하나의 입력과 하나의 출력을 가진 층만 사용하는 시퀀셜 API에는 사용할 수 없다.

 

훈련과 테스트에서 다르게 작동하는 층이 필요하다면 call() 메서드에 training 매개변수를 추가하여 훈련인지 테스트인지를 결정해야 한다.

훈련하는 동안 가우스 잡음을 추가하고 테스트 시에는 아무것도 하지 않는 층을 만들 수 있다. 

class MyGaussianNoise(tf.keras.layers.Layer):
    def __init__(self, stddev, **kwargs):
        super().__init__(**kwargs)
        self.stddev = stddev
        
    def call(self, X, training=False):
        if training:
            noise = tf.random.normal(tf.shape(X), stddev=self.stddev)
            return X + noise
        else:
            return X

12.2.6 사용자 정의 모델

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

 

입력이 첫 번째 완전 연결 층을 통과하여 두 개의 완전 연결 층과 스킵 연결로 구성된 잔차 블록으로 전달된다.

그다음 도일한 잔차 블록에 세 번 더 통과시킨다.

그리고 두 번쨰 잔차 블록을 지나 마지막 출력이 완전 연결된 출력 층에 전달된다.

class ResidualBlock(tf.keras.layers.Layer):
    def__init__(self, n_layers, n_neurons, **kwargs):
        super ().__init__(**kwargs)
        self.hidden = [tf.keras. layers.Dense(n_neurons, activation="relu",
                                              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 속성을 감지하고 필요한 변수를 자동으로 이 층의 변수 리스트에 추가한다.

class ResidualRegressor(tf.keras.Model):
    def __init_(self, output_dim, **kwargs):
        super()._init__(**kwargs)
        self.hidden1 = tf.keras. layers.Dense(30, activation="relu",
                                              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() 메서드를 사용해 모델을 저장하고 tf.keras.models.load_model() 함수를 사용해 저장된 모델을 로드하고 싶다면 ResidualBlock 클래스와 ResidualRegressor 클래스에 모두 get_config() 메서드를 구현해야 한다.

또한 save_weights()와 load_weights() 메서드를 사용해 가중치를 저장하고 로드할 수 있다. 

 

Model 클래스는 Layer 클래스의 서브클래스이므로 모델을 층처럼 정의할 수 있다.

하지만 모델은 compile(), fit(), evaluate(), predict() 메서드와 같은 추가적ㅇ닉 니ㅡㅇ이 있다.

또한 get_layers() 메서드와 save() 메서드가 있다. 

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

사용자 손실과 지표는 모두 레이블과 예측, 그리고 선택적으로 샘플 가중치를 기반으로 한다.

하지만 은닉 층의 가중치나 활성화 함수 등과 같이 모델의 구성 요소에 기반한 손실을 정의해야 할 때도 있다.

이런 손실은 규제나 모델의 내부 상황을 모니터링할 때 유용하다.

 

모델 구성 요소에 기반한 손실을 정의하고 계산하여 add_loss() 메서드에 그 결과를 전달한다.

재구성 손실(보조 출력에 연결된 손실)을 주 손실에 더하여 회귀 작업에 직접적으로 도움이 되지 않은 정보일지라도 모델이 은닉 층을 통과하면서 가능한 한 많은 정보를 유지하도록 유도한다.

이런 손실이 이따금 일반화 성능을 향상시킨다.

모델의 add_metric() 메서드를 사용해 사용자 정의 지표를 추가할 수도 있다. 

class ReconstructingRegressor(tf.keras.Model):
    def__init__(self, output_dim, **kwargs):
        super()._init__(**kwargs)
        self. hidden = [tf.keras.layers.Dense(30, activation="relu",
                                              kernel_initializer="he_normal")
                        for _ in range(5)]
        self.out = tf.keras.layers.Dense(output_dim)
        self.reconstruction_mean = tf.keras.metrics.Mean(
            name="reconstruction _error")

    def build(self, batch_input_shape):
        n_inputs = batch_input_shape[-1]
        self. reconstruct = tf.keras.layers.Dense(n_inputs)
    
    def call(self, inputs, training=False):
        Z = inputs
        for layer in self.hidden:
            Z = layer(Z)|
        reconstruction = self.reconstruct(Z)
        recon_loss = tf. reduce_mean(tf.square(reconstruction - inputs))
        self.add_loss(0.05 * recon_loss)
        if training:
            result = self.reconstruction_mean(recon_loss)
            self.add _metricresult)
        return self.out (Z)

생성자가 다섯 개의 은닉 층과 하나의 출력 층으로 구성된 심층 신경망을 만든다. 훈련하는 동안 재구성 오차를 추적하기 위해 Mean 스트리밍 지표도 만든다. 

build() 메서드에서 완전 연결 층을 하나 더 추가하여 모델의 입력을 재구성하는 데 사용한다. 이 완전 연결 층의 유닛 개수는 입력 개수와 같아야 한다. 이런 재구성 층을 build() 메서드에서 만드는 이유는 이 메서드가 호출되기 전까지는 입력 개수를 알 수 없기 때문이다. 

call() 메서드에서 입력이 다섯 개의 은닉 층에 모두 통과한다. 그다음 결괏값을 재구성 층에 전달하여 재구성을 만든다. 재구성 손실을 계산하고 add_loss() 메서드를 사용해 모델의 손실 리스트에 추가한다. 훈련 중에만 call() 메서드가 재구성 지표를 업데이트하고 화면에 출력되도독 모델에 추가한다. 케라스가 자동으로 평균을 추적한다. call() 메서드 마지막에서 은닉 층의 출력을 출력 층에 전달하여 얻은 출력값을 반환한다. 

 

훈련하는 동안 총 손실과 재구성 손실이 함께 계산된다.

Epoch 1/5
363/363 [========] - 1s 820us/step - loss: 0.7640 - reconstruction error: 1.2728
Epoch 2/5
363/363 [========] - Os 809us/step - loss: 0.4584 - reconstruction_error: 0.6340
[ . . . ]

12.2.8 자동 미분으로 그레이디언트 계산하기

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

두 변수 w1과 w2를 정의하고 tf.GradientTape 블록을 만들어 이 변수와 관현된 모든 연산을 자동으로 기록한다.

마지막으로 이 테이프에 두 변수 [w1, w2]에 대한 z의 그레이디언트를 요청한다.

 

변수의 개수에 상관없이 gradient() 메서드는 기록된 계산을 한 번만에 거꾸로 통과한다.

 

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

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

gradient() 메서드가 호출된 후에는 자동으로 테이프가 즉시 지워진다. 따라서 gradient() 메서드를 두 번 호출하면 예외가 발생한다.

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

gradient() 메서드를 한 번 이상 호출해야 한다면 지속 가능한 테이프를 만들고 사용이 끝난 후 테이프를 삭제하여 리소스를 해제해야 한다.

c1, c2 = tf.constant(5.), tf.constant (3.)
with tf.GradientTape() as tape:
    z = f(c1, c2)
    
gradients = tape.gradient(z, [c1, c2]) # [None, None] 반환

기본적으로 테이프는 변수가 포함된 연산만을 기록한다. 변수가 아닌 다른 객체에 대한 z의 그레이디언트를 계산하면 None이 반환된다.

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

gradinets = tape.gradient(z, [c1, c2]) # [36., 10. 텐서] 반환

필요한 경우 어떤 텐서든 감시하여 관련된 모든 연산을 기록하도록 강제할 수 있으며 변수처럼 이런 텐서에 대해 그레이디언트를 계산할 수 있다.

 

입력이 작을 때 변동 폭이 큰 활성화 함수에 대한 규제 손실을 구현하는 경우에 유용하다.

 

대부분의 경우 그레이디언트 테이프는 여러 값에 대한 한 값의 그레이디언트를 계산하는 데 사용된다.

이런 경우 후진 모드 자동 미분이 적합하며 한 번의 정방향 계산과 역방향 계산으로 모든 그레이디언트를 동시에 계산할 수 있다.

 

여러 손실이 포함된 벡터의 그레이디언트를 계산하면 텐서플로는 벡터의 합의 그레이디언트를 계산할 것이다.

개별 그레이디언트를 계산하고 싶다면 테이프의 jacobian() 메서드를 호출해야 한다.

벡터에 있는 각 손실마다 후진 자동 미분을 수행하며 이계도함수를 계산할 수 있다.

 

어떤 경우에는 신경망의 일부분에 그레이디언트가 역전파되지 않도록 막을 필요가 있다. 

tf.stop_gradient() 함수를 사용해야 한다.

이 함수는 정방향 계산을 할 때 입력을 반환하지만, 역전파 시에는 그레이디언트를 전파하지 않아 상수처럼 작동한다. 

def f(w1, w2):
    return 3 * w1 ** 2 + tf.stop_gradient(2 * w1 * w2)
    
with tf.gradientTape() as tape:
    z = f(w1, w2) # 정방향 계산은 stop_gradient()에 영향을 받지 않는다.

gradient = tape.gradient(z, [w1, w2]) # [30. 텐서, None] 반환

 

그레이디언트를 계산할 때 32비트 부동소수점으로 다룰 수 없는 수치적인 문제가 발생하기도 한다. 

x = tf.Variable(1e-50)
with tf.GradientTape() as tape:
... z = tf.sqrt(x)
... 
tape.gradient(z, [x])
# [<tf.Tensor: shape=(), dtype=float32, numpy=inf>]

이 문제를 해결하기 위해 제곱근을 계산할 때 작은 값을 추가한다.

 

지수 함수는 매우 빠르게 증가하기 때문에 지수 항이 폭발하지 않도록 my_softplus() 함수를 수치적으로 안정적이게 구현할 수 있다.

def my_softplus(z):
    return tf.math.log(1 + tf.exp(-tf.abs(z))) + tf.maximum(0., z)

 

수치적으로 안정적인 함수에도 수치적으로 불안정한 그레이디언트가 있을 수 있다.

이러한 경우 자동 미분을 사용하지 않고 그레이디언트 계산을 위해 사용할 식을 텐서플로에 알려주어야 한다.

함수를 정의할 때 @tf.custom_gradient 데코레이터를 사용하고 일반적인 함수 결과와 그레이디언트를 계산하는 함수를 모두 반환해야 한다.

@tf.custom_gradient
def my_softplus(z):
    def my_softplus_gradients(grads): # grads = 상위 층에서 역전파된 그레이디언트
        return grads * (1 - 1 / (1 + tf.exp(z))) # 안정적인 소프트플러스의 그레이디언트
        
    return = tf.math.log(1 + tf.exp(-tf.abs(z))) + tf.maximum(0., z)
    return result. my_softplus_gradients

12.2.9 사용자 정의 훈련 반복

l2_reg = tf.keras.regularizers.l2(0.05)
model = tf.keras.models.Sequential([
    tf.keras.layers.Dense(30, activation="relu", kernel_initializer="he_normal",
                          kernel_regularizer=l2_reg),
    tf.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]

훈련 세트에서 샘플 배치를 랜덤하게 추출하는 작은 함수를 만든다.

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

현재 스텝 수, 전체 스텝 수, 에포크 시작부터 평균 손실, 그 외 다른 지표를 포함하여 훈련 상태를 출력하는 함수를 만든다. 

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

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

 

for epoch in range(1, n_epochs + 1):
    print("Epoch {}/{}".format(epoch, n_epochs))
    for step in range(1, n_steps + 1):
        X_batch, y_batch = random_batch(X_train_scaled, y_train)
        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)
            
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.appy_gradient(zip(gradients, model.trainable_variables))
        mean_loss(loss)
        for metric in metrics:
            metirc(y_batch, y_pred)
            
        print_status_bar(step, n_steps, mean_loss, metrics)
        
    for metric in [mean_loss] + metrics:
        metric.reset_status()

훈련 세트에서 배치를 랜덤하게 샘플링한다.

tf.GradientTape() 블록 안에서 모델을 함수처럼 사용하여 배치 하나를 위한 예측을 만들고 손실을 계산한다. 이 손실은 주 손실에 다른 손실을 더한 것이다. mean_squred_error() 함수가 샘플마다 하나의 손실을 반환하기 때문에 tf.reduce_mean() 함수를 사용하여 배치에 대한 평균을 계산한다. 규제 손실은 하나의 스칼라값이므로 손실을 모두 더한다.

테이프를 사용해 훈련 가능한 각 변수에 대한 손실의 그레이디언트를 계산한다. 이를 옵티마이저에 적용하여 경사 하강법을 수행한다.

현재 에포크에 대한 평균 손실과 지표를 업데이트하고 상태 막대를 출력한다.

매 에포크 끝에서 평균 손실과 지표 값을 초기화한다. 

 

for variable in model.variables:
    if variable.constraint is not None:
        variable.assign(variable.consraint(variable))

만약 그레이디언트 클리핑을 하고 싶다면 clipnorm이나 clipvalue 하이퍼파라미터를 지정하면 된다.

가중치에 다른 변환을 적용하려면 apply_gradients() 메서드를 호출하기 전에 수행해야 한다.

모델에 가중치 제한을 추가하고 싶다면 apply_gradients() 다음에 이 제한을 적용하도록 훈련 반복을 수정해야 한다.

---

12.3 텐서플로 함수와 그래프

def cube(x):
    return x ** 3
    
tf_cube = tf.function(cube)
tf_cube
# <tensorflow.python.eager.def_function at 0x1546fc080>

tf.function()을 사용하여 파이썬 함수를 텐서플로 함수로 바꿀 수 있다.

tf_cube(2)
# <tf.Tensor: shape=(), dtype=int32, numpy=8>
tf_cube(tf.constant(2.0))
# <tf.Tensor: shape=().dtype=float32, numpy=8.0>

파이썬 함수로 사용할 수 있고 동일한 결과를 반환한다.

@tf.function
def tf_cube(x):
    return x ** 3

내부적으로 tf.function()은 cube() 함수에서 수행되는 계산을 분석하고 동일한 작업을 수행하는 계산 그래프를 생성한다.

tf.function 데코레이터를 사용하는 방법도 있다.

tf_cube.python_function(2)
# 8

원본 파이썬 함수는 필요할 때 텐서플로 함수의 python_function 속성으로 참조할 수 있다.

 

텐서플로는 사용하지 않는 노드를 제거하고 표현을 단순화하는 등 계산 그래프를 최적화한다.

최적화된 그래프가 준비되면 텐서플로 함수는 적절한 순서에 맞춰 그래프 내의 연산을 효율적으로 실행ㅎ나다.

일반적으로 텐서플로 함수는 원본 파이썬 함수보다 훨씬 빠르게 실행된다.

 

tf.function()을 호출할 때 jit_compile=True로 설정하면 텐서플로는 XLA를 사용하여 해당 그래프를 위한 전용 커널을 컴파일하며, 종종 여러 연산을 융합한다.

 

사용자 정의 손실 함수, 사용자 정의 지표, 사용자 정의 층 또는 다른 사용자 정의 함수를 작성하고 이를 케라스 모델에 사용할 때, 케라스는 자동으로 이 함수를 텐서플로 함수로 변환하므로 tf.function()을 사용할 필요가 없다.

케라스가 XLA를 사용하도록 하려면 compile() 메서드를 호출할 때 jit_compile=True로 설정하면 된다.

 

기본적으로 텐서플로 함수는 호출에 사용되는 입력 크기와 데이터 타입에 맞춰 매번 새로운 그래프를 생성한다. 

12.3.1 오토그래프와 트레이싱

오토그래프 : 텐서플로가 그래프를 생성하기 위해 먼저 파이썬 함수의 소스 코드를 분석하여 for문, while문, if문은 물론 break, continue, return 같은 제어문을 모두 찾는다.

텐서플로가 소스 코드를 분석하는 이유는 파이썬이 제어문을 찾는 방법을 제공하지 않기 때문이다. 

함수의 코드를 분석한 후 오토그래프는 이 함수의 모든 제어문을 텐서플로 연산으로 바꾸어 업그레이드된 버전을 만든다.

그다음 텐서플로가 이 업데이트된 함수를 호출한다.

하지만 매개변수 값을 전달하는 대신 심볼릭 텐서를 전달한다. 이 텐서는 실제하는 값이 없고 이름, 데이터 타입, 크기만 가진다. 

이 함수는 그래프 모드로 실행될 것이다. 각 텐서플로 연산이 해당 연산을 나타내고 텐서를 출력하기 위해 그래프에 노드를 추가한다는 의미이다. 

텐서플로 연산은 어떤 계산도 수행하지 않느낟. 

최종 그래프는 트레이싱 과정을 통해 생성된다. 노드는 연산을 나타내고 화살표는 텐서를 나타낸다.

12.3.2 텐서플로 함수 사용법

대부분의 경우 텐서플로 연산을 수행하는 파이썬 함수를 텐서플로 함수로 바꾸기 위해서는 @tf.function 데코레이터를 사용하면 된다.