10. Seq2Seq구현

구현

import os
import requests
import zipfile
import torch
import torch.nn as nn
import torch.optim as optim
import random
import re
import unicodedata
from torch.utils.data import Dataset, DataLoader
from collections import Counter
from torch.nn.utils.rnn import pad_sequence

프랑스어 영어 번역 데이터 셋

# 데이터 다운로드 및 압축 해제
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
} # 웹 서버가 봇 요청을 차단하지 않도록 하는 header

def download_zip(url, output_path):
    response = requests.get(url, headers=headers, stream=True)
    if response.status_code == 200:
        with open(output_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        print(f"ZIP file downloaded to {output_path}")
    else:
        print(f"Failed to download. HTTP Response Code: {response.status_code}")

url = "<http://www.manythings.org/anki/fra-eng.zip>"
output_path = "fra-eng.zip"
download_zip(url, output_path)

path = os.getcwd()
zipfilename = os.path.join(path, output_path)

with zipfile.ZipFile(zipfilename, 'r') as zip_ref:
    zip_ref.extractall(path)

웹 서버에서 프랑스어, 영어 번역 데이터 셋을 다운로드 하는 코드

• http://www.manythings.org/anki/fra-eng.zip 해당 사이트에서 다운로드

아스키 코드 전환 함수

# 데이터 로드 및 전처리
def unicode_to_ascii(s):
    return ''.join(c for c in unicodedata.normalize('NFD', s) if unicodedata.category(c) != 'Mn')

• 다운로드 받은 데이터를 NFD방식으로 정규화
  • 프랑스어의 악센트, 발음기호를 제거하는 역할

unicode_to_ascii("Café") # "Cafe"

• 위처럼 발음 기호가 제거된다.

문자열 정규화 함수

def normalize_string(s):
    s = unicode_to_ascii(s.lower().strip())
    s = re.sub(r"[^a-zA-Z.!?]+", " ", s)
    return s

• 문자열 앞 뒤 공백 제거 후 소문자로 변환

데이터 로드

def load_data(filepath, num_samples=50000):
    with open(filepath, encoding='utf-8') as f:
        lines = f.read().strip().split("\\n")
    pairs = [[normalize_string(s) for s in l.split('\\t')[:2]] for l in lines[:num_samples]]
    return pairs

• 위의 두 함수를 이용하여 데이터를 불러와 전처리하는 함수.

file_path = os.path.join(path, "fra.txt")
pairs = load_data(file_path, num_samples=20000)

• 그리고 실제로 데이터를 불러와준다.

단어 사전 생성 클래스

# 단어 사전 생성
class Lang:
    def __init__(self):
        self.word2index = {"<SOS>": 0, "<EOS>": 1, "<PAD>": 2}
        self.index2word = {0: "<SOS>", 1: "<EOS>", 2: "<PAD>"}
        self.word_count = Counter()
    
    def add_sentence(self, sentence):
        for word in sentence.split():
            self.word_count[word] += 1
    
    def build_vocab(self, min_count=2):
        for word, count in self.word_count.items():
            if count >= min_count:
                index = len(self.word2index)
                self.word2index[word] = index
                self.index2word[index] = word
    
    def sentence_to_indexes(self, sentence):
        return [self.word2index.get(word, self.word2index['<PAD>']) for word in sentence.split()]

    def __init__(self):
        self.word2index = {"<SOS>": 0, "<EOS>": 1, "<PAD>": 2}
        self.index2word = {0: "<SOS>", 1: "<EOS>", 2: "<PAD>"}
        self.word_count = Counter()

• word, index끼리 서로 매핑하는 사전을 미리 만들어둠
  • <SOS>: 문장 시작, <EOS>: 문장끝, <PAD>: 패딩 등의 특수 토큰을 사전 등록함
• word counter 변수를 만들어둠

    def add_sentence(self, sentence):
        for word in sentence.split():
            self.word_count[word] += 1

• 문장에 나온 단어를 세고 카운터를 올리는 일을 함.

    def build_vocab(self, min_count=2):
        for word, count in self.word_count.items():
            if count >= min_count:
                index = len(self.word2index)
                self.word2index[word] = index
                self.index2word[index] = word

• 단어 빈도수가 최소 n번 나온 단어만 사전에 추가할 것이다.
  • 희귀단어를 배제하여 모델 학습을 안정화한다.

    def sentence_to_indexes(self, sentence):
        return [self.word2index.get(word, self.word2index['<PAD>']) for word in sentence.split()]

• 이제 문장을 인덱스 리스트로 변환한다. 사전에 없는 단어는 <PAD>토큰으로 대체
  • 즉, hello world라는 문장에서 hello가 3번이고, world가 등록 안되어있으면, [3,2]

생성

input_lang = Lang()
target_lang = Lang()
for src, tgt in pairs:
    input_lang.add_sentence(src)
    target_lang.add_sentence(tgt)
input_lang.build_vocab()
target_lang.build_vocab()

• 위의 클래스를 이용하여 단어 사전을 생성

학습용 데이터 준비 클래스

class TranslationDataset(Dataset):
    def __init__(self, pairs, input_lang, target_lang, max_length=20):
        self.pairs = pairs
        self.input_lang = input_lang
        self.target_lang = target_lang
        self.max_length = max_length
    
    def __len__(self):
        return len(self.pairs)
    
    def __getitem__(self, idx):
        src, tgt = self.pairs[idx]
        src_idx = self.input_lang.sentence_to_indexes(src)[:self.max_length] + [self.input_lang.word2index['<EOS>']]
        tgt_idx = self.target_lang.sentence_to_indexes(tgt)[:self.max_length] + [self.target_lang.word2index['<EOS>']]
        return torch.tensor(src_idx), torch.tensor(tgt_idx)

• 생성자, 길이 반환, 데이터 반환

def collate_fn(batch):
    src_batch, tgt_batch = zip(*batch)
    src_batch = pad_sequence(src_batch, batch_first=True, padding_value=input_lang.word2index['<PAD>'])
    tgt_batch = pad_sequence(tgt_batch, batch_first=True, padding_value=target_lang.word2index['<PAD>'])
    return src_batch, tgt_batch

• 배치를 만들 때 서로 길이가 다른 문장들을 패딩 처리하여 같은 길이로 바꿔주는 역할.
• zip을 이용해 각 문장의 텐서들을 모아주고, pad_sequence로 패딩처리

dataset = TranslationDataset(pairs, input_lang, target_lang)
dataloader = DataLoader(dataset, batch_size=64, shuffle=True, collate_fn=collate_fn)

• 위 클래스로 데이터셋 준비+데이터로더 준비

인코더 모델 구현

모델은 GRU를 사용

# Encoder & Decoder 모델 정의
class Encoder(nn.Module):
    def __init__(self, input_size, embedding_size, hidden_size, num_layers=2, dropout=0.3):
        super().__init__()
        self.embedding = nn.Embedding(input_size, embedding_size)
        self.rnn = nn.GRU(embedding_size, hidden_size, num_layers=num_layers, dropout=dropout, batch_first=True, bidirectional=True)
        self.fc = nn.Linear(hidden_size * 2, hidden_size)
    
    def forward(self, x):
        embedded = self.embedding(x)
        outputs, hidden = self.rnn(embedded)
        hidden = torch.tanh(self.fc(torch.cat((hidden[-2,:,:], hidden[-1,:,:]), dim=1)))
        return hidden.unsqueeze(0).repeat(2, 1, 1)

init

• input_size: 단어 사전 크기
• embedding_size: 단어를 임베딩 벡터로 변환할 차원 크기
• hidden_size: 은닉 상태 개수
• num_layers: GRU 층 개수
• dropout: 과적합 방지로 껐다 켜주는 dropout의 비율
• nn.Embedding: 단어를 임베딩 벡터로 변환
• nn.GRU: GRU를 사용해 문장 인코딩(양방향: bidirectional=true)
• Linear: 출력을 하나의 hidden state로 변환

forward

1. 임베딩
2. GRU인코딩
3. 양방향 출력을 하나의 hiddenstate로 변환 후 tanh 비선형 활성화 함수적용
4. 동일한 hidden state복제

디코더

class Decoder(nn.Module):
    def __init__(self, output_size, embedding_size, hidden_size, num_layers=2, dropout=0.3):
        super().__init__()
        self.embedding = nn.Embedding(output_size, embedding_size)
        self.rnn = nn.GRU(embedding_size, hidden_size, num_layers=num_layers, dropout=dropout, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x, hidden):
        x = x.unsqueeze(1)
        embedded = self.embedding(x)
        output, hidden = self.rnn(embedded, hidden)
        prediction = self.fc(output.squeeze(1))
        return prediction, hidden

output_size: 출력 언어의 단어사전 크기

Forward

• unsqueeze: 입력 단어 인덱스를 (batch,1) 형태로 확장하여 GRU에 넣음
• 임베딩
• GRU 디코딩
• 예측: 각 단어에 대한 확률 분포
• 반환: 현시점에서 예측된 단어 분포, 다음 시점으로 전달할 은닉상태

학습 루프

# 학습 실행
def train(encoder, decoder, dataloader, optimizer, criterion, device, num_epochs=50, teacher_forcing_ratio=0.3):
    for epoch in range(num_epochs):
        total_loss = 0
        for src, tgt in dataloader:
            src, tgt = src.to(device), tgt.to(device)
            optimizer.zero_grad()

            encoder_hidden = encoder(src)
            decoder_input = torch.tensor([target_lang.word2index['<SOS>']] * src.shape[0], device=device)
            decoder_hidden = encoder_hidden
            loss = 0

            for t in range(tgt.shape[1]):
                output, decoder_hidden = decoder(decoder_input, decoder_hidden)
                loss += criterion(output, tgt[:, t])

                teacher_force = random.random() < teacher_forcing_ratio
                decoder_input = tgt[:, t] if teacher_force else output.argmax(1)
            
            loss.backward()
            optimizer.step()
            total_loss += loss.item()

        print(f"Epoch {epoch+1}, Loss: {total_loss / len(dataloader)}")

에폭별 아래의 사이클을 반복한다.

1. 배치를 가져온다(배치별로 아래 사이클 반복)
2. Optimizer 초기화(zero_grad)
3. Encoder실행: 입력 문장을 인코딩.
4. Decoder초기화: <SOS>토큰으로 문장 시작
5. Decoder반복으로 단어 생성: 목표 문장 길이 만큼
6. teacher_force: 30%확률로 정답 단어를 다음 입력으로 사용(0.3) > 안정적으로 학습하도록 도움
7. 역전파, 최적화: backward(), 파라미터 업데이트
8. 손실 기록

번역 테스트

# 번역 테스트 함수
def translate_sentence(sentence, encoder, decoder, input_lang, target_lang, device, max_length=20):
    encoder.eval()
    decoder.eval()
    with torch.no_grad():
        src_idx = input_lang.sentence_to_indexes(sentence) + [input_lang.word2index['<EOS>']]
        src_tensor = torch.tensor(src_idx, device=device).unsqueeze(0)
        encoder_hidden = encoder(src_tensor)
        decoder_input = torch.tensor([target_lang.word2index['<SOS>']], device=device)
        decoder_hidden = encoder_hidden
        translated_sentence = []
        for _ in range(max_length):
            output, decoder_hidden = decoder(decoder_input, decoder_hidden)
            top_word_idx = output.argmax(1).item()
            if top_word_idx == target_lang.word2index['<EOS>']:
                break
            translated_sentence.append(target_lang.index2word[top_word_idx])
            decoder_input = torch.tensor([top_word_idx], device=device)
    return " ".join(translated_sentence)

• 학습한 것을 이용하여 번역을 실제로 확인(평가모드)
• 입력 문장이 들어오면, 전처리 후 인코딩, 디코딩을 거쳐 결과를 반환

# 모델 학습 및 테스트 실행
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
encoder = Encoder(len(input_lang.word2index), 512, 512).to(device)
decoder = Decoder(len(target_lang.word2index), 512, 512).to(device)
optimizer = optim.Adam(list(encoder.parameters()) + list(decoder.parameters()), lr=0.0005)
criterion = nn.CrossEntropyLoss()
train(encoder, decoder, dataloader, optimizer, criterion, device, num_epochs=50)
print(translate_sentence("hello how are you", encoder, decoder, input_lang, target_lang, device))
# Epoch 49, Loss: 2.9901795874769315
# Epoch 50, Loss: 2.9577140469139755
# tu allez vous ?

• 위에서 정의한 학습 시퀀스를 실행한다.

학습을 거친 후

• hello how are you > tu allez vous ? 로 번역
• tu는 친근한 2인칭 단수형 인데 allez-vous라는 존칭 복수형이 나와 잘못된 문장이긴 하지만 의미 자체는 맞다.