뉴스 클리핑 분류 KoBERT 모델 설명

환경 : Colab Pro

import pandas as pd

# 엑셀 파일 로드
file_path = '/content/musma.xlsx'
data = pd.read_excel(file_path)

# 데이터의 처음 몇 행을 확인
data.head()
data.shape

데이터 전처리

# 데이터셋을 재구성하기 위해 각 카테고리 별로 데이터를 하나의 리스트에 저장합니다.
# 각 뉴스 제목에 해당 카테고리의 레이블을 부여합니다.

# 카테고리를 레이블로 매핑
categories = data.columns
label_map = {category: i for i, category in enumerate(categories)}

# (뉴스 제목, 레이블) 형식의 리스트 생성
news_data = []
for category in categories:
    news_titles = data[category].dropna().tolist()  # NaN 값 제거
    news_data.extend([(title, label_map[category]) for title in news_titles])

# 데이터셋을 DataFrame으로 변환
news_df = pd.DataFrame(news_data, columns=['news_title', 'label'])

news_df.head(), label_map

!pip install transformers

!pip install torch

import torch

from transformers import BertTokenizer, BertForSequenceClassification

from torch.utils.data import DataLoader, RandomSampler, SequentialSampler, TensorDataset, random_split

from sklearn.model_selection import train_test_split

from transformers import AdamW, get_linear_schedule_with_warmup

from tqdm import tqdm

import numpy as np

tokenizer = BertTokenizer.from_pretrained('monologg/kobert')
model = BertForSequenceClassification.from_pretrained('monologg/kobert', num_labels=4)

 

특수 토큰 추가

이유: 자연어 처리 모델에게 입력 데이터의 특정 구조나 상태를 알려주기 위함이다.

1. 시작 토큰 (예: [CLS], [START]):
   • 입력 시퀀스의 시작을 나타냅니다. 이 토큰은 모델이 입력 데이터의 시작을 인식하는 데 도움을 주며, 종종 분류 작업에서 출력의 기준점으로 사용됩니다. 예를 들어, BERT 모델에서 [CLS] 토큰은 종종 문장 분류 작업에서 결과를 예측하는 데 사용됩니다.
2. 종료 토큰 (예: [SEP], [END]):
   • 입력 시퀀스의 끝을 표시합니다. 이 토큰은 모델에게 입력 데이터의 끝을 알리는 역할을 합니다. 또한, 두 개 이상의 시퀀스를 구분하는 데에도 사용됩니다. 예를 들어, 질문과 답변 혹은 두 문장을 모델에 입력할 때 각각의 끝에 [SEP] 토큰을 추가하여 구분합니다.
3. 패딩 토큰 (예: [PAD]):
   • 모든 입력 시퀀스가 동일한 길이를 갖도록 하는 데 사용됩니다. 신경망은 일반적으로 고정된 크기의 입력을 요구하기 때문에, 짧은 시퀀스는 패딩 토큰을 사용하여 길이를 맞춥니다. 이 토큰은 모델이 실제 데이터와 패딩을 구분하도록 도와주며, 패딩된 부분은 처리 과정에서 무시됩니다.
4. 불명확한 토큰 (예: [UNK]):
   • 토크나이저의 어휘집에 없는 단어를 대체하는 데 사용됩니다. 모든 단어를 어휘집에 포함시킬 수 없기 때문에, 어휘집에 없는 단어들은 [UNK]와 같은 토큰으로 대체됩니다.

 

패딩 Padding

여기서 문장의 길이가 서로 다를 때 길이를 일정하게 만드는 과정, 즉 패딩(padding)을 하는 주된 이유는 대부분의 신경망 모델, 특히 배치(batch) 처리를 하는 모델들이 입력 데이터의 크리가 일정해야 효율적으로 작동하기 때문이다.

1. 배치 처리 최적화: 신경망은 보통 한 번에 여러 데이터 샘플을 처리하는 '배치' 단위로 작동합니다. 이 때, 모든 샘플이 동일한 길이를 가져야 각 샘플을 효율적으로 병렬 처리할 수 있습니다. 길이가 서로 다른 문장들을 하나의 배치로 묶으려면, 모든 문장이 같은 길이를 가져야 합니다.
2. 모델 구조의 일관성: 대부분의 신경망 모델은 고정된 크기의 입력을 요구합니다. 길이가 다른 입력들을 처리하기 위해서는 모든 입력이 동일한 차원을 가져야 모델이 제대로 작동할 수 있습니다.
3. 계산 효율성: 일정한 길이의 입력을 사용하면 모델이 더 효율적으로 학습하고 추론할 수 있습니다. 패딩은 불필요한 계산을 줄이고, 메모리 사용을 최적화하는 데 도움을 줍니다.

 

어텐션 마스크

이유 : 마스크는 모델이 입력 문장을 처리할 때 어떤 토큰에 집중해야 하는지를 알려주는 역할을 한다.

1. 패딩된 토큰 무시: 문장의 길이가 서로 다를 경우, 짧은 문장을 길이가 일정하도록 패딩(padding) 토큰으로 채웁니다. 어텐션 마스크는 모델이 이러한 패딩된 부분을 무시하도록 도와줍니다. 즉, 모델의 계산에서 이 부분을 고려하지 않도록 합니다.
2. 효율적인 계산: 마스크를 사용함으로써 모델은 실제 유의미한 데이터에만 집중할 수 있으며, 불필요한 계산을 줄일 수 있습니다.

# 뉴스 제목을 토큰화합니다.
input_ids = []
attention_masks = []

for news in news_df['news_title']:
    encoded_dict = tokenizer.encode_plus(
                        news,                      # 뉴스 제목을 인코딩
                        add_special_tokens = True, # 특수 토큰 추가
                        max_length = 64,           # 문장의 최대 토큰 수 설정
                        pad_to_max_length = True,  # 짧은 문장은 패딩으로 채움
                        return_attention_mask = True,   # 어텐션 마스크 반환
                        return_tensors = 'pt',     # 파이토치 텐서로 반환
                   )

    input_ids.append(encoded_dict['input_ids'])
    attention_masks.append(encoded_dict['attention_mask'])

# 토큰화된 데이터를 파이토치 텐서로 변환합니다.
input_ids = torch.cat(input_ids, dim=0)
attention_masks = torch.cat(attention_masks, dim=0)
labels = torch.tensor(news_df['label'].values)

# 훈련 세트와 검증 세트로 분할합니다.
train_inputs, validation_inputs, train_labels, validation_labels = train_test_split(input_ids, labels, random_state=2018, test_size=0.1)
train_masks, validation_masks, _, _ = train_test_split(attention_masks, labels, random_state=2018, test_size=0.1)

# 데이터 로더를 생성합니다.
train_data = TensorDataset(train_inputs, train_masks, train_labels)
train_sampler = RandomSampler(train_data)
train_dataloader = DataLoader(train_data, sampler=train_sampler, batch_size=32)

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

 

모델 로드

# KoBERT 토크나이저와 모델을 로드
tokenizer = BertTokenizer.from_pretrained('monologg/kobert')
model = BertForSequenceClassification.from_pretrained('monologg/kobert', num_labels=4)

# GPU 사용 설정
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)

# 학습 코드
epochs = 50
loss_values = []

# 옵티마이저 설정
optimizer = AdamW(model.parameters(),
                  lr = 2e-5, # 학습률
                  eps = 1e-8 # 0으로 나누는 것을 방지하기 위한 매우 작은 숫자
                )

# 총 훈련 스텝은 에폭 수와 배치 사이즈에 따라 결정
total_steps = len(train_dataloader) * epochs

# 학습률 스케줄러 설정
scheduler = get_linear_schedule_with_warmup(optimizer,
                                            num_warmup_steps = 0, # Warmup 스텝
                                            num_training_steps = total_steps)

 

from sklearn.metrics import f1_score

import matplotlib.pyplot as plt

# 손실과 정확도, F1 점수를 기록할 리스트 초기화

loss_values = []

validation_accuracy = []

validation_f1_scores = []

 

학습

for epoch_i in range(0, epochs):
    # 에폭당 학습
    print('======== Epoch {:} / {:} ========'.format(epoch_i + 1, epochs))
    total_loss = 0
    model.train()

    for step, batch in enumerate(train_dataloader):
        # 배치를 GPU에 로드
        b_input_ids = batch[0].to(device)
        b_input_mask = batch[1].to(device)
        b_labels = batch[2].to(device)

        # 그래디언트 초기화
        model.zero_grad()

        # 순전파
        outputs = model(b_input_ids,
                        token_type_ids=None,
                        attention_mask=b_input_mask,
                        labels=b_labels)

        # 손실 얻기
        loss = outputs.loss
        total_loss += loss.item()

        # 역전파
        loss.backward()

        # 그래디언트 클리핑
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)

        # 가중치 업데이트
        optimizer.step()

        # 학습률 감소
        scheduler.step()

    # 평균 손실 계산
    avg_train_loss = total_loss / len(train_dataloader)
    loss_values.append(avg_train_loss)
    print("  Average training loss: {0:.2f}".format(avg_train_loss))

    # 검증
    print("  Running Validation...")
    model.eval()
    # 변수 초기화
    total_eval_accuracy = 0
    total_eval_f1 = 0
    total_eval_examples = 0

    for batch in validation_dataloader:
        # 배치를 GPU에 로드
        b_input_ids = batch[0].to(device)
        b_input_mask = batch[1].to(device)
        b_labels = batch[2].to(device)

        # 그래디언트 계산을 하지 않도록 설정
        with torch.no_grad():
            # 순전파 진행
            outputs = model(b_input_ids,
                            token_type_ids=None,
                            attention_mask=b_input_mask)

        # 로그트와 레이블을 CPU로 이동
        logits = outputs.logits
        logits = logits.detach().cpu().numpy()
        label_ids = b_labels.to('cpu').numpy()

        # 출력 로그트의 가장 높은 점수를 받은 클래스를 예측값으로 선택
        preds = np.argmax(logits, axis=1)

        # 정확도 계산
        total_eval_accuracy += (preds == label_ids).mean()

        # F1 점수 계산
        total_eval_f1 += f1_score(label_ids, preds, average='weighted')

        # 처리한 예제 수 업데이트
        total_eval_examples += b_input_ids.size(0)

    # 평균 정확도 계산 및 기록
    avg_val_accuracy = total_eval_accuracy / len(validation_dataloader)
    validation_accuracy.append(avg_val_accuracy)
    print("  Validation Accuracy: {0:.2f}".format(avg_val_accuracy))

    # 평균 F1 점수 계산 및 기록
    avg_val_f1 = total_eval_f1 / len(validation_dataloader)
    validation_f1_scores.append(avg_val_f1)
    print("  Validation F1 Score: {0:.2f}".format(avg_val_f1))

print("Training complete!")

# 모델 저장
# model.save_pretrained('/content/drive/MyDrive/Model')
# tokenizer.save_pretrained('/content/drive/MyDrive/Model')

 

시각화

plt.figure(figsize=(12, 6))
# 학습 손실 그래프
plt.subplot(1, 3, 1)
plt.plot(loss_values, 'b-o')
plt.title("Training Loss")
plt.xlabel("Epoch")
plt.ylabel("Loss")

# 검증 정확도 그래프
plt.subplot(1, 3, 2)
plt.plot(validation_accuracy, 'r-o')
plt.title("Validation Accuracy")
plt.xlabel("Epoch")
plt.ylabel("Accuracy")

# 검증 F1 점수 그래프
plt.subplot(1, 3, 3)
plt.plot(validation_f1_scores, 'g-o')
plt.title("Validation F1 Score")
plt.xlabel("Epoch")
plt.ylabel("F1 Score")

plt.tight_layout()
plt.show()

print("Training complete!")

학습은 되지만 데이터가 적은상태

 

위 모델을 바탕으로 새로운 뉴스 데이터 태깅하기

new_news_df = pd.read_excel('/content/processed_news_articles.xlsx')

# 새 뉴스 데이터의 제목을 토크나이징 및 모델에 적합한 형태로 변환
new_input_ids = []
new_attention_masks = []

for news_title in new_news_df['Title']:
    encoded_dict = tokenizer.encode_plus(
                        news_title,                      
                        add_special_tokens = True, 
                        max_length = 64,           
                        pad_to_max_length = True,
                        return_attention_mask = True,
                        return_tensors = 'pt',     
                   )
    
    new_input_ids.append(encoded_dict['input_ids'])
    new_attention_masks.append(encoded_dict['attention_mask'])

new_input_ids = torch.cat(new_input_ids, dim=0)
new_attention_masks = torch.cat(new_attention_masks, dim=0)

# 배치 사이즈 설정
batch_size = 32

# 데이터 로더 생성
prediction_data = TensorDataset(new_input_ids, new_attention_masks)
prediction_sampler = SequentialSampler(prediction_data)
prediction_dataloader = DataLoader(prediction_data, sampler=prediction_sampler, batch_size=batch_size)

# 모델을 평가 모드로 설정
model.eval()

# 예측을 위한 변수 초기화
predictions = []

# 예측 시작
for batch in prediction_dataloader:
    # 배치를 디바이스에 추가
    batch = tuple(t.to(device) for t in batch)
    b_input_ids, b_input_mask = batch

    # 그래디언트 계산 안 함
    with torch.no_grad():
        # 순전파, 예측 수행
        outputs = model(b_input_ids, token_type_ids=None, attention_mask=b_input_mask)

    logits = outputs.logits

    # CPU로 이동
    logits = logits.detach().cpu().numpy()
    
    # 예측값 저장
    predictions.append(logits)

# 예측 결과에 대한 확신 임계값 설정 및 필터링
confidence_threshold = 0.9
filtered_prediction_labels = []
for prediction in predictions:
    probs = np.exp(prediction) / np.sum(np.exp(prediction), axis=1, keepdims=True)
    max_probs = np.max(probs, axis=1)
    preds = np.argmax(prediction, axis=1)
    filtered_preds = [pred if max_prob > confidence_threshold else 4 for pred, max_prob in zip(preds, max_probs)]
    filtered_prediction_labels.extend(filtered_preds)

new_news_df['Category'] = filtered_prediction_labels

# 결과를 Excel 파일로 저장
predicted_file_path = '/content/predicted_news_articles.xlsx'
new_news_df.to_excel(predicted_file_path, index=False)
predicted_file_path

 

추가적으로 crawling 데이터를 넣어서 모델을 학습시켜봤다.

확실히 좀더 다양한 데이터가 들어가다보니, acc가 그렇게 높지는 않다. 하지만 어느정도 일반화도 되어야하므로 이정도에서 만족해야할지도..?

ave acc = 0.62