Seq2Seq로 Chatbot 구현

 

https://www.youtube.com/watch?v=C9XLed6n6T4&t=1s

논문 리뷰 후 따로 실습을 하지 않았기에 구현 실습을 해보았다.

 

---

사용 라이브러리, 하이퍼 파라미터

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import random
import re
import pickle
import pandas as pd

딥러닝 모델 구축과 학습을 위해 pytorch를 사용하였고, 데이터 샘플링 및 텍스트 전처리를 위한 random과 re를 사용하였다. 또한, 모델과 데이터셋 저장을 위한 pickle을 사용하였다.

#하이퍼 파라미터
hidden_size = 256
PAD_token = 0
SOS_token = 1
EOS_token = 2
UNK_token = 3
MAX_LENGTH = 10
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

하이퍼파라미터는 다음과 같다.

---

데이터 전처리와 라벨링

def clean_text(text):
    if pd.isna(text):  # NaN값을 처리
        return ''
    text = text.lower()
    text = re.sub(r'\d+', ' ', text)   #숫자를 공백으로
    text = re.sub(r'([^\w\s])', r' \1 ', text)   # 마침표 앞 뒤로 공백 추가
    text = re.sub(r'\s+', ' ', text)  # 두개 이상의 공백은 하나의 공백으로..
    text = text.strip()  # 텍스트 앞 뒤의 공백 제거
    return text

clean_text 함수는 텍스트 데이터 전처리를 위한 함수로, 모델 학습에 적합한 형태로 정리를 해준다. 해당 실습에서 사용하는 데이터셋이 영어로 구성되어 있기에 영어 전처리에 알맞게 구성하였다.

def indexesFromSentence(vocab, sentence):
    return [vocab.get(word, vocab['<UNK>']) for word in sentence.split(' ')]

indexesFromSentence는 문장에서 각 단어를 사전에 정의된 vocab(어휘 사전)을 기반으로 인덱스 값으로 변환하는 함수이다. 쉽게 말해 단어 -> 숫자 인덱스로 매핑하는 과정이다.

def tensorFromSentence(vocab, sentence):
    indexes = indexesFromSentence(vocab, sentence)
    indexes.append(EOS_token)
    return torch.tensor(indexes, dtype=torch.long, device=device).view(-1, 1)

tensorFromSentence는 문장을 숫자 인덱스 시퀀스 텐서로 변환한 뒤, 마지막에 종료 토큰을 추가하여 문장의 종료를 명시하는 역할을 한다. 변환된 결과는 Pytorch 텐서 형태로 반환되며, 모델 학습에 사용된다.

---

 

Eocoder와 Decoder

class EncoderLSTM(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(EncoderLSTM, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(input_size, hidden_size)
        self.lstm = nn.LSTM(hidden_size, hidden_size, num_layers=2)

    def forward(self, input, hidden):
        embedded = self.embedding(input).view(1, 1, -1)
        output, hidden = self.lstm(embedded, hidden)
        return output, hidden

    def initHidden(self):
        return (torch.zeros(2, 1, self.hidden_size, device=device),
                torch.zeros(2, 1, self.hidden_size, device=device))

EncoderLSTM은 시퀀스 데이터를 고정된 크기의 컨텍스트 벡터로 인코딩하는 역할을 한다.

__init__ 메서드

input_size : 인코더에 입력되는 단어(혹은 토큰)의 개수

hidden_size : LSTM의 은닉 상태의 크기

self.embedding : 입력 데이터를 고정된 크기의 밀집 벡터로 변환하는 임베딩 레이어

self.lstm : 2개 레이어로 구성된 LSTM

forward 메서드

input, hidden : 현재 시점의 입력 단어 인덱스, 이전 시점의 은닉 상태 및 셀 상태

self.embedding으로 입력 단어를 임베딩 벡터로 변환 후 임베딩 벡터를 (1,1,-1) 크기로 변환 후 self.lstm에 입력하여 출력과 새로운 은닉 상태를 계산한다.

출력은 output, hidden으로 현재 시점의 출력 벡터와 새로운 은닉 상태 및 셀 상태이다.

initHidden 메서드

초기값을 0으로 설정하면 모델이 모든 입력에 대해 동일한 초기 상태에서 시작하므로 LSTM의 초기 은닉 상태와 셀 상태를 0으로 초기화한다.

class DecoderLSTM(nn.Module):
    def __init__(self, hidden_size, output_size):
        super(DecoderLSTM, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(output_size, hidden_size)
        self.lstm = nn.LSTM(hidden_size, hidden_size, num_layers=2)
        self.out = nn.Linear(hidden_size, output_size)

    def forward(self, input, hidden):
        output = self.embedding(input).view(1, 1, -1)
        output = F.relu(output)
        output, hidden = self.lstm(output, hidden)
        output = self.out(output[0])
        return output, hidden

    def initHidden(self):
        return (torch.zeros(2, 1, self.hidden_size, device=device),
                torch.zeros(2, 1, self.hidden_size, device=device))

DecoderLSTM은 고정된 크기의 컨텍스트 벡터를 입력으로 받아 시퀀스를 생성하는 역할을 한다.

__init__ 메서드

hidden_size : 디코더 LSTM의 은닉 상태 크기

output_size : 출력 단어의 개수

self.embedding : 출력 단어를 고정된 크기의 밀집 벡터로 변환

self.lstm : 2개 레이어로 구성된 LSTM. 임베딩 벡터를 처리하여 새로운 은닉 상태와 출력 계산

self.out : LSTM의 출력을 어휘 크기로 변환하는 선형 레이어

forward 메서드

input, hidden : 현재 시점의 디코더 입력 단어 인덱스, 이전 시점의 은닉 상태 및 셀 상태

입력 단어를 self.embedding으로 임베딩 벡터로 변환 후 ReLU 활성화 함수를 적용하여 비선형성을 추가한다. 이후 self.lstm에 입력하여 새로운 출력과 은닉 상태를 계산하고 LSTM의 출력을 self.out을 통해 어휘 크기의 벡터로 변환한다.

initHidden 메서드

Encoder와 마찬가지로 초기 은닉 상태와 셀 상태를 0으로 초기화한다.

 

---

 

train, trainIters, evaluate

def train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion):
    encoder_hidden = encoder.initHidden()
    
    encoder_optimizer.zero_grad()
    decoder_optimizer.zero_grad()
    
    input_length = input_tensor.size(0)
    target_length = target_tensor.size(0)
    
    loss = 0
    
    for ei in range(input_length):
        encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
    
    decoder_input = torch.tensor([[SOS_token]], device=device)
    decoder_hidden = encoder_hidden
    
    for di in range(target_length):
        decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
        topv, topi = decoder_output.topk(1)
        decoder_input = topi.squeeze().detach()
        loss += criterion(decoder_output, target_tensor[di])
        if decoder_input.item() == EOS_token:
            break
    
    loss.backward()    # backpropagation only 1 line!
    
    encoder_optimizer.step()
    decoder_optimizer.step()
    
    return loss.item() / target_length

1. 초기화

encoder_hidden = encoder.initHidden()

인코더의 초기 은닉 상태와 셀 상태를 0으로 설정한다. 첫 입력 데이터를 처리하기 위한 준비 단계이다.

encoder_optimizer.zero_grad()
decoder_optimizer.zero_grad()

이전 배치에서 계산된 그래디언트를 초기화한다. 이는 그래디언트가 누적되지 않도록 학습 전에 매번 초기화가 필요하기 때문이다.

2. Encoder 실행

for ei in range(input_length):
    encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)

입력 시퀀스의 각 단어를 인코더에 순차적으로 전달한다. 이 과정에서 Encoder는 출력 벡터(encoder_output)와 마지막 은닉 상태(encoder_hidden)를 생성한다. 마지막 은닉 상태는 이후 디코더로 전달된다.

3.Decoder 준비

decoder_input = torch.tensor([[SOS_token]], device=device)

시작 토큰을 첫 입력으로 받기에 SOS_token이 먼저 오게 된다.

decoder_hidden = encoder_hidden

인코더의 마지막 은닉 상태를 디코더의 초기 은닉 상태로 사용한다.

4.Decoder 실행

for di in range(target_length):
    decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)

Decoder는 한 번에 하나씩 목표 시퀀스의 단어를 생성한다.

topv, topi = decoder_output.topk(1)
decoder_input = topi.squeeze().detach()

topk(1)은 가장 높은 확률을 가진 단어를 선택한다. 선택된 단어는 다음 시점의 입력(decoder_input)으로 사용된다.

topi는 인덱스를 나타내고, topv는 확률을 나타내게된다. 이후 squeeze를 통해 불필요한 차원을 제거한다. detach의 경우 그래디언트 계산 그래프를 끊어 불필요한 연산을 방지한다고 한다.

loss += criterion(decoder_output, target_tensor[di])

디코더의 출력과 목표 단어(target_tensor[di]) 간의 손실을 계산하고 누적한다.

if decoder_input.item() == EOS_token:
    break

만약 디코더가 종료 토큰(EOS_token)을 생성하면 디코딩 과정을 종료한다.

loss.backward()

역전파를 통해 그래디언트를 계산한다. 손실로부터 시작해 모델의 모든 매개변수에 대해 미분값을 계산한다.

encoder_optimizer.step()
decoder_optimizer.step()

계산된 그래디언트를 사용해 인코더와 디코더의 가중치를 업데이트한다.

return loss.item() / target_length

총 손실을 목표 시퀀스 길이로 나누어 반환한다. 이는 학습의 안정성을 확인하는 데 사용한다.

# 학습을 반복해주는 코드
def trainIters(encoder, decoder, n_iters, print_every=1000, learning_rate=0.01):
    print_loss_total = 0
    
    for iter in range(1, n_iters+1):
        training_pair = random.choice(pairs)
        input_tensor = tensorFromSentence(word_to_ix, training_pair[0]).to(device)
        target_tensor = tensorFromSentence(word_to_ix, training_pair[1]).to(device)
        
        loss = train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion)
        print_loss_total += loss
        
        if iter % print_every == 0:
            print_loss_avg = print_loss_total / print_every
            print(f'Iteration: {iter}, Loss: {print_loss_avg: .4f}')
            print_loss_total = 0

trainIters 함수는 학습 루프를 구현하여 인코더와 디코더를 반복적으로 학습시키는 역할을 한다.

함수 정의 초기화

encoder,decoder : 학습시킬 인코더와 디코더 모델

n_iters : 총 반복 횟수

print_every : print_every번 반복마다 학습 손실을 출력 (초기값 1000으로 설정)

learning_rate : 학습률

print_loss_total : print_every 동안 누적된 손실을 저장 -> 평균 손실을 출력하기 위해 사용

for iter in range(1, n_iters+1):
	training_pair = random.choice(pairs)
	input_tensor = tensorFromSentence(word_to_ix, training_pair[0]).to(device)
	target_tensor = tensorFromSentence(word_to_ix, training_pair[1]).to(device)

iter : 1부터 시작하여 n_iters까지 반복, 반복마다 하나의 데이터 쌍(training_pair)을 랜덤으로 선택하여 학습 진행.

pairs : 입력 문장과 목표 문장의 쌍으로 이루어진 데이터셋.

random.choice(pairs) : 학습 데이터에서 무작위로 한 쌍을 선택

tensorFromSentence : 문장을 숫자 인덱스 시퀀스 텐서로 변환.

loss = train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion)
print_loss_total += loss

train 함수 : input_tensor와 target_tensor를 사용해 인코더와 디코더를 학습

누적 손실은 print_loss_total에 저장

if iter % print_every == 0:
    print_loss_avg = print_loss_total / print_every
    print(f'Iteration: {iter}, Loss: {print_loss_avg: .4f}')
    print_loss_total = 0

print_every 반복마다 평균 손실을 계산하고 출력한다.

def evaluate(encoder, decoder, sentence, max_length=MAX_LENGTH):
    with torch.no_grad():
        input_tensor = tensorFromSentence(word_to_ix, sentence).to(device)
        input_length = input_tensor.size(0)
        encoder_hidden = encoder.initHidden()
        encoder_hidden = tuple([e.to(device) for e in encoder_hidden])
        
        for ei in range(input_length):
            encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)

encoder, decoder : 학습된 인코더와 디코더 모델

sentence : 평가할 입력 문장

max_length : 디코더가 생성할 최대 출력 길이

torch.no_grad() : 평가 단계에서는 역전파가 필요 없으므로, 계산 그래프를 비활성화하여 메모리 사용량과 속도 최적화

tensorFromSentence : 입력 문장을 단어 인덱스 시퀀스 텐서로 변환.

encoder.initHidden : 인코더의 초기 은닉 상태와 셀 상태를 0으로 초기화 이후 입력 문장의 각 단어를 순차적으로 인코더에 전달해, 최종 은닉 상태 encoder_hidden 생성

decoder_input = torch.tensor([[SOS_token]], device=device)
decoder_hidden = encoder_hidden
decoded_words = []  # output sentence

디코더의 첫 입력 시작 토큰(SOS_token)으로 설정, 인코더의 마지막 은닉 상태를 디코더의 초기 은닉 상태로 사용

decoded_words : 생성된 출력 단어를 저장할 리스트.

for di in range(max_length):
    decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
    topv, topi = decoder_output.data.topk(1)
    if topi.item() == EOS_token:
        decoded_words.append('<EOS>')
        break
    else:
        decoded_words.append(ix_to_word[topi.item()])
    decoder_input = topi.squeeze().detach()

디코더가 최대 max_length까지 단어를 생성한다. decoder_output은 디코더 출력으로 각 단어의 확률 분포를 나타낸다. topv와 topi는 위에서 언급한대로 가장 높은 확률을 가진 단어의 값과 인덱스이다.

EOS_token이 생성되면 <EOS>를 추가하고 반복 종료한다. 종료하지 않을 경우 ix_to_word가 인덱스를 단어로 변환하는 사전으로 변환된 단어를 decoded_words에 추가한다. 이후 topi에서 인덱스를 추출 후 디코더의 다음 입력으로 설정한다.

return ' '.join(decoded_words)

생성된 단어 리스트 decoded_words를 공백으로 연결해 하나의 문장으로 반환한다.

# 채팅함수
def chat(encoder, decoder, max_length=MAX_LENGTH):
    print("Let's chat! (type 'bye' to exit)")
    while True:
        input_sentence = input("> ")
        if input_sentence == 'bye':
            break
        output_sentence = evaluate(encoder, decoder, input_sentence)
        print('<', output_sentence)

지금까지 학습한 인코더와 디코더 모델을 활용하여 채팅을 할 수 있게 해준다. bye를 입력하면 break를 사용해 반복문을 종료한다.

evaluate : 입력 문장(input_sentnece)에 대한 모델의 응답 생성 -> 입력 문장을 인코더로 처리 후 디코더로 응답 생성

---

 

실습 후기

 

준비된 데이터셋을 읽어와 출력했을 때 다음과 같이 나온다.

input_sentence = [sentence for sentence in df['Encoder Inputs']]
output_sentence = [sentence + "<EOS>" for sentence in df['Decoder Inputs']]

input_sentence와 output_sentence를 준비해준다.

# 단어 사전 생성
all_words = set(' '.join(df['Encoder Inputs'].tolist()+df['Decoder Inputs'].tolist()).split())
vocab = {'<PAD>': PAD_token, '<SOS>': SOS_token, '<EOS>': EOS_token, '<UNK>': UNK_token}
vocab.update({word: i+4 for i, word in enumerate(all_words)})
vocab_size = len(vocab)
# vocab 변수 저장
with open('vocab.pkl', 'wb') as f:
    pickle.dump(vocab, f)

 

DataFrame에서 encoder와 decoder의 입력 텍스트가 포함된 열을 리스트로 변환하여 두 열의 텍스트를 합치고 공백을 기준으로 분리하여 단어 리스트를 생성한다. 이후 set 함수를 통해 중복을 제거해 고유한 단어들만 모은 집합 형태로 변환한다.

vocab에는 사전에 특수 토큰들을 미리 할당한다.

<PAD> : 패딩 토큰 ( 문장의 길이를 맞추기 위해 사용)

<SOS> : 시작 토큰

<EOS> : 종료 토큰

<UNK> : 알 수 없는 단어를 나타내는 토큰 ( 모델이 학습하지 못한 단어에 대해 사용 )

이후 enumerate(all_words)를 통해 각 단어와 그 인덱스를 반환하여 새로운 단어들을 vacab에 추가한다. 이는 pickle을 통해 파일로 저장하여 추후에 다시 불러와서 사용할 수 있게 한다. 아래와 같은 형태의 사전으로 저장된다.

{
    '<PAD>': 0,
    '<SOS>': 1,
    '<EOS>': 2,
    '<UNK>': 3,
    'hello': 4,
    'world': 5,
    'how': 6,
    'are': 7,
    'you': 8,
    ...
}

word_to_ix = vocab
ix_to_word = {i: word for word, i in word_to_ix.items()}

word_to_ix가 '<PAD>' : 0 이라면 ix_to_word는 0: '<PAD>'이 된다.

이 두 딕셔너리는 모델 학습 시 단어를 인덱스로 변환하거나, 인덱스를 단어로 변환할 때 사용

encoder = EncoderLSTM(vocab_size, hidden_size).to(device)
decoder = DecoderLSTM(hidden_size, vocab_size).to(device)

encoder_optimizer = optim.Adam(encoder.parameters(), lr=0.005)
decoder_optimizer = optim.Adam(decoder.parameters(), lr=0.005)
criterion = nn.CrossEntropyLoss()

# pairs 리스트를 만들어서 학습 데이터를 준비
pairs = [list(x) for x in zip(df['Encoder Inputs'], df['Decoder Inputs'])]

#학습실행 def trainIters(encoder, decoder, n_iters, print_every=1000, learning_rate=0.01):
trainIters(encoder, decoder, 30000, print_every=1000)

encoder와 decoder 정의 후, Adam 옵티마이저를 사용하였고 학습률을 0.005로 설정하였다.

CrossEntropyLoss는 분류 문제에서 사용되는 손실 함수이다. 다중 클래스 분류에 적합하여, 모델이 예측한 확률 분포와 실제 레이블 간의 차이를 계산해준다.

pairs는 df에서 인코더 입력과 디코더 입력 열을 사용하여 (입력,출력) 쌍으로 생성하는 코드이다.

이후, 모델 학습을 30000번 반복하도록 설정하였고, print_every =1000을 통해 1000번마다 학습 손실을 출력하였다.

encoder.eval()
decoder.eval()

평가모드로 변경한 후,

chat(encoder, decoder)

위와 같은 코드를 입력하게 되면 대화가 가능하게 된다.

채팅 결과인데 좋지 않은 결과를 보이고 있다. 아무래도 Seq2Seq가 Transformer 이전 모델이다보니 성능적으로 좋지 않은 결과가 나온 것 같다.