1 KorNLI 분류

데이터 셋: https://github.com/kakaobrain/kor-nlu-datasets

2 데이터셋 로드 및 구조 확인


from datasets import load_dataset

cs = load_dataset("klue", "nli", split="train")
cs = cs.train_test_split(0.1)
test_cs = load_dataset("klue", "nli", split="validation")
train_cs = cs["train"]
valid_cs = cs["test"]

train_cs
valid_cs
test_cs

3 전처리


import pandas as pd
import numpy as np
import random
import time
import datetime
from tqdm import tqdm

import csv
import os

import tensorflow as tf
import torch

# BERT 사용을 위함
from transformers import BertTokenizer
from transformers import BertForSequenceClassification, AdamW, BertConfig
from transformers import get_linear_schedule_with_warmup
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, SequentialSampler

# for padding
from tensorflow.keras.preprocessing.sequence import pad_sequences 

# 전처리 및 평가 지표
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, roc_auc_score, accuracy_score, hamming_loss

훈련 데이터, 검증 데이터, 테스트 데이터에 대해서 [CLS] 문장 [SEP] 구조를 만듭니다. [CLS]는 분류를 하기 위해 BERT가 사용하는 첫번째 입력 토큰이며, [SEP]는 입력 문장의 종료를 나타내기 위해 사용하는 스페셜 토큰입니다.

# 훈련 데이터, 검증 데이터, 테스트 데이터에 대해서 `[CLS] 문장 [SEP]` 구조를 만듭니다.

train_sentences = list(map(lambda train_cs: '[CLS] ' + str(train_cs['premise']) + ' [SEP] ' + str(train_cs['hypothesis']) + ' [SEP]', train_cs))
validation_sentences = list(map(lambda valid_cs: '[CLS] ' + str(valid_cs['premise']) + ' [SEP] ' + str(valid_cs['hypothesis']) + ' [SEP]', valid_cs))
test_sentences = list(map(lambda test_cs: '[CLS] ' + str(test_cs['premise']) + ' [SEP] ' + str(test_cs['hypothesis']) + ' [SEP]', test_cs))

train_labels = train_cs['label']
validation_labels = valid_cs['label']
test_labels = test_cs['label']

test_sentences[:5]
test_labels[:5]

4 BERT 토크나이저를 이용한 전처리

BERT를 사용하기 위해서는 토크나이저와 모델이 반드시 맵핑 관계여야만 합니다. 다시 말해 아래의 이름에 들어가는 모델이름은 반드시 동일해야 합니다.

BertTokenizer.from_pretrained('모델이름')
BertForSequenceClassification.from_pretrained("모델이름")

토크나이저는 내부적으로 Vocabulary를 갖고 있어 정수 인코딩을 수행해주는 모듈입니다.

# 한국어 BERT 중 하나인 'klue/bert-base'를 사용.
tokenizer = BertTokenizer.from_pretrained('klue/bert-base')
tokenized_text = tokenizer.tokenize('안녕하세요. 자연어 처리를 배울거에요.')
input_id = tokenizer.convert_tokens_to_ids(tokenized_text)

print('토큰화 된 문장 :', tokenized_text)
print('정수 인코딩 된 문장 :', input_id)

MAX_LEN = 128

def data_to_tensor (sentences, labels):
  # 정수 인코딩 과정. 각 텍스트를 토큰화한 후에 Vocabulary에 맵핑되는 정수 시퀀스로 변환한다.
  # ex) ['안녕하세요'] ==> ['안', '녕', '하세요'] ==> [231, 52, 45]
  tokenized_texts = [tokenizer.tokenize(sent) for sent in sentences]
  input_ids = [tokenizer.convert_tokens_to_ids(x) for x in tokenized_texts]

  # pad_sequences는 패딩을 위한 모듈. 주어진 최대 길이를 위해서 뒤에서 0으로 채워준다.
  # ex) [231, 52, 45] ==> [231, 52, 45, 0, 0, 0]
  input_ids = pad_sequences(input_ids, maxlen=MAX_LEN, dtype="long", truncating="post", padding="post") 

  attention_masks = []

  for seq in input_ids:
      seq_mask = [float(i > 0) for i in seq]
      attention_masks.append(seq_mask)

  tensor_inputs = torch.tensor(input_ids)
  tensor_labels = torch.tensor(labels)
  tensor_masks = torch.tensor(attention_masks)

  return tensor_inputs, tensor_labels, tensor_masks

train_inputs, train_labels, train_masks = data_to_tensor(train_sentences, train_labels)
validation_inputs, validation_labels, validation_masks = data_to_tensor(validation_sentences, validation_labels)
test_inputs, test_labels, test_masks = data_to_tensor(test_sentences, test_labels)

tokenizer.decode([2])
tokenizer.decode([3])
test_inputs[0]
tokenizer.decode(test_inputs[0])

훈련 데이터, 검증 데이터, 텍스트 데이터에 대해서 data_to_tensor 함수를 통해서 정수 인코딩 된 데이터, 레이블, 어텐션 마스크를 얻습니다.

배치 크기는 32로 하고 파이토치의 데이터로더(배치 단위로 데이터를 꺼내올 수 있도록 하는 모듈)로 변환합니다.

batch_size = 32

train_data = TensorDataset(train_inputs, train_masks, train_labels)
train_sampler = RandomSampler(train_data)
train_dataloader = DataLoader(train_data, sampler=train_sampler, batch_size=batch_size)

validation_data = TensorDataset(validation_inputs, validation_masks, validation_labels)
validation_sampler = SequentialSampler(validation_data)
validation_dataloader = DataLoader(validation_data, sampler=validation_sampler, batch_size=batch_size)

test_data = TensorDataset(test_inputs, test_masks, test_labels)
test_sampler = RandomSampler(test_data)
test_dataloader = DataLoader(test_data, sampler=test_sampler, batch_size=batch_size)

5 GPU가 정상 셋팅되었는지 확인.

Colab에서 GPU를 사용하기 위해서는 아래와 같이 설정이 되어있어야만 합니다.

런타임 > 런타임 유형 변경 > 하드웨어 가속기 > ‘GPU’ 선택

if torch.cuda.is_available():    
    device = torch.device("cuda")
    print('There are %d GPU(s) available.' % torch.cuda.device_count())
    print('We will use the GPU:', torch.cuda.get_device_name(0))
else:
    device = torch.device("cpu")
    print('No GPU available, using the CPU instead.')

6 모델로드

BERT를 사용하여 텍스트를 분류하는 BERT 아키텍처는 BertForSequenceClassification.from_pretrained(“모델이름”)을 넣어서 가능합니다. 레이블 수로 num_labels라는 인자값에 레이블의 수를 기재해줍니다.

num_labels = 3

model = BertForSequenceClassification.from_pretrained("klue/bert-base", num_labels=num_labels)
model.cuda()

# 옵티마이저 선택
optimizer = AdamW(model.parameters(),
                  lr = 2e-5,
                  eps = 1e-8
                )
# 몇 번의 에포크(전체 데이터에 대한 학습 횟수)를 할 것인지 선택
epochs = 2
total_steps = len(train_dataloader) * epochs
scheduler = get_linear_schedule_with_warmup(optimizer, 
                                            num_warmup_steps = 0,
                                            num_training_steps = total_steps)

def format_time(elapsed):
    elapsed_rounded = int(round((elapsed)))
    return str(datetime.timedelta(seconds=elapsed_rounded))  # hh:mm:ss

def metrics(predictions, labels):
    y_pred = predictions
    y_true = labels

    # 사용 가능한 메트릭들을 사용한다.
    accuracy = accuracy_score(y_true, y_pred)
    f1_macro_average = f1_score(y_true=y_true, y_pred=y_pred, average='macro', zero_division=0)
    f1_micro_average = f1_score(y_true=y_true, y_pred=y_pred, average='micro', zero_division=0)
    f1_weighted_average = f1_score(y_true=y_true, y_pred=y_pred, average='weighted', zero_division=0)

    # 메트릭 결과에 대해서 리턴
    metrics = {'accuracy': accuracy,
               'f1_macro': f1_macro_average,
               'f1_micro': f1_micro_average,
               'f1_weighted': f1_weighted_average}

    return metrics

7 모델 학습

# 랜덤 시드값.
seed_val = 777
random.seed(seed_val)
np.random.seed(seed_val)
torch.manual_seed(seed_val)
torch.cuda.manual_seed_all(seed_val)

model.zero_grad()
for epoch_i in range(0, epochs):
    print('======== Epoch {:} / {:} ========'.format(epoch_i + 1, epochs))
    t0 = time.time()
    total_loss = 0

    model.train()

    for step, batch in tqdm(enumerate(train_dataloader)):
        if step % 500 == 0 and not step == 0:
            elapsed = format_time(time.time() - t0)
            print('  Batch {:>5,}  of  {:>5,}.    Elapsed: {:}.'.format(step, len(train_dataloader), elapsed))

        batch = tuple(t.to(device) for t in batch)
        b_input_ids, b_input_mask, b_labels = batch

        outputs = model(b_input_ids, 
                        token_type_ids=None, 
                        attention_mask=b_input_mask, 
                        labels=b_labels)
        
        loss = outputs[0]
        total_loss += loss.item()
        loss.backward()

        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)  # gradient clipping if it is over a threshold
        optimizer.step()
        scheduler.step()

        model.zero_grad()

    avg_train_loss = total_loss / len(train_dataloader)            

    print("")
    print("  Average training loss: {0:.4f}".format(avg_train_loss))
    print("  Training epcoh took: {:}".format(format_time(time.time() - t0)))

8 검증 데이터에 대한 평가

t0 = time.time()
model.eval()
accum_logits, accum_label_ids = [], []

for batch in validation_dataloader:
    batch = tuple(t.to(device) for t in batch)
    b_input_ids, b_input_mask, b_labels = batch

    with torch.no_grad():
        outputs = model(b_input_ids, 
                        token_type_ids=None, 
                        attention_mask=b_input_mask)

    logits = outputs[0]
    logits = logits.detach().cpu().numpy()
    label_ids = b_labels.to('cpu').numpy()

    for b in logits:
        # 3개의 값 중 가장 큰 값을 예측한 인덱스로 결정
        # ex) [ 3.5134246  -0.30875662 -2.111316  ] ==> 0
        accum_logits.append(np.argmax(b))

    for b in label_ids:
        accum_label_ids.append(b)

accum_logits = np.array(accum_logits)
accum_label_ids = np.array(accum_label_ids)
results = metrics(accum_logits, accum_label_ids)

print("Accuracy: {0:.4f}".format(results['accuracy']))
print("F1 (Macro) Score: {0:.4f}".format(results['f1_macro']))
print("F1 (Micro) Score: {0:.4f}".format(results['f1_micro']))
print("F1 (Weighted) Score: {0:.4f}".format(results['f1_weighted']))

9 모델 저장과 로드

%pwd

# 폴더 생성
%mkdir model
# 모델 저장
torch.save(model.state_dict(), path+"BERT_kornli.pt")

# 모델 로드
model.load_state_dict(torch.load(path+"BERT_kornli.pt"))

10 테스트 데이터에 대한 평가

t0 = time.time()
model.eval()
accum_logits, accum_label_ids = [], []

for step, batch in tqdm(enumerate(test_dataloader)):
    if step % 100 == 0 and not step == 0:
        elapsed = format_time(time.time() - t0)
        print('  Batch {:>5,}  of  {:>5,}.    Elapsed: {:}.'.format(step, len(test_dataloader), elapsed))

    batch = tuple(t.to(device) for t in batch)
    b_input_ids, b_input_mask, b_labels = batch

    with torch.no_grad():
        outputs = model(b_input_ids, 
                        token_type_ids=None, 
                        attention_mask=b_input_mask)

    logits = outputs[0]
    logits = logits.detach().cpu().numpy()
    label_ids = b_labels.to('cpu').numpy()
    
    for b in logits:
        # 3개의 값 중 가장 큰 값을 예측한 인덱스로 결정
        # ex) [ 3.5134246  -0.30875662 -2.111316  ] ==> 0
        accum_logits.append(np.argmax(b))

    for b in label_ids:
        accum_label_ids.append(b)

accum_logits = np.array(accum_logits)
accum_label_ids = np.array(accum_label_ids)
results = metrics(accum_logits, accum_label_ids)

print("Accuracy: {0:.4f}".format(results['accuracy']))
print("F1 (Macro) Score: {0:.4f}".format(results['f1_macro']))
print("F1 (Micro) Score: {0:.4f}".format(results['f1_micro']))
print("F1 (Weighted) Score: {0:.4f}".format(results['f1_weighted']))

11 예측


from transformers import pipeline

pipe = pipeline("text-classification", model=model.cuda(), tokenizer=tokenizer, device=0, max_length=512,
                return_all_scores=True, function_to_apply='softmax')
inputs = {"text" : "흡연자분들은 발코니가 있는 방이면 발코니에서 흡연이 가능합니다.", "text_pair" : "어떤 방에서도 흡연은 금지됩니다."}
result = pipe([inputs])
print(result)

return_all_scores를 제거하면 정답으로 확신하는 레이블만 리턴합니다.

# return_all_scores 제거
pipe = pipeline("text-classification", model=model.cuda(), tokenizer=tokenizer, device=0, max_length=512, function_to_apply='softmax')
result = pipe([inputs])
print(result)

한글로 된 레이블 예측을 얻기 위해서 label_dict를 만듭니다.

label_dict = {'LABEL_0' : '얽힘', 'LABEL_1' : '중립', 'LABEL_2' : '모순'}
def prediction(sent1, sent2):
  text = {"text" : sent1, "text_pair" : sent2}
  result = pipe(text)
  return [label_dict[result['label']]]

sent1 = "흡연자분들은 발코니가 있는 방이면 발코니에서 흡연이 가능합니다."
sent2 = "어떤 방에서도 흡연은 금지됩니다"

prediction(sent1, sent2)

sent1 = "저는, 그냥 알아내려고 거기 있었어요."
sent2 = "나는 돈이 어디로 갔는지 이해하려고 했어요."

prediction(sent1, sent2)

sent1 = "저는 그것을 이해하려고 거기 있었어요."
sent2 = "저는 이해하려고 노력하고 있었어요."

prediction(sent1, sent2)