트레이너 API 내부 학습 과정
# torch: PyTorch 라이브러리. 텐서 연산 및 모델 학습에 사용됨.
# transformers: Hugging Face의 트랜스포머 모델을 쉽게 불러올 수 있도록 해주는 라이브러리.
# AutoModelForSequenceClassification: 사전 학습된 트랜스포머 모델을 불러와서 **문장 분류(task)**를 수행할 수 있도록 함.
# AutoTokenizer: 텍스트 데이터를 토큰화할 수 있도록 해주는 클래스.
# AdamW: 트랜스포머 모델에서 일반적으로 사용되는 최적화 알고리즘(Adam의 변형).
# load_dataset: Hugging Face datasets 라이브러리에서 KLUE 데이터셋을 불러오기 위한 함수.
# tqdm.auto: 학습 진행 과정을 시각적으로 표시해주는 라이브러리.
# DataLoader: PyTorch에서 데이터를 배치 단위로 로드할 때 사용하는 클래스.
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from datasets import load_dataset
from tqdm.auto import tqdm
from torch.utils.data import DataLoader
from torch.optim.adamw import AdamW
import numpy as np
klue_tc_train = load_dataset('klue', 'ynat', split='train').remove_columns(['guid', 'url', 'date'])
klue_tc_eval = load_dataset('klue', 'ynat', split='validation').remove_columns(['guid', 'url', 'date'])
klue_tc_label = klue_tc_train.features['label']
def make_str_label(batch):
batch['label_str'] = klue_tc_label.int2str(batch['label'])
return batch
klue_tc_train = klue_tc_train.map(make_str_label, batched=True, batch_size=1000)
train_dataset = klue_tc_train.train_test_split(test_size=10000, shuffle=True, seed=42)['test']
dataset = klue_tc_eval.train_test_split(test_size=1000, shuffle=True, seed=42)
test_dataset = dataset['test']
valid_dataset = dataset['train'].train_test_split(test_size=1000, shuffle=True, seed=42)['test']
model_id = "klue/roberta-base"
# label 갯수 만큼 뉴런을 가지는 모델 생성
model = AutoModelForSequenceClassification.from_pretrained(model_id, num_labels=len(train_dataset.features['label'].names))
tokenizer = AutoTokenizer.from_pretrained(model_id)
def tokenize_function(examples): # 제목(title) 컬럼에 대한 토큰화
return tokenizer(examples["title"], padding="max_length", truncation=True)
# 모델과 토크나이저 불러오기
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model_id = "klue/roberta-base"
model = AutoModelForSequenceClassification.from_pretrained(model_id, num_labels=len(train_dataset.features['label'].names))
tokenizer = AutoTokenizer.from_pretrained(model_id)
# GPU로 모델 이동
model.to(device)
def make_dataloader(dataset, batch_size, shuffle=True):
dataset = dataset.map(tokenize_function, batched=True).with_format("torch") # 데이터셋에 토큰화 수행
# AutoModelForSequenceClassification에서 출력값을 labels로 학습하기 때문에 컬럼 이름 변경
dataset = dataset.rename_column("label", "labels")
dataset = dataset.remove_columns(column_names=['title']) # 불필요한 컬럼 제거
return DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)
# 데이터로더 만들기
train_dataloader = make_dataloader(train_dataset, batch_size=8, shuffle=True)
valid_dataloader = make_dataloader(valid_dataset, batch_size=8, shuffle=False)
test_dataloader = make_dataloader(test_dataset, batch_size=8, shuffle=False)
def train_epoch(model, data_loader, optimizer):
# 모델을 드롭 아웃 등을 활성화 하는 훈련 모드
model.train()
total_loss = 0
for batch in tqdm(data_loader): # tdqm을 사용하면 학습진행 상황을 프로그레스 바로 시각화
optimizer.zero_grad()
input_ids = batch['input_ids'].to(device) # 모델에 입력할 토큰 아이디
attention_mask = batch['attention_mask'].to(device) # 모델에 입력할 어텐션 마스크
labels = batch['labels'].to(device) # 모델에 입력할 레이블
outputs = model(input_ids, attention_mask=attention_mask, labels=labels) # 모델 계산
loss = outputs.loss # 손실
loss.backward() # 오차를 기반으로 모델의 가중치를 업데이트할 방향을 계산 (역전파)
optimizer.step() # 모델의 가중치 업데이트
total_loss += loss.item()
avg_loss = total_loss / len(data_loader)
return avg_loss
def evaluate(model, data_loader):
model.eval() # 평가 모드로 설정 (드롭아웃, 배치 정규화 비활성화)
total_loss = 0
predictions = []
true_labels = []
with torch.no_grad(): # 그래디언트 계산 비활성화 (평가할 때 필요 없음)
for batch in tqdm(data_loader): # 데이터 로더에서 배치 단위로 데이터 가져오기
input_ids = batch['input_ids'].to(device) # 입력 데이터
attention_mask = batch['attention_mask'].to(device) # 어텐션 마스크
labels = batch['labels'].to(device) # 실제 정답 라벨
outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
logits = outputs.logits # 모델의 예측값
loss = outputs.loss # 손실값
total_loss += loss.item()
preds = torch.argmax(logits, dim=-1) # 가장 높은 확률을 가진 클래스로 예측
predictions.extend(preds.cpu().numpy()) # 예측값을 리스트에 저장
true_labels.extend(labels.cpu().numpy()) # 실제 라벨을 리스트에 저장
avg_loss = total_loss / len(data_loader)
accuracy = np.mean(np.array(predictions) == np.array(true_labels))
return avg_loss, accuracy
num_epochs = 1
# 트랜스포머에서 흔히 사용되는 확률 값(0.00005)
optimizer = AdamW(model.parameters(), lr=5e-5)
# 학습 루프
for epoch in range(num_epochs):
print(f"Epoch {epoch+1}/{num_epochs}")
train_loss = train_epoch(model, train_dataloader, optimizer)
print(f"Training loss: {train_loss}")
valid_loss, valid_accuracy = evaluate(model, valid_dataloader)
print(f"Validation loss: {valid_loss}")
print(f"Validation accuracy: {valid_accuracy}")
# Testing
_, test_accuracy = evaluate(model, test_dataloader)
print(f"Test accuracy: {test_accuracy}") # 정확도 0.82.jpg)