[HAI] 2023 여름 방학 프로젝트 - 5편(학습 플로우 & BERT 계열 선정이유)

GitHub - dorae222/HAI_Kaggle_Competition: 한국 지역 방언 분류

---

1. 데이터 처리 및 학습 플로우

1. 데이터 불러오기 
   • train.csv, valid.csv 파일 로드
2. Tokenizer 선택 및 로딩
   • transformers 라이브러리의 Pre-trained Tokenizer 선택
     (BertTokenizerFast, AlbertTokenizer 등)
   • Tokenizer 로드
3. 데이터 전처리(BERT 계열 모델 기준)
   • 문장 앞뒤에 [CLS], [SEP] 토큰 추가
   • 문장 토큰화
   • 토큰을 숫자(ID)로 변환
   • Padding 및 Truncation 처리
   • Attention Mask 생성
4. Dataloader 생성
   • TensorDataset으로 데이터 변환
   • DataLoader로 배치 구성
5. 모델 선택 및 로딩
   • transformers 라이브러리의 Pre-trained Model 선택
     (BertForSequenceClassification, AlbertForSequenceClassification 등)
   • 모델 로드
6. 모델 학습 설정
   • Optimizer 설정 (예: AdamW)
   • Learning rate scheduler 설정
   • GPU 설정 (CUDA)
7. 모델 학습
   • Epoch 수만큼 반복
   • 각 배치에 대해 Forward pass 및 Backward pass
   • 학습 손실 및 정확도 계산
   • Validation 데이터로 검증
   • Early stopping 조건 확인
8. 모델 저장
   • 최적의 Validation 정확도를 가진 모델 저장
9. 테스트
   • 저장된 모델 로드
   • Test 데이터에 대해 예측
   • 결과 저장

# 필요한 라이브러리 import

import optuna
import os
import random
import easydict
import torch
import numpy as np
import pandas as pd
import torch.nn as nn
from sklearn.model_selection import train_test_split
from tqdm import tqdm
from transformers import AdamW, get_linear_schedule_with_warmup
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, SequentialSampler
from keras_preprocessing.sequence import pad_sequences
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from torch.utils.tensorboard import SummaryWriter
import matplotlib.pyplot as plt
import warnings

warnings.filterwarnings('ignore', category=FutureWarning)

def generate_data_loader(file_path, tokenizer, args):
    # 텍스트 데이터를 BERT 모델의 입력 형식에 맞게 변환하는 함수
    def get_input_ids(data):
        # 각 문장 앞에 "[CLS]", 뒤에 "[SEP]"를 추가함
        document_bert = ["[CLS] " + str(s) + " [SEP]" for s in data]
        # 문장을 토큰으로 분리함
        tokenized_texts = [tokenizer.tokenize(s) for s in tqdm(document_bert, "Tokenizing")]
        # 토큰을 숫자 인덱스로 변환함
        input_ids = [tokenizer.convert_tokens_to_ids(x) for x in tqdm(tokenized_texts, "Converting tokens to ids")]
        print("Padding sequences...")
        # 시퀀스를 일정한 길이로 패딩함
        input_ids = pad_sequences(input_ids, maxlen=args.maxlen, dtype='long', truncating='post', padding='post')
        return input_ids
    
    # 입력 시퀀스에 대한 attention mask를 생성하는 함수
    def get_attention_masks(input_ids):
        attention_masks = []
        for seq in tqdm(input_ids, "Generating attention masks"):
            # 토큰이 있는 위치는 1, 패딩 위치는 0으로 마스킹함
            seq_mask = [float(i > 0) for i in seq]
            attention_masks.append(seq_mask)
        return attention_masks

    # 입력 데이터로부터 PyTorch DataLoader를 생성하는 함수
    def get_data_loader(inputs, masks, labels, batch_size=args.batch):
        # TensorDataset을 생성함
        data = TensorDataset(torch.tensor(inputs), torch.tensor(masks), torch.tensor(labels))
        # 학습 모드일 경우 랜덤 샘플링, 그렇지 않을 경우 순차 샘플링을 사용함
        sampler = RandomSampler(data) if args.mode == 'train' else SequentialSampler(data)
        # DataLoader를 생성함
        data_loader = DataLoader(data, sampler=sampler, batch_size=batch_size)
        return data_loader
    
    # 지정한 경로에서 CSV 파일을 읽어 DataFrame으로 로드함
    data_df = pd.read_csv(file_path)

    # 학습 모드일 경우, 검증 데이터를 포함한 전체 데이터를 불러와 학습/검증 데이터로 분할함
    if args.mode == 'train':
        valid_data_df = pd.read_csv(args.valid_path)
        all_data_df = pd.concat([data_df, valid_data_df])
        
        train_data_df, valid_data_df = train_test_split(
            all_data_df,
            test_size=0.2,
            random_state=42,
            stratify=all_data_df['label'].values
        )
        data_df = train_data_df
    
    # 텍스트 데이터를 BERT 입력 형식으로 변환함
    input_ids = get_input_ids(data_df['text'].values)
    # Attention mask를 생성함
    attention_masks = get_attention_masks(input_ids)
    # DataLoader를 생성함
    data_loader = get_data_loader(input_ids, attention_masks, data_df['label'].values)

    return data_loader

def save(model, dir_name):
    # 주어진 디렉터리 이름으로 디렉터리를 생성함
    # exist_ok=True 옵션은 해당 디렉터리가 이미 존재할 경우 오류를 발생시키지 않도록 설정함
    os.makedirs(dir_name, exist_ok=True)
    # PyTorch 모델의 상태(state_dict)를 'model.pth' 파일로 저장함
    # 저장 경로는 지정한 디렉터리 내부임
    torch.save(model.state_dict(), os.path.join(dir_name, 'model.pth'))

def flat_accuracy(preds, labels):
    # 예측값에서 각 행(row)에 대해 최대값을 갖는 인덱스를 찾음
    pred_flat = np.argmax(preds, axis=1).flatten()
    # 실제 레이블을 1차원 배열로 변환함
    labels_flat = labels.flatten()
    # 예측값과 실제 레이블이 일치하는 요소의 개수를 전체 요소 개수로 나누어 정확도를 계산함
    return np.sum(pred_flat == labels_flat) / len(labels_flat)

def predict(model, args, data_loader):
    print('start predict')
    # 모델을 평가 모드로 설정함
    # 이는 Dropout, BatchNorm과 같은 레이어들이 평가 모드에서 동작하도록 설정하는 것임
    model.eval()
    
    # 평가 중에 계산될 손실과 정확도를 저장할 리스트를 초기화함
    eval_loss = []
    eval_accuracy = []
    # 모델의 예측 결과를 저장할 리스트를 초기화함
    logits = []

    # data_loader에서 배치 단위로 데이터를 불러옴
    for step, batch in tqdm(enumerate(data_loader)):
        # 배치 내의 모든 텐서를 지정된 디바이스로 이동시킴 (예: CUDA 디바이스)
        batch = tuple(t.to(args.device) for t in batch)
        b_input_ids, b_input_mask, b_labels = batch

        # 그래디언트 계산을 비활성화하여 메모리 사용량을 줄이고 속도를 향상시킴
        with torch.no_grad():
            if args.mode == 'test':
                # 테스트 모드일 경우, 레이블 없이 모델을 실행함
                outputs = model(b_input_ids, attention_mask=b_input_mask)
                logit = outputs[0]
            else:
                # 그렇지 않은 경우, 레이블과 함께 모델을 실행하여 손실을 계산함
                outputs = model(b_input_ids, attention_mask=b_input_mask, labels=b_labels)
                loss, logit = outputs[:2]
                # 손실 값을 eval_loss 리스트에 추가함
                eval_loss.append(loss.item())

        # logit (모델의 raw 출력 값)을 numpy 배열로 변환함
        logit = logit.detach().cpu().numpy()
        # 실제 레이블을 numpy 배열로 변환함
        label = b_labels.cpu().numpy()
        
        # logit 값을 logits 리스트에 추가함
        logits.append(logit)

        if args.mode != 'test':
            # 예측된 logit과 실제 레이블을 사용하여 정확도를 계산함
            accuracy = flat_accuracy(logit, label)
            # 정확도 값을 eval_accuracy 리스트에 추가함
            eval_accuracy.append(accuracy)

    # 모든 배치에 대한 logit 값을 하나의 numpy 배열로 합침
    logits = np.vstack(logits)
    # 각 예측 벡터에서 최대 값을 갖는 인덱스를 찾아 예측 레이블을 생성함
    predict_labels = np.argmax(logits, axis=1)

    if args.mode == 'test':
        # 테스트 모드일 경우, 예측 레이블만 반환함
        return predict_labels, None

    # 평균 손실과 평균 정확도를 계산함
    avg_eval_loss = np.mean(eval_loss)
    avg_eval_accuracy = np.mean(eval_accuracy)

    # 예측 레이블, 평균 손실, 평균 정확도를 반환함
    return predict_labels, avg_eval_loss, avg_eval_accuracy

def train(model, args, train_loader, valid_loader, patience=5):
    accumulation_steps = 4  # Gradient를 누적할 배치의 수
    experiment_name = f"v_10_model_ckpt_{args.model_ckpt.replace('/', '_')}_lr_{args.lr}_batch_{args.batch}_epochs_{args.epochs}_maxlen_{args.maxlen}_eps_{args.eps}"
    
    # TensorBoard를 위한 SummaryWriter 인스턴스를 생성함
    writer = SummaryWriter(f'hai_kaggle/{experiment_name}')

    # AdamW 옵티마이저를 초기화함
    optimizer = AdamW(model.parameters(),
                      lr=args.lr,
                      eps=args.eps
                      )
    
    total_steps = len(train_loader) * args.epochs

    # Learning rate scheduler를 초기화함
    scheduler = get_linear_schedule_with_warmup(optimizer,
                                                num_warmup_steps=0,
                                                num_training_steps=total_steps)

    # 난수 시드 설정
    seed_val = 42
    random.seed(seed_val)
    np.random.seed(seed_val)
    torch.manual_seed(seed_val)
    torch.cuda.manual_seed_all(seed_val)

    # 최고의 검증 정확도를 저장하기 위한 변수 초기화
    best_val_accuracy = 0.0
    best_train_accuracy = 0.0
    epochs_without_improvement = 0

    # train_loader로부터 레이블 추출
    train_labels_new = []
    for _, _, b_labels in train_loader:
        train_labels_new.extend(b_labels.numpy())
    train_labels_new = np.array(train_labels_new)

    # 클래스별 가중치 계산
    _, counts = np.unique(train_labels_new, return_counts=True)
    class_weights = 1. / counts
    class_weights = class_weights / np.sum(class_weights)
    class_weights_tensor = torch.tensor(class_weights, dtype=torch.float32).to(args.device)

    # 가중치를 사용하는 Loss 함수 정의
    criterion = nn.CrossEntropyLoss(weight=class_weights_tensor)

    train_accuracies = []
    val_accuracies = []

    print('start training')
    for epoch in range(args.epochs):
        model.train()
        train_loss = []

        optimizer.zero_grad()  # Gradient 초기화

        # 학습 데이터에 대해 배치 단위로 학습을 진행함
        for step, batch in tqdm(enumerate(train_loader), f"training epoch {epoch}", total=len(train_loader)):
            batch = tuple(t.to(args.device) for t in batch)
            b_input_ids, b_input_mask, b_labels = batch
            
            # 모델 실행 및 손실 계산
            outputs = model(b_input_ids,
                            attention_mask=b_input_mask,
                            labels=b_labels)
            loss = outputs[0] / accumulation_steps  # Gradient Accumulation을 위해 loss를 나눈다.
            train_loss.append(loss.item())
            loss.backward()

            # accumulation_steps마다 optimizer 업데이트
            if (step + 1) % accumulation_steps == 0:
                torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
                optimizer.step()
                scheduler.step()
                optimizer.zero_grad()

            # TensorBoard에 학습 손실값을 기록함
            writer.add_scalar('training loss',
                            loss.item(),
                            epoch * len(train_loader) + step)

        # 평균 학습 손실값을 계산함
        avg_train_loss = np.mean(train_loss)
        # 학습 정확도와 검증 정확도를 계산함
        _, _, avg_train_accuracy = predict(model, args, train_loader)
        _, _, avg_val_accuracy = predict(model, args, valid_loader)
        print("Epoch {0},  Average training loss: {1:.4f} , Train accuracy : {2:.4f}, Validation accuracy : {3:.4f}"
              .format(epoch, avg_train_loss, avg_train_accuracy, avg_val_accuracy))

        train_accuracies.append(avg_train_accuracy)
        val_accuracies.append(avg_val_accuracy)

        # TensorBoard에 학습 및 검증 정확도를 기록함
        writer.add_scalar('training accuracy',
                        avg_train_accuracy,
                        epoch)
        writer.add_scalar('validation accuracy',
                        avg_val_accuracy,
                        epoch)

        # 최고의 검증 정확도를 갱신하는 경우 모델 저장
        if avg_val_accuracy > best_val_accuracy:
            best_val_accuracy = avg_val_accuracy
            best_train_accuracy = avg_train_accuracy
            epochs_without_improvement = 0
            save_path = f"./saved_checkpoints/best_model/{experiment_name}_TrainAcc_{best_train_accuracy}_ValAcc_{best_val_accuracy}"
            save(model, save_path)
        else:
            epochs_without_improvement += 1

        # Early Stopping 조건 확인
        if epochs_without_improvement >= patience:
            print(f"[Early Stopping]{patience} epoch에서 중단.[Early Stopping]")
            break

    writer.close()
    # 학습된 모델과 학습/검증 정확도 기록을 반환함
    return model, best_train_accuracy, best_val_accuracy, train_accuracies, val_accuracies

def train_valid(args):
    # GPU가 사용 가능한 경우 'cuda'를, 그렇지 않은 경우 'cpu'를 사용하도록 설정함
    if torch.cuda.is_available():
        args.device = 'cuda'
    else:
        args.device = 'cpu'

    # 지정된 사전 학습 모델을 불러와서 분류 작업을 위한 모델로 초기화함
    # num_labels는 분류할 클래스의 수를 지정함
    model = AutoModelForSequenceClassification.from_pretrained(args.model_ckpt, num_labels=3)
    # 모델을 지정된 디바이스로 이동시킴 (예: 'cuda' 또는 'cpu')
    model.to(args.device)
    
    # 지정된 사전 학습 모델의 토크나이저를 불러옴
    tokenizer = AutoTokenizer.from_pretrained(args.model_ckpt)

    # 학습 데이터를 불러와서 DataLoader를 생성함
    train_dataloader = generate_data_loader(args.train_path, tokenizer, args)

    # 학습 데이터와 검증 데이터를 합친 데이터에서 다시 분리하여
    # 검증 데이터에 대한 DataLoader를 생성함
    valid_dataloader = generate_data_loader(args.train_path, tokenizer, easydict.EasyDict({'mode': 'valid', **args}))

    # train 함수를 사용하여 모델을 학습시키고, 최고의 학습/검증 정확도를 얻음
    model, best_train_accuracy, best_val_accuracy, train_accuracies, val_accuracies = train(model, args, train_dataloader, valid_dataloader)

    # 학습된 모델, 토크나이저, 학습/검증 정확도의 기록, 최고의 학습/검증 정확도를 반환함
    return model, tokenizer, train_accuracies, val_accuracies, best_train_accuracy, best_val_accuracy

# best_train_accuracy & best_val_accuracy는 모델 경로 때문에 주는 것임!
def test(model, tokenizer, test_args, file_path, args, best_train_accuracy, best_val_accuracy):
    # GPU가 사용 가능한 경우 'cuda'를, 그렇지 않은 경우 'cpu'를 사용하도록 설정함
    if torch.cuda.is_available():
        test_args.device = 'cuda'

    # 테스트 데이터를 불러와서 DataLoader를 생성함
    test_dataloader = generate_data_loader(file_path, tokenizer=tokenizer, args=test_args)
    
    # 테스트 데이터에 대한 예측을 수행함
    labels, _ = predict(model, test_args, test_dataloader)

    # 예측한 레이블을 저장하기 위한 DataFrame을 생성함
    submit_df = pd.DataFrame()
    submit_df["idx"] = range(len(labels))
    submit_df["label"] = labels
    
    # 예측 결과를 저장할 파일 경로를 생성함
    # 이 때, 학습 및 검증 정확도를 파일 이름에 포함시켜 모델의 성능을 쉽게 확인할 수 있도록 함
    save_path = f"./saved_checkpoints/best_model/model_ckpt_{args.model_ckpt.replace('/', '_')}_lr_{args.lr}_batch_{args.batch}_epochs_{args.epochs}_maxlen_{args.maxlen}_eps_{args.eps}_TrainAcc_{best_train_accuracy}_ValAcc_{best_val_accuracy}/submission.csv"
    
    # 예측 결과를 CSV 파일로 저장함
    submit_df.to_csv(save_path, index=False)
    
    # 저장한 파일 경로를 출력함
    print(f"Submission file saved to {save_path}")

# 이 함수는 주어진 PyTorch 모델을 사용하여 테스트 데이터에 대한 예측을 수행하고,
# 이를 CSV 파일로 저장하는 역할을 합니다.
# 이 때, CSV 파일의 이름에는 해당 모델의 학습 및 검증 정확도가 포함되어,
# 다양한 모델의 성능을 비교하기 용이하도록 되어 있습니다.

def objective(trial):
    # 하이퍼파라미터 범위 설정
    # 학습률(learning rate)의 범위를 log scale로 설정하고, 이 범위에서 하나의 값을 선택함
    lr = trial.suggest_loguniform('lr', 3e-07, 3e-05)
    
    # 배치 크기(batch size)의 범위를 설정하고, 이 범위에서 하나의 값을 선택함
    # log=True로 설정하여 로그 스케일로 탐색을 수행함
    batch = trial.suggest_int('batch', 64, 128, log=True)
    
    # 시퀀스의 최대 길이(max length) 범위를 설정하고, 이 범위에서 하나의 값을 선택함
    maxlen = trial.suggest_int('maxlen', 32, 128)
    
    # AdamW 옵티마이저의 epsilon 값 (일반적으로 고정값을 사용)
    eps = 1e-8

    # 하이퍼파라미터 및 기타 설정값을 담은 dictionary 생성
    # easydict.EasyDict는 dictionary를 속성 형태로 접근할 수 있게 해주는 라이브러리
    optuna_args = easydict.EasyDict({
        "train_path": "./train.csv",
        "valid_path": "./valid.csv",
        "device" : 'cpu',
        "mode" : "train",
        "batch" : batch,
        "maxlen" : maxlen,
        "lr" : lr,
        "eps" : eps,
        "epochs" : 50,
        "model_ckpt" : "monologg/koelectra-small-v3-discriminator",
    })

    # train_valid 함수를 사용하여 모델을 학습시키고, 검증 정확도를 얻음
    _, _, _, _, _, val_accuracy = train_valid(optuna_args)

    # 최적화 목표는 검증 정확도를 최대화하는 것입니다.
    # Optuna가 다음 반복에서 이 값을 기반으로 하이퍼파라미터를 업데이트합니다.
    return val_accuracy

# 이 함수는 Optuna 라이브러리를 사용하여 하이퍼파라미터 최적화를 수행하는 역할을 합니다.
# Optuna는 자동으로 이 함수를 여러 번 호출하며, 각 호출마다 다른 하이퍼파라미터 세트를 사용하여
# 검증 정확도를 최대화하려고 시도합니다.

# 최적화 수행
study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=10)

# 최적의 하이퍼파라미터 출력
print(study.best_params)

---

2. 모델 선택

모델을 선정함에 있어, 한국어 모델의 종류를 살펴보고,

이번 테스크에서는 왜 BERT 계열을 골랐는지 알아봅시다.

(이미지 출처: https://www.letr.ai/blog/tech-20221124)

각 내용은 직접 추가적으로 정리하였습니다.

 

Encoder-Centric Models: BERT 계열

Encoder-Centric

• 구조: Transformer의 인코더 구조만을 사용합니다.
• 학습 방식: 마스크된 언어 모델링(Masked Language Modeling)을 통해 학습합니다.
  즉, 일부 토큰을 가려놓고 이를 예측하도록 학습시킵니다.
• 특징:
  • 문장 내의 어떤 단어든지 그 문맥(context)을 기반으로 표현(representation)할 수 있습니다.
  • BERT는 문장의 앞뒤 문맥을 모두 고려하여 단어의 임베딩을 생성합니다.
  • 전이 학습(Transfer Learning)을 통해 다양한 자연어 처리 작업에 적용이 가능합니다.
• 주의점: BERT는 큰 모델 사이즈와 많은 양의 데이터로 학습되므로, 학습 시 많은 리소스와 시간이 소요됩니다.

Decoder-Centric Models: GPT 계열

Decoder-Centric

• 구조: Transformer의 디코더 구조만을 사용합니다.
• 학습 방식: 전통적인 언어 모델링 방식으로 학습합니다.
  이전 단어들만을 기반으로 다음 단어를 예측하도록 학습시킵니다.
• 특징:
  • 주어진 문장이나 문장의 일부를 기반으로 연속된 텍스트 생성에 강점을 가집니다.
  • GPT는 주로 왼쪽에서 오른쪽으로 문맥을 고려합니다.
  • BERT와 마찬가지로 전이 학습을 통해 다양한 자연어 처리 작업에 적용이 가능합니다.
• 주의점: GPT 역시 큰 모델 사이즈로 학습되므로, 학습 시 많은 리소스와 시간이 소요됩니다

Encoder-Decoder Models: Seq2seq 계열

Encoder-Decoder

• 구조: Transformer의 전체 구조(인코더와 디코더)를 사용합니다.
• 학습 방식: 주어진 입력 시퀀스를 바탕으로 목표 출력 시퀀스를 생성하도록 학습합니다.
  주로 기계 번역, 문장 요약 등에 사용됩니다.
• 특징:
  • 인코더는 입력 시퀀스의 정보를 컴팩트한 형태로 인코딩하고, 디코더는 이를 바탕으로 출력 시퀀스를 생성합니다.
  • 주어진 문장 또는 문서의 정보를 다른 형태의 문장으로 변환하는 작업에 특화되어 있습니다.
• 주의점: Seq2seq 모델은 학습이 복잡하며, 특히 문장 길이가 길어질수록 성능 저하가 발생할 수 있습니다.

---

BERT 계열이 분류 작업에 더 적합할까?

1. 고정된 길이의 출력: BERT는 문장의 길이와 상관없이 항상 고정된 길이의 출력 벡터를 생성합니다. 이 출력 벡터는 문장의 전체적인 의미를 잘 반영하므로, 분류 작업에 적합합니다.
2. 문맥 파악 능력: 언어의 뉘앙스나 문맥을 정확히 파악하는 것은 방언 분류와 같은 작업에서 중요합니다. BERT의 양방향 문맥 정보를 활용하는 능력은 이러한 작업에 큰 도움이 됩니다.

다른 계열과 비교하기

1. GPT 계열과의 비교: GPT는 주로 왼쪽에서 오른쪽으로의 문맥만을 고려합니다. 따라서 BERT가 양방향 문맥 정보를 활용하는 것에 비해 문맥 파악 능력에서 상대적으로 약할 수 있습니다. 텍스트 생성 작업에는 GPT가 더 적합하지만, 분류 작업에서는 BERT의 양방향 문맥 정보 활용 능력이 더 유리할 수 있습니다.
2. Seq2seq 계열과의 비교: Seq2seq 모델은 주로 입력 시퀀스를 다른 형태의 출력 시퀀스로 변환하는 작업에 적합합니다. 예를 들어, 기계 번역이나 문장 요약 같은 작업입니다. 텍스트 분류와 같은 작업에서는 BERT와 같은 인코더 중심의 모델이 더 간결하고 효과적일 수 있습니다.