[HandsOn]12. 텐서플로를 사용한 사용자 정의 모델과 훈련 - 연습문제
GitHub - rickiepark/handson-ml2: 핸즈온 머신러닝 2/E의 주피터 노트북
핸즈온 머신러닝 2/E의 주피터 노트북. Contribute to rickiepark/handson-ml2 development by creating an account on GitHub.
github.com
답안은 여기를 참고하였다!
정말 어렵지만 내맘대로 커스텀한 모델을 짜기 위해선 필수적이니 꼭 알아두도록 하자.
절대 첫 번째에 익숙해지지 않으니 보고 또 볼 것!
12. 사용자 정의 모델과 훈련 알고리즘
1. 사용자 정의 손실 함수
후버 손실(keras.losses.Huber)을 없다고 생각해보고 짜도록 하자.
이 때 모델을 저장하고 로드하는 것까지 생각해야 한다.
후버 손실에서 threshold를 매개변수로 받아야 한다면? ( 1.0 이하면 이 함수를 쓰고, 이상이면 이 함수를 써라 이런식으로 커스텀을 하고 싶다면)
1. keras.losses.Loss 클래스를 상속
2. get_config() 메서드를 구현
하여 해결할 수 있다. ( 이해가 안되므로 외우자 그냥 )
import tensorflow as tf
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
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}생성자에서 기본적인 하이퍼 파라미터를 상속을 받고, threshold도 지정을 해 줌.
call 메서드에서 손실 계산하여 반환.
get_config 메서드는 모르겠당.. 그냥 매개변수 받을 때 갖다 붙이는듯...
이렇게 해 놓으면 모델을 컴파일할때 이 클래스의 인스턴스를 사용할 수 있다.
모델을 저장할 때 threshold도 같이 저장되고, 로드할 때 클래스 이름과 클래스를 매핑해 주면 된다.
2. 사용자 정의 손실 지표
비슷함.. (지금까지 해 왔던 것처럼)지표를 정의만 하면 케라스가 자동으로 함수 호출하고 에포크동안 평균 기록해줌.
클래스를 일일히 다 정의해주는 거는
1. 매개변수를 저장할 때!
2. 배치가 계속될 때마다 전체 배치를 포함하는 '정밀도' 같은 '스트리밍 지표'를 구현할 때!
3. 사용자 정의 층
텐서플로에 없는 특이한 층을 가진 네트워크를 만들어야 할 때가 있음!
가중치를 가진 층을 만들려면 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) # must be at the end
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)}1. 생성자는 모든 하이퍼 파라미터를 매개변수로 받음. 부모 생성자 호출하고, 하이퍼파라미터를 속성으로 저장.
2. build() 메서드의 역할은 가중치를 만드는 것.
"kernel": 연결 가중치(입력의 마지막 차원 크기를 받음) ,
"bias" : 편향, 0으로 초기화
3. compute_output_shape(): 이 층의 출력 크기를 반환
4. get_config(): 앞서 보았던 것과 같음. keras.activations.serialize()를 사용하여 활성화 함수의 전체 설정 저장
model = keras.models.Sequential([
MyDense(30, activation="relu", input_shape=input_shape),
MyDense(1)
])4. 사용자 정의 모델
한 마디로, keras.Model 클래스를 상속하여 생성자에서 층과 변수를 만들고 모델이 해야 할 작업을 call()메서드에 구현한다.
먼저 Residual Block 층을 만들자.
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 에 층을 레이어 갯수만큼 지정해줌.
그 다음에 call()메서드에서 Z를 hidden 에 통과시키고, input 이랑 더해준다.
ResidualBlock 층 완성!
해당 층을 이용해서 모델을 만든다.
class ResidualRegressor(keras.models.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)
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):
Z = self.block1(Z)
Z = self.block2(Z)
return self.out(Z)모델을 상속받고, 생성자에 필요한 층을 정의하고( 내맘대로 적고있음 )
call() 메서드에 모델의 형태를 정의한다.
모델 완성!
5. 자동 미분을 사용하여 그레이디언트 계산하기
간단한 함수로 예를 살펴보자.
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])gradients
[<tf.Tensor: shape=(), dtype=float32, numpy=36.0>,
<tf.Tensor: shape=(), dtype=float32, numpy=10.0>]포인트 (5,3)에서의 그레이디언트 벡터가 (36,10) 으로 계산되었다!
6. 사용자 정의 훈련 반복
만약에 모델을 피팅할때, fit()의 유연성이 충분하지 않다면? (예를 들어 fit()메서드에서는 compile할 때 하나의 옵티마이저만 지원하는데, 두 개의 다른 옵티마이저를 레이어마다 쓰고 싶다면)
훈련 반복을 직접 구현해야 함! 중요한데 넘나어려운것...
간단한 모델을 만들어보자. 훈련 반복을 직접 다루기 때문에 컴파일할 필요는 없음.
그냥 간단한 시퀀셜 모델을 만들었다.
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)
return X[idx], y[idx]훈련 상태를 출력하는 함수도 만든다. (사실 내가만든건 아니고 책에 적혀있는거임)
간편하게 tqdm 라이브러리를 쓸 수도 있음.
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)
이제 몇 개의 하이퍼파라미터를 정의하고 옵티마이저, 손실 함수, 지표를 선택해야 함.
n_epochs = 5
batch_size = 32
n_steps = len(X_train) // batch_size
optimizer = keras.optimizers.Nadam(learning_rate=0.01)
loss_fn = keras.losses.mean_squared_error
mean_loss = keras.metrics.Mean()
metrics = [keras.metrics.MeanAbsoluteError()]
사용자 정의 훈련 반복문: print status bar를 사용하지 않고 tqdm을 사용하였다.
try:
from tqdm.notebook import trange
from collections import OrderedDict
with trange(1, n_epochs + 1, desc="All epochs") as epochs:
for epoch in epochs:
with trange(1, n_steps + 1, desc="Epoch {}/{}".format(epoch, n_epochs)) as steps:
for step in steps:
X_batch, y_batch = random_batch(X_train_scaled, y_train)
with tf.GradientTape() as tape:
y_pred = model(X_batch)
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.apply_gradients(zip(gradients, model.trainable_variables))
# for variable in model.variables:
# if variable.constraint is not None:
# variable.assign(variable.constraint(variable))
모델에 가중치 제한을 추가한 항.근데 잘 모르겠어서 주석처리함
status = OrderedDict()
mean_loss(loss)
status["loss"] = mean_loss.result().numpy()
for metric in metrics:
metric(y_batch, y_pred)
status[metric.name] = metric.result().numpy()
steps.set_postfix(status)
for metric in [mean_loss] + metrics:
metric.reset_states()
except ImportError as ex:
print("To run this cell, please install tqdm, ipywidgets and restart Jupyter")넘나 어려워보인다. 천천히 이해해보자.
한 에포크 안에서 배치가 돌고 있다.
랜덤 배치 함수를 이용해서 X 배치와 y 배치를 지정하고,
그레이디언트 테이프를 이용해서 배치마다 나온 loss를 "훈련 가능한 변수" 로 미분한다.
이 값들을 optimizer에 보냄.
status라는 dictionary를 만들고, 상태들을 여기에 저장.
한 에포크가 다 돌면 metric 을 리셋.
너무 어렵다...