[AICE PROFESSIONAL 준비] 한국어 혐오 댓글

캐글의 [Korean Extremist Website Womad Hate Speech Data] 를 이용하여 댓글 혐오 여부를 분류하는 모델을 만들었다. 실제 댓글 데이터를 모아놓은 데이터셋인데... 상당히 불쾌한 내용의 문장도 다수 있으니 참고하길 바란다.

 

https://www.kaggle.com/datasets/captainnemo9292/korean-extremist-website-womad-hate-speech-data

Korean Extremist Website Womad Hate Speech Data

 

 

데이터 로드 및 확인

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from sklearn.model_selection import train_test_split
from collections import Counter
import re

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using Device: {device}")

# 데이터셋 로드
df = pd.read_csv("./data/hate_speech_data.csv")
df

 

df.info()

 

# 비어있는 값 확인
df.isnull().sum()

 

# 중복 확인
df["문장"].duplicated().sum()

 

# 중복되는 문장 확인
df[df["문장"].duplicated()]

 

# 중복 문장은 삭제
# 붎필요한 컬럼인 "Unnamed: 0" 도 삭제
df_drop = df.drop_duplicates(subset=["문장"])
df_drop = df_drop.drop(columns=["Unnamed: 0"])
df_drop

 

df_drop.info()

 

 

환경 설정 및 하이퍼파라미터

DEVICE = device
MAX_LEN = 64
BATCH_SIZE = 64
EMBEDDING_DIM = 128
# HIDDEN_DIM = 256
HIDDEN_DIM = 64
EPOCHS = 5
LR = 0.001

 

# 텍스트 정제 및 토큰화 함수
def clean_and_tokenize(text):
    if not isinstance(text, str):
        return []
    # 한글, 영문, 숫자만 남기고 특수문자 제거
    text = re.sub(r"[^가-힣a-zA-Z0-9\s]", "", text)
    # 단순 공백 분리 (직관적인 baseline 방식)
    return text.split()

 

# 단어사전 구축
class Vocabulary:
    def __init__(self, min_freq=2):
        self.word2idx = {"<PAD>":0, "<UNK>":1}
        self.idx2word = {0:"<PAD>", 1:"<UNK>"}
        self.min_freq = min_freq

    def build_vocab(self, sentence_list):
        frequencies = Counter()
        for sentence in sentence_list:
            tokens = clean_and_tokenize(sentence)
            frequencies.update(tokens)

        idx = 2

        for word, freq in frequencies.items():
            if freq >= self.min_freq:
                self.word2idx[word] = idx
                self.idx2word[idx] = word
                idx += 1

    def numericalize(self, sentence):
        tokens = clean_and_tokenize(sentence)
        return [self.word2idx.get(token, self.word2idx["<UNK>"]) for token in tokens]

 

# 파이토치 커스텀 데이터셋
class KoreanCommentDataset(Dataset):
    def __init__(self, df, vocab, max_len):
        self.df = df.reset_index(drop=True)
        self.vocab = vocab
        self.max_len = max_len

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

    def __getitem__(self, idx):
        # 캐글 데이터셋 피처명이 "문장", "혐오여부" 이고 이를 sentence와 label로 변경
        sentence = self.df.loc[idx, "문장"]
        label = self.df.loc[idx, "혐오 여부"]

        # 정수 인코딩
        numericalized = self.vocab.numericalize(sentence)

        # 패딩 및 잘라내기로 길이 맞추기
        if len(numericalized) < self.max_len:
            numericalized += [self.vocab.word2idx["<PAD>"]] * (self.max_len - len(numericalized))
        else:
            numericalized = numericalized[:self.max_len]

        return torch.tensor(numericalized, dtype=torch.long), torch.tensor(label, dtype=torch.float32)

 

# PyTorch 내장 LSTM 기반 분류 모델
class ClassificationSpeechLSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim):
        super(ClassificationSpeechLSTM, self).__init__()
        # 임베딩 레이어 (패딩 토큰인 0은 학습에서 제외하도록 padding_idx 지정)
        self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=0)

        # 양방향 LSTM 활용하여 문맥을 앞뒤로 파악
        # self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True, bidirectional=True, num_layers=2, dropout=0.3)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True, bidirectional=True, num_layers=1, dropout=0.3)

        # Fully Connected Layer (양방향이므로 hidden_dim * 2)
        self.fc = nn.Linear(hidden_dim * 2, 1)
        self.sigmoid = nn.Sigmoid()

    def forward(self, x):
        embedded = self.embedding(x)

        # out: 모든 시점의 은닉 상태
        # (h_n, c_n): 마지막 시점의 은닉 상내 및 셀 상태
        out, (h_n, c_n) = self.lstm(embedded)

        # 순방향 마지막 은닉 상태와 역방향 마지막 은닉 상태를 결함
        # batch_first = True 일 때 h_n은 [num_layers * num_directions, batch, hidden_dim]
        # 마지막 레이어의 정방향(Top-layer forward)과 역방향(Top-layer backward)을 추출
        hidden_forward = h_n[-2, :, :]
        hiddne_backward = h_n[-1, :, :]
        combined_hidden = torch.cat((hidden_forward, hiddne_backward), dim=1)

        return self.sigmoid(self.fc(combined_hidden)).squeeze()

 

# 가상의 데이터 로드 및 실행 파이프라인
# 데이터셋 분할
train_df, test_df = train_test_split(df_drop, test_size=0.2, random_state=42)

 

# 단어 사전 구축
# vocab = Vocabulary(min_freq=1)
vocab = Vocabulary(min_freq=2)
vocab.build_vocab(train_df["문장"].tolist())
VOCAB_SIZE = len(vocab.word2idx)
print(f"Vocab Size: {VOCAB_SIZE}")

 

# 데이터셋 & 데이터로더 생성
train_dataset = KoreanCommentDataset(train_df, vocab, MAX_LEN)
test_dataset = KoreanCommentDataset(test_df, vocab, MAX_LEN)

train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE, shuffle=False)

 

# 모델, 손실함수, 옵티마이저 선언
model = ClassificationSpeechLSTM(VOCAB_SIZE, EMBEDDING_DIM, HIDDEN_DIM).to(DEVICE)
criterion = nn.BCELoss() # 이진 분류이므로 Binary Cross Entropy 사용
optimizer = torch.optim.Adam(model.parameters(), lr=LR)

 

# 학습 루프
for epoch in range(EPOCHS):
    model.train()
    train_loss = 0
    correct = 0
    total = 0

    for inputs, labels in train_loader:
        inputs, labels = inputs.to(DEVICE), labels.to(DEVICE)

        optimizer.zero_grad()
        outputs = model(inputs)

        # 데이터 개수가 1개일 때 차원 squeeze 에러 방지
        if outputs.dim() == 0:
            outputs = outputs.unsqueeze(0)

        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

        train_loss += loss.item() * inputs.size(0)

        # 0.5 기준으로 1과 0 분류
        predicted = (outputs > 0.5).float()
        correct += (predicted == labels).sum().item()
        total += labels.size(0)

    epoch_loss = train_loss / total
    epoch_acc = (correct / total) * 100
    print(f"Epoch [{epoch+1}/{EPOCHS}] - Loss: {epoch_loss:.4f}, Acc: {epoch_acc:.2f}%")

print("학습 완료")

 

# 모델 평가 모드로 전환
model.eval()

test_loss = 0
test_correct = 0
test_total = 0

# 기울기(Gradient) 계산 비활성화 (메모리 절약 및 속도 향상)
with torch.no_grad():
    for inputs, labels in test_loader:
        inputs, labels = inputs.to(DEVICE), labels.to(DEVICE)

        # 예측 수행
        outputs = model(inputs)

        # 데이터 개수가 1개일 때 차원 squeeze 에러 방지
        if outputs.dim() == 0:
            outputs = outputs.unsqueeze(0)

        # 손실(Loss) 계산
        loss = criterion(outputs, labels)
        test_loss += loss.item() * inputs.size(0)

        # 정확도(Accuracy) 계산 (0.5 기준으로 1과 0 분류)
        predicted = (outputs > 0.5).float()
        test_correct += (predicted == labels).sum().item()
        test_total += labels.size(0)

# 최종 테스트 Loss 및 Accuracy 계산
final_test_loss = test_loss / test_total
final_test_acc = (test_correct / test_total) * 100

print(f"Test Evaluation - Loss: {final_test_loss:.4f}, Acc: {final_test_acc:.2f}%")

 

 

모델 저장

# 저장할 경로 설정
MODEL_SAVE_PATH = "korean_hate_speech_model.pt"

# 예측(Inference)을 위해 모델의 가중치(state_dict)와 단어 사전(vocab)을 함께 딕셔너리로 저장
save_data = {
    "model_state_dict": model.state_dict(),
    "vocab": vocab,
    "max_len": MAX_LEN,
    "embedding_dim": EMBEDDING_DIM,
    "hidden_dim": HIDDEN_DIM,
    "vocab_size": VOCAB_SIZE
}

# torch.save를 사용하여 .pt 파일로 저장
torch.save(save_data, MODEL_SAVE_PATH)
print(f"모델과 단어 사전이 성공적으로 저장되었습니다: {MODEL_SAVE_PATH}")

 

 

가상으로 만든 데이터를 이용하여 예측하기

# 혐오 여부 값이 없는 데이터를 만들어서 예측
target_file_path = "./data/real_data.csv"
df_new = pd.read_csv(target_file_path)

 

# 예측 결과를 담을 빈 리스트 생성
predicted_labels = []
predicted_probs = []

 

# 모델 평가 모드 전환
model.eval()

# 기울기 계산 비활성화 (메모리 절약 및 속도 향상)
with torch.no_grad():
    for sentence in df_new["문장"]:
        # 결측치(NaN)나 문자열이 아닌 데이터 예외 처리
        if not isinstance(sentence, str):
            predicted_labels.append(0)
            predicted_probs.append(0.0)
            continue

        # 텍스트 정제 및 토큰화
        text_cleaned = re.sub(r"[^가-힣a-zA-Z0-9\s]", "", sentence)
        tokens = text_cleaned.split()

        # 정수 인코딩
        numericalized = [vocab.word2idx.get(token, vocab.word2idx["<UNK>"]) for token in tokens]

        # 패딩 및 잘라내기
        if len(numericalized) < MAX_LEN:
            numericalized += [vocab.word2idx["<PAD>"]] * (MAX_LEN - len(numericalized))
        else:
            numericalized = numericalized[:MAX_LEN]

        # 텐서 변환 및 디바이스 할당 (배치 차원 추가를 위해 리스트로 한 번 더 감쌈)
        inputs = torch.tensor([numericalized], dtype=torch.long).to(DEVICE)

        # 예측 수행
        output = model(inputs)
        prob = output.item()

        # 결과 저장 (0.5 기준)
        predicted_probs.append(prob)
        predicted_labels.append(1 if prob > 0.5 else 0)

 

# 데이터프레임에 예측 결과 파생 변수 추가
df_new["혐오 여부"] = predicted_labels # 1: 혐오, 0: 정상

# 새로운 CSV 파일로 저장
save_file_path = "./result.csv"
df_new.to_csv(save_file_path, index=False, encoding="utf-8-sig")