텐서플로우(Tensorflow) 자동 미분과 사용자정의 훈련

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

자동 미분이란?¶

• 신경망처럼 수만 개의 파라미터를 가진 복잡한 함수의 도함수(미분, 그래디언트)를 쉽게 계산할 수 있도록 해주는 도구
• tensorflow.GradientTape() 을 활용
• 예시: $w_1 = 5, w_2 = 3$ 일 때 도함수는 각각 36과 10임을 미분을 통해 구함

$$f(w_1, w_2) = 3w_1^2 + 2w_1w_2$$$${df \over {dw_1}} = 6w_1 + 2w_2$$$${df \over {dw_2}} = 2w_1$$

In [1]:

import tensorflow as tf

w1, w2 = tf.Variable(5.), tf.Variable(3.) # 두 변수 w1과 w2를 정의

def f(w1, w2): # 함수 f 정의
    return 3 * w1 ** 2 + 2 * w1 * w2

with tf.GradientTape(persistent = True) as tape: # tf.GradientTape 블록을 만들어, 관련된 연산을 기록
    z = f(w1, w2)

dz_dw1 = tape.gradient(z, w1) # 그래디언트 확인
dz_dw2 = tape.gradient(z, w2)
del tape # Tape 삭제하여 리소스 해제

print(dz_dw1, dz_dw2)

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

상수(constant)에 대한 자동 미분¶

• 원칙적으로는 변수(Variable)에 대해서만 미분이 가능하기 때문에, 상수에 대한 미분은 None으로 계산

In [2]:

c1, c2 = tf.constant(5.), tf.constant(3.) # 두 상수 c1, c2를 정의

with tf.GradientTape(persistent = True) as tape: # tf.GradientTape 블록을 만들어, 관련된 연산을 기록
    z = f(c1, c2)

dz_dc1 = tape.gradient(z, c1) # 그래디언트 확인
dz_dc2 = tape.gradient(z, c2)
del tape # Tape 삭제하여 리소스 해제

print(dz_dc1, dz_dc2)

None None

• 변수, 상수에 관계없이 그래디언트를 구하고자 한다면, watch 메소드를 사용

In [3]:

with tf.GradientTape(persistent = True) as tape: # tf.GradientTape 블록을 만들어, 관련된 연산을 기록
    tape.watch(c1) # Variable, constant 상관없이 어떤 텐서라도 감시하여 관련된 연산을 기록하도록 설정
    tape.watch(c2)
    z = f(c1, c2)

dz_dc1 = tape.gradient(z, c1) # 그래디언트 확인
dz_dc2 = tape.gradient(z, c2)
del tape # Tape 삭제하여 리소스 해제

print(dz_dc1, dz_dc2)

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

자동 미분 응용¶

• 그래디언트를 계산할 때 일부를 제외하고 역전파되도록 설정하기 위해 tf.stop_gradient를 활용

$$f(w_1, w_2) = 3w_1^2 + 2w_1w_2$$$${df \over {dw_1}} = 6w_1$$$${df \over {dw_2}} = None$$

In [4]:

def f(w1, w2): # 함수 f 정의
    return 3 * w1 ** 2 + tf.stop_gradient(2 * w1 * w2) # 그래디언트 일부분이 역전파되지 않도록 설정(tf.stop_gradient)

with tf.GradientTape(persistent = True) as tape: # tf.GradientTape 블록을 만들어, 관련된 연산을 기록
    z = f(w1, w2)

dz_dw1 = tape.gradient(z, w1) # 그래디언트 확인
dz_dw2 = tape.gradient(z, w2)
del tape # Tape 삭제하여 리소스 해제

print(dz_dw1, dz_dw2)

tf.Tensor(30.0, shape=(), dtype=float32) None

• 그래디언트를 해석적인 방법으로 구하기 위해서는 @tf.custom_gradient 데코레이터를 사용하고, 해석적으로 구한 도함수를 정의
• 예시: softplus

$$f(x) = log(exp(x) + 1)$$$${df \over {dx}} = {1 \over 1 + exp(-x)}$$

Softplus

In [5]:

x = tf.Variable([100.]) # 변수 x 정의

def my_softplus(z): # 활성화함수 정의(위 그림 참고)
    return tf.math.log(tf.exp(z) + 1.)

with tf.GradientTape() as tape:
    z = my_softplus(x)

tape.gradient(z, [x]) # 그래디언트가 nan으로 계산됨(수치적으로 불안정)

Out[5]:

[<tf.Tensor: shape=(1,), dtype=float32, numpy=array([nan], dtype=float32)>]

In [6]:

@tf.custom_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

In [7]:

with tf.GradientTape() as tape:
    z = my_better_softplus(x)

tape.gradient(z, [x]) # 그래디언트 계산됨

Out[7]:

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

12.3.9. 사용자 정의 훈련 반복¶

왜 필요한가?¶

1. fit() 메소드 대신 원하는대로 훈련을 수행하기 위해
2. 의도한대로 훈련이 잘 되고있는지 확신을 갖기 위해

• 예시: 두 개의 다른 옵티마이저를 하나의 신경망에 사용(와이드 네트워크에는 Adam, 딥 네트워크에는 SGD) ## 문제점
• 버그가 발생하기 쉬움
• 유지 보수가 어려움
• 팀 작업에 어려움이 발생

In [8]:

# 신경망 모델 생성
import numpy as np
from tensorflow import keras

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) # 균일 분포의 정수 난수 생성[0 ~ len(X)-1]
    return X[idx], y[idx]

# 훈련 상태를 출혁하는 함수 생성
def print_status_bar(iteration, total, loss, metrics = None):
    metrics = '-'.join([f'{m.name}: {m.result():.4f}' for m in [loss] + (metrics or [])])
    end     = '' if iteration < total else '\n'
    print(f'\r{iteration}/{total} - ' + metrics, end = end)

# 하이퍼파라미터 정의
n_epochs   = 5
batch_size = 32
n_steps    = len(X_train) // batch_size
optimizer  = keras.optimizer.Nadam(lr = .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(f'Epoch {epoch}/{n_epochs}')
    for step in range(1, n_steps + 1): # 에포크 내 훈련 스텝 반복
        X_batch, y_batch = random_batch(X_train, 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_fn: 샘플마다 하나의 손실을 반환, reduce_mean을 통해 배치에 대한 평균 손실을 계산
            loss      = tf.add_n([main_loss] + model.losses)
        gradients = tape.gradient(loss, model.trainable_variables) # 훈련 가능한 변수들에 대한 손실의 그래디언트를 계산
        optimizer.apply_gradients(zip(gradients, model.trainable_variables)) # 경사하강법 수행
        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() # 평균 손실과 지표들을 초기화