시계열 분석_2.순환 신경망(RNN)_2)RNN 셀 구현

### 라이브러리 호출

import os 
os.environ['TF_CPP_MIN_LOG_LEVEL'] = "2"

import tensorflow as tf
import numpy as np
from tensorflow import keras
from tensorflow.keras import layers
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
from tensorflow.keras.optimizers import Adam

---

### 값 초기화

tf.random.set_seed(22)
np.random.seed(22)
assert tf.__version__.startswith('2.')	# 텐서플로 버전이 2임을 확인

batch_size = 128
total_words = 10000
max_review_len = 80
embedding_len = 100

---

### 데이터셋 준비

(x_train, y_train), (x_test, y_test) = tf.keras.datasets.imdb.load_data(num_words=total_words)  # 설명 1

x_train = tf.keras.preprocessing.sequence.pad_sequences(x_train, maxlen=max_review_len) # 설명 2
x_test = tf.keras.preprocessing.sequence.pad_sequences(x_test, maxlen=max_review_len)

train_data = tf.data.Dataset.from_tensor_slices((x_train, y_train)) # 설명 3
train_data = train_data.shuffle(10000).batch(batch_size, drop_remainder=True) # 설명 4

test_data = tf.data.Dataset.from_tensor_slices((x_test, y_test))  
test_data = test_data.batch(batch_size, drop_remainder=True)

print('x_train_shape: ', x_train.shape , tf.reduce_max(y_train), tf.reduce_min(y_train))
print('x_test_shape: ', x_test.shape)

sample = next(iter(test_data))
print(sample[0].shape)

훈련과 테스트 용도의 데이터셋에 대한 형태를 출력한 결과

상세 설명

• 설명 1: 
  • imdb.load_data() 함수를 사용하여 IMDB 데이터셋 사용
  • num_words = 10000: 등장 빈도 순위가 1 ~ 10000에 해당하는 단어만 사용
• 설명 2: 
  • 패딩(padding): 각 문서 혹은 문장의 단어 수가 제각각이기 때문에 학습을 위해서 보통 숫자 0을 넣어 길이를 맞추어 줌
  • pad_sequence(): 정해 준 길이보다 길이가 긴 샘플은 값을 일부 자르고, 정해 준 길이보다 길이가 짧은 샘플은 값을 0으로 채움
    • x_train: 패딩을 진행할 데이터 
    • maxlen: 모든 데이터에 대해 정규화할 길이 
• 설명 3: 
  • 넘파이 배열을 Dataset으로 변환, 전제 데이터를 메모리로 로딩해야 하므로 큰 용량의 데이터가 필요
  • from_generator: 데이터를 한 번에 메모리에 로딩하는 것이 아니라, 필요할 때만 사용
• 설명 4: 
  • shuffle(): 데이터셋을 임으로 섞어 줌, 데이터셋이 완벽하게 섞이기 위해 전체 데이터셋의 크기에 비해 크거나 같은 버퍼 크기로 지정
  • batch(): 데이터셋의 항목들을 하나의 배치로 묶어줌
  • drop_remainder(): 마지막 배치 크기를 무시하고 지정한 배치 크기를 사용

---

### RNN 셀을 이용한 네트워크 생성

class RNN_Build(tf.keras.Model):  # 설명 1
  def __init__(self, units):  # 설명 2
    super(RNN_Build, self).__init__() # 설명 3

    self.state0 = [tf.zeros([batch_size, units])] # 설명 4
    self.state1 = [tf.zeros([batch_size, units])]
    self.embedding = tf.keras.layers.Embedding(total_words, embedding_len, input_length=max_review_len) # 설명 5

    self.RNNCell0 = tf.keras.layers.SimpleRNNCell(units, dropout=0.2) # 설명 6
    self.RNNCell1 = tf.keras.layers.SimpleRNNCell(units, dropout=0.2)
    self.outlayer = tf.keras.layers.Dense(1)

  def call(self, inputs, training=None):  
    x = inputs
    x = self.embedding(x)
    state0 = self.state0
    state1 = self.state1
    for word in tf.unstack(x, axis=1):
      out0, state0 = self.RNNCell0(word, state0, training)
      out1, state1 = self.RNNCell1(word, state1, training)
    x = self.outlayer(out1)
    prob = tf.sigmoid(x)
    return prob

상세 설명

• 설명 1: 
  • 객체 지향 프로그램으로 구현
  • 구조를 설계한 후 재사용성을 고려하거나 코드의 반복을 최소화
• 설명 2: 
  • 클래스 인스턴스 생성할 때 초기화하는 부분
  • init은 객체가 생성될 때 호출되며, call은 인스턴스가 생성될 때 호출 
  •  
• 설명 3: 
  • 부모(RNN_Build) 클래스의 __init__ 메서드 호출
  • super() 뒤에 '.'을 붙여 메서드 호출
• 설명 4: 
  • self는 자신의 인스턴스를 의미
  • tf.zeros를 사용하여 0 값으로 채워진 텐서를 생성해서 state0에 저장
• 설명 5:
  • 임베딩층(embedding layer)을 사용하려면 각 입력이 모두 정수로 변환된 상태로 인코딩 되어야 함
    • total_words: 텍스트 데이터의 전체 단어 집합 크기
    • embedding_len: 임베딩이 되고 난 후 단어의 차원
    • input_length: 입력 데이터의 길이 
• 설명 6:
  • SimpleRNN의 셀 클래스
    
    • units: 출력 공간의 차원
    • dropout: 0과 1의 부동 소수점. 입력 중에서 삭제할 유닛의 비율
• 설명 7:
  • 그룹으로 묶은 데이터를 행렬 형태로 전환하여 연산

---

### 모델 훈련

import time
units = 64
epochs = 4
t0 = time.time()

model = RNN_Build(units)
model.compile(optimizer=tf.keras.optimizers.Adam(0.001),
              loss=tf.losses.BinaryCrossentropy(),
              metrics=['accuracy'],
              experimental_run_tf_function=False) # 설명 1

model.fit(train_data, epochs=epochs, validation_data=test_data, validation_freq=2)

모델 훈련 출력한 결과

상세 설명

• 설명 1: 
  • model.compile()에서 다양한 파라미터를 정의
    
    • optimizer: 옵티마이저 설정 / 해당 모델에서는 0.001의 학습률을 적용한 Adam 사용
    • loss: 손실 함수 설정 / 해당 모델에서는 이진 분류인 BinaryCrossentropy 사용
    • metrics: 훈련을 모니터링하기 위한 지표 선택 / 해당 모델에서는 정확도(accuracy)를 사용  
    • experimental_run_tf_function: 모델을 인스턴스화하는 기능을 제송
• 설명 2: 
  • model.fit()은 모델을 학습하는 데 사용
    
    • train_data: 입력 데이터
    • epochs: 학습 데이터 반복 횟수
    • validation_data: 검증 데이터 
    • validation_freq: epoch마다 검증 데이터 셋에 대한 계산을 수행하지 않고 간격을 두어 계산

---

### 모델 평가

print("훈련 데이터셋 평가 >>> ")
(loss, accuracy) = model.evaluate(train_data, verbose=0)  # 설명 1
print("loss={:.4f}, accuracy:{:.4f}%".format(loss, accuracy * 100))
print("테스트 데이터셋 평가 >>> ")
(loss, accuracy) = model.evaluate(test_data, verbose=0)
print("loss={:.4f}, accuracy : {:4f}%".format(loss, accuracy * 100))
t1 = time.time()
print('시간: ', t1-t0)

모델 평가를 실행한 결과

상세 설명

• 설명 1: 
  • model.evaluate는 모델을 평가하기 위한 함수
    
    • train_data: 훈련 데이터
    • verbose: 얼마나 자세하게 정보를 표시할지 지정

 

참고: 서지영, 『딥러닝 텐서플로 교과서』, 길벗(2022)