- 텍스트가 token이라는 수치로 변환 되고 각 token은 Word Embedding을 통해 벡터로 변환됨
- 순환 하지 않고 데이터를 한번에 병렬 처리할 수 있어 순환 신경망(RNN)과 LSTM(장단기 메모리) 보다 학습 및 훈련 속도가 획기적으로 빨라짐
- 문장의 앞 뒤 거리에 상관없이 단어 간의 관계를 정확하게 이해할 수 있음
- 구성 : 인코더(Encoder) + 디코더(Decoder)
- 인코더(Encoder) : 데이터 이해
- 디코더(Decoder) : 결과 생성
import torch
import pandas as pd
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from transformers import AutoTokenizer
# !pip install transformers
# 데이터를 로드 ratings_train.txt 로드
df = pd.read_csv("./ratings_train.txt", sep='\t')
df.dropna(inplace=True)
df.drop_duplicates('document', inplace=True)
df = df[:1000]
# 토크나이저 로드
tokenizer = AutoTokenizer.from_pretrained('beomi/kcbert-base')
X = df['document'].tolist()
y = df['label'].tolist()
# AutoTokenizer를 이용해서 불러온 토큰화 함수는 입력값을 리스트형태로 받는다.
tokenized_inputs = tokenizer(
X,
padding = 'max_length',
max_length = 64,
truncation = True,
return_tensors = 'pt'
)
class SampleDataset(Dataset):
def __init__(self, tokenized_datas, labels):
# tokenized_datas : AutoTokenizer를 이용하여 토큰화한 데이터 (dict)
# labels : 종속 변수 데이터
self.input_ids = tokenized_datas['input_ids']
self.attention_mask = tokenized_datas['attention_mask']
self.labels = labels
def __len__(self):
return len(self.labels)
def __getitem__(self, idx):
return self.input_ids[idx], self.attention_mask[idx], self.labels[idx]
# DataLoader 생성
train_dataset = SampleDataset( tokenized_inputs, y )
train_loader = DataLoader(train_dataset, batch_size = 16, shuffle=True)
class TransformerCLF(nn.Module):
def __init__(
self, vocab_size, d_model=128, nhead = 4, num_layer = 2, num_classes = 2
):
# vocab_size -> 단어 사전의 길이
# d_model -> 태랜스포머 인코딩에 입력 차원의 수
# nhead -> 헤드의 개수 (attetion 갑을 구하기 위한 시점의 개수),
# num_layer -> 레이어의 개수
# 임베딩 : 인코딩된 데이터를 벡터화 작업
super().__init__()
self.emb = nn.Embedding(vocab_size, d_model)
encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=nhead, batch_first=True)
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers = num_layer)
self.fc = nn.Linear(d_model, num_classes)
# 순전파 함수
def forward(self, input_ids):
# input_ids의 크기는 [batch_size, seq_len]
# x의 크기 [batch_size, seq_len, emb_dim] -> [16, 64, 128]
x = self.emb(input_ids)
# 트랜스포머 레이어에 입력
# Q, K, V 벡터가 생성 -> atteion_score가 생성 -> softmax(attetion_score / k차원의수 ** 0.5) @ V
# 트랜스포머 레이어를 통과한 출력의 크기 [batch_size, seq_len, emb_dim]
x = self.transformer(x)
# 문장 전체를 단어 벡터들의 평균을 계산
# x의 크기는 [batch_size, emb_dim] -> [16, 128]
x = x.mean(dim=1)
logits = self.fc(x)
return logits
#모델 생성
model = TransformerCLF(vocab_size= tokenizer.vocab_size)
tokenizer.vocab_size
#손실 함수
criterion = nn.CrossEntropyLoss()
#옵티마이저
optimizer = optim.Adam(model.parameters(), lr = 0.0001)
#반복 학습
model.train()
for epoch in range(3):
total_loss = 0.0
# train_loader는 (input_ids, attnetion_mask, label)
for input_ids, attention_mask, label in train_loader:
# 기울기 초기화
optimizer.zero_grad()
# 순전파
output = model(input_ids)
# 손실 계산
loss = criterion(output, label)
# 역전파 : 손실을 기준으로 미분으로 가중치들의 오차 기여도를 생성
loss.backward()
# 가중치의 수정
optimizer.step()
# 손실값 total_loss에 누적합
total_loss += loss.item()
print(f"Epoch {epoch+1}/3 평균 손실 값 : {round(total_loss / len(train_loader), 4)} ")
>>> Epoch 1/3 평균 손실 값 : 0.706
Epoch 2/3 평균 손실 값 : 0.6951
Epoch 3/3 평균 손실 값 : 0.6763
#예측 값을 생성하고 정확도 계산
#데이터 프레임에서 하위 100개의 데이터를 추출
df2=df.tail(100)
X_test=df2['document'].tolist()
y_test=df2['label'].tolist()
tokenized_test = tokenizer(
X_test,
truncation = True,
padding = 'max_length',
max_length = 64,
return_tensors = 'pt'
)
tokenized_test
>>> {'input_ids': tensor([[ 2, 13991, 4163, ..., 0, 0, 0],
[ 2, 24543, 2535, ..., 0, 0, 0],
[ 2, 12296, 4358, ..., 0, 0, 0],
...,
[ 2, 2177, 4970, ..., 0, 0, 0],
[ 2, 2635, 4455, ..., 0, 0, 0],
[ 2, 8451, 24750, ..., 4327, 17, 3]]), 'token_type_ids': tensor([[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]]), 'attention_mask': tensor([[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
...,
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 1, 1, 1]])}
ids_list=tokenized_test['input_ids']
ids_list
>>> tensor([[ 2, 13991, 4163, ..., 0, 0, 0],
[ 2, 24543, 2535, ..., 0, 0, 0],
[ 2, 12296, 4358, ..., 0, 0, 0],
...,
[ 2, 2177, 4970, ..., 0, 0, 0],
[ 2, 2635, 4455, ..., 0, 0, 0],
[ 2, 8451, 24750, ..., 4327, 17, 3]])
correct_cnt = 0
for ids, label in zip(ids_list, y_test):
# print(ids)
# break
#모델에 데이터를 넣을때의 크기 -> [batch_size, seq_len]
#ids의 크기는 [seq_len] --> [1, seq_len]
ids = ids.unsqueeze(0)
output = model(ids)
correct_cnt += torch.argmax(output, dim=1)[0] == label
# print( correct_cnt )
# break
print(f"트랜스포머 모델의 정확도는 :", correct_cnt / len(ids_list))
>>> 트랜스포머 모델의 정확도는 : tensor(0.6000)
