[부스트코스] RNN seq2seq

파이토치로 시작하는 딥러닝기초

출처: https://www.boostcourse.org/ai214

• 순환신경망(Recurrent Neural Network)
• Sequence-To-Sequence

---

• 시퀀스를 입력받아서 시퀀스를 출력
• 대표적인 사용처: 번역이나 챗봇

• Encoder - Decoder 구조
• 인코더로 압축된 벡터를 디코더에 전달
• 스타트 플래그와 함께 모델 시작
• 아웃풋을 reply에 첫번째에 두고, 이 아웃풋이 다음으로 또 들어간다. -> 완전한 문장 생성
• 모든 문장을 들은 후에 답변을 생성한다
• RNN 2개를 생성해서 중간을 연결한 형태
• pytorch로 구현 가능

마지막 10줄의 코드가 전체 200줄 정도의 내용을 압축하고 있는 코드

이 예시는 번역 task를 수행하는 모델

source text (영문)-> target text(한국어)

train-test로 나눠서 검증

pre-process 보조함수

raw 원문

s, t 문장의 최대길이 제한 

encoder, decoder의 Hidden State도 같은 크기로 정의를 해준다.

encoder, decoder는 각각 클래스로, RNN layer이다. 각각선언하여 학습시킨다.

train 안에 인코더의 출력을 디코더의 첫 입력으로 연결해주는 부분이 있다.

전체 model학습이 끝나면 test를 가지고 evaluation

 

예제 데이터 입력

import random
import torch
import torch.nn as nn
import torch.optim as optim

torch.manual_seed(0)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

raw = ["I feel hungry. 나는배가고프다.",
 "Pytorch is very easy. 파이토치는매우쉽다.",
 "Pytorch is a framework for deep learning. 파이토치는딥러닝을위한프레임워크이다.",
 "Pytorch is very clear to use. 파이토치는사용하기매우직관적이다."]

SOS_token = 0
EOS_token = 1

SOS_token: Start of Sentence 인덱스를 0번으로 설정

End of Sentence: 문장이 끝났다라는 뜻에서 토큰을 붙여서 문장의 종료를 알려줌 (1번 인덱스를 가진 토큰)

 

Preprocessing

def preprocess(corpus, source_max_length, target_max_length):
 	print("reading corpus...")
 	pairs = []
 	for line in corpus:
 		pairs.append([s for s in line.strip().lower().split("\t")])
 	print("Read {} sentence pairs".format(len(pairs)))
 
 	pairs = [pair for pair in pairs if filter_pair(pair, source_max_length, target_max_length)]
 	print("Trimmed to {} sentence pairs".format(len(pairs)))
 
 	source_vocab = Vocab()
 	target_vocab = Vocab()
 
 	print("Counting words...")
 	for pair in pairs:
 		source_vocab.add_vocab(pair[0])
 		target_vocab.add_vocab(pair[1])
 	print("source vocab size =", source_vocab.n_vocab)
 	print("target vocab size =", target_vocab.n_vocab)
 	
    return pairs, source_vocab, target_vocab

데이터 전처리

vocab 선언 단어 갯수, 단어 딕셔너리

 

인코더, 디코더 클래스 선언

class Encoder(nn.Module):
 	def __init__(self, input_size, hidden_size):
 		super(Encoder, self).__init__()
 		self.hidden_size = hidden_size
 		self.embedding = nn.Embedding(input_size, hidden_size)
 		self.gru = nn.GRU(hidden_size, hidden_size)
 
 	def forward(self, x, hidden):
 		x = self.embedding(x).view(1, 1, -1)
 		x, hidden = self.gru(x, hidden)
 		return x, hidden
 
 class Decoder(nn.Module):
 	def __init__(self, hidden_size, output_size):
 		super(Decoder, self).__init__()
 		self.hidden_size = hidden_size
 		self.embedding = nn.Embedding(output_size, hidden_size)
 		self.gru = nn.GRU(hidden_size, hidden_size)
 		self.out = nn.Linear(hidden_size, output_size)
 		self.softmax = nn.LogSoftmax(dim=1)

	def forward(self, x, hidden):
 		x = self.embedding(x).view(1, 1, -1)
 		x, hidden = self.gru(x, hidden)
 		x = self.softmax(self.out(x[0]))
 		return x, hidden

인코더

copus의 길이 만큼의 0을 가지고 있는 원핫 인코딩

embedding이라는 matrix의 input이 된다.

임베딩(embedding)

들어온 input을 hidden size만큼의 vector로 줄인다(적은 차원을 가진 벡터로 표현한다) 

줄어든 벡터가 gru에 들어오게 된다. -> 처리

 

디코더

hidden size 차원의 벡터를 다시 taget text의 index만큼 복원

임베딩 통과, gru 통과, out 통과, softmax 통과

(attention, highway network 추가)

 

학습

def tensorize(vocab, sentence):
 	indexes = [vocab.vocab2index[word] for word in sentence.split(" ")]
	indexes.append(vocab.vocab2index["<EOS>"])
 	return torch.Tensor(indexes).long().to(device).view(-1, 1)

def train(pairs, source_vocab, target_vocab, encoder, decoder, n_iter, print_every=1000, learning_rate=0.01):
 	loss_total = 0
 
 	encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)
 	decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)
 
 	training_batch = [random.choice(pairs) for _ in range(n_iter)]
 	training_source = [tensorize(source_vocab, pair[0]) for pair in training_batch]
 	training_target = [tensorize(target_vocab, pair[1]) for pair in training_batch]
 	
    criterion = nn.NLLLoss()

tensorize 보조 함수 : sentence를 원핫벡터, tensor의 형태로 바꿔줌

카테고리 분류 NLL loss

def train(pairs, source_vocab, target_vocab, encoder, decoder, n_iter, print_every=1000, learning_rate=0.01):
 	for i in range(1, n_iter + 1):
		source_tensor = training_source[i - 1]
 		target_tensor = training_target[i - 1]
 
 		encoder_hidden = torch.zeros([1, 1, encoder.hidden_size]).to(device)
 
 		encoder_optimizer.zero_grad()
 		decoder_optimizer.zero_grad()
 
 		source_length = source_tensor.size(0)
 		target_length = target_tensor.size(0)
 
 		loss = 0
 
 		for enc_input in range(source_length):
 			_, encoder_hidden = encoder(source_tensor[enc_input], encoder_hidden)

for 문 루프

encoder_hidden 0 벡터를 넣어준다

encoder에서 hidden state를 꺼내온다.

 

def train(pairs, source_vocab, target_vocab, encoder, decoder, n_iter, print_every=1000, learning_rate=0.01):
 	
    decoder_input = torch.Tensor([[SOS_token]]).long().to(device)
 	decoder_hidden = encoder_hidden
 	
    for di in range(target_length):
 		decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
 		loss += criterion(decoder_output, target_tensor[di])
 		decoder_input = target_tensor[di] # teacher forcing
 	
    loss.backward()
 	
    encoder_optimizer.step()
 	decoder_optimizer.step()
 
 	loss_iter = loss.item() / target_length
 	loss_total += loss_iter
 	
    if i % print_every == 0:
 		loss_avg = loss_total / print_every
 		loss_total = 0
 		print("[{} - {}%] loss = {:05.4f}".format(i, i / n_iter * 100, loss_avg))

decoder_hidden = encoder_hidden

입력값 SOS_token 

# teacher forcing

 

#184~195번줄

SOURCE_MAX_LENGTH = 10
TARGET_MAX_LENGTH = 12
load_pairs, load_source_vocab, load_target_vocab = preprocess(raw, SOURCE_MAX_LENGTH, TARGET_MAX_LENGTH)
print(random.choice(load_pairs))

enc_hidden_size = 16
dec_hidden_size = enc_hidden_size
enc = Encoder(load_source_vocab.n_vocab, enc_hidden_size).to(device)
dec = Decoder(dec_hidden_size, load_target_vocab.n_vocab).to(device)

train(load_pairs, load_source_vocab, load_target_vocab, enc, dec, 5000, print_every=1000)
evaluate(load_pairs, load_source_vocab, load_target_vocab, enc, dec, TARGET_MAX_LENGTH)