Seq2Seq 모델: 간단한 구조로 문장을 번역

• 수업 내용 리마인드 및 아카이빙 목적의 업로드

---

---

 

• 기본 Seq2Seq 구조
• Attention 미사용
• 인코더의 마지막 hidden state만 디코더에 입력으로 사용

 

이번 글에서는 Seq2Seq(Sequence-to-Sequence) 모델을 활용하여 영어 문장을 프랑스어로 번역하는 모델을 구축하는 과정을 단계별로 설명해드리겠습니다. 데이터 다운로드, 전처리, 모델 구성, 학습, 그리고 번역까지의 전 과정을 다룹니다. 이 글을 통해 자연어 처리에서 Seq2Seq 모델을 어떻게 활용하는지 이해하고 직접 구현할 수 있을 거예요.

 

1. 필요 모듈 임포트

우선, 데이터 처리와 모델 구축을 위해 필요한 모듈을 모두 임포트합니다. PyTorch, Numpy, Pandas 등의 라이브러리를 사용하여 데이터를 처리하고, torchinfo를 이용해 모델 구조를 시각적으로 확인할 수 있습니다.

!pip install torchinfo
import re
import os
import unicodedata
import urllib3
import zipfile
import shutil
import numpy as np
import pandas as pd
import torch

# 학습에 GPU 사용 가능 여부 확인
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")

여기서 중요한 포인트는 device 설정입니다. GPU가 사용 가능한 경우에는 CUDA를 사용하고, 그렇지 않으면 CPU를 사용해 모델을 학습시킵니다. 이는 대규모 데이터와 복잡한 모델 학습에서 속도를 크게 향상시킵니다.

 

2. 데이터 준비

Seq2Seq 모델을 학습시키기 위해 영어-프랑스어 문장 쌍이 필요합니다. fra-eng.zip 파일을 다운로드하고, fra.txt 파일에서 문장 쌍 데이터를 추출합니다. 여기서는 ManyThings.org의 공개 데이터를 사용합니다.

# 데이터 다운로드
!wget -c http://www.manythings.org/anki/fra-eng.zip && unzip -o fra-eng.zip

# 데이터 샘플 확인
with open("fra.txt", "r") as lines:
    for i, line in enumerate(lines):
        if i == 10: break
        src_line, tar_line, _ = line.strip().split('\t')
        print(f"영어 : {src_line}, 프랑스어 : {tar_line}")

데이터 설명:

• 영어 문장: 번역의 원문이 되는 문장입니다.
• 프랑스어 문장: 영어 문장을 번역한 문장으로, 이 문장을 학습하여 예측하게 됩니다.

이 과정을 통해 데이터의 구조를 확인하고, 각 문장을 영어-프랑스어 쌍으로 분리할 수 있습니다.

 

3. 데이터 전처리

자연어 처리에서 문장을 모델에 입력하기 전에 전처리 과정을 거쳐야 합니다. 여기서는 악센트 제거, 소문자 변환, 구두점 처리 등의 작업을 수행합니다.

def unicode_to_ascii(s):
    # 악센트 제거
    return ''.join(c for c in unicodedata.normalize('NFD', s) if unicodedata.category(c) != 'Mn')

def preprocess_sentence(sent):
    # 악센트 제거 및 소문자 변환
    sent = unicode_to_ascii(sent.lower())
    
    # 구두점 처리
    sent = re.sub(r"([?.!,¿])", r" \1", sent)
    
    # 알파벳과 일부 구두점 제외한 모든 문자 제거
    sent = re.sub(r"[^a-zA-Z!.?]+", r" ", sent)
    
    # 다중 공백을 하나의 공백으로 변환
    sent = re.sub(r"\s+", " ", sent)
    return sent

# 전처리 확인
en_sent = "Have you had dinner?"
fr_sent = "Avez-vous déjà diné?"
print('전처리 후 영어 문장 :', preprocess_sentence(en_sent))
print('전처리 후 프랑스어 문장 :', preprocess_sentence(fr_sent))

전처리 작업:

• 악센트 제거: 프랑스어에는 악센트가 포함된 단어가 많습니다. 예를 들어, déjà는 deja로 변환됩니다.
• 구두점 처리: 문장 내의 구두점과 단어 사이에 공백을 추가하여, 모델이 단어와 구두점을 구분할 수 있도록 합니다.
• 다중 공백 제거: 문장 중간에 불필요한 다중 공백이 있을 경우 이를 하나로 축소합니다.

이러한 전처리 과정을 통해 문장을 깔끔하게 정리하여 모델에 입력할 수 있게 됩니다.

 

4. 정수 인코딩 및 단어 집합 생성

모델에 문장을 입력하기 위해서는 각 단어를 고유한 정수로 변환해야 합니다. 이를 위해 먼저 단어 집합(vocabulary)을 생성한 후, 각 단어를 정수로 인코딩합니다.

from collections import Counter

def build_vocab(sents):
    word_list = [word for sent in sents for word in sent]
    word_counts = Counter(word_list)
    vocab = sorted(word_counts, key=word_counts.get, reverse=True)
    
    word_to_index = {'<PAD>': 0, '<UNK>': 1}
    for index, word in enumerate(vocab):
        word_to_index[word] = index + 2
    return word_to_index

# 단어 집합 생성
src_vocab = build_vocab(sents_en_in)
tar_vocab = build_vocab(sents_fra_in + sents_fra_out)

# 인덱스 확인
index_to_src = {v: k for k, v in src_vocab.items()}
index_to_tar = {v: k for k, v in tar_vocab.items()}

주요 내용:

• 단어 집합 생성: 각 단어의 등장 빈도를 계산하고, 이를 기반으로 단어-정수 매핑을 만듭니다.
• 정수 인코딩: 각 문장에 있는 단어를 정수로 변환하여 모델에 입력할 수 있도록 준비합니다.

단어 집합을 사용하면 문장을 단순한 숫자의 나열로 변환할 수 있어, 모델이 이를 학습할 수 있게 됩니다.

 

5. 커스텀 데이터셋 및 데이터로더 생성

PyTorch의 Dataset 클래스를 활용해 커스텀 데이터셋을 생성하고, DataLoader를 통해 모델에 학습시킬 데이터를 배치 단위로 나눕니다.

from torch.utils.data import Dataset, DataLoader

class TranslationDataset(Dataset):
    def __init__(self, encoder_input, decoder_input, decoder_target):
        self.encoder_input = encoder_input
        self.decoder_input = decoder_input
        self.decoder_target = decoder_target

    def __len__(self):
        return len(self.encoder_input)

    def __getitem__(self, idx):
        return (self.encoder_input[idx], self.decoder_input[idx], self.decoder_target[idx])

# 데이터셋 생성 및 데이터로더 설정
dataset = TranslationDataset(encoder_input_padded, decoder_input_padded, decoder_target_padded)
train_dataset, val_dataset = random_split(dataset, [len(dataset) - n_of_val, n_of_val])
train_loader = DataLoader(train_dataset, batch_size=128, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=128, shuffle=False)

역할:

• Dataset: 데이터셋을 관리하고, 인덱스를 통해 각 데이터를 불러오는 역할을 합니다.
• DataLoader: 데이터셋을 배치 단위로 나누어 모델에 효율적으로 학습시키는 역할을 합니다.

 

6. Seq2Seq 모델 정의

Seq2Seq 모델은 인코더(Encoder)와 디코더(Decoder)로 구성됩니다. 인코더는 입력 문장을 인코딩하고, 디코더는 이를 기반으로 출력 문장을 예측합니다.

class Encoder(nn.Module):
    def __init__(self, src_vocab_size, embedding_dim, hidden_units):
        super(Encoder, self).__init__()
        self.embedding = nn.Embedding(src_vocab_size, embedding_dim, padding_idx=0)
        self.lstm = nn.LSTM(embedding_dim, hidden_units, batch_first=True)

    def forward(self, x):
        x = self.embedding(x)
        _, (hidden, cell) = self.lstm(x)
        return hidden, cell

class Decoder(nn.Module):
    def __init__(self, tar_vocab_size, embedding_dim, hidden_units):
        super(Decoder, self).__init__()
        self.embedding = nn.Embedding(tar_vocab_size, embedding_dim, padding_idx=0)
        self.lstm = nn.LSTM(embedding_dim, hidden_units, batch_first=True)
        self.fc = nn.Linear(hidden_units, tar_vocab_size)

    def forward(self, x, hidden, cell):
        x = self.embedding(x)
        output, (hidden, cell) = self.lstm(x, (hidden, cell))
        output = self.fc(output)
        return output, hidden, cell

class Seq2Seq(nn.Module):
    def __init__(self, encoder, decoder):
        super(Seq2Seq, self).__init__()
        self.encoder = encoder
        self.decoder = decoder

    def forward(self, src, trg):
        hidden, cell = self.encoder(src)
        output, _, _ = self.decoder(trg, hidden, cell)
        return output

# 모델 초기화
encoder = Encoder(src_vocab_size, embedding_dim, hidden_units)
decoder = Decoder(tar_vocab_size, embedding_dim, hidden_units)
model = Seq2Seq(encoder, decoder).to(device)

주요 구성:

• 인코더: 입력 문장을 임베딩하고, 이를 LSTM으로 처리해 hidden state와 cell state를 반환합니다.
• 디코더: 인코더에서 전달받은 상태를 바탕으로 번역 문장을 예측합니다.
• Seq2Seq: 인코더와 디코더를 연결하여 전체적인 번역 모델을 구성합니다.

 

7. 모델 학습

모델을 학습시키기 위한 train 함수와 evaluation 함수입니다. 학습 과정에서는 손실 함수(CrossEntropyLoss)를 사용해 출력과 정답을 비교하고, Adam Optimizer를 사용해 모델의 가중치를 업데이트합니다.

def train(model, dataloader, loss_function, optimizer):
    model.train()
    total_loss = 0
    for encoder_inputs, decoder_inputs, decoder_targets in dataloader:
        optimizer.zero_grad()
        encoder_inputs, decoder_inputs, decoder_targets = encoder_inputs.to(device), decoder_inputs.to(device), decoder_targets.to(device)
        outputs = model(encoder_inputs, decoder_inputs)
        loss = loss_function(outputs.view(-1, outputs.size(-1)), decoder_targets.view(-1))
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    return total_loss / len(dataloader)

학습 과정:

1. 순방향 전파: 입력 데이터를 모델에 전달하여 출력을 계산합니다.
2. 손실 계산: 예측된 출력과 실제 정답(번역)을 비교하여 손실을 계산합니다.
3. 역전파: 손실을 기반으로 모델의 가중치를 업데이트하여 학습을 진행합니다.

 

8. 번역 및 성능 평가

학습된 모델을 사용해 실제 번역을 수행합니다. 입력 문장을 모델에 전달하여 번역된 문장을 생성합니다.

def decode_sequence(input_seq, model, src_vocab_size, tar_vocab_size, max_output_len, int_to_src_token, int_to_tar_token):
    encoder_inputs = torch.tensor(input_seq, dtype=torch.long).unsqueeze(0).to(device)
    hidden, cell = model.encoder(encoder_inputs)
    decoder_input = torch.tensor([3], dtype=torch.long).unsqueeze(0).to(device)
    decoded_tokens = []

    for _ in range(max_output_len):
        output, hidden, cell = model.decoder(decoder_input, hidden, cell)
        output_token = output.argmax(dim=-1).item()
        if output_token == 4: break
        decoded_tokens.append(output_token)
        decoder_input = torch.tensor([output_token], dtype=torch.long).unsqueeze(0).to(device)

    return ' '.join(int_to_tar_token[token] for token in decoded_tokens)

# 번역 결과 확인
for seq_index in [20, 30, 110]:
    input_seq = val_dataset[seq_index][0].numpy()
    translated_text = decode_sequence(input_seq, model, src_vocab_size, tar_vocab_size, 20, index_to_src, index_to_tar)
    print("입력문장 :", seq_to_src(val_dataset[seq_index][0].numpy()))
    print("번역문장 :", translated_text)
    print("-" * 50)

번역 과정:

1. 인코딩: 입력 문장을 인코더에 전달하여 hidden state와 cell state를 얻습니다.
2. 디코딩: 첫 번째 디코더 입력으로 <sos>를 사용하여 번역을 시작하고, 각 단계에서 예측된 단어를 다음 단계의 입력으로 사용합니다.
3. 번역 결과 출력: 디코딩이 끝나면 번역된 문장을 출력합니다.

 

기본 Seq2Seq 모델을 구현하며, Attention 메커니즘을 사용하지 않습니다. 인코더의 마지막 hidden state만을 기반으로 디코더가 예측을 수행합니다. 이 방식은 문장이 길어질수록 중요한 정보가 손실될 가능성이 있습니다.