텍스트 데이터 EDA와 Fake News Detection 모델 파이썬으로 구현해보기 2

2024.09.01 - [일상/컴퓨터] - Fake News Detection 모델 파이썬으로 구현해보기

Fake News Detection 모델 파이썬으로 구현해보기

지난번에는 BART 모델을 라이브러리에서 불러와서 학습을 시키는 것으로 실험을 진행한 성능 평가에서 좋지 못한 성능을 보였다.

이번에는 직접 모델에 사용하는 마스킹 함수 등을 구현해서 라이브러리에서 불러오는 것이 아닌 가내수공업(?) 모델로 성능을 내볼 것이다.

 

 

 

이번에 사용하는 데이터는 이것이다. (마치 요리 유튜브에서 고기 소개해주는 멘트 같다)

https://www.kaggle.com/datasets/saurabhshahane/fake-news-classification/data

Fake News Classification

실행은 구글 코랩에서 실행을 진행하였다.

데이터를 사용하기 전에 EDA를 통해 데이터 상태를 살짝 확인해 보겠다.

데이터를 파일째로 사용하기에는 읽히는 방식에 문제가 있는지 파일이 아닌 주소 페이지에서 파일을 만들어 내는 형식으로 데이터를 불러오겠다.

import pandas as pd

# 웹 페이지에서 CSV 데이터를 읽을 때 첫 번째 열을 인덱스로 설정
url = "https://zenodo.org/records/4561253/files/WELFake_Dataset.csv"
data = pd.read_csv(url, sep=",", index_col=0, quoting=1)

데이터 개수와 타입을 확인하기 위해 아래 코드를 실행해준다.

print(data.info())
print(data.head())

<class 'pandas.core.frame.DataFrame'>
Index: 72134 entries, 0 to 72133
Data columns (total 3 columns):
 #   Column  Non-Null Count  Dtype 
---  ------  --------------  ----- 
 0   title   71576 non-null  object
 1   text    72095 non-null  object
 2   label   72134 non-null  int64 
dtypes: int64(1), object(2)
memory usage: 2.2+ MB
None
                                               title  \
0  LAW ENFORCEMENT ON HIGH ALERT Following Threat...   
1                                                NaN   
2  UNBELIEVABLE! OBAMA’S ATTORNEY GENERAL SAYS MO...   
3  Bobby Jindal, raised Hindu, uses story of Chri...   
4  SATAN 2: Russia unvelis an image of its terrif...   

                                                text  label  
0  No comment is expected from Barack Obama Membe...      1  
1     Did they post their votes for Hillary already?      1  
2   Now, most of the demonstrators gathered last ...      1  
3  A dozen politically active pastors came here f...      0  
4  The RS-28 Sarmat missile, dubbed Satan 2, will...      1

non-null 카운트를 보니 각 열별 카운트 개수가 맞지가 않는다. 아무래도 NULL값이 존재하는 것 같으니 좀 더 시각적인 방법으로 확인을 해보겠다.

# 결측치 확인
print(data.isnull().sum())

# 결측치 시각화
plt.figure(figsize=(10, 6))
sns.heatmap(data.isnull(), cbar=False, cmap='viridis')
plt.title("Missing Values Heatmap")
plt.show()

타이틀에는 558개, text에는 39개의 결측치가 존재한다. 이것을 사용하기 전에 미리 처리하는 것이 좋겠다.

그 다음에는 레이블 불균형이 존재하는지 확인하였다.

# 데이터 분포 확인
print(data['label'].value_counts())

# 데이터 분포 시각화
plt.figure(figsize=(6, 4))
sns.countplot(x='label', data=data)
plt.title("Label Distribution")
plt.show()

fake news인 0의 개수는 36509, real news인 1의 개수는 35028개이다. 데이터 총 개수에 비해 1천개 정도면 데이터 학습을 진행하는데에 있어서 딱히 불균형이 있다고는 생각이 들지 않아서 따로 조정은 하지 않았다. 

아무래도 텍스트 데이터니까 title과 text에서 주로 나타나는 단어에 대해 탐색해보기로 하였다.

from sklearn.feature_extraction.text import CountVectorizer

# 제목에서 가장 많이 등장하는 단어 확인
vectorizer = CountVectorizer(stop_words='english', max_features=20)
title_words = vectorizer.fit_transform(data['title'].dropna())
title_word_count = np.sum(title_words.toarray(), axis=0)

# 데이터프레임 생성
title_word_df = pd.DataFrame({
    'word': vectorizer.get_feature_names_out(),
    'count': title_word_count
}).sort_values(by='count', ascending=False)

plt.figure(figsize=(10, 6))
sns.barplot(x='count', y='word', data=title_word_df)
plt.title("Top 20 Most Frequent Words in Titles")
plt.show()

# TfidfVectorizer 사용하여 문서-단어 행렬 생성
from sklearn.feature_extraction.text import TfidfVectorizer

tfidf_vectorizer = TfidfVectorizer(max_features=100, stop_words='english')
tfidf_matrix = tfidf_vectorizer.fit_transform(data['text'].dropna())
tfidf_feature_names = tfidf_vectorizer.get_feature_names_out()

# TF-IDF 점수 시각화
tfidf_scores = np.sum(tfidf_matrix.toarray(), axis=0)
tfidf_df = pd.DataFrame({
    'word': tfidf_feature_names,
    'score': tfidf_scores
}).sort_values(by='score', ascending=False).head(20)

plt.figure(figsize=(10, 6))
sns.barplot(x='score', y='word', data=tfidf_df)
plt.title("Top 20 Words by TF-IDF Score in Texts")
plt.show()

타이틀은 길이가 짧고 중요한 핵심적인 내용만으로 구성이 되어 있기에 빈도수만 체크하였고, 본문 텍스트 데이터에서는 TF-IDF를 확인하였다. TF-IDF(Term Frequency-Inverse Document Frequency)는 문서 내 단어의 중요도를 계산하는 통계적 방법이다. 이 기법은 단어의 빈도와 역문서 빈도를 고려하여 단어가 문서에서 얼마나 중요한지를 측정한다. 단어 빈도(TF)는 특정 문서에서 해당 단어가 등장하는 횟수를 나타낸다. 역문서 빈도(IDF)는 전체 문서 집합에서 단어의 희소성을 나타낸다. TF-IDF 값이 높을수록 그 단어는 특정 문서에서 더 중요한 의미를 가진다. 즉, 자주 등장하면서 희소성이 높은 유의미한 단어를 찾는 것이다.

타이틀과 텍스트 모두 정치 관련 단어가 높게 나왔다. 뉴스에서 아무래도 주로 다루는 내용이다 보니 fake든 real이든 뉴스로 생산되기 쉬운 주제이다.

 

 

이제 분석한 데이터를 토대로 BART 기반 모델을 만들어서 fake news 분류 모델을 만들어보자.

import pandas as pd
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from transformers import RobertaTokenizer, RobertaModel
from tqdm import tqdm
import tensorflow as tf
from transformers import RobertaTokenizer, TFRobertaModel
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

먼저 모델을 사용하는데 있어서 필요한 라이브러리들을 불러와 준다.

X = pd.DataFrame(data=data['title'])
Y = pd.DataFrame(data=data['label'])

texts = X['title'].tolist()
labels = Y['label'].tolist()
texts = [str(text) if pd.notna(text) else "" for text in texts]

그 다음 데이터를 X와 Y로 나누어준다. 모든 텍스트 데이터를 문자열로 통일하여 처리 과정에서 데이터 타입에 따른 오류를 방지하고, 텍스트 데이터를 머신러닝 모델에 입력하기 전에 적절한 형태로 준비한다.

 

train_texts, val_texts, train_labels, val_labels = train_test_split(texts, labels, test_size=0.2, shuffle=True)

데이터를 훈련, 검증 데이터로 나누어 준다. 

tokenizer = RobertaTokenizer.from_pretrained('roberta-base')

# RoBERTa 토크나이저 로드
tokenizer = RobertaTokenizer.from_pretrained('roberta-base')

# RoBERTa 모델을 위한 전처리 함수 (배치 처리)
def preprocess_for_roberta(texts, tokenizer, batch_size=32, max_length=128):
    input_ids = []
    attention_masks = []
    for i in tqdm(range(0, len(texts), batch_size), desc="Processing"):
        batch_texts = texts[i:i + batch_size]
        # 수정된 부분: batch_encode_plus 사용
        batch_inputs = tokenizer.batch_encode_plus(
            batch_texts,
            max_length=max_length,
            truncation=True,
            padding='max_length',
            add_special_tokens=True,
            return_attention_mask=True,
            return_tensors='tf'
        )
        input_ids.append(batch_inputs['input_ids'])
        attention_masks.append(batch_inputs['attention_mask'])

    return tf.concat(input_ids, axis=0), tf.concat(attention_masks, axis=0)

train_input_ids, train_attention_masks = preprocess_for_roberta(train_texts, tokenizer)
val_input_ids, val_attention_masks = preprocess_for_roberta(val_texts, tokenizer)

def create_transformer_input(input_ids, attention_masks, batch_size=32):
    dataset = tf.data.Dataset.from_tensor_slices((input_ids, attention_masks)).batch(batch_size)

    features = []
    for batch_input_ids, batch_attention_masks in tqdm(dataset, desc="Generating Features"):
        outputs = roberta_model(batch_input_ids, attention_mask=batch_attention_masks)
        features.append(outputs.last_hidden_state[:, 0, :])  # 첫 번째 토큰(CLS)의 벡터 사용

    return tf.concat(features, axis=0)

train_features = create_transformer_input(train_input_ids, train_attention_masks)
val_features = create_transformer_input(val_input_ids, val_attention_masks)

RoBERTa 토크나이저를 불러와서 텍스트 데이터를 전처리를 해준다. 

import tensorflow as tf
from tensorflow.keras import layers, models
import numpy as np

# Denoising Autoencoder: Input에서 일부 토큰을 무작위로 마스킹
def add_noise_and_mask(inputs, mask_token_id, mask_prob=0.15):
    mask = np.random.rand(*inputs.shape) < mask_prob
    masked_inputs = np.where(mask, mask_token_id, inputs)
    return masked_inputs, mask

# Transformer Encoder
def transformer_encoder(inputs, head_size, num_heads, ff_dim, dropout=0):
    x = layers.MultiHeadAttention(key_dim=head_size, num_heads=num_heads, dropout=dropout)(inputs, inputs)
    x = layers.Dropout(dropout)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    res = x + inputs
    x = layers.Conv1D(filters=ff_dim, kernel_size=1, activation='relu')(res)
    x = layers.Dropout(dropout)(x)
    x = layers.Conv1D(filters=head_size, kernel_size=1)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    return x + res

# Transformer Decoder with masked self-attention
def transformer_decoder(inputs, encoder_outputs, head_size, num_heads, ff_dim, dropout=0):
    # Masked Self-Attention
    causal_mask = tf.linalg.band_part(tf.ones((inputs.shape[1], inputs.shape[1])), -1, 0)  # Lower triangular matrix
    x = layers.MultiHeadAttention(key_dim=head_size, num_heads=num_heads, dropout=dropout)(
        inputs, inputs, attention_mask=causal_mask
    )
    x = layers.Dropout(dropout)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    res = x + inputs

    # Encoder-Decoder Cross-Attention
    x = layers.MultiHeadAttention(key_dim=head_size, num_heads=num_heads, dropout=dropout)(res, encoder_outputs)
    x = layers.Dropout(dropout)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    res = x + res

    # Feed Forward
    x = layers.Conv1D(filters=ff_dim, kernel_size=1, activation='relu')(res)
    x = layers.Dropout(dropout)(x)
    x = layers.Conv1D(filters=head_size, kernel_size=1)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    return x + res

# FND-NS 모델 빌드 함수
def build_fnd_ns_model(input_shape, head_size, num_heads, ff_dim, num_encoder_layers, num_decoder_layers, dropout=0):
    encoder_inputs = layers.Input(shape=input_shape)
    decoder_inputs = layers.Input(shape=input_shape)

    # Segment Embeddings
    segment_inputs = layers.Input(shape=(input_shape[0],), dtype=tf.int32)
    segment_embeddings = layers.Embedding(2, input_shape[1])(segment_inputs)
    encoder_inputs_with_segment = layers.Add()([encoder_inputs, segment_embeddings])

    # Encoder
    x = encoder_inputs_with_segment
    for _ in range(num_encoder_layers):
        x = transformer_encoder(x, head_size, num_heads, ff_dim, dropout)
    encoder_outputs = x

    # Decoder
    y = decoder_inputs
    for _ in range(num_decoder_layers):
        y = transformer_decoder(y, encoder_outputs, head_size, num_heads, ff_dim, dropout)

    # Output Layer for binary classification
    outputs = layers.Dense(1, activation='sigmoid')(y[:, 0, :])  # Sigmoid activation for binary classification

    model = models.Model(inputs=[encoder_inputs, decoder_inputs, segment_inputs], outputs=outputs)
    return model

논문에서는 BART 모델을 사용한다고 했다. BART(Bidirectional and Auto-Regressive Transformers)는 양방향 인코더와 자동 회귀 디코더를 결합한 모델이다.인코더에서는 입력 시퀀스를 양방향으로 인코딩하여 모든 토큰이 전체 시퀀스의 정보를 활용할 수 있고, 디코더는 자동 회귀 방식으로, 각 토큰은 이전의 토큰들만 참조하여 다음 토큰을 예측한다. 이를 위해 디코더의 Self-Attention에 마스킹(causal masking)을 적용한다. 또한 입력에 랜덤하게 마스킹을 해서 BART의 사전 학습 방식인 노이즈 추가 및 복원 작업을 유사하게 구현해보고자 하였다. BART는 입력 시퀀스에 노이즈를 주입하고, 이를 복원하는 과정에서 학습한다.

# 모델 설정
input_shape = (1, 768)  # 시퀀스 길이 줄이기
head_size = 768
num_heads = 8
ff_dim = 3072
num_encoder_layers = 6
num_decoder_layers = 6
dropout = 0.1

# 모델 빌드 및 컴파일
model = build_fnd_ns_model(
    input_shape=input_shape,
    head_size=head_size,
    num_heads=num_heads,
    ff_dim=ff_dim,
    num_encoder_layers=num_encoder_layers,
    num_decoder_layers=num_decoder_layers,
    dropout=dropout
)

적절한 하이퍼파라미터를 설정하고 모델 구조를 확인한다.

model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=2e-5),
              loss='binary_crossentropy',  # Binary Cross Entropy for binary classification
              metrics=['accuracy'])

model.summary()

RoBERTa로 준비된 데이터를 다시 모델 학습에 맞게 마스킹 추가 및 전처리를 해준다.

# 마스킹된 입력 생성
mask_token_id = 1  # Assuming <mask> token id is 1
train_masked_inputs, train_masks = add_noise_and_mask(train_input_ids.numpy(), mask_token_id)
val_masked_inputs, val_masks = add_noise_and_mask(val_input_ids.numpy(), mask_token_id)

# 데이터 준비: train_masked_inputs와 val_masked_inputs을 올바른 모양으로 맞추기
train_masked_inputs = tf.expand_dims(train_features, 1)
val_masked_inputs = tf.expand_dims(val_features, 1)

# 입력 데이터 형태를 (batch_size, 1, 768)로 맞추기
train_masked_inputs = tf.reshape(train_masked_inputs, (-1, 1, 768))
val_masked_inputs = tf.reshape(val_masked_inputs, (-1, 1, 768))

# 세그먼트 입력 (0으로 채움)
train_segments = tf.zeros((train_features.shape[0], 1), dtype=tf.int32)
val_segments = tf.zeros((val_features.shape[0], 1), dtype=tf.int32)

# train_features와 val_features의 차원을 (batch_size, 1, 768)로 변경
train_features = tf.expand_dims(train_features, 1)
val_features = tf.expand_dims(val_features, 1)

# 데이터 형태 확인
print("train_masked_inputs shape:", train_masked_inputs.shape)
print("train_features shape:", train_features.shape)
print("val_masked_inputs shape:", val_masked_inputs.shape)
print("val_features shape:", val_features.shape)

최종적으로 구현된 모델에 학습을 진행한다.

# 모델 학습
history = model.fit(
    [train_masked_inputs, train_features, train_segments],
    tf.convert_to_tensor(train_labels),
    validation_data=([val_masked_inputs, val_features, val_segments], tf.convert_to_tensor(val_labels)),
    epochs=10,
    batch_size=32
)

이후 모델 평가와 confusion matrix까지 출력을 했다.

# 모델 평가
val_predictions = model.predict([val_masked_inputs, val_masked_inputs, val_segments])
val_predictions_binary = (val_predictions > 0.5).astype(int)

from sklearn.metrics import confusion_matrix, classification_report, ConfusionMatrixDisplay

# 혼동 행렬 및 성능 지표 출력
conf_matrix = confusion_matrix(val_labels, val_predictions_binary)
print("Confusion Matrix:")
print(conf_matrix)

# 정밀도, 재현율, F1 스코어 등 출력
print("\nClassification Report:")
print(classification_report(val_labels, val_predictions_binary))

import matplotlib.pyplot as plt

# 혼동 행렬 시각화
disp = ConfusionMatrixDisplay(confusion_matrix=conf_matrix)
disp.plot(cmap=plt.cm.Blues)
plt.show()

지난번 포스팅에 비하면 확실히 좋은 성능을 보이고 있다.

이번 모델로 새롭게 성능을 내보고자 여러 시도를 하면서 깨달았는데, 더 빠르게 학습을 진행하고자 시도한 혼합 정밀도 훈련(Mixed Precision Training)이 지난번 실험에서 정확도를 떨어뜨리게 한 주 요인 중 하나일 수 있다는 것이다. 이 기법은 부동소수점 수의 정밀도를 32비트(float32)에서 16비트(float16)로 낮추어 연산 효율을 높인다. 이를 통해 GPU의 계산 속도를 증가시키고 더 큰 배치 크기를 사용할 수 있어 학습 시간이 단축되는 장점이 있다. 이를 도입하여 모델 학습 속도를 향상시키고자 했지만, 실제로는 모델의 정확도가 떨어지는 현상을 경험했다. 이는 부동소수점 정밀도가 낮아지면서 계산 오차가 증가하거나 손실 스케일링 과정에서 문제가 발생했을 가능성이 있다. 낮은 정밀도로 인해 연산 과정에서의 숫자 표현 범위가 줄어들어 일부 연산의 정확도가 떨어질 수 있기 때문이다. 특히, 작은 값들을 다루는 모델이나 민감한 연산이 많은 경우 이러한 문제가 두드러질 수 있다. 이러한 이유로 혼합 정밀도 훈련을 사용할 때는 모델의 안정성과 성능 저하 여부를 면밀히 검토하는 것이 중요하다.

물론 이것을 도입하지 않고도 실행했을 때도 오히려 직접 만들어서 더 단순한 모델에서 성능이 좋게 나온 것은 신기하긴 하다.