Transformer (33-4 한국어 챗봇 구현하기)

■ 트랜스포머를 이용한 한국어 챗봇(Transformer Chatbot Tutorial)

앞서 구현한 트랜스포머 코드를 사용하여 일상 대화 챗봇을 구현해보려고합니다.

물론 성능이나 대화흐름이 엄청 자연스럽지는않지만 어느정도 트랜스포머로 구현하는것이 목표입니다.

데이터로더 & 전처리

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import re
import urllib.request
import time
import tensorflow_datasets as tfds  ## !! -> 파이토치중 데이터로더를 통해서 테스트,학습 분리해주는것인데 그런 비슷한것.
import tensorflow as tf

urllib.request.urlretrieve("https://raw.githubusercontent.com/songys/Chatbot_data/master/ChatbotData.csv", filename="ChatBotData.csv")
train_data = pd.read_csv('ChatBotData.csv')
train_data.head()

이번 예제에서는 토큰화를 위해서 형태소 분석기를 사용하지 않으나 학습기반의 토크나이저를 사용할 예정입니다.

 

구두점 처리 : 제거하지않고 구두점 앞에 공백을 추가하여 다른 문자들과 구분을 하려고 합니다.

예를들어서 '12시 땡!' -> '12시 땡 !' 으로요 이 작업은 정규표현식으로 가능하니 정규표현식으로 사용할겁니다.

questions = []
for sentense in train_data['Q']:
  sentence = re.sub(r"([?.!,])", r" \1 ", sentense)  # 구두점에 대해 띄어쓰기 처리
  sentence = sentence.strip()
  questions.append(sentence)
  
  answers = []
for sentense in train_data['A']:
  sentence = re.sub(r"([?.!,])", r" \1 ", sentense)  # 구두점에 대해 띄어쓰기 처리
  sentence = sentence.strip()
  answers.append(sentence)

이렇게 구두점에 대해서 띄어쓰기를하는데 questions, answers 둘다 모두 같은 구두점처리를 해주었습니다.

print(questions[:5])
print(answers[:5])

. , ! 등이 모두 공백으로 처리된것을 볼 수 있습니다.

---

단어사전

# 질문 + 답변 데이터로부터 단어 집합을 생성
tokenizer = tfds.deprecated.text.SubwordTextEncoder.build_from_corpus(questions + answers, target_vocab_size=2**13)

질문과 답변 데이터로부터 단어집합을 생성하도록 합니다.

 

subwordTextEncoder가 토큰화해주는 작업이 포함되어있기때문에돌리면 살짝 시간이 걸립니다.

 

위 단어집합에 <sos> <eos> 토큰도 추가해주어야합니다.

START_TOKEN, END_TOKEN = [tokenizer.vocab_size], [tokenizer.vocab_size + 1]

# 단어 집합의 크기 조정 +2
VOCAB_SIZE = tokenizer.vocab_size + 2

#확인코드
print('시작 토큰 번호 :',START_TOKEN)
print('종료 토큰 번호 :',END_TOKEN)
print('단어 집합의 크기 :',VOCAB_SIZE) #0~8179 해서 총 8180개


시작 토큰 번호 : [8178]
종료 토큰 번호 : [8179]
단어 집합의 크기 : 8180

 

이렇게보면 시작토큰과 종료토큰번호르 ㄹ확인할 수 있습니다.

단어 집합의 크기는 0부터 8179해서 총 8180으로됩니다.

---

정수 인코딩과 패딩

question[20]을 확인해보니 '가스비 비싼데 감기 걸리겠어'로 보입니다.

텍스트 시퀀스를 정수 시퀀스로 변환해보려고합니다.

tokenizer.encoder(question[20])을 하니 결과값을 [5766, 611, 3509, 141, 685, 3747, 849]로 받을 수 있었습니다.

# decode()
# 정수 시퀀스 -> 텍스트 시퀀스 복원

sample_string = questions[20]

tokenized_string = tokenizer.encode(sample_string)
print(f'정수 인코딩 {tokenized_string}')

original_string = tokenizer.decode(tokenized_string)
print(f'기존 문장 {original_string}')

정수인코딩 [5766, 611, 3509, 141, 685, 3747, 849]

기존문장 가스비 비싼데 감기 걸리겠어

로 출력됩니다.  형태소 분석을 하지 않았는데 이렇게 되는이유는 위에서 했던 SubwordTextEncoder 학습에 의해 분류된것입니다.

5766 ----> 가스 611 ----> 비  3509 ----> 비싼 141 ----> 데  685 ----> 감기  3747 ----> 걸리 849 ----> 겠어

 

이후에 패딩진행을 하려고합니다

# 패딩 진행
MAX_LENGTH = 40

# 토큰화 / 정수 인코딩 / 시작토큰, 종료토큰 추가 / 패딩
def tokenize_and_filter(inputs, outputs):
  tokenized_inputs, tokenized_outputs = [], []

  for (sentence1, sentence2) in zip(inputs, outputs):
    # 정수인코딩, 시작토큰 과 종료토큰 추가
    sentence1 = START_TOKEN + tokenizer.encode(sentence1) + END_TOKEN
    sentence2 = START_TOKEN + tokenizer.encode(sentence2) + END_TOKEN

    tokenized_inputs.append(sentence1)
    tokenized_outputs.append(sentence2)

  # 패딩
  tokenized_inputs = tf.keras.preprocessing.sequence.pad_sequences(
      tokenized_inputs, maxlen=MAX_LENGTH, padding='post')

  tokenized_outputs = tf.keras.preprocessing.sequence.pad_sequences(
    tokenized_outputs, maxlen=MAX_LENGTH, padding='post')

  return tokenized_inputs, tokenized_outputs

questions, answers = tokenize_and_filter(questions, answers)
questions.shape
(11823, 40)

answers.shape
(11823, 40)

으로 패딩작업으로 question answers까지 모두 차원, 길이가 맞춰진것을 확인할 수 있습니다.

---

인코더와 디코더의 입력, 그리고 레이블을 만들어보기

tf.data.Dataset을 사용해서 데이터를 배치 단위로 불러오겠습니다.

텐서플로우 dataset을 이용하여 셔플을 수행하되, 배치 크기로 데이터를 묶으려고합니다.

또한 이 과정에서 교사 강요를 사용하기 위해서 디코더의 입력과 실제값 시퀀스를 구성한다.

BATCH_SIZE = 64
BUFFER_SIZE = 20000

# 디코더의 실제값 시퀀스에서는 시작 토큰을 제거해야 한다.
dataset = tf.data.Dataset.from_tensor_slices((
    {
      'inputs': questions,
      'dec_inputs': answers[:, :-1]   # 디코더의 입력.  마지막 패딩 토큰 제거.
    },
    {
      'outputs': answers[:, 1:]   # 맨 처음 토근이 제거됨.  즉 시작토큰 제거
    }
))

dataset = dataset.cache()
dataset = dataset.shuffle(BUFFER_SIZE)
dataset = dataset.batch(BATCH_SIZE)
dataset = dataset.prefetch(tf.data.experimental.AUTOTUNE)

 

기존샘플을 뽑아보면

answers[0]

answers[:1][:, :-1]

answers[:1][:, 1:]

을 확인할 수 있습니다.

---

트렌스포머 만들기  & 훈련

이제 트랜스포머를 만들고 하이퍼파라미터를 조정하여서 실제 논문보다 작은 모델을 만들어보려고합니다.

여기서 선택한 주요 파라미터의 값은

d_model=256

num_layers=2

num_heads=8

dff=512

입니다.

tf.keras.backend.clear_session()

# 하이퍼파라미터
D_MODEL = 256
NUM_LAYERS = 2
NUM_HEADS = 8
DFF = 512
DROPOUT = 0.1

model = transformer(
    vocab_size=VOCAB_SIZE,
    num_layers=NUM_LAYERS,
    dff=DFF,
    d_model=D_MODEL,
    num_heads=NUM_HEADS,
    dropout=DROPOUT)
    
    
(1, 8180, 256)
(1, 8180, 256)

 

# 학습률과 옵티마이저 정의

learning_rate = CustomSchedule(D_MODEL)

optimizer = tf.keras.optimizers.Adam(
    learning_rate, beta_1=0.9, beta_2=0.98, epsilon=1e-9)

def accuracy(y_true, y_pred):
  # 레이블의 크기는 (batch_size, MAX_LENGTH - 1)
  y_true = tf.reshape(y_true, shape=(-1, MAX_LENGTH - 1))
  return tf.keras.metrics.sparse_categorical_accuracy(y_true, y_pred)

model.compile(optimizer=optimizer, loss=loss_function, metrics=[accuracy])

EPOCHS = 50
model.fit(dataset, epochs=EPOCHS)

에포크 50정도로 돌려보고 학습시킵니다.

---

 평가하기

def preprocess_sentence(sentence):
  # 단어와 구두점 사이에 공백 추가.
  # ex) 12시 땡! -> 12시 땡 !
  sentence = re.sub(r"([?.!,])", r" \1 ", sentence)
  sentence = sentence.strip()
  return sentence

def evaluate(sentence):
  # 입력 문장에 대한 전처리
  sentence = preprocess_sentence(sentence)

  # 입력 문장에 시작 토큰과 종료 토큰을 추가
  sentence = tf.expand_dims(
      START_TOKEN + tokenizer.encode(sentence) + END_TOKEN, axis=0)

  output = tf.expand_dims(START_TOKEN, 0)

  # 디코더의 예측 시작
  for i in range(MAX_LENGTH):
    predictions = model(inputs=[sentence, output], training=False)

    # 현재 시점의 예측 단어를 받아온다.
    predictions = predictions[:, -1:, :]
    predicted_id = tf.cast(tf.argmax(predictions, axis=-1), tf.int32)

    # 만약 현재 시점의 예측 단어가 종료 토큰이라면 예측을 중단
    if tf.equal(predicted_id, END_TOKEN[0]):
      break

    # 현재 시점의 예측 단어를 output(출력)에 연결한다.
    # output은 for문의 다음 루프에서 디코더의 입력이 된다.
    output = tf.concat([output, predicted_id], axis=-1)

  # 단어 예측이 모두 끝났다면 output을 리턴.
  return tf.squeeze(output, axis=0)

def predict(sentence):
  prediction = evaluate(sentence)

  # prediction == 디코더가 리턴한 챗봇의 대답에 해당하는 정수 시퀀스
  # tokenizer.decode()를 통해 정수 시퀀스를 문자열로 디코딩.
  predicted_sentence = tokenizer.decode(
      [i for i in prediction if i < tokenizer.vocab_size])

  print('Input: {}'.format(sentence))
  print('Output: {}'.format(predicted_sentence))

  return predicted_sentence

이제 예측함수를 만들었으니

실제로 예측을 해 본 결과를 첨부해놓겠습니다.

 

100% 자연스럽게 되지는 않지만 얼추? 잘 되는 챗봇이 생겼음을 확인할 수 있습니다.

---

마무리하면

RNN (LSTM, GRU, bi-LSTM)

↓

Seq2Seq (encoder - decoder)

↓

Attention

↓

Transformer (현재 가장 많이 사용하는 부분)

↓

BERT, GPT ...(사전 훈련된 모델)

↓

LLM (대형)

 

 

이 순서대로 정리를 해보았습니다.