A Multi-label Hate Speech Detection Dataset

Ryuzy 2025. 2. 13. 01:44

728x90

1. A Multi-label Hate Speech Detection Dataset

A Multi-label Hate Speech Detection Dataset은 혐오 발언(hate speech)을 더 세분화하여 여러 개의 혐오 범주를 동시에 라벨링한 데이터셋입니다. 기존의 혐오 발언 탐지 데이터셋은 보통 특정 문장이 혐오 발언인지 아닌지를 이진 분류(Binary Classification)로 다루었지만, 다중 라벨(Multi-label) 데이터셋은 한 문장이 여러 개의 혐오 범주(예: 인종 차별, 성차별, 종교적 혐오 등)에 동시에 해당할 수 있음을 반영합니다. 이러한 데이터셋은 혐오 발언 탐지 모델이 더 정교하게 학습할 수 있도록 도와주며, 소셜 미디어 및 온라인 플랫폼에서 보다 정확한 콘텐츠 모니터링을 가능하게 합니다. 연구에서는 일반적으로 NLP 기법(예: BERT, LSTM, Transformer 등)과 함께 활용되며, 다중 라벨 분류(Multi-label Classification)를 수행하기 위해 시그모이드 활성화 함수(Sigmoid Activation Function) 및 바이너리 크로스엔트로피(Binary Cross-Entropy) 손실 함수를 사용하여 각 혐오 범주별 확률을 독립적으로 예측합니다. 이 데이터셋은 사회적 이슈 분석, 콘텐츠 필터링, 온라인 플랫폼의 자동 모니터링 시스템 개발 등 다양한 응용 분야에서 활용됩니다.

!pip install transformers
!pip install datasets

from datasets import load_dataset
import pandas as pd
import numpy as np
import random
import time
import datetime
from tqdm import tqdm
from sklearn.preprocessing import MultiLabelBinarizer

import csv
import os
import torch

from transformers import pipeline
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

from torch.nn.utils.rnn import pad_sequence

from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, roc_auc_score, accuracy_score, hamming_loss

dataset = load_dataset("jeanlee/kmhas_korean_hate_speech")
dataset

dataset

dataset = load_dataset("jeanlee/kmhas_korean_hate_speech", split="test")

dataset

dataset.features

print('테스트 데이터 셋의 크기 :', len(dataset['text']))
print('첫번째 샘플 출력 :', dataset['text'][0])
print('첫번째 샘플의 레이블 출력 :', dataset['label'][0])

    0: origin (출신차별)
    1: physical (외모차별)
    2: politics (정치성향차별)
    3: profanity (혐오욕설)
    4: age (연령차별)
    5: gender (성차별)
    6: race (인종차별)
    7: religion (종교차별)
    8: not_hate_speech (혐오아님)

print('두번째 샘플 출력 :', dataset['text'][5])
print('두번째 샘플의 레이블 출력 :', dataset['label'][5])

dataset = load_dataset("jeanlee/kmhas_korean_hate_speech")

train = load_dataset("jeanlee/kmhas_korean_hate_speech", split="train")
validation = load_dataset("jeanlee/kmhas_korean_hate_speech", split="validation")
test = load_dataset("jeanlee/kmhas_korean_hate_speech", split="test")

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

train_sentences = list(map(lambda x: '[CLS] ' + str(x) + ' [SEP]', train['text']))
validation_sentences = list(map(lambda x: '[CLS] ' + str(x) + ' [SEP]', validation['text']))
test_sentences = list(map(lambda x: '[CLS] ' + str(x) + ' [SEP]', test['text']))

# 정답인 레이블의 위치에는 1, 나머지 위치에는 0을 기록합니다.
# 레이블 전처리 예시)
# [8]    -> [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0] : 의미적으로는 [no, no, no, no, no, no, no, no, no, yes]
# [2, 3] -> [0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0] : 의미적으로는 [no, no, yes, yes, no, no, no, no, no, no]

enc = MultiLabelBinarizer()

def multi_label(example):
    enc_label = enc.fit_transform(example['label'])
    float_arr = np.vstack(enc_label[:]).astype(float)
    update_label = float_arr.tolist()
    return update_label

train_labels = multi_label(train)
validation_labels = multi_label(validation)
test_labels = multi_label(test)

test_sentences[:5]

# 각 레이블은 기존에 [8], [2, 3], [2], [0], [8] 이었으며 전처리 후 아래와 같이 변경됨.
test_labels[:5]

# 한국어 BERT 중 하나인 'klue/bert-base'를 사용.
tokenizer = BertTokenizer.from_pretrained('klue/bert-base')

MAX_LEN = 128

def data_to_tensor(sentences, labels, tokenizer):
    # 정수 인코딩: 각 텍스트를 토큰화한 후 Vocabulary에 맵핑되는 정수 시퀀스로 변환
    tokenized_texts = [tokenizer.tokenize(sent) for sent in sentences]
    input_ids = [torch.tensor(tokenizer.convert_tokens_to_ids(x)) for x in tokenized_texts]

    # PyTorch의 pad_sequence 사용 (batch_first=True로 설정)
    padded_inputs = pad_sequence(input_ids, batch_first=True, padding_value=0)

    # 고정된 MAX_LEN 길이로 패딩 적용 (부족하면 0 추가, 길면 자름)
    if padded_inputs.shape[1] < MAX_LEN:
        padded_inputs = F.pad(padded_inputs, (0, MAX_LEN - padded_inputs.shape[1]), value=0)
    else:
        padded_inputs = padded_inputs[:, :MAX_LEN]

    # Attention Mask 생성 (0이 아닌 값은 1, 0은 0)
    attention_masks = (padded_inputs != 0).float()

    # 텐서 변환
    tensor_inputs = padded_inputs
    tensor_labels = torch.tensor(labels)
    tensor_masks = attention_masks

    return tensor_inputs, tensor_labels, tensor_masks

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

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)

print('훈련 데이터의 크기:', len(train_labels))
print('검증 데이터의 크기:', len(validation_labels))
print('테스트 데이터의 크기:', len(test_labels))

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

num_labels = 9

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

# 옵티마이저 선택
optimizer = AdamW(model.parameters(),
                  lr = 2e-5,
                  eps = 1e-8
                )

# 몇 번의 에포크(전체 데이터에 대한 학습 횟수)를 할 것인지 선택
epochs = 10
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 multi_label_metrics(predictions, labels, threshold=0.5):

    # 모델의 예측에 대해서 시그모이드 함수값을 통과시킨다. (batch_size, num_labels)
    sigmoid = torch.nn.Sigmoid()
    probs = sigmoid(torch.Tensor(predictions))

    # 만약 threshold 값을 넘는 경우에는 1로 예측했다고 간주한다.
    # threshold 값은 일반적으로 로지스틱 회귀 방식에 의하여 0.5를 선택하는 것이 일반적이다.
    y_pred = np.zeros(probs.shape)
    y_pred[np.where(probs >= threshold)] = 1

    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)
    roc_auc = roc_auc_score(y_true, y_pred, average = 'micro')

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

    return metrics

# 랜덤 시드값.
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)))

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:
        accum_logits.append(list(b))

    for b in label_ids:
        accum_label_ids.append(list(b))

accum_logits = np.array(accum_logits)
accum_label_ids = np.array(accum_label_ids)
results = multi_label_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']))
print("ROC-AUC: {0:.4f}".format(results['roc_auc']))

%pwd
%mkdir model

path = '/content/model/'

torch.save(model.state_dict(), path+"BERT_multilabel_model.pt")

model.load_state_dict(torch.load(path+"BERT_multilabel_model.pt"))

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:
        accum_logits.append(list(b))

    for b in label_ids:
        accum_label_ids.append(list(b))

accum_logits = np.array(accum_logits)
accum_label_ids = np.array(accum_label_ids)
results = multi_label_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']))
print("ROC-AUC: {0:.4f}".format(results['roc_auc']))

pipe = pipeline("text-classification", model=model.cuda(), tokenizer=tokenizer, device=0, max_length=512,
                return_all_scores=True, function_to_apply='sigmoid')

result = pipe('잼민이들 왜 그렇게 민폐를 끼치냐?')
print(result)

label_dict = {'LABEL_0' : '출신차별', 'LABEL_1' : '외모차별', 'LABEL_2' : '정치성향차별', \
              'LABEL_3': '혐오욕설', 'LABEL_4': '연령차별', 'LABEL_5': '성차별', 'LABEL_6' : '인종차별', \
              'LABEL_7': '종교차별', 'LABEL_8': '해당사항없음'}

def prediction(text):
  result = pipe(text)
  return [label_dict[res['label']] for res in result[0] if res['score'] > 0.5]

prediction('잼민이들 왜 그렇게 민폐를 끼치냐?')

2. NLP 태스크

NLP(Task, 자연어 처리 태스크)는 인간의 언어를 컴퓨터가 이해하고 처리할 수 있도록 하는 다양한 작업들을 의미합니다. 여기에는 텍스트 분석, 언어 모델링, 기계 번역, 감정 분석, 질의응답, 텍스트 요약, 개체명 인식, 문장 생성 등 여러 가지가 포함됩니다. 각 태스크는 언어의 의미를 추출하거나 생성하는 과정으로, 주로 텍스트 데이터에서 정보를 얻고 이를 특정 목적에 맞게 처리하는 데 사용됩니다. NLP 태스크는 기계 학습 및 딥러닝 기술을 활용하여 더욱 정확하고 효율적으로 수행됩니다.

자동 음성 인식 (ASR): 음성을 텍스트로 변환하는 기술입니다.
음성 합성 (TTS): 텍스트를 음성으로 변환하는 기술입니다.
명명된 엔터티 인식 (NER): 텍스트에서 사람, 장소, 조직 등을 식별하는 작업입니다.
품사 태깅 (POS): 단어의 품사를 태깅하는 작업입니다.
텍스트 분류: 텍스트를 미리 정의된 카테고리로 분류하는 작업입니다.
코어퍼스 해결 (Coreference Resolution): 텍스트 내에서 지시 대명사나 다른 표현들이 무엇을 참조하는지 해결하는 작업입니다.
구문 분석 (Syntactic Parsing): 문장의 구문 구조를 분석하는 작업입니다.
기계 번역 (Machine Translation): 한 언어의 텍스트를 다른 언어로 번역하는 작업입니다.
패러프레이즈 및 자연어 추론 (Paraphrase & Natural Language Inference): 두 문장이 같은 의미를 지니는지, 또는 한 문장이 다른 문장에서 유추할 수 있는지 확인하는 작업입니다.
대화형 에이전트 (Dialogue Agents): 대화형 인공지능 시스템을 구축하는 작업입니다.
의미적 구문 분석 (Semantic Parsing): 문장의 의미를 분석하고 추출하는 작업입니다.
질문 응답 (QA): 사용자의 질문에 대한 정확한 답변을 제공하는 작업입니다.
요약 (Summarization): 텍스트의 핵심 내용을 간결하게 요약하는 작업입니다.
감정 분석 (Sentiment Analysis): 텍스트에서 긍정적, 부정적 감정을 분석하는 작업입니다.

728x90