CHAPTER7 순차적인 데이터를 처리하는 RNN 1

'quick brown for jumps over the lazy dogs.'

['quick', 'brown', 'for', 'jumps', 'over', 'the', 'lazy','dogs.']

# 1. 필수 라이브러리 import
import os
import torch
import torch.nn as nn
import torch.nn.functional as F
from torchtext import data, datasets # 이번엔 자연어 데이터셋이라 torchtext import

# 2. 하이퍼 파라미터 정의
BATCH_SIZE = 64
lr = 0.001
EPOCHS = 40
USE_CUDA = torch.cuda.is_available()
DEVICE = torch.device("cuda" if USE_CUDA else "cpu")

# 3. IMDB 데이터셋 로딩 -> 텐서로 변환
TEXT = data.Field(sequential = True, batch_first = True, lower = True)
LABEL = data.Field(sequential = False, batch_first = True)

import torch
import torchtext

print(torch.__version__)
print(torchtext.__version__)

# 4. datasets 객체의 splits() 함수를 사용해 모델에 입력되는 데이터셋 생성
trainset, testset = datasets.IMDB.splits(TEXT, LABEL)

# 5. 워드 임베딩에 필요한 단어 사전 생성

# 최소 5번 이상 나온 단어만 사전에 넣음
# 5번 미만 출현 단어는 unk(=Unknown) 토큰으로 대체.
TEXT.build_vocab(trainset, min_freq = 5)
LABEL.build_vocab(trainset)

# 6. 검증셋 생성, 데이터를 한 번에 처리하기 위한 반복자 구현

trainset, valset = trainset.split(split_ratio = 0.8)
train_iter, val_iter, test_iter = data.BucketIterator.splits(
(trainset, valset, testset),
batch_size = BATCH_SIZE,
shuffle = True, repeat = False
)

# 7. 사전 속 단어들의 개수와 레이블의 수 정해주는 변수 생성
vocab_size = len(TEXT.vocab)
n_classes = 2

# 학습셋, 검증셋, 테스트셋 예제 개수 출력하여 확인해보기
print(f"[학습셋] : {len(trainset)} [검증셋] : {len(valset)} [테스트셋] : {len(testset)}")
print(f"[단어수] : {vocab_size} [클래스] : {n_classes}")

# BasicGRU 라는 RNN을 포함하는 신경망 모델 생성
class BasicGRU(nn.Module):
    def __init__(self, n_layers, hidden_dim, n_vocab, embed_dim,
                n_classes, dropout_p = 0.2):
        super(BasicGRU, self).__init__()
        print("Building Basic GRU model...")

        # 은닉 벡터들의 '층'.
        # 아주 복잡한 모델이 아닌 이상 2 이하로 설정.
        self.n_layers = n_layers

        # n_vocab : 사전에 등재된 단어 수
        # embed_dim = 임베딩된 단어 텐서가 지니는 차원값
        self.embed = nn.Embedding(n_vocab, embed_dim)

        self.hidden_dim = hidden_dim # 생성되는 은닉 벡터 차원값
        self.dropout = nn.Dropout(dropout_p) # 드롭아웃 정의

        # RNN 모델 정의
        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 = 한 배치 속에 있는 모든 영화평 -> 시계열 데이터로 변환
        x = self.embed(x)

        # 보통은 여기서 forward() 했겠지만,
        # RNN은 첫번째 은닉 벡터 H0 을 정의해 x와 함께 입력해줘야함.
        h_0 = self._init_state(batch_size = x.size(0))

        # 입력 x를 첫번째 은닉벡터와 핳ㅁ께 입력하면 은닉 벡터들이 시계열 배열 형태로 반환.
        x, _ = self.gru(x, h_0) # (batch_size, x길이, hidden_dim) 형태의 3d 텐서
        h_t = x[:, -1, :] # 맨 마지막 출력값
        self.dropout(h_t) # 드롭아웃 설정

        logit = self.out(h_t) # 모든 리뷰 압축한 h_t 를 신경망에 넣어 결과 출력
        return logit

    # 첫 번째 은닉 벡터 정의하는 함수
    def _init_state(self, batch_size = 1):
        # parameters() : 신경망 모듈의 가중치 정보를 반복자 형태로 반환.
        # 얘가 생성하는 원소들은 실제 신경망의 가중치 텐서를 지닌 객체.
        weight = next(self.parameters()).data # 첫번째 가중치 텐서 추출

        # new()로 텐서로 변환 후 zero_()로 텐서 내 모든 값 0으로 초기화
        # -> 첫번째 은닉벡터는 모든 특성값이 0인 벡터로 설정됨.
        return weight.new(self.n_layers, batch_size, self.hidden_dim).zero_()

# 학습 함수 구현
def train(model, optimizer, train_iter):
    model.train()
    for b, batch in enumerate(train_iter):
        x, y = batch.text.to(DEVICE), batch.label.to(DEVICE)

        # 라벨값을 보기 쉽게 1/2 에서 0/1로 바꾸기
        y.data.sub_(1)

        optimizer.zero_grad() # 기울기 초기화
        logit = model(x) # 예측값 계산

        # 오차 구하고 최적화
        loss = F.cross_entropy(logit, y)
        loss.backward()
        optimizer.step()

# 성능 측정 함수
def evaluate(model, val_iter):
    model.eval()
    corrects, total_loss = 0, 0
    for batch in val_iter:
        x, y = batch.text.to(DEVICE), batch.label.to(DEVICE)
        y.data_sub_(1) # 앞이랑 똑같이 레이블값 0과 1로 바꿔주기
        logit = model(x)
        loss = F.cross_entropy(logit, y, reduction = 'sum')
        total_loss += loss.item()
        corrects += (logit.max(1)[1].view(y.size()).data == y.data).sum()
    size = len(val_iter.dataset)
    avg_loss = total_loss / size
    avg_accuracy = 100.0 * corrects / size
    return avg_loss, avg_accuracy

# 모델 정의

# 은닉벡터 차원값 = 256, 임베딩된 토큰 차원값 = 128
model = BasicGRU(1, 256, vocab_size, 128, n_classes, 0.5).to(DEVICE)

# 최적화 알고리즘 뭐 쓸지 모르겠다 = Adam -> 빠른 학습 속도
optimizer = torch.optim.Adam(model.parameters(), lr = lr)

# 학습 실행
best_val_loss = None
for e in range(1, EPOCHS + 1):
    train(model, optimizer, train_iter)
    val_loss, val_accuracy = evaluate(model, val_iter)

    print(f"[{e}] 검증 오차: {val_loss:5.2f} | 검증 정확도: {val_accuracy:5.2f}")

    # 검증 오차가 최소화된 모델 저장
    if not best_val_loss or val_loss < best_val_loss:
        if not os.path.isdir("snapshot"):
            os.makedirs("snapshot")
        torch.save(model.state_dict(), './snapshot/txtclassificatoion.pt')
        best_val_loss = val_loss

model.load_state_dict(torch.load('./snapshot/txtclassification.pt'))
test_loss, test_acc = evaluate(model, test_iter)
print('테스트 오차: %5.2f | 테스트 정확도: %5.2f' % (test_loss, test_acc))