[작성자:] dlgndud

이번 예제에서는 간단한 시나리오 기반 챗봇을 구현해보겠습니다.

해당 예제를 실행한 결과는 아래의 영상과 같습니다.

이 대화는 아래와 같이 4개의 턴(Turn)으로 이루어져있습니다. 대화의 흐름은 “인사-간단한 일상 대화-주문-끝인사”로 이뤄져있습니다. 각 턴을 수행하면 자연스럽게 다음 턴으로 연결됩니다. 대화가 예상된 흐름으로 넘어가지 않을 때는 사전에 정의된 간단한 대화를 출력하고 다시 이전 질문을 다시 수행합니다.

Dialog Flow : Greeting – Where – Order – Bye

테스트에 사용할 간단한 시나리오는 아래와 같습니다. 아래에 order – bye가 화면상에는 표시되어 있지 않지만 내용은 위와 다르지 않습니다.

category에 NaN으로 되어 있는 부분은 시스템의 발화 부분입니다. 그 외의 부분은 시스템에 입력되는 기대값들입니다.

예를 들어 greeting 카테고리를 살펴보면 시스템이 “안녕하세요”라고 발화 했을 때에 해당 발화에 답변으로는 기대되는 값들을 greeting 카테고리에 등록합니다. 현재 시스템 발화에는 하나만 등록했지만 만약 시스템 발화 부분을 다양하게 하고자 한다면 여러개의 답변을 넣고 그중에서 하나의 답을 랜덤하게 표시해주는 방법으로 해도 됩니다. 실제로 많은 채팅 시나리오가 같은 방법으로 제작되고 있습니다. 여기서는 간단하게 시스템에서는 하나의 답변만 낼 수 있도록 합니다.

시스템 메세지에 대한 사용자의 기대되는 “안녕하세요”, “안녕”, “헬로”, “네 반갑습니다”, “hi hello” 5가지 중에 하나로 입력된다고 가정합니다.

이와 같은 방법으로 “어디서 오셨나요?”라는 시스템의 질문에도 사용자는 몇가지 대답을 할 수 있습니다. 그에 대한 답변을 미리 등록해봅니다.

동일한 방법으로 나머지 시나리오도 입력해봅니다.

그렇지만 안타깝게도 위와 같이 정의된 답변만 사용자가 입력하지는 않습니다. 사용자는 여러가지 답변을 입력할 수 있습니다. 기본적으로는 챗봇에게 많은 내용을 학습시킬 수 있다면 좋겠지만 실제로 그렇게 하기는 쉽지 않습니다. 또 하나의 문제는 시스템은 사용자가 어떤 순서로 답변을 낼지 알지 못한다는 것입니다.

그렇기 때문에 챗봇 시스템은 사용자의 입력한 답변이 입력한 시나리오에 있는지 그렇다면 어떤 질문인지 만약에 아니라면 어떻게 예외적인 사항을 처리해야 하는지 알아야합니다. 즉, NLU(자연어이해, Natural Language Understanding)가 필요합니다.

이부분에 형태소 분석과 구문분석 등의 과정이 필요하고 더 높은 이해를 위해서 사전 구축등의 작업이 필요합니다. 하지만 비슷한 예제를 이미 구현했기 때문에 여기서는 간단히 해당 문장이 어떤 카테고리에 속하는지 분류하는 분류기 정도로만 구현해 보겠습니다.

딥러닝을 활용하여 텍스트 분류를 수행할 수 있습니다.

위와 같은 텍스트 분류 예제를 참고하시기 바랍니다.

간단히 작성한 시나리오를 통해서 학습을 위해 아래와 같이 각 카테고리(Category 혹은 Intent)에 코드값을 부여해줍니다. 이때 시스템의 발화는 제외하고 사용자의 발화만 코드값을 부여합니다.

index2category = {0:'greeting',1:'where',2:'ask',3:'bye'}
def category_define(x):
    code = ''
    if x=='greeting': code=0
    elif x=='where': code=1
    elif x=='ask': code=2
    elif x=='bye': code=3
    else: code='NaN'
    return code

# category code!
df['code'] = df['category'].apply(category_define)

# only answer
df=df[df['category'].notnull()]

사용자가 입력할 예상 문장들을 아래와 같이 추출할 수 있습니다.

sentence = df['text'].values
print(sentence)
array(['안녕하세요', '안녕', '헬로', '네 반갑습니다', 'hi hello', '세종에서 왔습니다',
       '세종에서 살아요', '대전 살아요', '세종이요', '세종요', '세종시에서 왔지', '서울요', '부산요',
       '빵을 사고 싶어요', '음료수 사고 싶어요', '커피 주세요', '빵 주세요', '케이크 주세요',
       '아이스 아메리카노 주세요', '베이글 주세요', '감사합니다', '고맙습니다', '잘먹을께요', 'Thank you'],
      dtype=object)

각 문장을 형태소분석이나 구분분석 등의 과정을 생략하고 단순히 문장을 공백으로 분리하여 각 단어의 집합을 생성합니다. 집합 생성시에 입력되는 문장에 단어가 없는 경우를 위해서 unk 코드와 자리수를 맞추기 위한 padding 값을 부여합니다.

sentence = df['text'].values

words = list(set([w for word in sentence for w in word.split(' ')]))
words = np.insert(words,0,'!') # padding 1
words = np.insert(words,0,'#') # unk 0

이제 생성한 문장을 단어 단위로 분리하고 각각 Index 값을 부여했기 때문에 각 문장을 숫자로 표현할 수 있습니다. word2index의 경우는 입력되는 단어들을 Index 값으로 바꿔주는 python dictionary이고 index2word는 그 반대의 경우입니다.

해당 과정을 거치면 문장은 숫자의 형태로 변경됩니다. 이렇게 하는 이유는 컴퓨터가 인간의 문장을 이해하지 못하기 때문입니다. 이제 학습을 위해 각 단어의 입력 Sequence Length를 맞춰주는 작업이 필요합니다. 이때 지나치게 패딩을 많이 입력하면 훈련 데이터에 노이즈가 많이 들어가기 때문에 예측 결과가 좋지 않을 수 있습니다. 그리고 마지막에 각 문장이 어떤 카테고리 혹은 의도(Intent)에 속하는지 Label 데이터를 입력해줍니다.

word2index = {w:i for i,w in enumerate(words)}
index2word = {i:w for i,w in enumerate(words)}

def xgenerator(x):
    return [ word2index['#'] if x_ not in word2index else word2index[x_] for x_ in x]

x_data = [xgenerator(words.split(' ')) for words in sentence]

for ndx,d in enumerate(x_data):
    x_data[ndx] = np.pad(d, (0, config.max_length), 'constant', constant_values=0)[:config.max_length]

y_data = df['code'].values

아래와 같이 학습에 필요한 파이토치 라이브러리를 import하고 DataLoader를 통해서 학습용 데이터셋을 만들어줍니다. 배치 사이즈는 데이터 셋이 크지 않기 때문에 한번에 학습을 하는 것으로 설정하시면 됩니다.

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F

config.vocab_size = len(word2index)
config.input_size = 30
config.hidden_size = len(df['code'].unique())

from torch.utils.data import Dataset, DataLoader

class TxtDataSet(Dataset):
    def __init__(self, data, labels):
        super().__init__()
        self.data = data
        self.labels = labels
        
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        return self.data[idx], self.labels[idx]

train_loader = DataLoader(dataset=TxtDataSet(x_data, y_data), batch_size=config.batch_size, shuffle=True)

학습용 모델을 생성합니다. 학습은 아래의 3개의 레이어를 통과하고 나온 결과 값을 사용합니다. 제가 작성한 여러 예제에 해당 모델에 대한 설명이 있기 때문에 자세한 설명은 하지 않고 넘어가겠습니다.

Embedding Layer – LSTM Layer – Linear Layer

class RNN(nn.Module):
    def __init__(self, vocab_size, input_size, hidden_size):
        super().__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.vocab_size = vocab_size
        
        self.embedding = nn.Embedding(self.vocab_size, self.input_size)
        self.rnn = nn.LSTM(
            input_size = self.input_size, 
            hidden_size = self.hidden_size, 
            num_layers=4, 
            batch_first=True, 
            bidirectional=True
        )
        
        self.layers = nn.Sequential(
            nn.ReLU(), 
            nn.Linear(hidden_size*2, hidden_size),
            
        )
        
    def forward(self,x):
        x = self.embedding(x)
        y, _ = self.rnn(x)
        y = self.layers(y[:,-1]) # last output dim...

        return F.softmax(y, dim=-1)
    
model = RNN(config.vocab_size, config.input_size, config.hidden_size)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters())
print(model)

생성한 모델을 출력하면 아래와 같이 표시됩니다.

RNN(
  (embedding): Embedding(35, 30)
  (rnn): LSTM(30, 4, num_layers=4, batch_first=True, bidirectional=True)
  (layers): Sequential(
    (0): ReLU()
    (1): Linear(in_features=8, out_features=4, bias=True)
  )
)

model.train()

hist_loss = []
hist_accr = []

for epoch in range(config.number_of_epochs):
    epoch_loss = 0
    for x_i, y_i in train_loader:
        y_hat = model(x_i) 
        loss = criterion(y_hat, y_i)
        
        accr = torch.argmax(y_hat, axis=1)== y_i
        accr = accr.data.numpy()
        accr = accr.sum()/len(y_i)
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        epoch_loss += float(loss)
    
    if epoch % 100 == 0:
        print('epoch:{}, loss:{:.5f}, accr:{:.5f}'.format(epoch, epoch_loss/config.number_of_epochs, accr))

epoch:0, loss:0.00070, accr:0.20833
epoch:100, loss:0.00068, accr:0.29167
epoch:200, loss:0.00065, accr:0.50000
epoch:300, loss:0.00062, accr:0.62500
epoch:400, loss:0.00056, accr:0.62500
epoch:500, loss:0.00054, accr:0.62500
epoch:600, loss:0.00052, accr:0.62500
epoch:700, loss:0.00051, accr:0.83333
epoch:800, loss:0.00050, accr:0.83333
epoch:900, loss:0.00049, accr:0.83333
epoch:1000, loss:0.00048, accr:0.83333
epoch:1100, loss:0.00048, accr:0.83333
epoch:1200, loss:0.00047, accr:1.00000
epoch:1300, loss:0.00046, accr:1.00000
epoch:1400, loss:0.00046, accr:1.00000
epoch:1500, loss:0.00045, accr:1.00000
epoch:1600, loss:0.00044, accr:1.00000
epoch:1700, loss:0.00044, accr:1.00000
epoch:1800, loss:0.00043, accr:1.00000
epoch:1900, loss:0.00043, accr:1.00000

학습한 모델을 통해서 입력한 어떤 내용으로 발화한 것인지를 예측해봅니다.

test_sentence = ['잘먹을께요']
x_test = [xgenerator(words.split(' ')) for words in test_sentence]
for ndx,d in enumerate(x_test):
    x_test[ndx] = np.pad(d, (0, config.max_length), 'constant', constant_values=0)[:config.max_length]
    
with torch.no_grad():
    x_test = torch.tensor(x_test, dtype=torch.long)
    predict = model(x_test)
    print(predict)
    result = torch.argmax(predict,dim=-1).data.numpy()
    print([index2category[p] for p in result])

학습을 완료한 후 모델을 아래와 같이 저장합니다.

torch.save({
  'model': model.state_dict(), 'config':config
}, './model/model.scenario')

import pickle
def save_obj(obj, name):
    with open('./pkl/'+ name + '.pkl', 'wb') as f:
        pickle.dump(obj, f, pickle.HIGHEST_PROTOCOL)
    
save_obj(index2category,'index2category')
save_obj(word2index,'word2index')
save_obj({'vocab_size':config.vocab_size,'input_size':config.input_size,'hidden_size':config.hidden_size,'max_length':config.max_length},'config')

django를 통해서 간단한 웹서버를 만들고 해당 모델을 활용해서 간단한 챗봇을 만들어봅니다.

웹서버의 사용자의 입력을 받아서 시나리오에서 어떤 흐름에 속하는 대화인지를 찾아내고 그 흐름에 맞으면 다음 대화를 진행하고 맞지 않을 경우 다시 발화를 할 수 있도록 유도합니다.