Bert모델 Fine-tuning: Single Sentence Classification Task

•  모델에 입력되는 단일 문장의 종류를 분류하는 문제
• Task에 대해 모델을 평가하기 위한 데이터셋으로 CoLA (The Corpus of Linguisic Acceptability)를 사용
  • CoLA는 문장마다 문법적으로 올바르거나 잘못된 것으로 레이블이 지정된 데이터
• 평가지표는 Matthews correlation을 사용

 

하나씩 손코딩으로 따라 쳐보았다(소소한 오타때문에 에러남 ㅠㅠ) 

눈으로만 봤을때는 이해한 줄 알고 그냥 지나쳤었는데 

실제로 하나씩 타이핑해보니 모르는 부분이 많았고

모르는 내용을 검색해가며 정리했다. 

 

작성하고 나서 코드가 복잡해서 엑셀로 정리해 보았다(다이어그램을 그리려니 시간이 너무 오래걸림...)

 

엑셀로 모든 함수와 변수를 정리해보니 흐름이 이해가 되었다. 

예를 들어, prepare_dataset 함수에서는 3가지 함수가 사용된다 

1. train_test_split: dataset을 넣어서 -> train_dataset, valid_dataset 변수를 생성

2. tokenized_dataset: train_dataset, tokenizer를 넣어서 -> tokenized_train 변수를 생성

    tokenized_dataset: valid_dataset, tokenizer를 넣어서 -> tokenized_valid 변수를 생성

3. ClsDataset: tokenized_train, train_label을 넣어서 -> cls_train_dataset 변수를 생성

    ClsDataset: tokenized_valid, valid_label을 넣어서 -> cls_valid_dataset 변수를 생성

 

남이 작성한 코드, 남이 정리해놓은 다이어그램 백만개 보는것 보다

내가 직접 코드 따라 쳐보고, 함수와 변수의 흐름을 그려보는게 확실하게 이해가 된다. 

 

 

# 데이터 토큰화 
def load_tokenizer_and_model_for_train():
    MODEL_NAME = args_train.model_name
    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
    model = BertForSequenceClassification.from_pretrained(MODEL_NAME, num_labels=2)
    print("-------Modeling Done-------")
    return tokenizer, model
    
    
    
# 데이터셋을 입력으로 받아 토큰화 -> 토큰화된 결과를 반환 
def tokenized_dataset(dataset, tokenizer):
    print()
    print("tokenizer에 들어가는 데이터 형태")
    print(list(dataset["sentence"])[-1])  # 'sentence' 칼럼에서 마지막 문장을 출력하여 토크나이저에 들어가는 데이터 형태 확인 
    tokenized_sentences = tokenizer(list(dataset["sentence"]),   # 문장 리스트를 토큰화 
                                         add_special_tokens=True,  # [CLS], [SEP] 토큰 추가
                                         max_length=64,            # 패딩/잘라내기를 위한 최대 문장 길이
                                         pad_to_max_length=True,
                                         return_attention_mask=True,  # attention masks 출력
                                         return_tensors='pt')   # 토큰화된 결과를 PyTorch 텐서 형태로 반환 
    print("tokenizing 된 데이터 형태") 
    print(tokenizer.convert_ids_to_tokens(tokenized_sentences['input_ids'][-1]))  # 토큰화된 결과의 마지막 문장을 토큰ID로 출력 -> 토큰ID를 다시 토큰으로 변환

    print(tokenized_sentences['input_ids'][-1])
    print()
    return tokenized_sentences
#>> list(dataset["sentence"]): dataset['sentence'] 칼럼의 모든 문장을 리스트로 변환 
#>> 문장을 토큰 단위로 분리, 특수 토큰 추가, 최대 길이에 맞춰 패딩 



class ClsDataset(torch.utils.data.Dataset):  # 데이터프레임을 torch dataset class로 변환 
    def __init__(self, news_dataset, labels):
        self.dataset = news_dataset
        self.labels = labels 

    def __getitem__(self, idx):
        item = {key: val[idx].clone().detach() for key, val in self.dataset.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item 

    def __len__(self):
        return len(self.labels)

#>> ClsDataset 클래스는 DataLoader와 함께 사용되어 모델 학습 및 평가에 사용됨
#>> DataLoader는 ClsDataset에서 배치 크기만큼의 데이터를 가져와서 모델에 공급



# train/val set split 9:1 -> 각 데이터셋을 토큰화 -> 데이터셋 클래스로 변환 
def prepare_dataset(dataset, tokenizer):
    train_dataset, valid_dataset = train_test_split(dataset, test_size=0.1, random_state=42)

    # 토큰화: input_ids, token_type_ids, attention_mask 생성
    tokenized_train = tokenized_dataset(train_dataset, tokenizer)
    tokenized_valid = tokenized_dataset(valid_dataset, tokenizer)
    
    # split label 
    train_label = train_dataset['label'].values
    valid_label = valid_dataset['label'].values
    print('------tokenizing Done------')
    
    # make dataset for pytorch
    cls_train_dataset = ClsDataset(tokenized_train, train_label)
    cls_valid_dataset = ClsDataset(tokenized_valid, valid_label)
    print('------dataset class Done------')
    
    return cls_train_dataset, cls_valid_dataset
    
    
    
# 모델 학습을 위한 HuggingFace의 Trainer 객체를 생성 & 설정 
def load_trainer_for_train(model, train_dataset, test_dataset, tokenizer):
    training_args = TrainingArguments(output_dir='./',   # train을 위한 huggingface trainer 설정 
                                      per_device_train_batch_size=16,
                                      num_train_epochs=1)
    
    print('------Set training arguments Done-----')
    
    # Trainier 객체 생성 
    trainer = Trainer(model=model,     
                       args=training_args,
                       train_dataset=train_dataset)
    
    print('------Set Trainer Done------')
    
    return trainer
#>> HuggingFace의 Trainer 클래스



# 모델을 학습하고 best model을 저장
def train(dataset):
    
    # fix a seed
    pl.seed_everything(seed=42, workers=False)
    
    # set device
    device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
    print('device: ', device)
    
    # set model and tokenizer
    tokenizer, model = load_tokenizer_and_model_for_train()
    model.to(device)
    print('Classifier 확인: ', model.classifier)
    
    # set data
    cls_train_dataset, cls_valid_dataset = prepare_dataset(dataset, tokenizer)
    
    # set trainer
    trainer = load_trainer_for_train(model, cls_train_dataset, cls_valid_dataset, tokenizer)
    
    # train model
    print("------Start train------")
    trainer.train()
    print("------Finish train------")
    ## model.save_pretrained("./best_model")
    
    
class args_train():
    """학습(train)에 사용되는 arguments 관리하는 class"""
    model_name = "bert-base-uncased"
    # classifier의 weight는 fine-tuning을 통해서 새로 학습시켜야!

train(df_train[:10])

df_test = pd.read_csv("cola_public_1.1/cola_public/raw/out_of_domain_dev.tsv", delimiter='\t', header=None, names=['sentence_source', 'label', 'label_notes', 'sentence'])
df_test.head()



def load_model_for_inference():
    """추론(infer)에 필요한 모델과 토크나이저 load """
    MODEL_NAME = args_test.model_name
    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
    model = BertForSequenceClassification.from_pretrained(MODEL_NAME,
                                                          num_labels=2)
    print("--- Modeling Done ---")
    return tokenizer, model
    
    
    

# 학습된 모델을 사용하여 입력 문장에 대한 결과를 추론
def inference(model, tokenized_sentence, device):
    dataloader = DataLoader(tokenized_sentence, batch_size=args_test.batch_size, shuffle=False) 
    # tokenized_sentence를 DataLoader에 넣어 배치단위로 데이터를 로드 
    
    model.eval()  # 평가모드 
    
    output_pred = []
    for i, data in enumerate(tqdm(dataloader)):  # dataloader에서 배치단위로 데이터를 가져와 추론을 수행
        with torch.no_grad():
            outputs = model(input_ids=data['input_ids'].to(device),
                            attention_mask=data['attention_mask'].to(device))
        
        logits = outputs[0]
        logits = logits.detach().cpu().numpy()  # detach(): 그래디언트 추적을 중단(torch.no_grad 블럭 안에 있기 때문에 여기서는 불필요)
        result = np.argmax(logits, axis=-1)  # logits에서 가장 높은 값을 가지는 인덱스를 result에 저장 
        #>> axis=-1: 마지막 차원을 따라 최댓값의 인덱스를 찾도록 지정 
        
        output_pred.append(result)
        
    return (np.concatenate(output_pred).tolist(),)



# 추론 후 예측한 결과(pred)를 평가(eval)
def infer_and_eval(test_dataset):
    # set device 
    device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
    
    # set model & tokenizer
    tokenizer, model = load_model_for_inference()
    model.to(device)
    
    # set data
    tokenized_test = tokenized_dataset(test_dataset, tokenizer)  # 테스트셋을 토큰화 
    test_label = test_dataset['label'].values       # 데스트셋의 'label'을 test_label로 저장 
    cls_test_dataset = ClsDataset(tokenized_test, test_label)  # 위 둘을 합쳐서 cls_test_dataset 생성 
    
    # predict answer
    pred_answer = inference(model, cls_test_dataset, device)  # 모델에서 class 추론
    print('------Prediction done------')
    
    print('test_label: ', list(test_label[:30]))
    print('pred_answer: ', pred_answer[0][:30])
    # return model, cls_test_dataset, device 
    
    mcc = matthews_corrcoef(test_label, pred_answer[0])
    print('\n------Total MCC: $.3f------' % mcc)
    #>> MCC(Mathews Correlation Coefficient)를 사용하여 CoLA 벤치마크 정확도 측정 
    #>> MCC: 분류 문제의 성능 평가 지표 
    
    

# arguments 관리하는 class 
class args_test():
    model_name = "Ruizhou/bert-base-uncased-finetuned-cola"
    # cola처럼 유명한 데이터셋은 학습된 weight가 이미 올라와있다 -> 그걸 가져와서 사용 
    batch_size = 64

infer_and_eval(df_test)