[ SK 네트웍스 Family AI 캠프 수업 내용 복습 ] 자연어 딥러닝 응용(신경망 기계번역) 2025-04-29

Seq2Seq (Sequence-to-Sequence)

정의:
Seq2Seq는 입력 시퀀스를 받아서 출력 시퀀스로 변환하는 딥러닝 기반 모델 구조로, 기계 번역, 문장 요약, 질의응답 등 입력과 출력 길이가 다를 수 있는 시퀀스 간 변환 문제에 사용됨.

---

구조 구성

1. Encoder

• 역할: 입력 문장 전체를 순차적으로 읽고, 그 의미를 하나의 벡터(h = hidden state)에 요약 및 압축함.
• 과정:
  • 각 입력 단어를 임베딩(embedding) 벡터로 변환
  • RNN/LSTM/GRU 등의 순환 구조를 통해 순차적으로 처리
  • 마지막 hidden state는 입력 문장의 **의미 요약 벡터(context vector)**로 사용됨
• 출력: 문장의 의미를 함축한 고정 길이 벡터 (보통 마지막 hidden state)

📎 보충 설명:
전통적인 Seq2Seq에서는 마지막 시점의 hidden state 하나만 context로 사용하므로, 문장이 길거나 복잡할수록 정보 손실이 발생할 수 있음. 이를 보완하기 위해 Attention 메커니즘이 추가되기도 함 (아래 참고).

---

2. Decoder

• 역할: Encoder가 만든 요약 벡터(context vector)를 바탕으로 출력 시퀀스를 생성
• 과정:
  1. 시작 토큰 <SOS>를 입력으로 받음
  2. RNN 구조를 사용해 다음 단어를 예측
  3. 예측한 단어를 다시 입력으로 넣고, 다음 단어를 반복해서 예측
  4. 종료 토큰 <EOS>가 나올 때까지 반복
• 출력: 예측된 출력 문장 시퀀스

보충 설명:
Decoder는 매 시점마다 이전의 출력 결과를 입력으로 삼기 때문에, 시퀀스 생성에 시간이 걸리고 누적 오류가 발생할 수 있음. 이를 개선하기 위해 Teacher Forcing이라는 기법을 학습 시 사용할 수 있음 — 이전 시점 예측값 대신 실제 정답을 넣는 방식.

---

전체 흐름 요약

입력 문장 → Encoder → Context Vector → Decoder → 출력 문장

즉,

1. Encoder는 입력 문장을 모두 읽고 의미를 압축한 context vector를 생성
2. Decoder는 이 context vector를 바탕으로 문장을 한 단어씩 생성

---

추가 개념: Attention ( 주의 메커니즘 )

기존 Seq2Seq에서는 전체 문장의 의미를 고정된 하나의 벡터로만 전달하기 때문에 정보 손실이 있음.
이를 보완하기 위해 Attention은 매 시점마다 입력 문장의 모든 단어와 관련된 정보를 동적으로 계산하여 Decoder에 제공함.

 

---

단어 사전 생성 및 번역용 데이터셋 정의

from typing import List, Tuple

# 단어 사전 생성 함수 정의
def build_vocab(sentences: List[str]) -> Tuple[dict, dict]:
    words = set()  # 모든 문장에서 단어들을 수집할 집합
    for sent in sentences:
        words.update(sent.split())  # 공백 기준으로 단어를 나누어 추가

    # 특수 토큰 포함 초기 단어 사전
    word2idx = {"<PAD>": 0, "<SOS>": 1, "<EOS>": 2}
    for word in words:
        if word not in word2idx:
            word2idx[word] = len(word2idx)  # 고유 인덱스 부여

    # 인덱스 → 단어 변환을 위한 역 사전 생성
    idx2word = {idx: word for word, idx in word2idx.items()}

    return word2idx, idx2word  # (단어→인덱스, 인덱스→단어) 반환
    
# 영문과 국문의 단어 사전 생성
en_word2idx, en_idx2word = build_vocab(df["en"])
ko_word2idx, ko_idx2word = build_vocab(df["ko"])


# 번역용 데이터셋 정의 (PyTorch Dataset 상속)
class TranslationDataset(Dataset):
    def __init__(self, df, en_word2idx, ko_word2idx, max_len=15):
        self.df = df  # 데이터프레임 (영문-국문 병렬 데이터)
        self.en_word2idx = en_word2idx  # 영문 단어 → 인덱스 사전
        self.ko_word2idx = ko_word2idx  # 국문 단어 → 인덱스 사전
        self.max_len = max_len  # 시퀀스 최대 길이 (패딩/잘림 기준)

    def __len__(self):
        return len(self.df)  # 전체 샘플 수 반환

    def __getitem__(self, idx):
        # 해당 인덱스의 문장 불러오기
        en_sent = self.df.iloc[idx]["en"].split()
        ko_sent = self.df.iloc[idx]["ko"].split()

        # 영문 문장: 단어 인덱스 + <EOS>
        en_ids = [self.en_word2idx.get(word, 0) for word in en_sent] + [self.en_word2idx["<EOS>"]]

        # 국문 문장: <SOS> + 단어 인덱스 + <EOS>
        ko_ids = [self.ko_word2idx["<SOS>"]] + [self.ko_word2idx.get(word, 0) for word in ko_sent] + [self.ko_word2idx["<EOS>"]]

        # 최대 길이에 맞게 패딩하거나 잘라내기
        en_ids = (en_ids + [self.en_word2idx["<PAD>"]] * self.max_len)[:self.max_len]
        ko_ids = (ko_ids + [self.ko_word2idx["<PAD>"]] * self.max_len)[:self.max_len]

        # 텐서로 변환하여 반환
        return torch.tensor(en_ids), torch.tensor(ko_ids)

전체 구조 요약

1. build_vocab: 문장 리스트에서 고유 단어를 추출하고, 이를 인덱스로 매핑하는 단어 사전을 생성.
2. TranslationDataset: PyTorch에서 사용할 수 있는 데이터셋 클래스.
   • 영문 입력 시퀀스는 <EOS> 토큰을 붙여 마무리.
   • 국문 출력 시퀀스는 <SOS>부터 시작하여 <EOS>로 끝남.
   • 고정된 max_len에 맞춰 패딩 처리.
3. 반환된 두 개의 텐서(en_ids, ko_ids)는 각각 입력 시퀀스와 타겟 시퀀스로 사용됨.

 

Seq2Seq 모델 정의

# Seq2Seq 모델 정의
class Encoder(nn.Module):
    def __init__(self, vocab_size, embed_size=64, hidden_state=128):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)  # 단어 임베딩
        self.rnn = nn.GRU(embed_size, hidden_state, batch_first=True)  # GRU 사용

    def forward(self, x):
        embeded = self.embedding(x)  # 입력 시퀀스 → 임베딩 벡터 (batch, seq_len, embed_size)
        output, hidden = self.rnn(embeded)  # RNN 실행: output은 전체 시퀀스의 출력, hidden은 마지막 상태
        return hidden  # 최종 hidden state만 반환 → Decoder의 초기 상태로 사용

class Decoder(nn.Module):
    def __init__(self, vocab_size, embed_size=64, hidden_size=128):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)  # 단어 임베딩
        self.rnn = nn.GRU(embed_size, hidden_size, batch_first=True)  # GRU 사용
        self.fc = nn.Linear(hidden_size, vocab_size)  # GRU 출력 → 단어 예측을 위한 선형층

    def forward(self, x, hidden):
        embeded = self.embedding(x)  # 입력 문장 → 임베딩 벡터
        output, hidden = self.rnn(embeded, hidden)  # hidden은 encoder의 마지막 상태
        logits = self.fc(output)  # GRU 출력 → 어휘 수만큼의 로짓(logits) 반환
        return logits, hidden  # 로짓은 softmax 없이 반환됨 (CrossEntropyLoss에서 처리)
        
class Seq2seq(nn.Module):
    def __init__(self, encoder, decoder):
        super().__init__()
        self.encoder = encoder  # Encoder 객체
        self.decoder = decoder  # Decoder 객체

    def forward(self, src, tgt):
        hidden = self.encoder(src)  # src 문장을 인코더에 입력 → 요약된 hidden state 얻기
        # decoder 입력은 <SOS>부터 시작해서 마지막 <EOS> 전까지 (tgt[:, :-1])
        # 정답은 <SOS> 다음부터 <EOS>까지 (tgt[:, 1:])가 정답 시퀀스가 됨
        logits, _ = self.decoder(tgt[:, :-1], hidden)  
        return logits  # 디코더가 출력한 로짓 시퀀스 반환

구조 설명

1. Encoder

• 입력: src 문장 (batch, seq_len)
• 임베딩 후 GRU로 시퀀스를 처리하여 마지막 hidden state를 반환
• 이 hidden state는 전체 입력 문장의 요약 정보 (context vector) 역할

2. Decoder

• 입력: tgt 문장의 <SOS>부터 <EOS> 전까지의 시퀀스
• 인코더에서 전달받은 hidden state를 초기 상태로 사용
• GRU → Linear을 통해 각 시점마다 다음 단어의 logit을 출력
• 학습 시 CrossEntropyLoss를 통해 logits와 정답 시퀀스를 비교

3. Seq2seq 모델

• forward(src, tgt)는 학습 시 사용
• tgt[:, :-1]는 디코더 입력, tgt[:, 1:]은 정답 시퀀스
• 디코더 출력은 각 시점마다 단어 예측에 사용될 로짓(logits) 값

---

보충 설명: GRU와 CrossEntropyLoss

• GRU는 RNN 계열로, hidden_state를 인코더와 디코더 사이에서 공유함으로써 정보 흐름을 유지
• 디코더 출력은 softmax를 적용하지 않고 raw logits를 반환 → nn.CrossEntropyLoss가 내부적으로 softmax와 log를 적용하므로 중복 계산을 피함

 

데이터셋 준비, 모델 학습 루프 및 손실 함수 계산

import torch
from torch.utils.data import DataLoader
from torch import nn, optim
from tqdm import tqdm

# 데이터셋 준비
dataset = TranslationDataset(df, en_word2idx, ko_word2idx)  # TranslationDataset 객체 생성
dataloader = DataLoader(dataset, batch_size=8, shuffle=True)  # DataLoader로 배치 단위 데이터 처리

# 모델, 손실 함수, 옵티마이저 설정
encoder = Encoder(vocab_size=len(en_word2idx))  # Encoder 모델 생성
decoder = Decoder(vocab_size=len(ko_word2idx))  # Decoder 모델 생성

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")  # GPU 또는 CPU 설정
model = Seq2seq(encoder, decoder).to(device)  # Seq2Seq 모델을 device에 맞게 이동
criterion = nn.CrossEntropyLoss(ignore_index=ko_word2idx["<PAD>"])  # CrossEntropyLoss, PAD 토큰은 무시
optimizer = optim.Adam(model.parameters(), lr=0.001)  # Adam 옵티마이저 설정

# 학습 설정
num_epochs = 100  # 학습 epoch 수
train_loss = []  # 학습 손실을 저장할 리스트

model.train()  # 모델을 학습 모드로 설정

# tqdm을 이용해 학습 진행 상황을 시각적으로 표시
iterator = tqdm(range(num_epochs))

# 학습 루프
for epoch in iterator:
    epoch_loss = 0  # epoch마다 손실 초기화

    for batch_idx, (src, tgt) in enumerate(dataloader):
        src, tgt = src.to(device), tgt.to(device)  # 배치 데이터를 device로 이동
        optimizer.zero_grad()  # 이전 gradient 초기화

        logits = model(src, tgt)  # 모델에 입력하여 예측 결과(logits) 얻기

        # CrossEntropyLoss 계산
        # logits의 크기: (batch_size, seq_len, vocab_size)
        # tgt의 크기: (batch_size, seq_len)
        # 로그와 레이블을 비교하여 손실 계산
        loss = criterion(logits.view(-1, len(ko_word2idx)), tgt[:,1:].contiguous().view(-1))

        loss.backward()  # 손실에 대한 gradient 계산
        optimizer.step()  # optimizer에 의한 파라미터 업데이트

        epoch_loss += loss.item()  # epoch 손실에 더하기

    train_loss.append(epoch_loss / len(dataloader))  # 평균 손실 기록
    iterator.set_description(f"Epoch {epoch+1}/{num_epochs} Loss: {epoch_loss / len(dataloader):.4f}")

전체 구조 요약

1. 데이터셋 준비

• TranslationDataset을 사용해 데이터를 로드하고, DataLoader로 배치 처리 및 셔플링.

2. 모델, 손실 함수, 옵티마이저 설정

• Seq2Seq 모델을 정의한 뒤, Encoder와 Decoder를 합쳐 모델을 구축.
• 손실 함수로 CrossEntropyLoss 사용, 여기서 ignore_index를 설정해 <PAD> 토큰은 손실 계산에서 제외.
• 옵티마이저는 Adam을 사용하여 모델 파라미터를 업데이트.

3. 학습 루프

• num_epochs에 대해 학습을 진행하면서, 각 배치에 대해:
  • 입력(src)과 타겟(tgt)을 모델에 넣어 예측값(logits)을 얻음.
  • 예측값과 실제값을 비교하여 손실을 계산하고 역전파(loss.backward())를 통해 그래디언트를 업데이트.
  • optimizer.step()으로 파라미터를 업데이트.

4. 손실 계산 및 출력

• 매 epoch마다 평균 손실을 계산하여 train_loss에 저장하고, tqdm으로 진행 상황을 시각적으로 표시.

---

보충 설명

• logits.view(-1, vocab_size):
  • logits의 크기는 (batch_size, seq_len, vocab_size)이므로 이를 (batch_size * seq_len, vocab_size) 형태로 펼쳐서 CrossEntropyLoss에 적합하게 만듦.
• tgt[:, 1:].contiguous().view(-1):
  • tgt[:, 1:]: tgt의 첫 번째 토큰(<SOS>)을 제외한 부분을 사용. 디코더는 <SOS> 다음 토큰을 예측해야 하므로, 실제 예측값은 tgt의 두 번째 토큰부터 마지막까지.
  • contiguous().view(-1): 텐서를 평탄화하여 손실 함수에 맞는 형태로 만듦.

 

손실 시각화

import matplotlib.pyplot as plt

# 손실 시각화
plt.plot(range(1, num_epochs + 1), train_loss)  # 학습 에폭(epoch) 별 평균 손실 값 그래프
plt.xlabel('Epoch')  # x축 라벨: Epoch
plt.ylabel('Loss')  # y축 라벨: Loss
plt.title('Training Loss')  # 그래프 제목: Training Loss
plt.show()  # 그래프를 화면에 출력

코드 설명

1. plt.plot(range(1, num_epochs + 1), train_loss):
   • train_loss 리스트에는 각 epoch마다 계산된 평균 손실 값이 저장돼 있어. 이 값들을 range(1, num_epochs + 1)(1부터 num_epochs까지의 숫자)와 함께 그래프에 그려.
   • range(1, num_epochs + 1)는 x축에 epoch 번호를 표시하고, train_loss는 y축에 손실 값을 표시.
2. plt.xlabel('Epoch'):
   • x축에 Epoch를 나타내는 라벨을 추가.
3. plt.ylabel('Loss'):
   • y축에 Loss를 나타내는 라벨을 추가.
4. plt.title('Training Loss'):
   • 그래프 제목을 Training Loss로 설정.
5. plt.show():
   • 그래프를 실제로 화면에 표시.

---

손실 시각화의 의미

이 그래프는 학습 과정에서 손실 함수 값이 어떻게 변화하는지를 보여줍니다. 일반적으로 학습이 진행됨에 따라 손실 값은 감소해야 하며, 이를 통해 모델이 점점 더 잘 학습하고 있음을 확인할 수 있습니다. 만약 손실 값이 줄어들지 않거나 급격히 증가하는 경우, 하이퍼파라미터 튜닝이나 데이터 전처리 등을 다시 확인해야 할 필요가 있습니다.

---

추가 팁

• 학습이 제대로 진행되었는지 확인할 때 손실 값 외에도 검증 손실(validation loss) 을 함께 시각화하는 것이 유용합니다. 이는 모델이 훈련 데이터에 과적합(overfitting)되지 않도록 돕기 때문입니다.

 

추론 수행

max_len = 15  # 문장의 최대 길이를 설정
model.eval()  # 모델을 평가 모드로 설정 (dropout, batchnorm 등의 동작을 평가 모드로 변경)
test_sentence = ["I said let's watch a movie tonight, I have a lot of homework to do."]

# 입력 문장을 처리하는 루프
for sent in test_sentence:
    # 문장에 <EOS> 추가
    src_tokens =  sent.split() + ["<EOS>"]
    # 각 단어를 인덱스로 변환, 최대 길이에 맞게 패딩 처리
    src_ids = [en_word2idx.get(word, 0) for word in src_tokens][:max_len]
    src_ids += [en_word2idx["<PAD>"]] * (max_len - len(src_ids))
    src = torch.tensor([src_ids]).to(device)  # 입력 문장을 텐서로 변환하여 device로 이동

    # encoder
    hidden = model.encoder(src)  # 입력 문장을 인코더에 넣어서 hidden state 얻음

    # decoder
    tgt_ids = [ko_word2idx['<SOS>']]  # 시작 토큰(<SOS>)으로 디코더 시작
    for _ in range(max_len):  # 최대 길이만큼 예측 반복
        tgt = torch.tensor([tgt_ids]).to(device)  # 현재까지 예측된 단어들을 텐서로 변환
        with torch.no_grad():  # 추론 시에는 기울기 계산을 하지 않음
            logits, hidden = model.decoder(tgt, hidden)  # 디코더에서 예측
            pred = logits[:, -1, :].argmax(-1).item()  # 예측한 마지막 단어의 인덱스를 선택
        tgt_ids.append(pred)  # 예측한 단어를 tgt_ids에 추가
        if pred == ko_word2idx['<EOS>']:  # <EOS> 토큰을 만나면 번역 종료
            break

    # 예측된 단어들의 인덱스를 단어로 변환하여 출력
    translation = [ko_idx2word[idx] for idx in tgt_ids[1:]]  # <SOS>는 제외
    print(" ".join(translation))  # 번역된 문장을 출력

코드 설명

1. max_len = 15:
   • 입력 문장의 최대 길이를 설정. 이보다 긴 문장은 잘리고, 짧은 문장은 <PAD>로 채워짐.
2. model.eval():
   • 모델을 평가 모드로 설정. dropout이나 batch normalization 등을 학습 모드와 다르게 동작하도록 설정.
3. test_sentence = ["I said let's watch a movie tonight, I have a lot of homework to do."]:
   • 번역할 입력 문장을 설정.
4. src_tokens = sent.split() + ["<EOS>"]:
   • 입력 문장을 공백 기준으로 나누고, 문장의 끝에 <EOS> 토큰을 추가.
5. src_ids = [en_word2idx.get(word, 0) for word in src_tokens][:max_len]:
   • 각 단어를 인덱스로 변환. 인덱스가 없으면 0 (PAD)으로 처리.
6. src_ids += [en_word2idx["<PAD>"]] * (max_len - len(src_ids)):
   • 문장이 max_len보다 짧으면 남은 부분을 <PAD>로 채움.
7. src = torch.tensor([src_ids]).to(device):
   • src_ids를 텐서로 변환하고, 모델이 학습된 device (CPU 또는 GPU)로 이동.
8. hidden = model.encoder(src):
   • 입력 문장을 인코더에 넣어 hidden state를 얻음.
9. tgt_ids = [ko_word2idx['<SOS>']]:
   • 디코더 입력의 시작은 <SOS> 토큰으로 시작.
10. for _ in range(max_len)::
    • 디코더에서 최대 길이만큼 예측을 반복.
11. logits, hidden = model.decoder(tgt, hidden):
    • 디코더에서 예측 결과를 얻음.
12. pred = logits[:, -1, :].argmax(-1).item():
    • 디코더에서 나온 마지막 단어의 확률 분포에서 가장 확률이 높은 단어의 인덱스를 선택.
13. tgt_ids.append(pred):
    • 예측한 단어를 tgt_ids에 추가.
14. if pred == ko_word2idx['<EOS>']::
    • <EOS> 토큰이 예측되면 번역을 종료.
15. translation = [ko_idx2word[idx] for idx in tgt_ids[1:]]:
    • tgt_ids에 있는 인덱스를 단어로 변환하여 번역된 문장을 생성. 첫 번째 <SOS> 토큰은 제외.
16. print(" ".join(translation)):
    • 번역된 문장을 출력.

---

예측 결과

위 코드는 주어진 영어 문장을 한국어로 번역하는 작업을 수행합니다. 모델이 인코더를 통해 입력 문장을 요약(hidden state) 하고, 디코더가 이를 기반으로 단어를 하나씩 예측하여 번역된 문장을 생성하는 방식입니다.

• 이 과정에서 "<SOS>" 토큰을 입력으로, "<EOS>" 토큰을 만나면 번역을 종료하는 방식으로 동작합니다.
• 추론 시, 기울기 계산을 하지 않음 (torch.no_grad())으로 속도를 높이고, 추론에만 집중합니다.

이 코드에서 번역 결과는 학습된 모델에 따라 달라지며, 모델이 얼마나 잘 학습되었는지에 따라 정확도가 달라질 수 있습니다.

결과값: 내일 눈이 온다고 들었어요. <EOS>

 

---

1. 기계 번역의 핵심: Attention

• Seq2Seq 모델의 한계:
  • Seq2Seq 모델에서는 Encoder가 입력 문장을 고정된 벡터로 압축하여 Decoder로 전달합니다. 그러나 이 방법은 문장의 길이가 길어질수록 정보 손실이 발생할 수 있습니다.
  • Encoder가 입력 문장을 하나의 벡터로 요약하기 때문에 긴 문장에서는 중요한 정보를 놓칠 가능성이 있습니다.
• Attention의 해결책:
  • Attention 메커니즘은 Decoder가 출력 단어를 생성할 때마다 입력 문장의 모든 단어를 동적으로 참고하여, 문맥을 보다 정확히 반영할 수 있도록 합니다.
  • 이를 통해 각각의 출력 단어를 생성할 때 입력 문장의 특정 부분에 집중(attend)하게 되어 정보 손실을 줄이고, 더 정확한 번역을 가능하게 합니다.

---

2. Bahdanau Attention vs Luong Attention

1) Bahdanau Attention

• 특징: Encoder의 hidden state와 Decoder의 hidden state를 결합하여 Attention 점수를 계산합니다.
• 작동 방식:
  • Decoder의 hidden state와 Encoder의 각 시점에서의 hidden state를 사용하여 가중치(attention weights)를 계산합니다.
  • 각 입력 단어에 대한 가중치를 계산한 후, 이 가중치에 따라 Context vector를 만들고, 이를 기반으로 출력 단어를 생성합니다.
• 장점:
  • 입력 문장의 각 단어에 대해 더 세밀하게 집중할 수 있습니다.
• 단점:
  • 계산이 상대적으로 더 복잡하고 느릴 수 있습니다.

2) Luong Attention

• 특징: Dot-product 방식으로 점수를 계산하여 간단하고 빠르게 처리합니다.
• 작동 방식:
  • Dot-product 점수 계산 방법을 사용하여 Encoder의 hidden state와 Decoder의 hidden state 간의 유사도를 계산합니다.
  • 이 점수를 바탕으로 가중치를 계산하여 Context vector를 생성하고, 이를 기반으로 Decoder가 다음 단어를 예측합니다.
• 장점:
  • 계산이 간단하고 빠르며, 큰 데이터셋에 대해서도 잘 동작합니다.
• 단점:
  • Bahdanau Attention보다 표현력이 떨어질 수 있습니다. 예를 들어, 긴 문장을 번역할 때 주의 집중이 덜 될 수 있습니다.

---

3. Seq2Seq + Attention 모델

• Encoder:
  • 입력 문장을 Encoder에 전달하여 문장을 hidden state로 변환합니다. 이 상태는 Decoder에서 사용되며, 문장의 정보를 압축하여 표현합니다.
• Attention Layer:
  • Attention layer는 Decoder의 hidden state와 Encoder의 hidden state를 비교하여 가중치(importance scores)를 계산합니다.
  • 이 가중치들은 각각의 입력 단어가 출력에 얼마나 중요한지를 나타냅니다. 각 단어에 가중치를 부여하여, 더 중요한 단어에 더 많은 비중을 두고 예측을 할 수 있습니다.
• Decoder:
  • Decoder는 Attention 문맥 벡터와 이전 출력으로부터 다음 단어를 예측합니다.
  • Attention 메커니즘 덕분에, Decoder는 매 시점에서 문장의 다양한 부분에 동적으로 집중할 수 있습니다.

---

요약

• Seq2Seq 모델은 입력 문장을 고정된 벡터로 압축하는 방식에 한계가 있지만, Attention 메커니즘을 도입하여 이 문제를 해결할 수 있습니다.
• Bahdanau Attention은 Encoder와 Decoder의 hidden state를 결합하여 Attention 점수를 계산하고, Luong Attention은 더 간단한 Dot-product 방식으로 점수를 계산합니다.
• Seq2Seq + Attention 모델은 Encoder가 입력 문장을 요약하고, Attention Layer가 이 요약과 Decoder의 상태를 비교하여 중요한 정보를 동적으로 반영합니다. 그 후 Decoder는 이 정보를 기반으로 다음 단어를 예측합니다.

 

---

 

Transformer 구현 순서

1. 필요한 라이브러리 임포트
   • 필요한 라이브러리들을 import합니다.
2. Multi-Head Attention 구현
   • Multi-Head Attention 모듈을 구현합니다.
     • 여러 개의 Attention을 병렬로 사용하여 정보를 처리합니다.
3. Encoder Layer 구현
   • Multi-Head Attention을 사용하여 Encoder의 Attention을 구현합니다.
   • FeedForward Network를 추가하여 문맥을 처리합니다.
   • Layer Normalization과 Dropout을 적용하여 안정적인 학습을 도와줍니다.
4. Decoder Layer 구현
   • Self-Attention을 적용하여 Decoder 내부에서의 Attention을 처리합니다.
   • Encoder-Decoder Attention을 사용하여 Encoder의 정보와 Decoder의 정보를 결합합니다.
   • FeedForward Network를 추가하고, Layer Normalization과 Dropout을 적용합니다.
5. Transformer 모델 전체 구현
   • 임베딩을 통해 단어를 벡터로 변환합니다.
   • 포지셔널 인코딩을 통해 단어의 위치 정보를 모델에 제공하여 순서 정보가 반영되도록 합니다.
   • 여러 Encoder Stack을 쌓아 Encoder를 구성합니다.
   • 여러 Decoder Stack을 쌓아 Decoder를 구성합니다.
   • 출력층을 통해 최종적인 예측 결과(logit)를 출력합니다.
   • forward 함수:
     • 마스크 생성: Masking을 통해 PAD 토큰을 무시하고, 디코더에서는 미래의 정보를 차단합니다.
     • 임베딩 + 포지셔널 인코딩: 임베딩된 단어에 위치 정보를 더합니다.
     • Encoder 처리: Encoder Stack을 통해 입력을 처리합니다.
     • Decoder 처리: Decoder Stack을 통해 출력 예측을 처리합니다.
     • 최종출력층: 최종적으로 logits을 출력하여 예측을 만듭니다.

 

---

 

Multi-Head Attention 구현

Multi-Head Attention 클래스

class MutliHeadAttention(nn.Module):
  def __init__(self,d_model, num_header):
    super().__init__()
    assert d_model % num_header == 0  # 헤드당 차원이 같아야 함
    self.d_model = d_model   # 전체 임베딩 차원
    self.num_header = num_header  # 헤드의 개수
    self.d_k = d_model // num_header  # 헤드당 임베딩 차원

    # 각 query, key, value를 위한 선형 변환
    self.W_q = nn.Linear(d_model, d_model)
    self.W_k = nn.Linear(d_model, d_model)
    self.W_v = nn.Linear(d_model, d_model)

    # 다중 헤드 결과를 합친 후 다시 변환하는 선형 계층
    self.W_o = nn.Linear(d_model, d_model)

설명

1. d_model: 전체 모델의 임베딩 차원. 예를 들어, d_model = 512일 수 있습니다.
2. num_header: Attention의 헤드 개수. 예를 들어, num_header = 8일 수 있습니다.
3. d_k: 각 헤드의 차원 수. d_k = d_model // num_header로 계산됩니다.
4. W_q, W_k, W_v: 각각 Query, Key, Value에 대해 선형 변환을 적용하는 가중치 행렬입니다.
5. W_o: 다중 헤드 Attention의 출력을 결합한 후 최종 출력을 계산하는 선형 변환입니다.

forward 함수

def forward(self, Q, K, V, mask=None):
  batch_size = Q.size(0)  # 현재 배치크기

  # query, key, value에 선형 변환을 적용하고 차원 변경
  Q = self.W_q(Q).view(batch_size, -1, self.num_header, self.d_k).transpose(1, 2)
  K = self.W_k(K).view(batch_size, -1, self.num_header, self.d_k).transpose(1, 2)
  V = self.W_v(V).view(batch_size, -1, self.num_header, self.d_k).transpose(1, 2)

  # scaled dot-product attention 계산
  scores = torch.matmul(Q, K.transpose(-2, -1)) / np.sqrt(self.d_k)  # Q와 K의 내적
  if mask is not None:  # 마스크가 있을 경우 처리
    scores = scores.masked_fill(mask == 0, -1e9)  # mask가 0인 위치에 -1e9를 넣어 softmax 이후 0이 되게 함
  attn_weight = torch.softmax(scores, dim=-1)  # attention weight 계산

  # attention weight과 value를 곱해서 최종 context vector 계산
  context = torch.matmul(attn_weight, V)

  # 여러 헤드를 합침
  context = context.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)

  # 출력에 대해 선형 변환
  output = self.W_o(context)

  return output, attn_weight

설명

1. Q, K, V 선형 변환 및 차원 변경:
   • Q, K, V는 각각 Query, Key, Value를 의미합니다.
   • W_q, W_k, W_v를 사용하여 각 텐서에 선형 변환을 적용하고, 각 텐서를 (batch_size, num_header, seq_len, d_k) 형태로 변경합니다. 이때 num_header 차원은 각 Attention 헤드를 나타냅니다.
   • transpose(1, 2)는 각 헤드에 대한 차원을 올바르게 맞추기 위해 수행합니다.
2. Scaled Dot-Product Attention 계산:
   • Q와 K의 내적을 계산하여, attention scores를 구합니다.
   • scores를 sqrt(d_k)로 나누어 scaling을 수행합니다. 이는 내적 값이 너무 커지지 않도록 조정하는 과정입니다.
3. Mask 처리:
   • Mask가 제공되면, masked_fill을 사용하여 마스크된 부분을 매우 작은 값(-1e9)으로 설정해 softmax 후 attention이 적용되지 않도록 합니다.
4. Softmax:
   • scores에 대해 softmax를 적용하여, 각 단어가 다른 단어에 얼마나 집중할지를 결정하는 attention weights를 구합니다.
5. Context Vector:
   • attn_weight와 V를 곱하여, 각 단어에 대한 Context Vector를 계산합니다.
6. 헤드 결합:
   • 여러 개의 헤드를 처리한 후, 그 결과를 합쳐서 (batch_size, seq_len, d_model) 형태로 변환합니다. 이때, transpose 및 contiguous()와 같은 텐서 변환을 통해 차원을 정리합니다.
7. 최종 출력 계산:
   • 다중 헤드 Attention을 통해 나온 Context Vector를 W_o를 통해 선형 변환하여 최종 출력을 계산합니다.

---

핵심 개념

• Multi-Head Attention은 여러 개의 독립적인 Attention 헤드를 사용하여, 입력 데이터의 다양한 부분에 대해 여러 가지 관계를 동시에 학습합니다. 이렇게 함으로써 모델은 다양한 문맥을 한 번에 처리할 수 있습니다.
• Attention Weight는 각 단어가 다른 단어에 얼마나 집중할지를 결정하는 확률 값입니다.
• Masking은 특정 단어에 대한 Attention을 계산하지 않도록 마스크를 적용하여 처리합니다.

Multi-Head Attention을 통해 더 풍부한 표현을 학습하고, 문장의 여러 부분에 동시에 주의를 기울일 수 있습니다.

 

---

 

Encoder Layer

class EncoderLayer(nn.Module):
  '''
  Args:
    d_model : 전체 임베딩 차원
    num_head : 병렬로 수행할 Attention 헤드 수
    d_ff : Feed Forward Network의 은닉층 차원
  '''
  def __init__(self, d_model, num_head, d_ff, dropout=0.1) -> None:
    super().__init__()
    # MultiHeadAttention 객체 생성
    self.mha = MutliHeadAttention(d_model, num_head)
    # FeedForward Network (FFN)
    self.ffn = nn.Sequential(
        nn.Linear(d_model, d_ff),   # d_model 차원 -> d_ff 차원으로 변환
        nn.ReLU(),                  # 활성화 함수
        nn.Linear(d_ff, d_model)    # d_ff 차원 -> d_model 차원으로 변환
    )
    # Layer Normalization 설정
    self.layernormal1 = nn.LayerNorm(d_model)
    self.layernormal2 = nn.LayerNorm(d_model)
    # Dropout 설정
    self.dropout = nn.Dropout(dropout)

  def forward(self, x, mask=None):
    # MultiHeadAttention을 통해 attention 값과 가중치를 계산
    attn_output, attn_weights = self.mha(x, x, x, mask)
    # Residual Connection + Layer Normalization + Dropout
    x = self.layernormal1(x + self.dropout(attn_output))
    # FeedForward Network를 통과시킴
    ffn_output = self.ffn(x)
    # Residual Connection + Layer Normalization + Dropout
    x = self.layernormal2(x + self.dropout(ffn_output))
    return x, attn_weights

설명:

1. EncoderLayer는 Transformer의 인코더에 해당하는 한 레이어입니다.
2. MultiHeadAttention과 FeedForward Network (FFN)을 결합하여, 입력 텍스트에 대한 처리를 합니다.
3. Residual Connection: 각 레이어는 입력값을 그대로 다음 레이어로 전달하고, 그 위에 현재 레이어의 출력을 더해주어 Vanishing Gradient 문제를 완화합니다.
4. Layer Normalization은 출력값을 정규화하여 안정성을 높여줍니다.
5. Dropout은 과적합을 방지하기 위해 적용됩니다.

 

Decoder Layer

class DecoderLayer(nn.Module):
  def __init__(self, d_model, num_heads, d_ff, dropout=0.1):
    super().__init__()
    # 첫 번째 MultiHeadAttention: 디코더 자신의 출력에 대한 self-attention
    self.mha1 = MutliHeadAttention(d_model, num_heads) # 디코더는 자기자신을 참조
    # 두 번째 MultiHeadAttention: 디코더가 인코더의 출력과 비교하여 attention 수행
    self.mha2 = MutliHeadAttention(d_model, num_heads) # 소스 문장의 정보 참조
    # FeedForward Network (FFN)
    self.ffn = nn.Sequential(
        nn.Linear(d_model, d_ff),
        nn.ReLU(),
        nn.Linear(d_ff, d_model)
    )
    # Layer Normalization 설정
    self.layernormal1 = nn.LayerNorm(d_model)
    self.layernormal2 = nn.LayerNorm(d_model)
    self.layernormal3 = nn.LayerNorm(d_model)
    # Dropout 설정
    self.dropout = nn.Dropout(dropout)

  def forward(self, x, enc_output, src_mask=None, tgt_mask=None):
    # 첫 번째 Self-Attention (디코더의 입력에 대해 자기자신을 참조)
    attn_output, attn_weights1 = self.mha1(x, x, x, tgt_mask)
    # Residual Connection + Layer Normalization + Dropout
    x = self.layernormal1(x + self.dropout(attn_output))
    # 두 번째 Attention (디코더의 출력과 인코더의 출력을 비교)
    attn_output, attn_weights2 = self.mha2(x, enc_output, enc_output, src_mask)
    # Residual Connection + Layer Normalization + Dropout
    x = self.layernormal2(x + self.dropout(attn_output))
    # FeedForward Network 통과
    ffn_output = self.ffn(x)
    # Residual Connection + Layer Normalization + Dropout
    x = self.layernormal3(x + self.dropout(ffn_output))
    return x, attn_weights1, attn_weights2

설명:

1. DecoderLayer는 Transformer의 디코더에 해당하는 한 레이어입니다.
2. Self-Attention: 디코더 내부에서 자신의 출력을 참조하며, 미래 단어를 보지 않도록 masking을 합니다.
3. Encoder-Decoder Attention: 디코더는 인코더에서 나온 출력을 참조하여 해당 정보를 기반으로 출력을 생성합니다.
4. FeedForward Network (FFN)을 통과시켜 더 높은 수준의 특징을 추출합니다.

 

---

 

Transformer 모델

class Transformer(nn.Module):
  '''
  Args:
    src_vocab_size, tgt_vocab_size : 단어 사전 크기
    d_model : 전체 임베딩 차원
    num_heads : 병렬로 수행할 Attention 헤드 수
    d_ff : Feed Forward Network의 은닉층 차원
    num_layers : Encoder와 Decoder의 레이어 수
  '''
  def __init__(self, src_vocab_size, tgt_vocab_size, d_model=128, num_heads=4, d_ff=512, num_layers=2, dropout=0.1):
      super(Transformer, self).__init__()
      
      # Source와 Target 입력에 대한 임베딩 레이어 생성
      self.src_embedding = nn.Embedding(src_vocab_size, d_model)
      self.tgt_embedding = nn.Embedding(tgt_vocab_size, d_model)
      
      # Positional Encoding을 생성 (순서 정보 추가)
      self.positional_encoding = self.create_positional_encoding(max_len=100, d_model=d_model)
      
      # Encoder와 Decoder 레이어 스택 생성
      self.encoder_layers = nn.ModuleList([EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)])
      self.decoder_layers = nn.ModuleList([DecoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)])
      
      # 출력층 (디코더의 출력을 단어사전 크기만큼 변환)
      self.fc = nn.Linear(d_model, tgt_vocab_size)
      
      # Dropout 레이어
      self.dropout = nn.Dropout(dropout)

  def create_positional_encoding(self, max_len, d_model):  # Attention is All You Need
      # Positional Encoding 생성 (각 단어에 순서 정보를 제공)
      pe = torch.zeros(max_len, d_model)
      position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)  # [max_len, 1]
      div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-np.log(10000.0) / d_model))  # 주파수 설정
      pe[:, 0::2] = torch.sin(position * div_term)  # 짝수 인덱스는 sin
      pe[:, 1::2] = torch.cos(position * div_term)  # 홀수 인덱스는 cos
      return pe.to(device)  # 계산 장치(device)에 맞게 반환

  def forward(self, src, tgt):
      # Masks 생성
      src_mask = (src != 0).unsqueeze(1).unsqueeze(2)  # [batch, 1, 1, src_len] : 패딩이 아닌 부분에만 어텐션을 적용
      tgt_mask = self.create_subsequent_mask(tgt.size(1)).to(device)  # [1, tgt_len, tgt_len] : 디코더의 미래 단어를 참조하지 않도록 마스킹

      # 임베딩 + 위치 인코딩
      # Source와 Target에 Positional Encoding 추가
      pos_enc = self.positional_encoding[:src.size(1), :].unsqueeze(0)
      src = self.src_embedding(src) + pos_enc  # 입력 텍스트에 대한 임베딩 + 순서 정보
      tgt_enc = self.positional_encoding[:tgt.size(1), :].unsqueeze(0)
      tgt = self.tgt_embedding(tgt) + tgt_enc  # 타겟 텍스트에 대한 임베딩 + 순서 정보

      # Dropout 적용
      src = self.dropout(src)
      tgt = self.dropout(tgt)

      # Encoder 처리 (여러 Encoder Layer를 순차적으로 통과)
      enc_attn_weights = []
      for layer in self.encoder_layers:
          src, attn_weights = layer(src, src_mask)
          enc_attn_weights.append(attn_weights)

      # Decoder 처리 (여러 Decoder Layer를 순차적으로 통과)
      dec_attn_weights = []
      for layer in self.decoder_layers:
          tgt, attn_weights1, attn_weights2 = layer(tgt, src, src_mask, tgt_mask)
          dec_attn_weights.append(attn_weights2)  # Encoder-Decoder Attention (디코더가 인코더의 출력을 참조)

      # 출력층 (디코더 출력을 단어사전 크기만큼 변환하여 최종 예측)
      logits = self.fc(tgt)  # [batch, seq_len, tgt_vocab_size]
      return logits, enc_attn_weights, dec_attn_weights

  def create_subsequent_mask(self, size):
      # 디코더에서 미래 단어를 참조하지 않도록 마스크 생성
      mask = torch.triu(torch.ones(size, size), diagonal=1).bool()  # 상삼각 행렬 생성
      return ~mask.unsqueeze(0)  # [1, size, size] : 마스크 반전 (True가 마스킹되게)

주요 개념 설명:

1. src_embedding과 tgt_embedding:
   • src_embedding과 tgt_embedding은 각각 입력 텍스트와 타겟 텍스트에 대한 임베딩을 수행합니다. 단어 사전 크기와 임베딩 차원(d_model)을 기반으로 임베딩 레이어를 설정합니다.
2. positional_encoding:
   • Transformer는 순서 정보가 없기 때문에 위치 인코딩을 추가하여 각 단어가 문장에서 어떤 위치에 있는지 정보를 제공합니다. 위치 인코딩은 사인과 코사인 함수로 주어진 차원에서 계산됩니다.
3. encoder_layers와 decoder_layers:
   • 각각 Encoder와 Decoder의 여러 층을 생성하여 ModuleList로 관리합니다. 각 레이어는 MultiHeadAttention, Feed Forward Network, Layer Normalization 등을 포함합니다.
4. create_subsequent_mask:
   • 이 함수는 디코더에서 미래 단어를 참조하지 않도록 마스크를 생성합니다. 마스크는 상삼각 행렬 형태로, 현재 단어보다 뒤에 있는 단어들은 참조하지 않도록 설정됩니다.
5. forward:
   • 입력 데이터를 임베딩하고 위치 인코딩을 추가합니다.
   • Encoder와 Decoder를 순차적으로 처리하며, 각 레이어에서 생성되는 attention 가중치들을 저장합니다.
   • 마지막으로 출력층에서 디코더의 출력을 기반으로 최종 예측값을 계산합니다.

 

---

 

Transformer 모델 학습을 위한 기본적인 설정을 제공

# Dataset 초기화: 데이터프레임(df), 영어 단어 인덱스(en_word2idx), 한국어 단어 인덱스(ko_word2idx)
dataset = TranslationDataset(df, en_word2idx, ko_word2idx)
# DataLoader를 사용하여 데이터를 배치 단위로 처리 (배치 크기 8, 셔플 적용)
dataloader = DataLoader(dataset, batch_size=8, shuffle=True)

# Transformer 모델 초기화 (영어-한국어 번역 모델)
model = Transformer(
    src_vocab_size=len(en_word2idx),  # 영어 단어사전 크기
    tgt_vocab_size=len(ko_word2idx),  # 한국어 단어사전 크기
    d_model=128,  # 임베딩 차원
    num_heads=4,  # 병렬 Attention 헤드 수
    d_ff=512,  # Feed Forward Network의 은닉층 차원
    num_layers=2  # Encoder와 Decoder 레이어 수
).to(device)  # 모델을 장치(device)로 이동 (GPU 또는 CPU)

# 손실 함수 설정: CrossEntropyLoss, PAD 토큰은 무시 (ko_word2idx["<PAD>"])
criterion = nn.CrossEntropyLoss(ignore_index=ko_word2idx["<PAD>"])

# 옵티마이저 설정: Adam, 학습률은 0.0001
optimizer = optim.Adam(model.parameters(), lr=0.0001)

# 학습 설정
num_epochs = 10  # 총 학습 epoch 수
train_losses = []  # 학습 중 손실값 기록용 리스트

model.train()  # 모델을 학습 모드로 설정
for epoch in range(num_epochs):
    epoch_loss = 0  # 한 epoch의 손실을 누적할 변수
    for src, tgt in dataloader:
        # 데이터를 장치(device)로 이동
        src, tgt = src.to(device), tgt.to(device)

        # 순전파: 모델에 입력 데이터를 넣어 예측값(logits) 계산
        logits, _, _ = model(src, tgt)  # 모델의 출력: logits (배치 크기, 시퀀스 길이, vocab 크기)

        # 손실 계산: CrossEntropyLoss에 맞게 logits과 tgt를 reshape하고 loss 계산
        loss = criterion(logits.view(-1, logits.size(-1)), tgt.contiguous().view(-1))

        # 역전파: 기울기 초기화, 역전파 실행, 가중치 업데이트
        optimizer.zero_grad()  # 이전 기울기 초기화
        loss.backward()  # 역전파
        optimizer.step()  # 가중치 업데이트

        # epoch_loss에 현재 배치의 손실값을 더해줌
        epoch_loss += loss.item()

    # 한 epoch의 평균 손실값 계산
    avg_loss = epoch_loss / len(dataloader)
    train_losses.append(avg_loss)  # 평균 손실값을 기록
    # epoch의 손실값 출력
    print(f"Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.4f}")

주요 개념 설명:

1. TranslationDataset:
   • 주어진 데이터프레임(df), 영어 단어 인덱스(en_word2idx), 한국어 단어 인덱스(ko_word2idx)를 기반으로 번역 데이터셋을 생성합니다.
2. DataLoader:
   • 데이터셋을 배치 단위로 나누고, 배치 크기(batch_size=8)와 셔플 옵션(shuffle=True)을 통해 데이터가 무작위로 섞여서 학습에 사용되도록 합니다.
3. 모델 초기화:
   • Transformer 모델을 초기화할 때, 단어 사전 크기와 임베딩 차원, Attention 헤드 수, Feed Forward Network의 크기 등을 설정합니다. 모델을 학습에 사용할 장치(device)로 이동합니다.
4. 손실 함수 (CrossEntropyLoss):
   • CrossEntropyLoss는 다중 클래스 분류에서 자주 사용되는 손실 함수입니다. 여기서는 출력값(logits)과 실제 타겟(tgt) 간의 차이를 계산합니다. PAD 토큰은 학습에 영향을 미치지 않도록 무시됩니다.
5. 옵티마이저 (Adam):
   • Adam 옵티마이저는 기울기를 기반으로 가중치를 업데이트하는 알고리즘입니다. 학습률(lr=0.0001)을 설정하여 파라미터를 업데이트합니다.
6. 학습 과정:
   • 학습 과정은 총 num_epochs 동안 반복됩니다.
   • 각 배치에 대해 순전파, 손실 계산, 역전파 및 가중치 업데이트를 진행하며, epoch마다 평균 손실을 출력합니다.

 

---

 

학습 손실 값 시각화

import matplotlib.pyplot as plt

# 학습 손실 시각화
plt.plot(train_losses)
plt.title("Training Loss")
plt.xlabel("Epochs")
plt.ylabel("Loss")
plt.show()

• plt.plot(train_losses) 코드를 실행하면 train_losses 리스트에 저장된 학습 손실 값을 그래프 형태로 시각화할 수 있습니다. 이는 모델이 학습을 진행하면서 손실 값이 어떻게 변화하는지 확인하는 데 유용합니다.