pytorch 공부 (3) - RNN 구현하기

2024.03.12 - [데이터 분석/딥러닝] - pytorch 공부 (3) - CNN 구현하기

IMBD 데이터

`torchtext==0.17.0`이라 legacy가 없어져 github에 적힌 코드를 사용하였다.

import torch
from torch.nn.utils.rnn import pad_sequence
from torch.utils.data import random_split
from torch.utils.data import DataLoader

from torchtext.datasets import IMDB
from torchtext.data.functional import to_map_style_dataset
from torchtext.data.utils import get_tokenizer
from torchtext.vocab import build_vocab_from_iterator

USE_CUDA = torch.cuda.is_available()
DEVICE = torch.device("cuda" if USE_CUDA else "cpu")

tokenizer = get_tokenizer('basic_english')
train_iter, test_iter = IMDB(root='.data', split=('train', 'test'))

def train_valid_split(train_iterator, split_ratio=0.8, seed=42):
    train_count = int(split_ratio * len(train_iterator))
    valid_count = len(train_iterator) - train_count
    generator = torch.Generator().manual_seed(seed)
    train_set, valid_set = random_split(
        train_iterator, lengths=[train_count, valid_count],
        generator=generator)
    return train_set, valid_set

train_iter = to_map_style_dataset(train_iter)
test_iter = to_map_style_dataset(test_iter)

train_set, val_set = train_valid_split(train_iter)

def yield_tokens(data_iter):
    for _, text in data_iter:
        yield tokenizer(text)

vocab = build_vocab_from_iterator(
    iterator=yield_tokens(train_iter),
    min_freq=5,
    specials=['<unk>'],)
vocab.set_default_index(vocab['<unk>'])

def collate_batch(batch):
    text_pipeline = lambda x: vocab(tokenizer(x))
    label_pipeline = lambda x: int(x)

    label_list, text_list = [], []
    for (_label, _text) in batch:
        label_list.append(label_pipeline(_label))
        processed_text = torch.tensor(text_pipeline(_text), dtype=torch.int64)
        text_list.append(processed_text)
    label_list = torch.tensor(label_list, dtype=torch.int64)
    text_tensor = pad_sequence(text_list, padding_value=1, batch_first=True)
    return text_tensor, label_list

train_dataloader = DataLoader(
    train_set, batch_size=64, shuffle=True, collate_fn=collate_batch)
val_dataloader = DataLoader(
    val_set, batch_size=64, shuffle=True, collate_fn=collate_batch)
test_dataloader = DataLoader(
    test_iter, batch_size=64, shuffle=True, collate_fn=collate_batch)

 

len(train_set), len(val_set) # (20000, 5000)

len(train_dataloader), len(val_dataloader), len(test_dataloader) # 배치 사이즈 64
# (313, 79, 391)

위에서 배치 사이즈를 64로 했기 때문에 각 dataloader 사이즈는 그만큼 줄어들었다.

 

 

RNN 구현하기

GRU 모델 정의

import torch.nn as nn

class Gru(nn.Module):
    def __init__(self, n_layers, hidden_dim, n_vocab, embed_dim, n_classes, dropout_p=0.2):
        """
        n_layers: 히든 레이어 개수
        hidden_dim: 히든 레이어 차원
        n_vocab: 사전 사이즈
        embed_dim: 임베딩된 데이터의 차원
        n_classes: 레이블 수
        dropout_p: 드랍아웃 비율
        """
        super(Gru, self).__init__()
        self.n_layers = n_layers
        self.hidden_dim = hidden_dim
        self.embed = nn.Embedding(n_vocab, embed_dim)
        self.dropout = nn.Dropout(dropout_p)
        self.gru = nn.GRU(embed_dim, self.hidden_dim, num_layers=self.n_layers, batch_first=True)
        self.out = nn.Linear(self.hidden_dim, n_classes)
        
    def forward(self, x):
        x = self.embed(x) # 텍스트를 단어 단위인 토큰으로 벡터 변환
        h_0 = self._init_state(batch_size=x.size(0)) # 아래 함수 참고
        x, _ = self.gru(x, h_0) # GRU의 리턴 값(배치 사이즈, 입력값 길이, 히든 스테이트의 길이) -> 텐서 형태
        h_t = x[:, -1, :] # 텐서로 크기가 변경, 마지막 히든 스테이트만 가져옴
        self.dropout(h_t)
        logit = self.out(h_t) # 배치 사이즈와 히든 스테이트의 크기 -> 배치 사이즈의 출력층의 크기로 변환
        return logit
    
    def _init_state(self, batch_size=1):
        "첫 번째 히든 스테이트를 0 벡터로 초기화"
        weight = next(self.parameters()).data # 첫 가중치 데이터 추출(텐서)
        return weight.new(self.n_layers, batch_size, self.hidden_dim).zero_() # 현재 모델의 가중치와 같은 모양의 텐서의 값을 0으로 초기화

 

모델 학습 및 검증

n_classes = 2
vocab_size = len(vocab)
lr = 0.001
EPOCHS = 10

model = Gru(1, 256, vocab_size, 128, n_classes, 0.5).to(DEVICE)
optimizer = torch.optim.Adam(model.parameters(), lr=lr)

import torch.nn.functional as f

for epoch in range(1, EPOCHS+1):
    model.train()
    for i, (text, label) in enumerate(train_dataloader):
        text, label = text.to(DEVICE), label.to(DEVICE)
        optimizer.zero_grad()
        output = model(text)
        
        label.data.sub_(1) # <unk>:0 인 token 값 제거
        loss = f.cross_entropy(output, label)
        loss.backward()
        optimizer.step()
        
        if i == 100: break
    
    model.eval()

    val_loss_sum = 0
    val_correct = 0

    with torch.no_grad():
        for text, label in val_dataloader:
            text, label = text.to(DEVICE), label.to(DEVICE)
            label.data.sub_(1) # <unk>:0 인 token 값 제거
            output = model(text)
            val_loss_sum += f.cross_entropy(output, label, reduction='sum').item()
            pred = output.max(1)[1].view(label.size()).data
            val_correct += (pred == label.data).sum()

    val_loss = val_loss_sum / len(val_set)
    val_acc = val_correct / len(val_set) * 100
    
    print(f'[Epoch: {epoch:2d}] train loss: {loss.item():.4f} | val loss: {val_loss:.4f} val accuracy: {val_acc:.2f}%')

 

최적 모델 저장 및 EarlyStopping

위 코드에서 학습 부분과 평가 부분을 함수화하여 가독성을 높였다.

def train(model, optimizer, train_iter):
    model.train()
    for i, (text, label) in enumerate(train_iter):
        text, label = text.to(DEVICE), label.to(DEVICE)
        optimizer.zero_grad()
        output = model(text)
        
        label.data.sub_(1) # <unk>:0 인 token 값 제거
        loss = f.cross_entropy(output, label)
        loss.backward()
        optimizer.step()
            
    return loss

def evaluate(model, val_iter, len_val_data):
    model.eval()

    loss_sum = 0
    correct = 0

    with torch.no_grad():
        for text, label in val_iter:
            text, label = text.to(DEVICE), label.to(DEVICE)
            label.data.sub_(1) # <unk>:0 인 token 값 제거
            output = model(text)
            loss_sum += f.cross_entropy(output, label, reduction='sum').item()
            pred = output.max(1)[1].view(label.size()).data
            correct += (pred == label.data).sum()

    loss = loss_sum / len_val_data
    acc = correct / len_val_data * 100
    
    return loss, acc

class EarlyStopping:
    def __init__(self, patience=5):
        self.loss = float('inf')
        self.patience = 0
        self.patience_limit = patience
        
    def step(self, loss):
        if self.loss > loss:
            self.loss = loss
            self.patience = 0
        else:
            self.patience += 1
    
    def is_stop(self):
        return self.patience >= self.patience_limit

lr = 0.001
EPOCHS = 50

model = Gru(1, 256, vocab_size, 128, n_classes, 0.5).to(DEVICE)
optimizer = torch.optim.Adam(model.parameters(), lr=lr)

best_model = None
best_val_loss = None

es = EarlyStopping(patience=10)

for epoch in range(1, EPOCHS+1):
    loss = train(model, optimizer, train_dataloader)
    
    val_loss, val_acc = evaluate(model, val_dataloader, len(val_set))
    
    print(f'[Epoch: {epoch:2d}] train loss: {loss.item():.4f} | val loss: {val_loss:.4f} val accuracy: {val_acc:.2f}%')
    
    es.step(val_loss)
    if es.is_stop():
        break
    
    if not best_val_loss or val_loss < best_val_loss:
        best_model = model
        best_val_loss = val_loss
        torch.save(model.state_dict(), 'bestmodel.pt')

- `EarlyStopping(patience=10)`: 검증 loss가 epoch 10회 동안 개선되지 않으면 학습을 종료

- `torch.save(model.state_dict(), 'bestmodel.pt')`: 모델의 파라미터를 `bestmodel.pt` 파일로 저장

- 5회 째의 loss가 가장 낮고 이후로 개선되지 않아 15회에서 학습 종료

 

모델 테스트

test_loss, test_acc = evaluate(best_model, test_dataloader, len(test_iter))
print(f'test loss: {test_loss:.2f} | test accuracy: {test_acc:.2f}')

 

모델 불러오기

best_model_state = torch.load('bestmodel.pt')
best_model_loaded = Gru(1, 256, vocab_size, 128, n_classes, 0.5).to(DEVICE)
best_model_loaded.load_state_dict(best_model_state)

test_loss, test_acc = evaluate(best_model_loaded, test_dataloader, len(test_iter))
print(f'test loss: {test_loss:.2f} | test accuracy: {test_acc:.2f}')

- 파라미터만 저장하였기 때문에 모델 자체는 선언이 되어야 한다.

- `best_model_state = torch.load('bestmodel.pt')`: 앞서 저장했던 파라미터 불러오기

- `best_model_loaded.load_state_dict(best_model_state)`: 모델에 파라미터 주입

분명 위와 아래는 같은 모델인데 왜 loss와 정확도가 다르게 나오는지 그 이유를 모르겠다.