Seq2Seq

1. 자연어 이해(NLU)

자연어 이해(NLU, Natural Language Understanding)는 기계가 인간의 언어를 의미적으로 해석하고 이해하는 기술을 의미합니다. 이는 단순한 단어 분석을 넘어 문맥을 파악하고, 개체 인식(NER), 감성 분석, 의미적 유사성 판단, 문장 의도 분류 등 다양한 과업을 포함합니다. NLU는 기계 학습과 딥러닝 기술을 활용하여 문장의 의미를 추론하며, 챗봇, 음성 비서, 기계 번역, 자동 요약 등의 응용 분야에서 중요한 역할을 합니다.

 

2. 자연어 생성(NLG)

자연어 생성(NLG, Natural Language Generation)은 기계가 사람이 이해할 수 있는 자연스러운 텍스트를 생성하는 기술을 의미합니다. 이는 데이터를 분석하고 적절한 문맥을 반영하여 문장을 구성하는 과정으로, 보고서 작성, 뉴스 요약, 대화형 AI 응답 생성, 소설 및 시 작성 등 다양한 응용 분야에서 활용됩니다. NLG 시스템은 일반적으로 데이터 해석, 문장 계획, 언어 실현의 세 단계를 거치며, 최근에는 GPT와 같은 대규모 언어 모델이 등장하면서 더욱 자연스럽고 창의적인 텍스트 생성이 가능해졌습니다.

 

3. 시퀀스투시퀀스

시퀀스투시퀀스(Sequence-to-Sequence, Seq2Seq) 모델은 입력 시퀀스를 받아 출력 시퀀스를 생성하는 신경망 아키텍처로, 주로 자연어 처리 분야에서 번역, 요약, 질의응답 등에 활용됩니다. Seq2Seq 모델은 일반적으로 인코더(Encoder)와 디코더(Decoder) 두 부분으로 구성되며, 인코더는 입력 문장을 고정된 크기의 컨텍스트 벡터로 변환하고, 디코더는 이를 기반으로 출력 문장을 생성합니다. RNN, LSTM, GRU 등의 순환 신경망을 활용하며, 최근에는 어텐션 메커니즘과 트랜스포머 모델이 도입되어 더 긴 문맥을 효과적으로 처리할 수 있게 되었습니다. [논문]

 

※ 인코더

인코더는 입력 시퀀스를 읽고 중요한 정보를 하나의 고정된 크기의 벡터(컨텍스트 벡터, Context Vector)로 변환하는 역할을 합니다.
쉽게 말하면, 입력 문장을 하나의 "압축된 의미"로 바꿔주는 역할을 합니다.

1. 입력 시퀀스를 처리
   예를 들어, 한글 문장 "나는 학교에 간다"를 영어로 번역한다고 가정하면,
   인코더는 문장을 단어 단위 또는 문자 단위로 받아들입니다.
2. 순환 신경망(RNN, LSTM, GRU) 처리
   인코더는 RNN 계열(LSTM, GRU 등) 구조를 사용하여, 입력 단어를 하나씩 처리하면서 숨겨진 상태(hidden state)를 업데이트합니다.
3. 최종 상태 저장모든 단어를 처리한 후, 마지막 숨겨진 상태가 만들어지는데, 이를 컨텍스트 벡터(Context Vector)라고 합니다.
   이 컨텍스트 벡터에는 입력 시퀀스의 중요한 정보가 압축되어 있습니다.

※ 디코더

디코더는 인코더가 생성한 컨텍스트 벡터를 받아서, 원하는 출력 시퀀스를 생성하는 역할을 합니다.
즉, 인코더가 "기억한 내용"을 바탕으로 새로운 문장을 생성하는 과정입니다.

1. 컨텍스트 벡터를 받음
   디코더는 인코더에서 생성된 컨텍스트 벡터를 첫 번째 입력으로 받습니다.
2. 출력 시퀀스 생성
   • 디코더는 한 단어씩 출력을 예측하면서, 이전에 생성된 단어를 다음 입력으로 사용합니다.
   • 예를 들어, "나는 학교에 간다"라는 입력을 받았다면, 디코더는 순차적으로 "I" → "go" → "to" → "school"을 생성합니다.
3. RNN 계열 사용
   디코더도 LSTM 또는 GRU와 같은 RNN 계열 모델을 사용하여 문장을 생성합니다.
4. 종료 토큰(EOS, End of Sequence)디코더는 문장이 끝났음을 알리는 종료 토큰(EOS)을 생성할 때까지 출력을 계속합니다.

 

시퀀스 투 시퀀스(Seq2Seq) 모델은 자연어 처리에서 널리 사용되지만 몇 가지 단점이 있습니다. 먼저, 인코더가 입력을 하나의 고정된 컨텍스트 벡터로 변환하기 때문에 긴 문장에서 정보 손실이 발생할 수 있으며, 이는 성능 저하로 이어집니다. 또한, RNN 기반 구조는 장기 의존성 문제로 인해 문장이 길어질수록 초반과 후반 단어 간의 관계를 잘 학습하지 못하는 한계가 있습니다. 병렬 처리가 어려워 속도가 느리고, 디코더가 단어를 순차적으로 생성해야 하므로 생성 속도가 느리며 예측 오류가 이후 결과에 영향을 미치는 오토리그레시브 문제도 존재합니다. 마지막으로, 훈련 데이터에 의존적이어서 새로운 도메인이나 문장에서 일반화가 어렵습니다. 이를 해결하기 위해 어텐션 기법, 트랜스포머 모델, 사전 훈련된 대규모 언어 모델(GPT, BERT 등)이 등장하며 자연어 처리 기술이 발전하고 있습니다.

 

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'
}

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)

 

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

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

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

 

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

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

 

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

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

 

# 학습 실행
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)}")

 

# 번역 테스트 함수
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))

 

test_sentences = [
    "good morning",
    "i love you",
    "where are you?",
    "what is your name?",
    "this is a pen"
]

for sentence in test_sentences:
    translation = translate_sentence(sentence, encoder, decoder, input_lang, target_lang, device)
    print(f"Input: {sentence}\nTranslated: {translation}\n")