Attention 기반 모델: 중요한 정보에 집중하는 번역

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

---

---

 

• Attention 메커니즘이 적용된 Seq2Seq 구조
• Attention 사용
• 디코더는 인코더의 모든 출력(hidden state)을 활용하여 더 정확한 번역을 수행

 

이번 글에서는 Seq2Seq 모델을 사용하여 영어-프랑스어 번역 모델을 구축하고 학습하는 과정을 단계별로 설명합니다. 이 프로젝트는 자연어 처리(NLP)에서 많이 사용되는 Seq2Seq 구조를 적용해 언어 간 번역을 수행하는 모델을 만드는 과정입니다.

 

1. 필요한 모듈 임포트

프로젝트에서 사용되는 주요 모듈들을 먼저 설치하고 불러옵니다. 이 과정에서는 PyTorch 라이브러리를 이용해 딥러닝 모델을 구현하고, 데이터 처리에는 Pandas와 Numpy를 사용합니다.

# 필요 모듈 설치
!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
from torch.utils.data import Dataset, DataLoader, random_split
from collections import Counter
from torch.nn.utils.rnn import pad_sequence
import torch.nn as nn
import torch.optim as optim
from torchinfo import summary

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

• torchinfo: 딥러닝 모델의 구조와 파라미터 정보를 요약해서 출력하는 모듈입니다.
• unicodedata: 텍스트 데이터의 특수 문자를 처리하고, Unicode 데이터를 다룰 수 있도록 해줍니다.
• PyTorch: GPU 가속 딥러닝 프레임워크로, 본 프로젝트에서 Seq2Seq 모델 학습에 사용됩니다.

 

2. 데이터 준비

Seq2Seq 모델 학습을 위해, 영어-프랑스어 번역 데이터가 필요합니다. 이번 프로젝트에서는 인터넷에서 제공되는 fra-eng.zip 파일을 다운로드하여 fra.txt 파일을 사용합니다. 이 파일은 각각의 문장이 탭(\t)으로 구분되어 있는 형식입니다.

# 데이터 다운로드 및 압축 해제
!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}")

• 이 데이터는 영어 문장과 프랑스어 문장이 탭으로 구분되어 있으며, 이를 기반으로 학습 데이터를 생성합니다.
• 데이터 파일은 fra.txt로 압축을 해제하며, 샘플 출력 결과는 다음과 같습니다.

영어 : Go., 프랑스어 : Va !
영어 : Go., 프랑스어 : Marche.

 

 

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 = u"Have you had dinner?"
fr_sent = u"Avez-vous déjà diné?"

print('전처리 전 영어 문장 :', en_sent)
print('전처리 후 영어 문장 :', preprocess_sentence(en_sent))
print('전처리 전 프랑스어 문장 :', fr_sent)
print('전처리 후 프랑스어 문장 :', preprocess_sentence(fr_sent))

• 전처리 전: "Avez-vous déjà diné?"
• 전처리 후: "avez vous deja dine ?"

악센트가 제거되고 소문자로 변환되었으며, 문장 부호와 단어 사이에 공백이 추가되었습니다.

 

4. 정수 인코딩 및 단어 사전 생성

이제 모델에 데이터를 입력하기 위해 각 단어를 고유한 정수로 변환합니다. 이를 위해서는 먼저 각 문장에서 등장하는 단어들의 리스트(사전)를 생성한 후, 이를 기반으로 단어와 인덱스를 매핑하는 작업을 수행해야 합니다.

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)

# 영어 단어 집합 크기와 프랑스어 단어 집합 크기 확인
src_vocab_size = len(src_vocab)
tar_vocab_size = len(tar_vocab)
print(f"영어 단어 집합의 크기: {src_vocab_size}, 프랑스어 단어 집합의 크기: {tar_vocab_size}")

• 이 과정에서는 문장에서 단어를 추출하여 각 단어에 대해 고유한 인덱스를 부여합니다.
• 이렇게 만들어진 단어 집합(vocabulary)을 이용해 텍스트 데이터를 정수 인코딩하여 모델에서 처리할 수 있는 형태로 변환합니다.

 

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

PyTorch의 Dataset 클래스를 이용해 커스텀 데이터셋을 만들고, 이를 DataLoader에 전달하여 학습과 검증 데이터를 배치(batch) 단위로 나누어 처리합니다.

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])

# 패딩 처리 후 데이터셋 생성
encoder_input_padded = pad_sequence([torch.tensor(seq) for seq in encoder_input], batch_first=True, padding_value=0)
decoder_input_padded = pad_sequence([torch.tensor(seq) for seq in decoder_input], batch_first=True, padding_value=0)
decoder_target_padded = pad_sequence([torch.tensor(seq) for seq in decoder_target], batch_first=True, padding_value=0)

# 데이터셋 및 데이터로더 생성
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)

• pad_sequence 함수는 시퀀스의 길이를 맞추기 위해 패딩을 추가합니다.
• 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)
        outputs, (hidden, cell) = self.lstm(x)
        return outputs, 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, hidden_units, batch_first=True)
        self.fc = nn.Linear(hidden_units, tar_vocab_size)

    def forward(self, x, encoder_outputs, hidden, cell):
        x = self.embedding(x)
        attention_scores = torch.bmm(encoder_outputs, hidden.transpose(0, 1).transpose(1, 2))
        attention_weights = nn.Softmax(dim=2)(attention_scores)
        context_vector = torch.bmm(attention_weights.transpose(1, 2), encoder_outputs)
        context_vector_repeated = context_vector.repeat(1, x.shape[1], 1)
        x = torch.cat((x, context_vector_repeated), dim=2)
        output, (hidden, cell) = self.lstm(x, (hidden, cell))
        output = self.fc(output)
        return output, hidden, cell

• Encoder는 입력 시퀀스를 임베딩 벡터로 변환하고 LSTM을 통해 잠재 벡터를 생성합니다.
• Decoder는 인코더에서 생성된 잠재 벡터를 입력으로 받아 타겟 언어의 문장을 생성합니다.

 

7. 모델 학습

이제 Seq2Seq 모델을 학습시킵니다. train 함수는 모델의 학습 과정, evaluation 함수는 검증 데이터를 통해 모델의 성능을 평가하는 역할을 합니다.

def train(model, dataloader, optimizer, loss_function):
    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)

def evaluation(model, dataloader, loss_function):
    model.eval()
    total_loss = 0
    total_correct = 0
    total_count = 0

    with torch.no_grad():
        for encoder_inputs, decoder_inputs, decoder_targets in dataloader:
            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))
            total_loss += loss.item()
            mask = decoder_targets != 0
            total_correct += ((outputs.argmax(dim=-1) == decoder_targets) * mask).sum().item()
            total_count += mask.sum().item()

    return total_loss / len(dataloader), total_correct / total_count

# 학습 실행
num_epochs = 30
for epoch in range(num_epochs):
    train_loss = train(model, train_loader, optimizer, loss_function)
    val_loss, val_acc = evaluation(model, val_loader, loss_function)
    print(f'Epoch: {epoch+1}/{num_epochs}, Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}')

• 학습 과정: 모델을 학습 모드로 설정하고, 순전파-역전파 과정을 통해 가중치를 업데이트합니다.
• 평가 과정: 검증 데이터를 통해 모델의 손실과 정확도를 계산하여 모델의 성능을 평가합니다.

 

8. 번역 수행

학습된 모델을 사용해 실제 번역을 수행합니다. 입력된 영어 문장을 프랑스어로 번역하는 예시입니다.

def decode_sequence(input_seq, model, max_output_len, int_to_tar_token):
    encoder_inputs = torch.tensor(input_seq, dtype=torch.long).unsqueeze(0).to(device)
    encoder_outputs, hidden, cell = model.encoder(encoder_inputs)
    decoder_input = torch.tensor([tar_vocab['<sos>']], dtype=torch.long).unsqueeze(0).to(device)
    decoded_tokens = []

    for _ in range(max_output_len):
        output, hidden, cell = model.decoder(decoder_input, encoder_outputs, hidden, cell)
        output_token = output.argmax(dim=-1).item()
        if output_token == tar_vocab['<eos>']:
            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, 300]:
    input_seq = val_dataset[seq_index][0].numpy()
    translated_text = decode_sequence(input_seq, model, max_output_len=20, int_to_tar_token=index_to_tar)
    print(f"입력 문장: {seq_to_src(val_dataset[seq_index][0].numpy())}")
    print(f"번역 문장: {translated_text}")

 

 

Attention 메커니즘이 적용되어 있으며, 인코더의 모든 타임스텝의 출력을 고려합니다. Attention은 특히 긴 문장의 번역에서 성능을 크게 향상시킬 수 있으며, 인코더의 모든 타임스텝에서 가장 중요한 정보를 선택적으로 활용하는 것이 가능합니다.