핸즈온 머신러닝 - Part2. 신경망과 딥러닝 ch16. RNN과 어텐션을 사용한 자연어 처리

govlKH·2023년 9월 6일

목록 보기

14/15

핸즈온 머신러닝 - Part2. 신경망과 딥러닝

ch16. RNN과 어텐션을 사용한 자연어 처리

16.1 셰익스피어 작품

1) 셰익스피어 작품 다운로드

import keras

shakespeare_url = "https://homl.info/shakespeare"
filepath = keras.utils.get_file("shajkespeare.txt", shakespeare_url)
with open(filepath) as f:
  shakespeare_text = f.read()

2) 확인하기

shakespeare_text

3) tokenizer 만들기 - fit 시키기

import tensorflow as tf

# char_level=True : 단어 수준 인코딩 대신 글자 수준 인코딩
tokenizer = tf.keras.preprocessing.text.Tokenizer(char_level=True) 
tokenizer.fit_on_texts(shakespeare_text)

# 고유 글자 개수
max_id = len(tokenizer.word_index)

# 전체 글자 개수
dataset_size = tokenizer.document_count

print(max_id, dataset_size)

=> 39, 1115394

4) fit시킨 tokenizer 적용

# 전체 텍스트를 인코딩하여 각 글자를 ID로 나타내기
# 1~39 대신 0~38 을 얻기 위해 1을 빼기

import numpy as np

[encoded] = np.array(tokenizer.texts_to_sequences([shakespeare_text])) - 1

5) dataset을 나누기(text에 있는 글자를 섞어서는 X!)

시계열을 다룰 때, 주로 시간의 순서에 따라 나누는 것이 안전하다.

암묵적으로는 RNN이 과거에서 학습하는 패턴이 미래에도 등장한다고 가정한다. 즉, 변하지 않는다는 statioinary를 가정한다.

하지만 금융시장같이 굉장히 불안정한 경우는 패턴을 발견한 뒤 적용하면 바로 사라지기도 한다.

따라서 모델을 검증해야 하는데, 만약 dev set에서 초반에 성능이 오히려 좋고 나중에 성능이 좋지 않다면 시계열이 충분히 안정되지 않은 것 일 수 있다.

이럴 때는 더 짧은 시간 간격으로 모델을 훈련하는 것이 좋다.

# 0.9 0.05 0.05

train_size = dataset_size * 90 // 100
dataset = tf.data.Dataset.from_tensor_slices(encoded[:train_size])

이렇게 만들어진 dataset은 백만 개 이상의 글자로 이루어진 시퀀스 중 하나이다.

이것을 바로 훈련시키기에는 너무 시간이 오래걸리기에 window로 나누어 이용한다.

RNN은 부분 문자열 길이만큼만 역전파를 위해 펼치는데, 이를 TBPTT(Truncated backpropagation through time) 이라고 부른다.

n_step = 100
window_length = n_step + 1 # target = 1글자 앞의 input
# shift : 얼마나 이동해갈지, drop_remainder : 남는건 버리기
dataset = dataset.window(window_length, shift=1, drop_remainder=True) 

# 모델은 tensor를 원하기에 다시 바꿔주기
dataset = dataset.flat_map(lambda window: window.batch(window_length))

경사하강법은 훈련 셋 샘플이 동일독립분포(iid)를 따를 때 가장 잘 작동하기 때문에 윈도우를 섞는다.

그 다음 윈도우를 배치로 만들고 입력(처음 100개 글자)과 타깃(마지막 100개 글자)를 분리한다.

batch_size = 32
dataset = dataset.shuffle(10000).batch(batch_size)
dataset = dataset.map(lambda windows: (windows[:, :-1], windows[:, 1:]))


# 원핫벡터를 사용해 글자를 인코딩

dataset = dataset.map(
    lambda X_batch, Y_batch: (tf.one_hot(X_batch, depth=max_id), Y_batch)
) # max_id = 39

dataset = dataset.prefetch(1)

Q. dataset = dataset.prefetch(1) 를 사용하는 이유?
dataset.prefetch(1)은 TensorFlow에서 데이터셋을 효율적으로 처리하기 위한 기능 중 하나입니다. 이 메서드는 데이터셋에서 배치를 추출하는 동안 데이터를 사전에 가져와서 메모리로 미리 로드하도록 도와줍니다.

설명:

prefetch 메서드는 모델의 학습 루프 동안 데이터셋을 효율적으로 처리하기 위해 사용됩니다.
(1)은 매개변수로서, 메모리에 미리 가져올 배치의 개수를 나타냅니다. 일반적으로 1을 사용하면 됩니다.
prefetch를 사용하면 GPU 또는 CPU와 같은 계산 장치가 데이터를 처리하는 동안 데이터 로딩과 전처리가 병렬로 수행됩니다. 이렇게 하면 데이터 로딩이 모델 학습에 대한 병목 현상을 줄일 수 있습니다.
예를 들어, 데이터셋이 매우 크고 데이터를 로드하고 전처리하는 데 시간이 오래 걸릴 때 prefetch를 사용하면 학습 속도를 높일 수 있습니다. 이렇게 하면 모델이 학습 중에 데이터가 준비될 때 대기하지 않고 지속적으로 데이터를 처리할 수 있습니다.

일반적으로 데이터셋 파이프라인에서 prefetch를 사용하는 것은 데이터 로딩과 모델 학습을 효율적으로 조율하는 데 도움이 되는 좋은 방법입니다.

6) Char-RNN 모델 만들고 훈련하기

model = keras.models.Sequential([
    keras.layers.GRU(128, return_sequences=True, input_shape=[None, max_id],
                     dropout=0.2, recurrent_dropout=0.2),
    keras.layers.GRU(128, return_sequences=True,
                     dropout=0.2, recurrent_dropout=0.2),
    keras.layers.TimeDistributed(keras.layers.Dense(max_id,
                                                    activation='softmax'))
])

model.compile(loss="sparse_categorical_crossentropy", optimizer='adam')
history = model.fit(dataset, epochs=1)

7) Char-RNN 모델 사용하기

def preprocess(texts):
  X = np.array(tokenizer.texts_to_sequences(texts)) - 1
  return tf.one_hot(X, max_id)

X_new = preprocess(['How are yo'])
Y_pred = np.argmax(model(X_new), axis=-1)
tokenizer.sequences_to_texts(Y_pred + 1)[0][-1]

8) 가짜 셰익스피어 text 생성하기

새롭게 generate한 글자를 맨 끝에 추가하고, 늘어난 텍스트를 모델에 전달하여 다음 글자를 예측하는 식이다.

실제로는 이렇게 하면 같은 단어가 계속 반복되어 나오기에, tf.random.categorical() 함수를 이용하여 모델이 추정한 확률을 기반으로 단어를 뽑게 된다. 이를 통해 exploration을 주어 다채로운 글을 만들 수 있게 한다.

또한 temperature을 이용하여 0에 가까우면 높은 확률을 갖는 단어를 뽑게 하고, 반대로 매우 높으면 모든 글자가 동일한 확률을 갖게 한다.

# 다음 글자 만드는 함수
def next_char(text, temperature=1):
  X_new = preprocess([text])
  y_proba = model(X_new)[0. -1:, :]
  rescaled_logits = tf.math.log(y_proba) / temperature
  char_id = tf.random.categorical(rescaled_logits, num_samples=1) + 1
  return tokenizer.seqeunces_to_texts(char_id.numpy())[0]
  
  
# 위에서 만든 함수를 통해 text generate하는 함수
def complete_text(text, n_chars=50, temperature=1):
  for _ in range(n_chars):
    text += next_char(text, temperature)
    return text

위에서 만든 함수를 통해 text generate하기

print(complete_text("t", temperature=0.2))

print(complete_text("w", temperature=0.1))

print(complete_text("w", temperature=0.15))

print(complete_text("w", temperature=1))

print(complete_text("w", temperature=2))

temperature =1 에서 잘 작동하는 것으로 보인다.