네이버 영화 리뷰 감정 분석 데이터세트로 분류 모델을 학습

#데이터 불러오기
import numpy as np
import pandas as pd
from Korpora import Korpora

# Korpora 라이브러리를 사용하여 네이버 영화 리뷰 데이터셋을 불러오기
corpus = Korpora.load("nsmc")
# 테스트 데이터셋에서 샘플을 랜덤하게 선택하고 20000개의 샘플을 추출한 후 데이터프레임으로 변환
df = pd.DataFrame(corpus.test).sample(20000, random_state=42)
# 전체 데이터셋을 훈련, 검증, 테스트 세트로 나눔
train, valid, test = np.split(
    df.sample(frac=1, random_state=42), [int(0.6 * len(df)), int(0.8 * len(df))]
)

# 훈련, 검증, 테스트 세트의 크기 출력
print(train.head(5).to_markdown())
print(f"Training Data Size : {len(train)}")
print(f"Validation Data Size : {len(valid)}")
print(f"Testing Data Size : {len(test)}")

# 전처리된 데이터를 Electra 모델에 적합한 형식으로 변환하는 함수 정의
def make_dataset(data, tokenizer, device):
    # 입력 텍스트를 토큰화하고 텐서로 변환
    tokenized = tokenizer(
        text=data.text.tolist(),
        padding="longest",
        truncation=True,
        return_tensors="pt"
    )
    # 토큰화된 입력 데이터와 라벨을 텐서 데이터셋으로 변환하여 반환
    input_ids = tokenized["input_ids"].to(device)
    attention_mask = tokenized["attention_mask"].to(device)
    labels = torch.tensor(data.label.values, dtype=torch.long).to(device)
    return TensorDataset(input_ids, attention_mask, labels)

# 데이터로더 생성 함수 정의
def get_datalodader(dataset, sampler, batch_size):
    data_sampler = sampler(dataset)
    dataloader = DataLoader(dataset, sampler=data_sampler, batch_size=batch_size)
    return dataloader

# 하이퍼파라미터 설정
epochs = 5
batch_size = 32
device = "cuda" if torch.cuda.is_available() else "cpu"
# Electra 토크나이저 로드
tokenizer = ElectraTokenizer.from_pretrained(
    pretrained_model_name_or_path="monologg/koelectra-base-v3-discriminator",
    do_lower_case=False,
)

# 훈련 데이터셋 전처리 및 데이터로더 생성
train_dataset = make_dataset(train, tokenizer, device)
train_dataloader = get_datalodader(train_dataset, RandomSampler, batch_size)

# 검증 데이터셋 전처리 및 데이터로더 생성
valid_dataset = make_dataset(valid, tokenizer, device)
valid_dataloader = get_datalodader(valid_dataset, SequentialSampler, batch_size)

# 테스트 데이터셋 전처리 및 데이터로더 생성
test_dataset = make_dataset(test, tokenizer, device)
test_dataloader = get_datalodader(test_dataset, SequentialSampler, batch_size)

# 훈련 데이터셋의 첫 번째 샘플 출력 (입력 텐서, 어텐션 마스크, 라벨)
print(train_dataset[0])

#KoELECTRA 모델 선언
from torch import optim
from transformers import ElectraForSequenceClassification

# KoELECTRA 모델 선언
model = ElectraForSequenceClassification.from_pretrained(
    pretrained_model_name_or_path="monologg/koelectra-base-v3-discriminator",  # 사전 학습된 KoELECTRA 모델 로드
    num_labels=2  # 분류할 클래스 수 설정 (긍정, 부정)
).to(device)  # 모델을 GPU 또는 CPU로 이동

# 옵티마이저 설정 (AdamW 사용)
optimizer = optim.AdamW(
    model.parameters(),  # 모델의 파라미터를 최적화 대상으로 지정
    lr=1e-5,  # 학습률 설정
    eps=1e-8  # AdamW 옵티마이저의 epsilon 값 설정
)

#모델 구조 출력
for main_name, main_module in model.named_children():
    print(main_name)
    for sub_name, sub_module in main_module.named_children():
        print("└", sub_name)
        for ssub_name, ssub_module in sub_module.named_children():
            print("│  └", ssub_name)
            for sssub_name, sssub_module in ssub_module.named_children():
                print("│  │  └", sssub_name)
#학습
import numpy as np
from torch import nn

# 정확도 계산 함수 정의
def calc_accuracy(preds, labels):
    pred_flat = np.argmax(preds, axis=1).flatten()
    labels_flat = labels.flatten()
    return np.sum(pred_flat == labels_flat) / len(labels_flat)

# 모델 훈련 함수 정의
def train(model, optimizer, dataloader):
    model.train()
    train_loss = 0.0

    for input_ids, attention_mask, labels in dataloader:
        outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels)

        loss = outputs.loss
        train_loss += loss.item()
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

    train_loss = train_loss / len(dataloader)
    return train_loss

# 모델 평가 함수 정의
def evaluation(model, dataloader):
    with torch.no_grad():
        model.eval()
        criterion = nn.CrossEntropyLoss()
        val_loss, val_accuracy = 0.0, 0.0
        
        for input_ids, attention_mask, labels in dataloader:
            outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels)
            logits = outputs.logits

            loss = criterion(logits, labels)
            logits = logits.detach().cpu().numpy()
            label_ids = labels.to("cpu").numpy()
            accuracy = calc_accuracy(logits, label_ids)
            
            val_loss += loss
            val_accuracy += accuracy
    
    val_loss = val_loss/len(dataloader)
    val_accuracy = val_accuracy/len(dataloader)
    return val_loss, val_accuracy

# 최적의 검증 손실 기록 및 모델 저장
best_loss = 10000
for epoch in range(epochs):
    train_loss = train(model, optimizer, train_dataloader)
    val_loss, val_accuracy = evaluation(model, valid_dataloader)
    print(f"Epoch {epoch + 1}: Train Loss: {train_loss:.4f} Val Loss: {val_loss:.4f} Val Accuracy {val_accuracy:.4f}")

    if val_loss < best_loss:
        best_loss = val_loss
        torch.save(model.state_dict(), "../models/ElectraForSequenceClassification.pt")
        print("Saved the model weights")

# 저장된 모델 가중치 로드 및 테스트 데이터셋 평가
model = ElectraForSequenceClassification.from_pretrained(
    pretrained_model_name_or_path="monologg/koelectra-base-v3-discriminator",
    num_labels=2
).to(device)
model.load_state_dict(torch.load("../models/ElectraForSequenceClassification.pt"))

test_loss, test_accuracy = evaluation(model, test_dataloader)
print(f"Test Loss : {test_loss:.4f}")
print(f"Test Accuracy : {test_accuracy:.4f}")