텐서플로란?
[머신러닝] 텐서플로(TensorFlow)란?
텐서플로(TensorFlow)란? 텐서플로(TensorFlow)는 구글에서 개발한 오픈소스 머신러닝 프레임워크이다. 주로 딥러닝 모델을 만들고 학습시키는 데 사용되며, 다양한 플랫폼에서 실행이 가능하다. 텐
puppy-foot-it.tistory.com
넘파이처럼 텐서플로 사용하기
텐서플로 API는 텐서를 순환시킨다. 텐서는 한 연산에서 다른 연산으로 흐르기 때문에 '텐서플로(tensorflow)'라고 부른다. 텐서는 넘파이 ndarray와 매우 비슷하여 일반적으로 다차원 배열이다. (스칼라값도 가질 수 있다)
사용자 정의 손실 함수, 사용자 정의 지표, 사용자 정의 층 등을 만들 때 텐서가 중요하다.
◆ 텐서와 연산
tf.constant() 함수로 텐서를 만들 수 있다.
import tensorflow as tf
# 두 개의 행과 세 개의 열을 가진 실수 행렬
t = tf.constant([[1., 2., 3.], [4., 5., 6.]]) # 행렬
t
ndarray와 마찬가지로 tf.Tensor는 크기(shape)와 데이터 타입(dtype)을 가진다.
print(t.shape)
print(t.dtype)
인덱스 참조도 넘파이와 매우 비슷하게 작동한다.
t[:, 1:]
t[..., 1, tf.newaxis]
가장 중요한 점은 모든 종류의 텐서 연산이 가능하다는 것이다.
1) 사칙연산
t + 10
t + 10은 tf.add(t,10)을 호출하는 것과 같다.
2) 제곱
tf.square(t)
3) 행렬 곱셉
t @ tf.transpose(t)
이 연산은 tf.matmul() 함수를 호출하는 것과 동일하다.
텐서는 스칼라 값도 가질 수 있는데, 이 경우 크기는 비어 있다.
tf.constant(42)
◆ 텐서와 넘파이
텐서는 넘파이와 함께 사용하기 편리한데, 넘파이 배열로 텐서를 만들 수 있고 그 반대도 가능하다.
넘파이 배열에 텐서플로 연산을 적용할 수 있고 텐서에 넘파이 연산을 적용할 수도 있다.
import numpy as np
a = np.array([2., 4., 5.])
tf.constant(a)
t.numpy()
tf.square(a)
np.square(t)
※ 넘파이는 기본으로 64비트 정밀도를 사용하지만 텐서플로는 32비트 정밀도를 사용한다. 일반적으로 신경망은 32비트 정밀도로 충분하고 더 빠르며 메모리도 적게 사용하기 때문이다. 따라서 넘파이 배열로 텐서를 만들때 dtype=tf.float32로 지정해야 한다.
◆ 타입 변환
타입 변환은 성능을 크게 감소시킬 수 있다. 타입이 자동으로 변환되면 사용자가 눈치채지 못할 수 있다. 이를 방지하기 위해 텐서플로는 어떤 타입 변환도 자동으로 수행하지 않는다. 호환되지 않는 타입의 텐서로 연산을 실행하면 예외가 발생한다.
예)
- 실수 텐서와 정수 텐서 더하기 불가
- 32비트 실수와 64비트 실수 더하기 불가
타입 변환이 필요할 때는 tf.cast() 함수를 사용할 수 있다.
t2 = tf.constant(40., dtype=tf.float64)
tf.constant(2.0) + tf.cast(t2, tf.float32)
◆ 변수 (tf.Variable)
tf.Tensor는 변경이 불가능한 객체라 텐서의 내용을 바꿀 수 없다. 따라서 일반적인 텐서로는 역전파로 변경되어야 하는 신경망의 가중치를 구현할 수 없다. 또한 시간에 따라 변경되어야 하는 파라미터도 있다.
v = tf.Variable([[1., 2., 3.], [4., 5., 6.]])
v
tf.Variable은 tf.Tensor와 비슷하게 작동한다. 동일한 연산을 수행할 수 있고 넘파이와도 잘 호환된다. 까다로운 데이터 타입도 마찬가지이다. 그러나 assign() 메서드를 사용하여 변숫값을 바꿀 수 있고, 또한 원소의 assign() 메서드나 scatter_update(), scatter_nd_update() 메서드를 사용하여 개별 원소를 수정할 수도 있다.
(assign_add()나 assign_sub() 메서드를 사용하면 주어진 값만큼 변수를 증가 혹은 감소시킬 수 있다.)
v.assign(2 * v)
v[0, 1].assign(42)
v[:, 2].assign([0., 1.])
v.scatter_nd_update(indices=[[0, 0], [1, 2]], updates=[100., 200.])
아래와 같이 직접 수정은 안 된다.
v[1] = [7., 8., 9.]◆ 다른 데이터 구조
1) 희소 텐서 (tf.SparseTensor)
대부분 0으로 채워진 텐서를 효율적으로 나타낸다. tf.sparse 패키지는 희소 텐서를 위한 연산을 제공한다.
2) 텐서 배열 (tf.TensorArray)
텐서의 리스트이다. 기본적으로 고정된 길이를 가지지만 동적으로 바꿀 수 있다. 리스트에 포함된 모든 텐서는 크기와 데이터 타입이 동일해야 한다.
3) 래그드 텐서(tf.RaggedTensor)
래그드 텐서는 리스트를 나타낸다. 모든 텐서는 랭크와 데이터 타입이 같아야 하지만 크기는 다를 수 있다. 텐서의 크기가 달라지는 차원을 래그드 차원이라고 부르며, tf.ragged 패키지는 래그드 텐서를 위한 연산을 제공한다.
4) 문자열 텐서
- tf.sring: 유니코드가 아니라 바이트 문자열을 나타낸다. 유니코드 문자열을 사용해 문자열 텐서를 만들면 자동으로 UTF-8로 인코딩 된다. 기본 데이터 타입이기 때문에 문자열의 길이가 텐서 크기에 나타나지 않는다.
- tf.int32: 유니코드 코드 포인트를 나타내며, 이를 사용해 유니코드 문자열을 표현할 수 있다. 문자열 길이가 텐서 크기에 표현된다.
- tf.strings: 바이트 문자열, 유니코드 문자열과 이런 텐서 사이의 변환을 위한 연산을 제공한다.
5) 집합
집합은 일반적인 텐서(또는 희소 텐서)로 나타낸다. 일반적으로 각 집합은 텐서의 마지막 축에 있는 벡터에 의해 표현된다. tf.sets 패키지의 연산을 사용해 집합을 다룰 수 있다.
6) 큐(tf.queue 패키지)
큐는 단계별로 텐서를 저장하며, 텐서플로는 여러 종류의 큐를 제공하며, 이는 tf.queue 패키지에 포함되어 있다.
- FIFOQueue: 간단한 FIFO(first in, first out)
- PriorityQueue: 어떤 원소에 우선권을 주는 큐
- RandomShuffleQueue: 원소를 섞는 큐
- PaddingFIFOQueue: 패딩을 추가하여 크기가 다른 원소의 배치를 만드는 큐
사용자 정의 모델과 훈련 알고리즘
◆ 사용자 정의 손실 함수
후버 손실이 케라스에 없다고 생각하고 구현해 보는데, 이를 구현하려면 레이블과 모델의 예측을 매개변수로 받는 함수를 만들고 텐서플로 연산을 사용해 각 샘플의 손실을 모두 담은 텐서를 계산한다.
※ 케라스에서는 후버 손실을 사용하려면 tf.keras.losses.Huber 클래스를 사용하면 된다.
※ 후버손실(Huber Loss): L1과 L2의 장점을 취하면서 단점을 보완하기 위해 제안된 것이 Huber Loss이다. Huber loss는 모든 지점에서 미분이 가능하면서 이상치에 강건한(robust) 성격을 보인다.
def huber_fn(y_true, y_pred):
error = y_true - y_pred
is_small_error = tf.abs(error) < 1
squared_loss = tf.square(error) / 2
linear_loss = tf.abs(error) - 0.5
return tf.where(is_small_error, squared_loss, linear_loss)
사용자 정의 손실 함수를 테스트하기 위해 기본 케라스 모델을 만들고 캘리포니아 주택 데이터셋으로 훈련해 본다.
# 캘리포니아 주택 데이터셋을 로드, 분할 및 스케일을 조정한 다음 간단한 케라스 모델을 생성.
from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
housing = fetch_california_housing()
X_train_full, X_test, y_train_full, y_test = train_test_split(
housing.data, housing.target.reshape(-1, 1), random_state=42)
X_train, X_valid, y_train, y_valid = train_test_split(
X_train_full, y_train_full, random_state=42)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_valid_scaled = scaler.transform(X_valid)
X_test_scaled = scaler.transform(X_test)
input_shape = X_train.shape[1:]
tf.keras.utils.set_random_seed(42)
model = tf.keras.Sequential([
tf.keras.layers.Dense(30, activation="relu", kernel_initializer="he_normal",
input_shape=input_shape),
tf.keras.layers.Dense(1),
])
후버 손실을 사용해 케라스 모델의 컴파일 메서드를 호출하고 모델을 훈련한다.
model.compile(loss=huber_fn, optimizer="nadam", metrics=["mae"])
model.fit(X_train_scaled, y_train, epochs=2,
validation_data=(X_valid_scaled, y_valid))
훈련하는 동안 배치마다 케라스는 huber_fn() 함수를 호출하여 손실을 계산하고 후진 모드 자동 미분을 사용해 모든 모델 파라미터에 대한 손실의 그레이디언트를 계산한다. 그리고 Nadam 옵티마이저를 사용해 경사 하강법 단계를 수행하고, 에포크 시작부터 전체 손실을 기록하여 평균 손실을 출력한다.
◆ 사용자 정의 요소를 가진 모델을 저장하고 로드하기
사용자 정의 손실 함수를 사용하는 모델을 저장하는 데는 아무런 문제가 없으나, 모델을 로드할 때는 함수 이름과 실제 함수를 매핑한 딕셔너리를 전달해야 한다. 사용자 정의 객체를 포함한 모델을 로드할 때는 그 이름과 객체를 매핑해야 한다.
model.save("my_model_with_a_custom_loss.keras")model = tf.keras.models.load_model("my_model_with_a_custom_loss.keras",
custom_objects={"huber_fn": huber_fn})
앞서 구현한 함수는 -과 1 사이의 오차는 작은 것으로 간주하는 데, 다른 기준이 필요한 경우에는 매개변수를 받을 수 있는 함수를 만들면 된다.
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_error, squared_loss, linear_loss)
return huber_fn
model.compile(loss=create_huber(2.0), optimizer='nadam')
모델을 저장할 때 이 threshold 값은 저장되지 않기 때문에 모델을 로드할 때 threshold 값을 지정해야 한다.
model.save("my_model_with_a_custom_loss_threshold_2.keras")model = tf.keras.models.load_model("my_model_with_a_custom_loss_threshold_2.keras",
custom_objects={"huber_fn": create_huber(2.0)})
이 문제는 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로 받은 매개변수 값을 부모 클래스의 생성자에게 전달한다. 손실 함수의 name과 개별 샘플의 손실을 모으기 위해 사용할 reduction 알고리즘이다. 기본값은 'AUTO'이며, 'SUM_OVER_BATCH_SIZE'에 해당한다. 샘플 손실에 가중치를 곱하여 더하고 배치 크기로 나눈다. 샘플 가중치가 없다면 1.0으로 간주하며, 다른 값으로는 'SUM'과 'NONE'이 있다.
- call() 메서드는 레이블과 예측을 받고 모든 샘플의 손실을 계산하여 반환한다
- get_config() 메서드는 하이퍼파라미터 이름과 같이 매핑된 딕셔너리를 반환한다. 먼저 부모 클래스의 get_config() 메서드를 호출하고 그다음 반환된 딕셔너리에 새로운 하이퍼파라미터를 추가한다
그리고 모델을 컴파일할 때 이 클래스의 인스턴스를 사용할 수 있다.
model.compile(loss=HuberLoss(2.), optimizer="nadam", metrics=["mae"])
모델을 저장할 때 케라스는 손실 객체의 get_config() 메서드를 호출하여 반환된 설정을 HDF5 파일에 JSON 형태로 저장한다.
model.save("my_model_with_a_custom_loss_class.keras")
모델을 로드하면 HuberLoss 클래스의 from_config() 클래스 메서드를 호출한다. 이 메서드는 기본 손실 클래스(Loss)에 구현되어 있고 생성자에게 **config 매개변수를 전달해 클래스의 인스턴스를 만든다.
model = tf.keras.models.load_model('my_model_with_a_custom_loss_class.keras',
custom_objects={'HuberLoss':HuberLoss})◆ 활성화 함수, 초기화, 규제, 제한을 커스터마이징 하기
손실, 규제, 제한, 초기화, 지표, 활성화 함수, 층, 모델과 같은 대부분의 케라스 기능은 유사한 방법으로 커스터마이징할 수 있는데, 대부분 적절한 입력과 출력을 가진 간단한 함수를 작성하면 된다.
사용자 정의에 대한 예
- 사용자 정의 활성화 함수: tf.keras.activations.softplus(), tf.nn.softplus() 과 동일
- 사용자 정의 글로럿 초기화: tf.keras.initializers.glorot_normal()과 동일
- 사용자 정의 L1 규제: tf.keras.regularizers.l1(0.01)과 동일
- 양수인 가중치만 남기는 사용자 정의 제한: tf.keras.constraints.nonneg()나 tf.nn.relu()와 동일
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): # tf.nn.relu(weight)와 반환값 동일
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, # 사용자 정의 L1 규제
kernel_constraint=my_positive_weights) # 양수인 가중치만 남기는 사용자 정의 제한▶ 이 활성화 함수는 Dense 층의 출력에 적용되고 다음 층에 그 결과가 전달된다. 층의 가중치는 초기화 함수에서 반환된 값으로 초기화된다. 훈련 스텝마다 가중치가 규제 함수에 전달되어 규제 손실을 계산하고 전체 손실에 추가되어 훈련을 위한 최종 손실을 만든다. 마지막으로 제한 함수가 훈련 스텝마다 호출되어 층의 가중치를 제한한 가중치 값으로 바뀐다.
함수가 모델과 함께 저장해야 할 하이퍼파라미터를 가지고 있다면 tf.keras.regularizers.Regularizer, tf.keras.constraints.Constraint, tf.keras.initializer.Initializer, tf.keras.layers.Layer와 같이 적절한 클래스를 상속한다.
아래는 사용자 정의 손실을 만들었던 것처럼 factor 하이퍼파라미터를 저장하는 L1 규제를 위한 간단한 클래스의 예이다.
(이 경우에는 부모 클래스에 생성자와 get_config() 메서드가 정의되어 있지 않기 때문에 호출할 필요가 없다.)
class MyL1Regularizer(tf.keras.regularizers.Regularizer):
def __init__(self, factor):
self.factor = factor
def __call__(self, weights):
return tf.reduce_sum(tf.abs(self.factor * weights))
def get_config(self):
return {'factor': self.factor}- 손실, 활성화 함수를 포함한 층, 모델의 경우 call() 메서드를 구현해야 한다.
- 규제, 초기화, 제한의 경우에는 __call__() 메서드를 구현해야 한다.
◆ 사용자 정의 지표
[손실과 지표]
- 손실: 모델을 훈련하기 위해 경사 하강법에서 사용되므로 적어도 평가할 지점에서는 미분 가능해야 하고 그레이디언트가 모든 곳에서 0이 아니어야 한다. 예. 크로스 엔트로피
- 지표: 모델을 평가할 때 사용된다. 지표는 훨씬 이해하기 쉬워야 하며, 미분이 가능하지 않거나 모든 곳에서 그레이디언트가 0이어도 괜찮다. 예. 정확도
대부분의 경우 사용자 지표 함수를 만드는 것은 사용자 손실 함수를 만드는 것과 동일하다.
tf.keras.metrics.Precision 클래스는 진짜 양성 개수와 거짓 양성 개수를 기록하고 필요할 때 정밀도를 계산할 수 있다.
[스트리밍 지표]
- 개념:
훈련 중 데이터 배치를 처리하면서 실시간으로 누적 데이터를 기반으로 성능 지표를 계산하는 방법이다. 이 방식은 각 배치에서 계산한 값을 누적하여 최종적으로 한 epoch 동안의 평균 지표를 제공한다.
- 특징:
- 배치 단위로 누적 계산: 각 배치에서 계산된 값을 업데이트하고, epoch가 끝난 뒤 누적 평균을 반환한다.
- 대규모 데이터 처리: 데이터셋이 크거나 한 번에 전체 데이터를 메모리에 로드하기 어려운 경우 유용하다.
- 사용자 정의 가능: keras.metrics.Metric 클래스를 상속하여 스트리밍 지표를 직접 정의할 수 있다.
Precision 클래스 객체를 만들고 이를 함수처럼 사용하여 아래와 같이 두 개의 배치의 레이블과 예측을 각각 첫 번째 매개변수와 두 번째 매개변수로 전달했다.
precision = tf.keras.metrics.Precision()
precision([0, 1, 1, 1, 0, 1, 0, 1], [1, 1, 0, 1, 0, 1, 0, 1])precision([0, 1, 0, 0, 1, 0, 1, 1], [1, 0, 1, 1, 0, 0, 0, 0])
이 지점에서 result() 메서드를 호출하여 현재 지표 값을 얻을 수 있고, variables 속성을 사용하여 진짜 양성과 거짓 양성을 기록한 변수를 확인할 수 있다. 또한, reset_states() 메서드를 사용해 이 변수를 초기화할 수 있다.
precision.reset_states() # 두 변수가 0.0으로 초기화된다
만약 사용자 정의 스트리밍 지표를 만들고 싶다면 tf.keras.metrics.Metric 클래스를 상속한다.
아래의 예는 전체 후버 손실과 지금까지 처리한 샘플 수를 기록하는 클래스이다. 결괏값을 요청하면 평균 후버 손실이 반환된다.
class HuberMetric(tf.keras.metrics.Metric):
def __init__(self, threshold=1.0, **kwargs):
super().__init__(**kwargs) # 기본 매개변수 처리(ex. 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):
sample_metrics = self.huber_fn(y_true, y_pred)
self.total.assign_add(tf.reduce_sum(sample_metrics))
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}[코드 설명]
- 생성자는 add_weight() 메서드를 사용해 여러 배치에 걸쳐 지표의 상태를 기록하기 위한 변수를 만든다. 이 예에서는 후버 손실의 합(total)과 지금까지 처리한 샘플 수(count)를 기록한다. 원한다면 수동으로 변수를 만들 수 있다. 케라스는 속성으로 만들어진 모든 tf.Variable을 관리한다.
- update_state() 메서드는 이 클래스를 함수처럼 사용할 때 호출된다. 배치의 레이블과 예측을 바탕으로 변수를 업데이트 한다.
- result() 메서드는 최종 결과를 계산하고 반환한다. 이 예에서는 모든 샘플에 대한 평균 후버 손실값이다. 이 지표 클래스를 함수처럼 사용하면 먼저 update_state() 메서드가 호출되고 그다음 result() 메서드가 호출되어 출력이 반환된다.
- get_config() 메서드를 구현하여 threshold 변수를 모델과 함께 저장한다.
◆ 사용자 정의 층
텐서플로에 없는 특이한 층을 가진 네트워크를 만들거나 동일한 층 블록이 여러 번 반복되는 네트워크를 만들 경우 각각의 블록을 하나의 층으로 다루는 것이 편리하다. 이러한 경우 사용자 정의 층을 만들다.
tf.keras.layers.Flatten 이나 tf.keras.layers.ReLU와 같이 가중치가 필요 없는 사용자 정의 층을 만드는 가장 간단한 방법은 파이썬 함수를 만든 후 tf.keras.layers.Lambda 층으로 감싸는 것이다.
아래는 입력에 지수 함수를 적용하는 층이다.
exponential_layer = tf.keras.layers.Lambda(lambda x: tf.exp(x))
상태가 있는 가중치를 가진 층을 만들려면 tf.keras.layers.Layer를 상속해야 한다.
아래 예는 Dense 층의 간소화 버전을 구현한 클래스이다.
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):
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')
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.serializer(self.activation)}[코드 설명]
- 생성자는 모든 하이퍼파라미터를 매개변수로 받는다. **kwargs 매개변수를 추가하는 것도 중요하다. 부모 생성자를 호출하면서 kwargs를 전달한다. 이를 통해 input_shape, trainable, name과 같은 기본 매개변수들을 처리할 수 있다. 그다음 하이퍼파라미터를 속성으로 저장하고 activation 매개변수를 tf.keras.activations.get() 함수를 사용해 적절한 활성화 함수로 바꾼다.
- build() 메서드의 역할은 가중치마다 add_weight() 메서드를 호출하여 층의 변수를 만드는 것이며, build() 메서드는 층이 처음 사용될 때 호출된다. 케라스가 층의 입력 크기를 알고 있을 것이므로 build() 메서드의 입력으로 크기를 전달한다. 이 크기는 입력의 마지막 차원 크기에 해당한다. 반드시 build() 메서드 끝에서 부모의 build() 메서드를 호출해야 한다. 이렇게 하면 층이 만들어졌다는 것을 케라스가 인식한다.
- call() 메서드는 이 층에 필요한 연산을 수행한다. 이 경우 입력 X와 층의 커널을 행렬 곱셉하고 편향을 더한다. 그다음 결과에 활성화 함수를 적용한다. 이 값이 층의 출력이다.
- get_config() 메서드는 tf.keras.activations.serializer() 를 사용하여 활성화 함수의 전체 설정을 저장한다.
Concatenate 층과 같은 여러 가지 입력을 받는 층을 만들려면 call() 메서드에 모든 입력이 포함된 튜플을 매개변수 값으로 전달해야 한다. 여러 출력을 가진 층을 만들려면 call() 메서드가 출력의 리스트를 반환해야 한다.
예로 아래는 두 개의 입력과 세 개의 출력을 만드는 층이다. 이 층은 다른 일반적인 층처럼 사용할 수 있는데, 하나의 입력과 하나의 출력을 가진 층만 사용하는 시퀀셜 API에는 사용할 수 없다. (함수형 API와 서브클래싱 API에만 사용 가능)
class MyMultiLayer(tf.keras.layers.Layer):
def call(self, X):
X1, X2 = X
return X1 + X2, X1 * X2, X1 / X2
훈련과 테스트에서 다르게 작동하는 층이 필요하다면 call() 메서드에 training 매개변수를 추가하여 훈련인지 테스트인지를 결정해야 한다. 훈련하는 동안 가우스 잡음을 추가하고 테스트 시에는 아무것도 하지 않는 층을 만들어 본다.
(케라스에는 동일한 작업을 하는 tf.keras.layers.GaussianNoise 층이 있다)
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◆ 사용자 정의 모델
사용자 정의 모델 클래스를 만드는 방법은 tf.keras.Model 클래스를 상속하여 생성자에서 층과 변수를 만들고, 모델이 해야 할 작업을 call() 메서드에 구현한다.
예를 들어 스킵 연결이 있는 사용자 정의 잔차 블록층을 가진 모델을 만들어야 한다고 가정할 경우,
이 모델을 구현하려면 동일한 블록을 여러 개 만들어야 하거나, 다른 모델에 재사용 하기 위해 먼저 ResidualBlock층을 만든다.
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
서브클래싱 API 를 사용해 이 모델을 정의해본다.
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 = inputs
for _ in range(1 + 3):
Z = self.block1(Z)
Z = self.block2(Z)
return self.out(Z)
◆ 모델 구성 요소에 기반한 손실과 지표
아래는 사용자 정의 재구성 손실과 이에 해당하는 사용자 정의 지표를 가지는 모델을 만드는 코드이다.
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)
self.built = True # super().build(batch_input_shape)에 대한 해결 방법
def call(self, inputs, training=None):
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_metric(result)
return self.out(Z)
◆ 자동 미분으로 그레이디언트 계산하기
자동 미분을 사용하여 그레이디언트를 자동으로 계산하는 방법을 이해하기 위해 간단한 함수를 살펴본다.
def f(w1, w2):
return 3 * w1 ** 2 + 2 * w1 * w2
예를 들어 포인트 (w1, w2) = (5, 3)에서 그레이디언트 벡트 구하기
w1, w2 = 5, 3
eps = 1e-6
(f(w1 + eps, w2) - f(w1, w2)) / eps
(f(w1, w2 + eps) - f(w1, w2)) / eps
텐서플로에서 계산하기
w1, w2 = tf.Variable(5.), tf.Variable(3.)
with tf.GradientTape() as tape:
z = f(w1, w2)
gradients = tape.gradient(z, [w1, w2])
gradients
gradient() 메서드가 호출된 후에는 자동으로 테이프가 즉지 지워지기 때문에 gradient() 메서드를 두 번 호출하면 예외가 발생한다. 따라서 gradient() 메서드를 한 번 이상 호출해야 한다면 지속 가능한 테이프를 만들고 사용이 끝난 후 테이프를 삭제하여 리소스를 해제해야 한다.
with tf.GradientTape(persistent=True) as tape:
z = f(w1, w2)
dz_w1 = tape.gradient(z, w1)
dz_w2 = tape.gradient(z, w2)
del tape
기본적으로 테이프는 변수가 포함된 연산만을 기록한다. 만약 변수가 아닌 다른 객체에 대한 z의 그레이디언트를 계산하면 None이 반환된다.
c1, c2 = tf.constant(5.), tf.constant(3.)
with tf.GradientTape() as tape:
z = f(c1, c2)
gradients = tape.gradient(z, [c1, c2])
gradients
그러나 필요한 경우 어떤 텐서든 감시하여 관련된 모든 연산을 기록하도록 강제할 수 있다. 그리고 변수처럼 이런 텐서에 대해 그레이디언트를 계산할 수 있다.
with tf.GradientTape() as tape:
tape.watch(c1)
tape.watch(c2)
z = f(c1, c2)
gradients = tape.gradient(z, [c1, c2])
gradients
대부분의 경우 그레이디언트 테이프는 여러 값에 대한 한 값의 그레이디언트를 계산하는 데 사용된다. 이런 경우 후진 모드 자동 미분이 적합한데, 한 번의 정방향 계산과 역방향 계산으로 모든 그레이디언트를 동시에 계산할 수 있기 때문이다.
만약 개별 그레이디언트를 계산하고 싶다면 테이프의 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)
gradients = tape.gradient(z, [w1, w2])
gradients
이따금 그레이디언트를 계산할 때 수치적인 이슈가 발생할 수 있다. 수치적으로 안정적인 그레이디언트 함수도 반환하도록 my_softplus() 함수를 업데이트 해본다.
@tf.custom_gradient
def my_softplus(z):
def my_softplus_gradients(grads): # grads = 상위 층에서 역전파된 그레이디언트
return grads * (1 - 1 / (1 + tf.exp(z))) # 안정적인 소프트플러스 그레이디언트
result = tf.math.log(1 + tf.exp(-tf.abs(z))) + tf.maximum(0., z)
return result, my_softplus_gradients
my_softplus_gradients() 함수는 이 식을 사용하여 그레이디언트를 계산하고, 이 함수는 지금까지 my_softplus() 함수까지 역전파된 그레이디언트를 입력으로 받으며 연쇄 법칙에 따라 이 함수의 그레이디언트를 곱해야 한다. 이제 큰 입력값에서도 my_better_sotfplus() 함수의 그레이디언트를 올바르게 계산할 수 있다.
◆ 사용자 정의 훈련 반복
먼저 간단한 모델을 만들어 본다.
tf.keras.utils.set_random_seed(42) # 재현성 보장
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)
이제 실제로 적용해보기 위해, 몇 개의 하이퍼파라미터를 정의하고, 옵티마이저, 손실 함수, 지표 (여기서는 MAE)를 선택해야 한다.
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.MeanSquaredError
mean_loss = tf.keras.metrics.Mean()
metrics = [tf.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_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.apply_gradients(zip(gradients, model.trainable_variables))
# 추가 코드 - 모델 변수에 제약이 있는 경우
for variable in model.variables:
if variable.constraint is not None:
variable.assign(variable.constraint(variable))
mean_loss(loss)
for metric in metrics:
metric(y_batch, y_pred)
print_status_bar(step, n_steps, mean_loss, metrics)
for metric in [mean_loss] + metrics:
metric.reset_states()
만약 그레이디언트 클리핑을 하고 싶다면 clipnorm이나 clipvalue 하이퍼파라미터를 지정
만약 가중치에 다른 변환을 적용하려면 apply_gradients() 메서드를 호출하기 전에 수행
모델에 가중치 제한을 추가하고 싶다면 apply_gradients() 다음에 이 제한을 적용하도록 훈련 반복을 수정
for variable in model.variables:
if variable.constraint is not None:
variable.assign(variable.constraint(variable))다음 내용
[딥러닝] 텐서플로 함수와 그래프
텐서플로 [머신러닝] 텐서플로(TensorFlow)란?텐서플로(TensorFlow)란? 텐서플로(TensorFlow)는 구글에서 개발한 오픈소스 머신러닝 프레임워크이다. 주로 딥러닝 모델을 만들고 학습시키는 데 사용되
puppy-foot-it.tistory.com
[출처]
핸즈 온 머신러닝
https://velog.io/@gjtang/Huber-Loss%EB%9E%80
'[파이썬 Projects] > <파이썬 딥러닝, 신경망>' 카테고리의 다른 글
| [딥러닝] 텐서플로 함수와 그래프 (0) | 2024.11.25 |
|---|---|
| [딥러닝] 심층 신경망 훈련 - 3 (1) | 2024.11.25 |
| [딥러닝] 심층 신경망 훈련 (1) | 2024.11.24 |
| [딥러닝] 심층 신경망 훈련 - 2 (2) | 2024.11.24 |
| [딥러닝] 심층 신경망 훈련 - 1 (0) | 2024.11.23 |
