[Imple] simple chatbot

이론

챗봇 : 사람과 대화하듯 질문에 알맞는 답이나 각종 연관정보를 제공하는 AI 기반 소프트웨어

명령방식에 따라 메시지 챗봇, 음성인식 봇, 개인 비서 봇 등으로 나눌 수 있다.

 

데이터의 예시들로는

어떤 카테고리인지 분류하고, 그 카테고리마다 input data와 target data 를 준비하는 Category 분류방식과

ex)

[{ 'tag' : 'greeting',

    'patterns' : ['안녕하세요', '안녕', '무슨일이야?', '오랜만이야']

    'responses' : ['안녕!', '어서오세요', '도와드릴까요?', '반가워요']

}]

 

심플하게 질문-대답 방식으로 된 Encoder-Decoder 방식의 데이터가 있다. (오픈 데이터셋 구글링...)

 

소스 코드

데이터는 https://github.com/songys/Chatbot_data 를 이용한다. (약 만개 이상)

GitHub - songys/Chatbot_data: Chatbot_data_for_Korean

 

실행환경은 구글 코랩에 GPU 스탠다드 환경을 이용하였다.

 

한글의 경우는 띄어쓰기로 나누는 케라스 토크나이저는 잘 맞지는 않음

워드피스가 더 좋은 성능을 나타내므로 이번엔  sentenceplece를 사용해보고자 한다.

 

필요한 sentencepiece 다운로드

!pip install -q sentencepiece

필요한 라이브러리 임포트

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import tensorflow as tf
import sentencepiece as spm
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Dense, Embedding, Input, LSTM
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.utils import to_categorical, plot_model

필요한 파라미터 지정

LATENT_DIM = 128 # encoding space 의 latent dimensionality
EMBEDDING_DIM = 100

데이터 불러오기

미리 데이터를 csv 로 변환해서 준비하고 read_csv로 불러오기

df = pd.read_csv('ChatbotData.csv')
df.head()

데이터 head 확인

데이터 결측치나 이상치 확인

df.info()

info 결과\

깔끔하게 정리된 데이터라는걸 볼 수 있음,, 제공해주신분께 감사,,,

 

 

학습용 질문-답변 데이터를 작성한다.

All_texts = [] # sentencepiece tokenizer 학습에 사용
Q_texts = [] # Encoder 질문 입력에 사용
A_texts = [] # Decoder 답변 입력에 사용

for Q, A in df.iloc[:, [0, 1]].values:
	Q_texts.append(Q)
	A_texts.append(A)
    
All_texts = Q_texts + A_texts

print(len(All_texts), len(Q_texts), len(A_texts))
print("Question :")
print(Q_texts[:5])
print("Answer :")
print(A_texts[:5])

질문-답변 데이터 확인해보기

 

 

다음으로 Sentencepiece 토큰화 학습을 위한 텍스트파일을 생성해준다

with open('chatbot_qna.txt', 'w', encoding='utf-8') as f:
	for line in All_texts:
		f.write(line + '\n')

만들어진 chatbot_qna.txt

 

파라미터를 지정하고 코맨드라인으로 만들기

input_file = 'chatbot_qna.txt'

pad_id = 0 #<pad> token을 0으로 설정
vocab_size = 5000 # vocab 사이즈
prefix = 'chatbot_qna' # 저장될 tokenizer 모델에 붙는 이름
bos_id=1 #<s> token을 1으로 설정
eos_id=2 #</s> token을 2으로 설정
unk_id=3 #<unknown> token을 3으로 설정

cmd = f'--input={input_file} \
--pad_id={pad_id} \
--bos_id={bos_id} \
--eos_id={eos_id} \
--unk_id={unk_id} \
--model_prefix={prefix} \
--vocab_size={vocab_size}'

spm.SentencePieceTrainer.Train(cmd)

 

학습완료된 sentencepiece 토큰화 로드 진행

sp = spm.SentencePieceProcessor()
sp.Load(f'{prefix}.model')

True로 반환되면 훈련이 잘 됐는지 확인해보고자 DecodeIds 를 확인해본다

sp.DecodeIds([170, 367, 10, 129, 16, 4])

실행 결과

 

토큰화 진행, 이 때 데이터 자체에 <s>,</s>를 붙여줘도 되지만 SetEncodeExtraOptions으로 설정해주어도 된다.

sp.SetEncodeExtraOptions('bos:eos')  이렇게 사용하면 문장 양 끝에 <s>, </s>를 추가해주는 것

sp.SetEncodeExtraOptions('bos:') # 문장 처음에 <s>추가 스타트토큰
pieces = sp.encode_as_pieces('아버지가 방에 들어가신다')
print(pieces)

ids = sp.encode_as_ids('아버지가 방에 들어가신다')

print(ids)
print(sp.DecodePieces(pieces))
print(sp.DecodeIds(ids))

 

bos 붙어서 토큰화 결과 확인

sp.SetEncodeExtraOptions(':eos') # 문장 끝에 </s>추가 --> Decoder targ
pieces = sp.encode_as_pieces('아버지가 방에 들어가신다')
print(pieces)

ids = sp.encode_as_ids('아버지가 방에 들어가신다')

print(ids)
print(sp.DecodePieces(pieces))
print(sp.DecodeIds(ids))

eos 붙어서 나오는 결과 확인

 

토큰화가 완료되면 리스트 컴프리핸션을 이용해서 질문 sequence를 작성한다

#encode_as_ids로 수열로 변환
Q_sequences = [sp.encode_as_ids(sent) for sent in Q_texts]

print(Q_sequences[:5])

답변(Answer) sequence input을 작성한다

sp.SetEncodeExtraOptions('bos:') # 1로 시작

#encode_as_ids로 수열로 변환
A_sequences_inputs = [sp.encode_as_ids(sent) for sent in A_texts]

print(A_sequences_inputs[:5])

답변(Answer) sequence target을 작성한다

sp.SetEncodeExtraOptions(':eos') # 2로 종료
A_sequences_targets = [sp.encode_as_ids(sent) for sent in A_texts]

print(A_sequences_targets[:5])

 

input text의 최대 길이를 지정하고자 시각화로 데이터를 확인해본다.

max_len_Q = max(len(s) for s in Q_texts)
print("Target Text 의 최대 길이 :", max_len_Q)

max_len_A = max(len(s) for s in A_texts)
print("Target Text 의 최대 길이 :", max_len_A)

plt.hist([len(s) for s in All_texts]);

 

데이터 결과확인

 

 

Input sequence 의 최대 길이는 30 으로 정한다.

MAX_LEN = 30

 

이후 sequence padding을 진행한다

encoder는 thought vector 생성을 목적으로 하므로 pre padding

decoder는 teacher forcing 해야하므로 post 로 padding

encoder_inputs = pad_sequences(Q_sequences, maxlen=MAX_LEN)
print("encoder input shape :", encoder_inputs.shape)
print("encoder_inputs[0] : ", encoder_inputs[1500])

decoder_inputs = pad_sequences(A_sequences_inputs,
				maxlen=MAX_LEN, padding="post")
print("\ndecoder input shape :", decoder_inputs.shape)
print("decoder_inputs[0] : ", decoder_inputs[1500])

decoder_targets = pad_sequences(A_sequences_targets,
				maxlen=MAX_LEN, padding="post")
print("\nencoder target shape :", decoder_targets.shape)
print("encoder_targets[0] : ", decoder_targets[1500])

 

 

 

그러고 이제 모델을 만든다

encoder 와 decoder의 embedding, lstm 및 dense layer를 학습할 목적의 모델을 작성한다

encoder는 decoder에 states인 [h, c] 만 전달한다

사용을 위한 모델은 훈련모델에서 만들어진 layer들의 weight을 이용해 별도로 작성한다.

 

훈련모델 : Encoder + Teacher Forcing 모델

#### Encoder
encoder_inputs_ = Input(shape=(MAX_LEN,), name='Encoder_Input')

# encoder 의 embedding layer
embedding_encoder = Embedding(vocab_size + 1, EMBEDDING_DIM)
x = embedding_encoder(encoder_inputs_)
encoder_outputs, h, c = LSTM(LATENT_DIM, return_state=True)(x)

# encoder 는 hidden state and cell state 만 decoder 로 전달
encoder_states = [h, c] #thought vector

encoder_model = Model(encoder_inputs_, encoder_states)
encoder_model.summary()

encoder summary 결과

 

 

decoder 모델 (teacher forcing)

# decoder 는 [h, c] 를 initial state 로 사용
decoder_inputs_ = Input(shape=(MAX_LEN,), name='Decoder_Input')

# decoder 의 embedding layer
embedding_decoder = Embedding(vocab_size + 1, EMBEDDING_DIM)
x = embedding_decoder(decoder_inputs_)

#many to many 모델이므로 return_sequences=True
decoder_lstm = LSTM(LATENT_DIM, return_sequences=True, return_state=True)
# initial state = encoder [h, c]
decoder_outputs, _, _ = decoder_lstm(x, initial_state=encoder_states)

# final dense layer
decoder_dense = Dense(vocab_size, activation='softmax', name='Decoder_Output')
decoder_outputs = decoder_dense(decoder_outputs)

# Teacher-forceing model 생성
model_teacher_forcing = Model([encoder_inputs_, decoder_inputs_], decoder_outputs)

#Compile the model and train it
model_teacher_forcing.compile(optimizer='rmsprop', loss='sparse_categorical_crossentropy',
			metrics=['accuracy'])
model_teacher_forcing.summary()

summary 결과 확인

 

모델 구조도 확인

#shape까지 확인하기 위해 show_shapes=True
plot_model(model_teacher_forcing, show_shapes=True)

teacher forcing 모델 구조도 확인

 

 

이제 모델학습을 시작한다

EPOCHS = 40
BATCH_SIZE = 64
history = model_teacher_forcing.fit([encoder_inputs, decoder_inputs],
		decoder_targets, batch_size=BATCH_SIZE, epochs=EPOCHS,
		validation_split=0.2)

 

모델 학습 완료후 성능을 위해 accuracy랑 loss를 시각화한다

# plot some data
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
plt.plot(history.history['loss'], label='loss')
plt.plot(history.history['val_loss'], label='val_loss')
plt.legend()

# accuracies
plt.subplot(1, 2, 2)
plt.plot(history.history['accuracy'], label='acc')
plt.plot(history.history['val_accuracy'], label='val_acc')
plt.legend()
plt.show()

결과 :&nbsp; 데이터가 적어서 오버피팅의 문제발생..ㅠㅠ

 

끝나면 모델을 파일로 저장해준다

model_teacher_forcing.save('s2s.h5')

 

---

 

 

사용 즉 prediction을 위해 별도의 encoder 모델을 만든다

encoder의 states를 initial state로 받는 decoder model 을 작성

encoder 는 training 단계와 동일하게 input_text 를 입력으로 받고 encoder_states 를 출력으로 하므로 이전에 define 한 encoder_input_ 과 encoder_states 변수를 재사용한다.

 

# Decoder for inference
decoder_state_input_h = Input(shape=(LATENT_DIM,), name='Decoder_hidden_h')
decoder_state_input_c = Input(shape=(LATENT_DIM,), name='Decoder_hidden_c')
decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c]

decoder_inputs_single = Input(shape=(1,), name='Decoder_input')
x = embedding_decoder(decoder_inputs_single)

# output, hidden states 를 저장
decoder_outputs, h, c = decoder_lstm(x, initial_state=decoder_states_inputs)
decoder_states = [h, c]

decoder_outputs = decoder_dense(decoder_outputs)

decoder_model = Model(
	[decoder_inputs_single] + decoder_states_inputs, #decoder_model.predict([target_seq]+state_value)
	[decoder_outputs] + decoder_states
)

decoder_model.summary()

사용 모델의 decoder summary 결과

 

편의를 위해 구조도로 확인

plot_model(decoder_model, show_shapes=True)

사용모델의 디코더 구조도 확인

 

다음으로 실제로 text generation 하는 함수를 만든다

실행되는 스텝 순서를 살펴보면

1. question 을 encoder 에 입력하여 thought vector 반환한다.

2. start token 설정한다.

3. 가장 확률 높은 next word 선택한다

4. end token 생성 될 때까지 반복한다

5. predict 결과가 sequence 로 반환되므로 index 를 word 로 변환한다

def decode_sequence(input_seq):
	# 입력된 질문을 encoder 에 입력하여 state vector 생성
	states_value = encoder_model.predict(input_seq)
	# size 1 짜리 빈 target sequence 생성
	target_seq = np.zeros((1, 1))
	# target sequence 의 첫번째 character 를 <s> 로 assign -> 1
	target_seq[0, 0] = 1
	eos = 2 #</s>
    
	# 답변 생성 시작
	output_ids = []
	for _ in range(max_len_A):
		output_tokens, h, c = decoder_model.predict([target_seq] + states_value)

		# argmax 로 가장 확률 높은 단어 선택(greedy selection)
		idx = np.argmax(output_tokens[0, 0, :])

		if eos == idx: # End sentence of EOS
			break
            
		if idx > 0: # idx 0 은 zero padding 이므로 skip
			output_ids.append(int(idx))

		# 생성된 word 를 decoder 의 다음 input 으로 사용
		target_seq[0, 0] = idx
		# Update states
		states_value = [h, c]
	
	return output_ids

 

이후에 이제 완성된 모델을 가지고 실제 샘플 Q&A를 테스트 (이미 훈련했던 데이터 가지고) 해보기

sp.Load(f'{prefix}.model')

for _ in range(5):
	i = np.random.choice(len(Q_texts))
	input_seq = encoder_inputs[i:i+1]
	response = decode_sequence(input_seq)
	print('-')
	print('질문 :' Q texts[i])
        #DecodeIds : id를 단어로 바꿔주는 메소드
	print('답변 : ', sp.DecodeIds(response))

결과 출력

 

 

없던 샘플 데이터 만들어서 테스트 해보기

txt = "정말 미워"
input_sequence = sp.encode_as_ids(txt)
encoder_input = pad_sequences([input_sequence], maxien=MAX_LEN)

response = decode_sequence(encoder_input)

print('-')
print('질문 : ',txt)
print('답변 : ',sp.DecodeIds(response))

결과 출력 1

결과 출력 2

 

 

 

챗봇으로 만들어 보기

print("챗봇 대화 시작...")

while True:
	txt = input("질문 : ")
    #quit 입력 전까지는 무한루프
    if txt.lower() == "quit":
    	break
        
    input_sequence = sp.encode_as_ids(txt)
    encoder_input = pad_sequences([input_sequence], maxien=MAX_LEN)
    
    answer = decode_sequence(encoder_input)
    
    print("답변 : ",sp.DecodeIds(answer))

 

챗봇 실행 결과

 

 

 

 

데이터가 적어서 성능적인 면에는 문제가 있고,, 대답도 좀 이상하게 하는데 

어쨌든 풍부한 데이터로 학습한다면 더 좋은 결과는 나올 듯 하다

한글로 자연어 처리 하는데 데이터 관련 장벽이 높다..ㅠㅠ