Seq2Seq 실습1

1. 이전 코드

import random
from tqdm import tqdm

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader


class Encoder(nn.Module):
    def __init__(self, input_dim, emb_dim, hidden_dim, n_layers, dropout):
        '''
        input_dim: 단어의 개수
        emb_dim: 임베딩 차원
        hidden_dim: gru의 h의 차원
        n_layers: gru의 층의 수
        dropout: dropout 비율
        '''
        super().__init__()
        self.embedding = nn.Embedding(input_dim, emb_dim) # 임베딩 층
        self.dropout = nn.Dropout(dropout)     # 드랍 아웃
        self.gru = nn.GRU(input_size=emb_dim,  # GRU
                          hidden_size=hidden_dim,
                          num_layers=n_layers,
                          batch_first=True,
                          dropout=dropout)

    def forward(self, src):
        '''
        src: 인코더에 입력되는 문장
        '''
        embedded = self.dropout(self.embedding(src))  # 임베딩 층, 드롭아웃 적용
        outputs, hidden = self.gru(embedded)  # GRU 층

        return outputs, hidden  # 인코더 출력


class Decoder(nn.Module):
    def __init__(self, output_dim, emb_dim, hidden_dim, n_layers, dropout):
        '''
        output_dim: 출력 단어의 개수 (타겟 어휘 크기)
        emb_dim: 임베딩 차원
        hidden_dim: GRU의 h의 차원
        n_layers: GRU의 층의 수
        dropout: 드롭아웃 비율
        '''
        super().__init__()
        self.output_dim = output_dim # 단어사전 개수
        self.embedding = nn.Embedding(output_dim, emb_dim) # 임베딩 층
        self.gru = nn.GRU(input_size=emb_dim, # GRU 층
                          hidden_size=hidden_dim,
                          num_layers=n_layers,
                          batch_first=True,
                          dropout=dropout if n_layers > 1 else 0)
        self.fc_out = nn.Linear(hidden_dim, output_dim) # 완전 연결 층
        self.dropout = nn.Dropout(dropout) # 드랍아웃 층

    def forward(self, trg, hidden):
        '''
        trg: 타겟 언어의 이전단어
        hidden: 컨텍스트 벡터
        '''
        trg = trg.unsqueeze(1)  # 입력 차원 증가: [batch_size, 1]
        embedded = self.dropout(self.embedding(trg))  # 드롭아웃 적용 임베딩
        output, hidden = self.gru(embedded, hidden)  # GRU 실행
        prediction = self.fc_out(output.squeeze(1))  # 선형 레이어를 통한 로짓 계산
        prediction = F.softmax(prediction, dim=1)  # 소프트맥스 적용

        return prediction, hidden


class Seq2Seq(nn.Module):
    def __init__(self, encoder, decoder, device, trg_vocab_size, tokens):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder
        self.device = device
        self.sos_token = tokens['sos_token']
        self.eos_token = tokens['eos_token']
        self.trg_vocab_size = trg_vocab_size

    def forward(self, src, trg=None, teacher_force_ratio=1.0, max_len=50):
        _, hidden = self.encoder(src)  # 인코더 실행, 컨텍스트 벡터 준비

        # 첫 디코더 입력을 <sos> 토큰으로 설정
        current_token = torch.tensor([self.sos_token] * src.size(0)).to(self.device)

        # 최종 출력값 설정
        batch_size = src.shape[0]
        outputs = torch.zeros(batch_size, max_len, self.trg_vocab_size).to(self.device)
        all_tokens = torch.zeros(batch_size, max_len).to(self.device)

        for t in range(1, max_len):
            output, hidden = self.decoder(current_token, hidden)
            outputs[:, t, :] = output
            top1 = output.argmax(1)
            all_tokens[:, t] = top1

            if teacher_force_ratio > 0 and trg is not None and t < trg.size(1):
                current_token = trg[:, t] if random.random() < teacher_force_ratio else top1
            else:
                current_token = top1

            # 종료 토큰을 만난 경우 디코딩 종료
            if (current_token == self.eos_token).all():
                break

        # 실제로 사용된 시퀀스 길이를 구합니다.
        actual_lengths = (all_tokens != self.eos_token).sum(dim=1)

        return outputs, actual_lengths

 

---

2. 모델 학습

 

※ 모델 학습 순서

1) 옵티마이저 초기화
2) 모델의 예측
3) Loss 값 계산
4) 역전파 실행
5) 모델의 가중치 업데이트
6) 위 행동 반복

 

def train_model(model, train_dataloader, optimizer, criterion, clip, device):
    model.train()
    epoch_loss = 0
    progress_bar = tqdm(train_dataloader, desc='Training', leave=False)

    for src, trg in progress_bar:
        src, trg = src.to(device), trg.to(device)

        optimizer.zero_grad()
        output, actual_lengths = model(src, trg)

        # 출력 차원 변경: [batch_size, max_len, trg_vocab_size] -> [total_tokens, trg_vocab_size]
        output_dim = output.shape[-1]
        max_len = trg.shape[1]

        output = output[:, 1:max_len].reshape(-1, output_dim)  # <sos> 토큰 제거
        trg = trg[:, 1:max_len].reshape(-1)  # 타겟에서 <sos> 토큰 제거 및 길이 맞추기

        # 실제 사용된 시퀀스 길이만 손실 계산에 포함
        valid_idx = (trg != model.eos_token).nonzero(as_tuple=True)[0]
        output = output[valid_idx]
        trg = trg[valid_idx]

        loss = criterion(output, trg)
        loss.backward()

        # 그라디언트 폭발 방지
        torch.nn.utils.clip_grad_norm_(model.parameters(), clip)

        optimizer.step()
        epoch_loss += loss.item()

    return epoch_loss / len(train_dataloader)

 

---

3. 모델 평가

※ 모델 평가 순서

1) 모델의 예측

2) Loss 값 계산

 

def evaluate_model(model, test_dataloader, criterion, device):
    model.eval()
    epoch_loss = 0
    progress_bar = tqdm(test_dataloader, desc='Evaluating', leave=False)

    with torch.no_grad():
        for src, trg in progress_bar:
            src, trg = src.to(device), trg.to(device)
            
            # 티처 포싱 없이 모델 실행
            output, actual_lengths = model(src, trg, teacher_force_ratio=0)

            # 출력 차원 변경
            # [batch_size, max_len, trg_vocab_size] 
            #    ->  [total_tokens, trg_vocab_size]
            output_dim = output.shape[-1]
            max_len = trg.shape[1]

            output = output[:, 1:max_len].reshape(-1, output_dim)  # <sos> 토큰 제거
            trg = trg[:, 1:max_len].reshape(-1)  # 타겟에서 <sos> 토큰 제거 및 길이 맞추기

            # 실제 사용된 시퀀스 길이만 손실 계산에 포함
            valid_idx = (trg != model.eos_token).nonzero(as_tuple=True)[0]
            output = output[valid_idx]
            trg = trg[valid_idx]

            loss = criterion(output, trg)
            epoch_loss += loss.item()

    return epoch_loss / len(test_dataloader)

 

---

4. 예측

위에서 학습한 모델을 불러와 예측을 실행합니다.
예측을 진행할 때에는, trg (타겟변수) 값을 사용하지 않는 점을 주의하도록 합니다. 그리고 티처포싱(Teacher Forcing)은 적용하지 않도록 합니다.
마지막으로 aromax 함수를 이용해 해당 차원에서 가장 큰 값의 인덱스를 추출하도록 합니다. 인덱스는 토큰값을 의미하므로 단어를 예측할 때, 가장 확률이 높은 단어를 선택한다는 의미 입니다.

# 예측 함수
def predict(model, src, max_len, device):
    model.eval()
    with torch.no_grad():
        outputs, _ = model(src, trg=None, teacher_force_ratio=0, max_len=max_len)
        predictions = outputs.argmax(2)
        return predictions

---

5. 랜덤 데이터 생성

5.1. 연습용 데이터

정의한 모델을 학습하기 위해 연습용 데이터를 만듭니다. 데이터의차원은 아래와 같습니다.
[시퀀스 개수, 시퀀스의 길이]
시퀀스 개수: 시퀀스 개수는 데이터셋에 포함된 시퀀스의 총 개수를 의미합니다. 예를 들어, 텍스트 데이터의 경우 시퀀스 개수는 문장이나 문 단의 수가 될 수 있습니다. 만약 우리가 100개의 문장을 가지고 있다면, 시퀀스 개수는 100이 됩니다.
시퀀스의 길이는 각 시퀀스 내에 포함된 요소(단어, 음성 프레임 등)의 개수를 의미합니다. 예를 들어, 텍스트 데이터에서 시퀀스의 길이는 문 장 내 단어의 수가 될 수 있습니다.

5.2. 텐서 데이터셋

먼저 리스트로 구성되어 있던 데이터를, 텐서(Tensor)로 만들어 줍니다. 그리고 toroh.utila.data.TensorDataset 를 이용해 텐서를 데이터셋으로 만들어 줍니다.

 

5.3. 데이터 로더

마지막으로 toroh.util.data.DataLoader 객체를 만들어 학습할 준비를 해줍니다. DataLoader는 미니배치 단위로 데이터를 사용할 수 있도록 합니다.

import torch
from torch.utils.data import DataLoader, TensorDataset

# 연습용 데이터
src_data = [[1, 2, 3, 4, 5],
            [6, 7, 8, 9, 10],
            [11, 12, 13, 14, 15],]

trg_data = [[101, 102, 103, 104, 105],
            [106, 107, 108, 109, 110],
            [111, 112, 113, 114, 115],]

# 테스트 데이터
src_data_test = [[1, 2, 3, 4, 5],
                [6, 7, 8, 9, 10],
                [11, 12, 13, 14, 15],]

trg_data_test = [[101, 102, 103, 104, 105],
                [106, 107, 108, 109, 110],
                [111, 112, 113, 114, 115],]

# 텐서
src = torch.tensor(src_data, dtype=torch.long)
trg = torch.tensor(trg_data, dtype=torch.long)

src_test = torch.tensor(src_data_test, dtype=torch.long)
trg_test = torch.tensor(trg_data_test, dtype=torch.long)

# 데이터셋 만들기
dataset = TensorDataset(src, trg)
dataset_test = TensorDataset(src_test, trg_test)

# DataLoader 만들기
train_dataloader = DataLoader(dataset, batch_size=3, shuffle=True)
test_dataloader = DataLoader(dataset_test, batch_size=3, shuffle=False)

# Check the DataLoader
for src, trg in train_dataloader:
    print("Train batch:")
    print(src)
    print(trg)

for src, trg in test_dataloader:
    print("Train batch:")
    print(src)
    print(trg)

Train batch:
tensor([[11, 12, 13, 14, 15],
        [ 1,  2,  3,  4,  5],
        [ 6,  7,  8,  9, 10]])
tensor([[111, 112, 113, 114, 115],
        [101, 102, 103, 104, 105],
        [106, 107, 108, 109, 110]])
Train batch:
tensor([[ 1,  2,  3,  4,  5],
        [ 6,  7,  8,  9, 10],
        [11, 12, 13, 14, 15]])
tensor([[101, 102, 103, 104, 105],
        [106, 107, 108, 109, 110],
        [111, 112, 113, 114, 115]])

 

---

6. 코드 실행

6.1. 하이퍼 파라미터 정의

Seq2Seq 모델도 학습 방법은 다른 딥러닝 모델과 같습니다. 가장 먼저 모델을 학습하기 앞서 하이퍼파라미터를 정의합니다. 하이퍼파라미터에는 벡터의 차원 수, 드랍아웃 비율, 에포크 수 등이 있습니다. 그리고 사용할 GPU에 대해 설정하게 됩니다.

6.2. 객체 생성

그리고 학습에 필요한 모델, 옵티마이저, 손실함수를 정의합니다. 여기서 손실함수는 CrossEntropyLoss를 사용합니다. Seq2Seq 모델은 다중 클래스 분류 문제를 해결하는 모델입니다. 각 시점에서 가능한 다음 단어 후보가 여러 개 존재하기 때문에, CrossEntropyLoss는 이러한 다중 클래스 분류 문제를 해결하는 데 적합합니다.

 

6.3. 모델 학습과 예측

이 예제에서는 모델을 학습할 때 위에서 정의한 train_model, evaluate_model 메소드를 사용합니다. 그리고 전체 데이터에 대해 여러 에포크 만큼 진행하기 위해 for문을 이용해 줍니다.
모델 학습이 끝나면, 학습한 모델을 이용해 예측을 해봅니다. 이번 예제에서는 연습용 데이터를 이용합니다. 또한 위에서 작성한 predict 메소드를 이용해 모델을 실행해 봅니다.

 

import math

# 모델의 하이퍼파라미터 정의
INPUT_DIM = max(max(src_data)) + 1  # 단어 사전의 크기 (최대 인덱스 + 1)
OUTPUT_DIM = max(max(trg_data)) + 1  # 출력 단어 사전의 크기 (최대 인덱스 + 1)
ENC_EMB_DIM = 32  # 인코더 임베딩 차원
DEC_EMB_DIM = 32  # 디코더 임베딩 차원
HIDDEN_DIM = 64  # GRU 히든 상태 차원
N_LAYERS = 2  # GRU 레이어 수
ENC_DROPOUT = 0.5  # 인코더 드롭아웃 비율
DEC_DROPOUT = 0.5  # 디코더 드롭아웃 비율
N_EPOCHS = 10

tokens = {
    'sos_token': 0,  
    'eos_token': 1   
}

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HIDDEN_DIM, N_LAYERS, ENC_DROPOUT)
dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HIDDEN_DIM, N_LAYERS, DEC_DROPOUT)
model = Seq2Seq(enc, dec, device, OUTPUT_DIM, tokens).to(device)

optimizer = optim.Adam(model.parameters()) # Adam 옵티마이저
criterion = nn.CrossEntropyLoss() # 크로스엔트로피 손실 사용

for epoch in range(N_EPOCHS):
    # 모델 학습
    train_loss = train_model(model, train_dataloader, optimizer, criterion, 1.0, device)

    # 모델 평가
    test_loss = evaluate_model(model, test_dataloader, criterion, device)
    
    print(f'Epoch: {epoch+1:02}')
    print(f'\tTrain Loss: {train_loss:.3f} | Train PPL: {math.exp(train_loss):7.3f}')
    print(f'\t Val. Loss: {test_loss:.3f} |  Val. PPL: {math.exp(test_loss):7.3f}')

Epoch: 01
	Train Loss: 4.754 | Train PPL: 115.990
	 Val. Loss: 4.754 |  Val. PPL: 115.991

 

Epoch: 02
	Train Loss: 4.754 | Train PPL: 115.993
	 Val. Loss: 4.753 |  Val. PPL: 115.972

 

Epoch: 03
	Train Loss: 4.753 | Train PPL: 115.912
	 Val. Loss: 4.753 |  Val. PPL: 115.951

 

Epoch: 04
	Train Loss: 4.753 | Train PPL: 115.947
	 Val. Loss: 4.753 |  Val. PPL: 115.942

 

Epoch: 05
	Train Loss: 4.753 | Train PPL: 115.891
	 Val. Loss: 4.753 |  Val. PPL: 115.912

 

Epoch: 06
	Train Loss: 4.753 | Train PPL: 115.922
	 Val. Loss: 4.752 |  Val. PPL: 115.874

 

Epoch: 07
	Train Loss: 4.752 | Train PPL: 115.850
	 Val. Loss: 4.752 |  Val. PPL: 115.842

 

Epoch: 08
	Train Loss: 4.752 | Train PPL: 115.840
	 Val. Loss: 4.752 |  Val. PPL: 115.815

 

Epoch: 09
	Train Loss: 4.752 | Train PPL: 115.821
	 Val. Loss: 4.752 |  Val. PPL: 115.782

 

Epoch: 10
	Train Loss: 4.752 | Train PPL: 115.793
	 Val. Loss: 4.751 |  Val. PPL: 115.738

 

※ 예측

max_len = 10  # 최대 예측 길이 설정
src = [[1, 2, 3, 4, 5]]
src_example = torch.tensor(src, dtype=torch.long).to(device)

predictions = predict(model, src_example, max_len, device)
print(predictions)

tensor([[  0, 112, 112, 115, 115, 112, 115, 112, 115, 115]])