TensworFlow 2.0을 이용한 Char-RNN구현

from __future__ import absolute_import, division, print_function, unicode_literals

from absl import app
import tensorflow as tf

import numpy as np
import os
import time

# input 데이터와 input 데이터를 한글자씩 뒤로 민 target 데이터를 생성하는 utility 함수를 정의합니다.
def split_input_target(chunk):
  input_text = chunk[:-1]
  target_text = chunk[1:]

  return input_text, target_text

# 학습에 필요한 설정값들을 지정합니다.
data_dir = tf.keras.utils.get_file('shakespeare.txt', 'https://storage.googleapis.com/download.tensorflow.org/data/shakespeare.txt')  # shakespeare
#data_dir = './data/linux/input.txt'  # linux
batch_size = 64      # Training : 64, Sampling : 1
seq_length = 100     # Training : 100, Sampling : 1
embedding_dim = 256  # Embedding 차원
hidden_size = 1024   # 히든 레이어의 노드 개수
num_epochs = 10

# 학습에 사용할 txt 파일을 읽습니다.
text = open(data_dir, 'rb').read().decode(encoding='utf-8')
# 학습데이터에 포함된 모든 character들을 나타내는 변수인 vocab과
# vocab에 id를 부여해 dict 형태로 만든 char2idx를 선언합니다.
vocab = sorted(set(text))  # 유니크한 character 개수
vocab_size = len(vocab)
print('{} unique characters'.format(vocab_size))
char2idx = {u: i for i, u in enumerate(vocab)}
idx2char = np.array(vocab)

# 학습 데이터를 character에서 integer로 변환합니다.
text_as_int = np.array([char2idx[c] for c in text])

# split_input_target 함수를 이용해서 input 데이터와 input 데이터를 한글자씩 뒤로 민 target 데이터를 생성합니다.
char_dataset = tf.data.Dataset.from_tensor_slices(text_as_int)
sequences = char_dataset.batch(seq_length+1, drop_remainder=True)
dataset = sequences.map(split_input_target)

# tf.data API를 이용해서 데이터를 섞고 batch 형태로 가져옵니다.
dataset = dataset.shuffle(10000).batch(batch_size, drop_remainder=True)

# tf.keras.Model을 이용해서 RNN 모델을 정의합니다.
class RNN(tf.keras.Model):
 def __init__(self, batch_size):
   super(RNN, self).__init__()
   self.embedding_layer = tf.keras.layers.Embedding(vocab_size, embedding_dim,
                                                    batch_input_shape=[batch_size, None])
   self.hidden_layer_1 = tf.keras.layers.LSTM(hidden_size,
                                             return_sequences=True,
                                             stateful=True,
                                             recurrent_initializer='glorot_uniform')
   self.output_layer = tf.keras.layers.Dense(vocab_size)

 def call(self, x):
   embedded_input = self.embedding_layer(x)
   features = self.hidden_layer_1(embedded_input)
   logits = self.output_layer(features)

   return logits

# sparse cross-entropy 손실 함수를 정의합니다.
def sparse_cross_entropy_loss(labels, logits):
  return tf.reduce_mean(tf.keras.losses.sparse_categorical_crossentropy(labels, logits, from_logits=True))

# 최적화를 위한 Adam 옵티마이저를 정의합니다.
optimizer = tf.keras.optimizers.Adam()

# 최적화를 위한 function을 정의합니다.
@tf.function
def train_step(model, input, target):
  with tf.GradientTape() as tape:
    logits = model(input)
    loss = sparse_cross_entropy_loss(target, logits)
  grads = tape.gradient(loss, model.trainable_variables)
  optimizer.apply_gradients(zip(grads, model.trainable_variables))

  return loss

def generate_text(model, start_string):
  num_sampling = 4000  # 생성할 글자(Character)의 개수를 지정합니다.

  # start_sting을 integer 형태로 변환합니다.
  input_eval = [char2idx[s] for s in start_string]
  input_eval = tf.expand_dims(input_eval, 0)

  # 샘플링 결과로 생성된 string을 저장할 배열을 초기화합니다.
  text_generated = []

  # 낮은 temperature 값은 더욱 정확한 텍스트를 생성합니다.
  # 높은 temperature 값은 더욱 다양한 텍스트를 생성합니다.
  temperature = 1.0

  # 여기서 batch size = 1 입니다.
  model.reset_states()
  for i in range(num_sampling):
    predictions = model(input_eval)
    # 불필요한 batch dimension을 삭제합니다.
    predictions = tf.squeeze(predictions, 0)

    # 모델의 예측결과에 기반해서 랜덤 샘플링을 하기위해 categorical distribution을 사용합니다.
    predictions = predictions / temperature
    predicted_id = tf.random.categorical(predictions, num_samples=1)[-1,0].numpy()

    # 예측된 character를 다음 input으로 사용합니다.
    input_eval = tf.expand_dims([predicted_id], 0)
    # 샘플링 결과를 text_generated 배열에 추가합니다.
    text_generated.append(idx2char[predicted_id])

  return (start_string + ''.join(text_generated))

def main(_):
  # Recurrent Neural Networks(RNN) 모델을 선언합니다.
  RNN_model = RNN(batch_size=batch_size)

  # 데이터 구조 파악을 위해서 예제로 임의의 하나의 배치 데이터 에측하고, 예측결과를 출력합니다.
  for input_example_batch, target_example_batch in dataset.take(1):
    example_batch_predictions = RNN_model(input_example_batch)
    print(example_batch_predictions.shape, "# (batch_size, sequence_length, vocab_size)")

  # 모델 정보를 출력합니다.
  RNN_model.summary()

  # checkpoint 데이터를 저장할 경로를 지정합니다.
  checkpoint_dir = './training_checkpoints'
  checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt_{epoch}")

  for epoch in range(num_epochs):
    start = time.time()

    # 매 반복마다 hidden state를 초기화합니다. (최초의 hidden 값은 None입니다.)
    hidden = RNN_model.reset_states()

    for (batch_n, (input, target)) in enumerate(dataset):
      loss = train_step(RNN_model, input, target)

      if batch_n % 100 == 0:
        template = 'Epoch {} Batch {} Loss {}'
        print(template.format(epoch+1, batch_n, loss))

    # 5회 반복마다 파라미터를 checkpoint로 저장합니다.
    if (epoch + 1) % 5 == 0:
      RNN_model.save_weights(checkpoint_prefix.format(epoch=epoch))

    print ('Epoch {} Loss {:.4f}'.format(epoch+1, loss))
    print ('Time taken for 1 epoch {} sec\n'.format(time.time() - start))

  RNN_model.save_weights(checkpoint_prefix.format(epoch=epoch))
  print("트레이닝이 끝났습니다!")

  sampling_RNN_model = RNN(batch_size=1)
  sampling_RNN_model.load_weights(tf.train.latest_checkpoint(checkpoint_dir))
  sampling_RNN_model.build(tf.TensorShape([1, None]))
  sampling_RNN_model.summary()

  # 샘플링을 시작합니다.
  print("샘플링을 시작합니다!")
  print(generate_text(sampling_RNN_model, start_string=u' '))

if __name__ == '__main__':
  # main 함수를 호출합니다.
  app.run(main)

 

 

동작 방식을 순서대로 대략 설명하면,

처음에 띄어쓰기 한개가 들어간다.

 

띄어쓰기 다음 단어에 가장 가까운 단어 한개를 출력한다.

그리고 그 단어에 가장 가까운 단어를 출력한다. 를 반복하면서 모든 문장을 생성하는 것이다.

 

이 결과로 완벽하지는 않지만, 셰익스피어 어투의 문장이 생성되는 것이다.

https://namu.wiki/w/%EC%A4%91%EA%B5%AD%EC%96%B4%20%EB%B0%A9

중국어 방

위 링크를 참고해서 읽어보면 대강 어떤느낌으로 문장이 생성되고 연결되는지 알 수 있다.

---

1. 데이터 준비 및 전처리

# input 데이터와 input 데이터를 한글자씩 뒤로 민 target 데이터를 생성하는 utility 함수를 정의합니다.
def split_input_target(chunk):
  input_text = chunk[:-1]
  target_text = chunk[1:]

  return input_text, target_text

# 학습에 필요한 설정값들을 지정합니다.
data_dir = tf.keras.utils.get_file('shakespeare.txt', 'https://storage.googleapis.com/download.tensorflow.org/data/shakespeare.txt')  # shakespeare
batch_size = 64      # Training : 64, Sampling : 1
seq_length = 100     # Training : 100, Sampling : 1
embedding_dim = 256  # Embedding 차원
hidden_size = 1024   # 히든 레이어의 노드 개수

# 학습에 사용할 txt 파일을 읽습니다.
text = open(data_dir, 'rb').read().decode(encoding='utf-8')
vocab = sorted(set(text))  # 유니크한 character 개수
vocab_size = len(vocab)
char2idx = {u: i for i, u in enumerate(vocab)}
idx2char = np.array(vocab)

# 학습 데이터를 character에서 integer로 변환합니다.
text_as_int = np.array([char2idx[c] for c in text])

# split_input_target 함수를 이용해서 input 데이터와 input 데이터를 한글자씩 뒤로 민 target 데이터를 생성합니다.
char_dataset = tf.data.Dataset.from_tensor_slices(text_as_int)
sequences = char_dataset.batch(seq_length+1, drop_remainder=True)
dataset = sequences.map(split_input_target)

# tf.data API를 이용해서 데이터를 섞고 batch 형태로 가져옵니다.
dataset = dataset.shuffle(10000).batch(batch_size, drop_remainder=True)

 

• split_input_target 함수: 이 함수는 입력 시퀀스를 받아서 마지막 문자를 제외한 부분을 input_text로, 첫 번째 문자를 제외한 부분을 target_text로만든다. 이 함수는 데이터셋을 생성할 때 사용.
• data_dir: 텍스트 파일의 경로나 URL을 지정하여 데이터를 가져온다.
  여기서는 셰익스피어의 작품을 다운로드 받는다.
• text: 파일에서 읽은 데이터를 utf-8로 디코딩하여 전체 텍스트로 저장.
• vocab: 텍스트에서 사용된 고유한 문자들의 집합을 정렬하여 저장.
• char2idx와 idx2char: 각 문자를 해당 인덱스로 매핑하는 딕셔너리를 생성.
  이 딕셔너리는 텍스트를 정수 배열로 변환하거나 반대로 변환할 때 사용한다.
• text_as_int: 전체 텍스트를 정수 배열로 변환하여 저장
  각 문자가 그에 해당하는 정수로 매핑된다. 키 Value값 형태로 기억하자
• char_dataset과 sequences: tf.data API를 사용하여 정수 배열을 입력 데이터와 타겟 데이터로 나누어
  데이터셋을 생성.
  sequences는 입력과 타겟 시퀀스를 나누고, dataset은 데이터를 섞고 배치 형태로 준비.

---

2. 모델 정의

# tf.keras.Model을 이용해서 RNN 모델을 정의합니다.
class RNN(tf.keras.Model):
    def __init__(self, batch_size):
        super(RNN, self).__init__()
        self.embedding_layer = tf.keras.layers.Embedding(vocab_size, embedding_dim)
        self.hidden_layer_1 = tf.keras.layers.LSTM(hidden_size,
                                                   return_sequences=True,
                                                   stateful=True,
                                                   recurrent_initializer='glorot_uniform')
        self.output_layer = tf.keras.layers.Dense(vocab_size)

    def call(self, x):
        embedded_input = self.embedding_layer(x)
        features = self.hidden_layer_1(embedded_input)
        logits = self.output_layer(features)
        return logits

• RNN 클래스: tf.keras.Model을 상속받아 RNN 모델을 정의
  
  
• embedding_layer: Embedding 레이어로, 문자 인덱스를 임베딩 벡터로 변환
  
  
• hidden_layer_1: LSTM 레이어로, 임베딩된 입력을 받아 시퀀스를 처리하고 hidden state를 유지
  
  
• output_layer: 출력 레이어로, LSTM의 출력을 받아 각 문자의 다음 글자일 확률을 출력
  
  
• call 함수: 입력 데이터 x를 받아 임베딩, LSTM, 출력 레이어를 순차적으로 적용하여 logits(확률 분포)를 반환

---

3. 손실 함수와 최적화

def sparse_cross_entropy_loss(labels, logits):
  return tf.reduce_mean(tf.keras.losses.sparse_categorical_crossentropy(labels, logits, from_logits=True))

optimizer = tf.keras.optimizers.Adam()

 

• sparse_cross_entropy_loss : 희소 교차 엔트로피 손실 함수 정의.
  
  
• optimizer : Adam 옵티마이저 정의.

---

4. 훈련 함수

# 최적화를 위한 function을 정의합니다.
@tf.function
def train_step(model, input, target):
  with tf.GradientTape() as tape:
    logits = model(input)
    loss = sparse_cross_entropy_loss(target, logits)
  grads = tape.gradient(loss, model.trainable_variables)
  optimizer.apply_gradients(zip(grads, model.trainable_variables))

  return loss

 

• train_step : 훈련 단계를 정의.
  
  
• model : 훈련할 모델.
  
  
• input : 입력 데이터.
  
  
• target : 목표 데이터.
  
  
• tape : 그래디언트 계산을 위한 테이프.
• grads : 계산된 그래디언트.

 

---

5. 텍스트 생성 함수

def generate_text(model, start_string):
  num_sampling = 4000  # 생성할 글자(Character)의 개수를 지정합니다.

  # start_sting을 integer 형태로 변환합니다.
  input_eval = [char2idx[s] for s in start_string]
  input_eval = tf.expand_dims(input_eval, 0)

  # 샘플링 결과로 생성된 string을 저장할 배열을 초기화합니다.
  text_generated = []

  # 낮은 temperature 값은 더욱 정확한 텍스트를 생성합니다.
  # 높은 temperature 값은 더욱 다양한 텍스트를 생성합니다.
  temperature = 1.0

  # 여기서 batch size = 1 입니다.
  model.reset_states()
  for i in range(num_sampling):
    predictions = model(input_eval)
    # 불필요한 batch dimension을 삭제합니다.
    predictions = tf.squeeze(predictions, 0)

    # 모델의 예측결과에 기반해서 랜덤 샘플링을 하기위해 categorical distribution을 사용합니다.
    predictions = predictions / temperature
    predicted_id = tf.random.categorical(predictions, num_samples=1)[-1,0].numpy()

    # 예측된 character를 다음 input으로 사용합니다.
    input_eval = tf.expand_dims([predicted_id], 0)
    # 샘플링 결과를 text_generated 배열에 추가합니다.
    text_generated.append(idx2char[predicted_id])

  return (start_string + ''.join(text_generated))

• generate_text : 텍스트를 생성하는 함수.
• num_sampling : 생성할 글자의 개수.
• input_eval : 시작 문자열을 정수로 변환한 값.
• text_generated : 생성된 텍스트를 저장할 리스트.
• temperature : 텍스트 생성의 다양성을 조절하는 파라미터.
• predicted_id : 예측된 다음 문자.

---

6. 메인 함수

def main(_):
  # Recurrent Neural Networks(RNN) 모델을 선언합니다.
  RNN_model = RNN(batch_size=batch_size)

  # 데이터 구조 파악을 위해서 예제로 임의의 하나의 배치 데이터 에측하고, 예측결과를 출력합니다.
  for input_example_batch, target_example_batch in dataset.take(1):
    example_batch_predictions = RNN_model(input_example_batch)
    print(example_batch_predictions.shape, "# (batch_size, sequence_length, vocab_size)")

  # 모델 정보를 출력합니다.
  RNN_model.summary()

  # checkpoint 데이터를 저장할 경로를 지정합니다.
  checkpoint_dir = './training_checkpoints'
  checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt_{epoch}")

  for epoch in range(num_epochs):
    start = time.time()

    # 매 반복마다 hidden state를 초기화합니다. (최초의 hidden 값은 None입니다.)
    hidden = RNN_model.reset_states()

    for (batch_n, (input, target)) in enumerate(dataset):
      loss = train_step(RNN_model, input, target)

      if batch_n % 100 == 0:
        template = 'Epoch {} Batch {} Loss {}'
        print(template.format(epoch+1, batch_n, loss))

    # 5회 반복마다 파라미터를 checkpoint로 저장합니다.
    if (epoch + 1) % 5 == 0:
      RNN_model.save_weights(checkpoint_prefix.format(epoch=epoch))

    print ('Epoch {} Loss {:.4f}'.format(epoch+1, loss))
    print ('Time taken for 1 epoch {} sec\n'.format(time.time() - start))

  RNN_model.save_weights(checkpoint_prefix.format(epoch=epoch))
  print("트레이닝이 끝났습니다!")

  sampling_RNN_model = RNN(batch_size=1)
  sampling_RNN_model.load_weights(tf.train.latest_checkpoint(checkpoint_dir))
  sampling_RNN_model.build(tf.TensorShape([1, None]))
  sampling_RNN_model.summary()

  # 샘플링을 시작합니다.
  print("샘플링을 시작합니다!")
  print(generate_text(sampling_RNN_model, start_string=u' '))

if __name__ == '__main__':
  # main 함수를 호출합니다.
  app.run(main)

• main : 메인 함수.
• RNN_model : 훈련할 RNN 모델.
• example_batch_predictions : 예제 배치의 예측 결과.
• checkpoint_dir : 체크포인트 저장 디렉터리.
• checkpoint_prefix : 체크포인트 파일명 접두사.
• start : 에포크 시작 시간.
• hidden : 초기 hidden state.
• sampling_RNN_model : 샘플링을 위한 RNN 모델.