반응형
아이펠 과정에서 진행하는 세 번의 해커톤 중 하나인 DLthon을 11월 6일부터 11일까지 총 6일 동안 진행했다.
나를 포함한 네 명의 팀을 구성해서 캐글 competition으로 참가했는데, 주제는 아래와 같다.
Overview
대화의 성격을 위협 세부 클래스 4개 또는 일반 대화 중 하나로 예측하는 과제
- 학습 데이터는 '협박', '갈취', '직장 내 괴롭힘', '기타 괴롭힘' 등 4개 클래스 각 1,000개 내외로 구성
- 테스트 데이터는 '협박', '갈취', '직장 내 괴롭힘', '기타 괴롭힘', '일반 대화' 등 5개 클래스 각 100개로 구성
- train data에는 없지만, test data는 일반 대화 클래스가 존재합니다. 5개 문장을 분류할 수 있게 train data에 일반 대화 데이터셋을 추가합니다.
- 4개의 위협 세부 클래스는 Augmentation만 가능(새로운 데이터 추가/생성 불가)
- 일반대화 클래스는 합성데이터로 구성
- 다양한 프롬프트로 문장을 생성하고 학습에 활용
- 최종 결과물로 제출할 수 있는 건 합성데이터 기반의 성능뿐입니다.
- 합성데이터 생성 및 활용(필수)
- 기 확보된 데이터 활용(AI hub 등 활용, 추가실험)
- 학습 결과를 확인하며 성능을 높이는데 영향을 미치는 요소는 무엇이 있는지, 어떻게 수정해야하는지 고민합니다.
- 위 기준에서 벗어나지 않는 범위 내에서 데이터셋의 구성은 자유입니다. 성능을 비교/기록해보세요 :)
- 실험 결과를 Ablation study형식으로 기록합니다
- 위 기준에서 벗어나지 않는 범위 내에서 데이터셋의 구성은 자유입니다. 성능을 비교/기록해보세요 :)
# 일반대화 예시
{
"id": {
"text": "이거 들어봐 와 이 노래 진짜 좋다 그치 요즘 이 것만 들어 진짜 너무 좋다 내가 요즘 듣는 것도 들어봐 음 난 좀 별론데 좋을 줄 알았는데 아쉽네 내 취향은 아닌 듯 배고프다 밥이나 먹으러 가자 그래"
}
}
아래 링크는 이번 프로젝트에서 사용했던 Github 레포지토리이다.
https://github.com/choiwonjini/DLthon_pepero_day.git
GitHub - choiwonjini/DLthon_pepero_day: This is a repository for Aiffel DLthon
This is a repository for Aiffel DLthon. Contribute to choiwonjini/DLthon_pepero_day development by creating an account on GitHub.
github.com
이 해커톤을 끝내고 느낀 점부터 말하자면,
데이터:
- GIGO (Garbage In Garbage Out)를 절실히 느꼈다.
- LLM이 생성하는 데이터는 한계가 명확하다.
- EDA는 필수다.
협업:
- 협업을 위한 Github 관리 능력의 중요성을 깨달았다.
- soft skill도 hard skill 만큼 중요하다.
그 이유들은 아래에서 하나하나 적어보겠다.
팀 plan
- 데이터는
- 위협 대화 데이터는 역번역 진행하고
- 일반 대화 데이터는 Gemini CLI를 이용해 생성하자
- 전처리, 토큰화 코드는 통일해서 Github에 올려두고 각자 사용하고,
- 네 명의 팀원이 서로 다른 모델을 만들고,
- Ablation study를 통해 loss가 가장 낮은 모델을 선별해서 제출하자.
사용한 깃헙 구조도는 아래와 같다.
DLthon_pepero_day/
├── configs/
├── Data/
│ └── aiffel-dl-thon-dktc-online-15
│ └── ~.csv
├── Images/
├── models/
│ ├── 1D_CNN.ipynb
│ ├── bi_GRU.ipynb
│ ├── boemikbert_base.ipynb
│ └── decoder_only.ipynb
├── dataset.py
├── preprocessing.py
├── README.md
├── tokenization.py
├── eda.py
└── utils.py
- folders
- Data : Kaggle에서 다운받은 원본 데이터셋
- configs : 모델 설정, 데이터 경로 등 프로젝트의 주요 설정 값들을 저장하는 파일을 담는 디렉토리
- Images : 결과 리포트나 발표 자료에 사용될 이미지 파일을 담는 디렉토리
- modles : 다양한 모델 아키텍처 실험 및 관리를 위한 디렉토리
- files
- dataset.py : 데이터셋을 불러오고, 모델 입력으로 사용할 수 있는 형태로 변환
- preprocessing.py : 원본 데이터에 대한 전처리
- tokenization.py : 텍스트 데이터에 대한 토큰화 진행
- eda.py : 데이터셋의 분포 확인
- utils.py : 여러 파일에서 공통적으로 사용되는 유용한 함수를 모아놓은 모듈
1. preprocessing.py
전처리는 간단히 세 가지만 수행했다.
- 자모/한글, 영어, 숫자, 공백, 구두점, 'ㅠ', 'ㅜ' 이외의 문자는 모두 제거
- 연속 공백 or 구두점 -> 하나로 줄이기
- 앞뒤 공백 제거
import pandas as pd
import re
def preprocess_sentence(sentence):
"""
간단한 전처리 함수
"""
if pd.isna(sentence) or sentence is None:
return ""
sentence = str(sentence)
sentence = re.sub(r'[^ㄱ-ㅎㅏ-ㅣ가-힣a-zA-Z0-9\s.,!?~ㅠㅜ]', ' ', sentence)
sentence = re.sub(r'\s+', ' ', sentence)
sentence = sentence.strip()
sentence = re.sub(r'([!?.])\1+', r'\1', sentence)
return sentence
def load_and_preprocess_data(train_path, test_path):
"""
.csv 파일에서 데이터 로드하고 간단한 전처리를 진행하는 함수
"""
print("=" * 50)
print("데이터 로드 및 전처리 중...")
print("=" * 50)
# pandas로 CSV 파일 읽기
train_df = pd.read_csv(train_path)
test_df = pd.read_csv(test_path)
print(f"Train 데이터: {len(train_df)} 개의 conversation")
print(f"Test 데이터: {len(test_df)} 개의 conversation")
# 레이블 매핑
class_to_idx = {
'협박 대화': 0,
'갈취 대화': 1,
'직장 내 괴롭힘 대화': 2,
'기타 괴롭힘 대화': 3,
'일반 대화': 4
}
# Train 데이터 전처리
train_conversations = []
train_labels = []
for i, row in train_df.iterrows():
conv = preprocess_sentence(row['conversation'])
label = class_to_idx[row['class']]
if conv:
train_conversations.append(conv)
train_labels.append(label)
# Test 데이터 전처리
test_conversations = []
test_ids = []
for i, row in test_df.iterrows():
conv = preprocess_sentence(row['conversation'])
test_id = row['idx']
if conv:
test_conversations.append(conv)
test_ids.append(test_id)
# 샘플 데이터 출력
print("\n샘플 데이터:")
for i in range(min(3, len(train_conversations))):
print(f"Conversation: {train_conversations[i]}")
print(f"Label: {train_labels[i]}\n")
return train_conversations, train_labels, test_conversations, test_ids, class_to_idx
2. tokenization.py
토큰화는 SentencePiece를 사용했다. (SentencePiece는 이전 포스팅에서 다루었으므로 자세한 설명은 생략)
import sentencepiece as spm
def train_sentencepiece_model(conversations,
model_prefix='./configs/spm_dktc',
all_sentences_path='./configs/sentences.txt',
vocab_size=1300):
"""
주어진 conversations를 통해 SentencePiece 모델 학습
Args:
conversations
model_prefix
vocab_size: 개발자 지정 vocab의 크기. default=1200
Return:
model_file (str): 학습된 SentencePiece 모델 파일의 path
"""
print("=" * 50)
print("SentencePiece 모델 학습 중...")
print("=" * 50)
# 모든 문장을 하나의 텍스트 파일로 저장
with open(all_sentences_path, 'w', encoding='utf-8') as f:
f.write('\n'.join(conversations))
# SentencePiece 학습 명령어 설정
# [CLS] 토큰 추가 : 분류를 위한 시작 토큰
# (user_defined_symbols 명령어로 특수 토큰을 설정하면 자동으로 기본 특수 토큰 다음에 ID를 할당함)
# --minloglevel=1: INFO 로그를 제외하고 WARNING, ERROR만 출력
cmd = f'--input={all_sentences_path} \
--model_prefix={model_prefix} \
--vocab_size={vocab_size} \
--model_type=unigram \
--max_sentence_length=999999 \
--pad_id=0 \
--unk_id=1 \
--bos_id=2 \
--eos_id=3 \
--user_defined_symbols=[CLS] \
--minloglevel=1'
# SentencePiece 모델 학습 실행
spm.SentencePieceTrainer.Train(cmd)
# 학습된 모델 파일 경로 생성
model_file = f"{model_prefix}.model"
print(f"\n모델 저장됨: {model_file}")
print(f"Vocab 크기: {vocab_size}")
return model_file
class SentencePieceVocab:
"""
SentencePiece 모델을 쉽게 사용하기 위한 wrapper 클래스
텍스트를 토큰 ID로 encoding하거나 토큰 ID를 다시 텍스트로 decoding하는 기능 제공
"""
def __init__(self, sp_model_path):
"""
Args:
sp_model_path: 학습된 SentencePiece 모델 파일의 path
"""
# SentencePiece 프로세서 초기화
self.sp = spm.SentencePieceProcessor()
# 학습된 모델 로드
self.sp.Load(sp_model_path)
# 특수 토큰 ID 정의
self.PAD_ID = 0 # 패딩
self.UNK_ID = 1 # 미등록 단어
self.BOS_ID = 2 # 문장 시작 (BOS)
self.EOS_ID = 3 # 문장 끝 (EOS)
self.CLS_ID = 4 # Classification 토큰
# 토큰 문자열 -> ID 매핑
# <s>, </s> : 각각 BOS, EOS를 의미
self.stoi = {'<pad>': 0, '<unk>': 1, '<s>': 2, '</s>': 3, '[CLS]': 4}
# ID -> 토큰 문자열 매핑 (전체 어휘)
self.itos = [self.sp.IdToPiece(i) for i in range(self.sp.GetPieceSize())]
def encode(self, sentence):
"""
문장을 토큰 ID 리스트로 인코딩
Args:
sentence: 인코딩할 문자열
"""
return self.sp.EncodeAsIds(sentence)
def decode(self, ids):
"""
토큰 ID 리스트를 문장으로 디코딩(특수 토큰은 제외)
Args:
ids: 인코딩된 토큰 ID list
"""
return self.sp.DecodeIds([i for i in ids if i not in [0, 2, 3, 4]])
def __len__(self):
"""어휘 사전 크기 반환"""
return self.sp.GetPieceSize()3. Dataset.py
데이터셋을 불러오고, 위의 두 파일을 import해서 전처리, 토큰화를 진행한 다음 모델이 학습할 수 있게 Dataloader에 담는 과정을 담고 있다.
from preprocessing import load_and_preprocess_data
from tokenization import train_sentencepiece_model
from tokenization import SentencePieceVocab
import torch
from torch.utils.data import Dataset
from torch.utils.data import DataLoader, random_split, random_split
class DKTCDataset(Dataset):
"""
Pytorch Dataset class를 상속받아, conversations, label을 모델 학습에 적합한 텐서 형태로 변환하는 클래스
"""
def __init__(self, conversations, labels, vocab, max_length=400, is_test=False):
"""
Args:
conversations
labels
vocab: SentencePieceVocab 객체
max_length: 시퀀스의 최대 길이(tokenization 이후의 길이). default=400
is_test: 테스트 데이터셋 여부. default=False
"""
self.vocab = vocab # SentencePiece vocab 객체
self.max_length = max_length # 최대 시퀀스 길이 (잘림 방지)
self.is_test = is_test # 테스트 데이터 여부
self.sequences = []
self.labels = labels if not is_test else None
# conversation -> sequence
for conv in conversations:
# [CLS] + conversation tokens + [EOS]
sequence = [self.vocab.CLS_ID] \
+ self.vocab.encode(conv) \
+ [self.vocab.EOS_ID]
# sequence의 길이 조절
if len(sequence) > max_length:
sequence = sequence[:max_length]
else:
pad_length = max_length - len(sequence)
sequence = sequence + [self.vocab.PAD_ID] * pad_length
self.sequences.append(sequence)
def __len__(self):
"""데이터셋에 포함된 총 샘플의 개수 반환"""
return len(self.sequences)
def __getitem__(self, idx):
"""
GPT-1 방식: Next-token prediction을 위한 shifted sequences
+ classification label도 함께 반환
Returns:
dict: {
'input_ids': 모델 입력으로 사용될 텐서 (마지막 토큰 제외),
'target_ids': 예측 대상이 되는 텐서 (첫 토큰 제외),
'labels': 분류 레이블 (test data인 경우 제외)
}
"""
sequence = self.sequences[idx]
tokens = torch.tensor(sequence, dtype=torch.long)
input_ids = tokens[:-1] # 마지막 토큰 제외
target_ids = tokens[1:] # 첫 토큰 제외
result = {
'input_ids': input_ids,
'target_ids': target_ids
}
# Test가 아닌 경우 레이블 추가
if not self.is_test:
result["labels"] = torch.tensor(self.labels[idx], dtype=torch.long)
return result
def collate_fn(batch, pad_idx=0):
"""
DataLoader의 배치 생성 함수
"""
input_batch = [item['input_ids'] for item in batch]
target_batch = [item['target_ids'] for item in batch]
result = {
'input_ids': torch.stack(input_batch),
'target_ids': torch.stack(target_batch)
}
# labels가 있는 경우(즉 Test가 아닌 경우) result에 labels 추가
if 'labels' in batch[0]:
label_batch = [item['labels'] for item in batch]
result['labels'] = torch.stack(label_batch)
return result
def create_dataloaders(train_path, test_path, vocab_size=1320, max_length=400, batch_size=64, validation_split=0.1):
"""
데이터를 로드, 전처리, 토큰화하고 PyTorch train/validation/test DataLoader를 생성하는 메인 함수.
"""
# 1. 데이터 로드 및 전처리
train_conversations, \
train_labels, \
test_conversations, \
test_ids, \
class_to_idx = load_and_preprocess_data(train_path, test_path)
# 2. SentencePiece 토크나이저 모델 학습
model_prefix = './configs/sentences'
sp_model_path = train_sentencepiece_model(
train_conversations, model_prefix=model_prefix, vocab_size=vocab_size
)
# 3. SentencePiece Vocab 로드
vocab = SentencePieceVocab(sp_model_path)
# 4. 전체 학습 데이터셋 생성
full_train_dataset = DKTCDataset(
train_conversations,
train_labels,
vocab,
max_length=max_length,
is_test=False
)
# 5. Train / Validation 데이터셋으로 분리
num_train = len(full_train_dataset)
val_size = int(num_train * validation_split)
train_size = num_train - val_size
train_dataset, val_dataset = random_split(full_train_dataset, [train_size, val_size])
# 6. 테스트 데이터셋 생성
test_dataset = DKTCDataset(
test_conversations,
labels=None, # 테스트 데이터에는 레이블이 없습니다.
vocab=vocab,
max_length=max_length,
is_test=True
)
# 7. DataLoader 생성
train_loader = DataLoader(
train_dataset,
batch_size=batch_size,
shuffle=True,
collate_fn=lambda batch: collate_fn(batch, pad_idx=vocab.PAD_ID),
)
val_loader = DataLoader(
val_dataset,
batch_size=batch_size,
shuffle=False,
collate_fn=lambda batch: collate_fn(batch, pad_idx=vocab.PAD_ID),
)
test_loader = DataLoader(
test_dataset,
batch_size=batch_size,
shuffle=False, # 테스트 데이터는 셔플 X
collate_fn=lambda batch: collate_fn(batch, pad_idx=vocab.PAD_ID),
)
print(f"\nTrain DataLoader 준비 완료: 총 {len(train_dataset)}개 conversations")
print(f"Validation DataLoader 준비 완료: 총 {len(val_dataset)}개 conversations")
print(f"Test DataLoader 준비 완료: 총 {len(test_dataset)}개 conversations.")
return train_loader, val_loader, test_loader, vocab4. 1D CNN
필자는 예측 모델로 1D CNN을 선택했는데, 그 이유는 아래와 같다.
- 1D CNN은 n-gram 방법을 통해 문장 내의 핵심 키워드를 잘 잡아낼 수 있는 특징이 있다.
- 그래서 "돈 내놔", "죽어" 등 명확한 키워드가 존재하는 '협박', '갈취' 등 자극적이고 폭력적인 클래스를 분류하는 데 적합하다고 판단.
1D CNN의 장단점은 아래와 같다.
장점
- n-gram으로 "돈 내놔", "죽어" 등 짧고 폭력성이 있어 보이는 위협 키워드를 잘 잡아내고,
- 다른 모델에 비해 학습 속도가 상대적으로 빠르다.
단점
- Max Pooling 때문에 문장의 순서 정보가 없어지고, 위협 단어가 존재한다는 사실만 남기므로 위협 대화와 “공격적인 단어가 들어가있는 일반 대화”를 구분하기 어렵다.
아래는 모델 구현 코드이다.
class CNNClassifier(nn.Module):
"""
1D CNN 기반 텍스트 분류 모델
"""
def __init__(self,
vocab_size, # 어휘 사전의 크기 (vocab 객체로부터 받음)
embed_dim, # 임베딩 벡터의 차원
num_classes, # 분류할 클래스의 개수 (5)
num_filters, # 각 필터 크기별 컨볼루션 필터의 수
filter_sizes, # 사용할 컨볼루션 필터의 크기
dropout_prob # 드롭아웃 확률
):
super(CNNClassifier, self).__init__()
# 1. 임베딩 레이어
# padding_idx=0: <PAD> 토큰은 0 벡터로 임베딩하고 학습하지 않음
self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
# 2. 1D Convolution 레이어들 (다른 커널 크기를 사용)
# filter_sizes 개수만큼의 Conv1d 레이어를 ModuleList로 생성
# Conv1d는 (batch_size, in_channels, seq_len)을 입력으로 받음
# 우리 임베딩은 (batch_size, seq_len, embed_dim)이므로, permute(0, 2, 1) 필요
self.convs = nn.ModuleList([
nn.Conv1d(in_channels=embed_dim,
out_channels=num_filters,
kernel_size=k) # n-gram 크기
for k in filter_sizes
])
# 3. 드롭아웃
self.dropout = nn.Dropout(dropout_prob)
# 4. FC 레이어 (분류기)
# 각 필터에서 하나씩의 피처(max-pooling)가 나오므로,
# 총 num_filters * len(filter_sizes) 개의 피처가 입력됨
self.fc = nn.Sequential(
nn.Linear(num_filters * len(filter_sizes), num_filters * len(filter_sizes)),
nn.ReLU(),
nn.Dropout(dropout_prob),
nn.Linear(num_filters * len(filter_sizes), num_classes)
)
def forward(self, input_ids):
"""
모델의 순전파 로직
Args:
input_ids (torch.Tensor): (batch_size, seq_len)
dataset.py에 의해 seq_len은 max_length-1이 됨
Returns:
torch.Tensor: (batch_size, num_classes)
각 클래스에 대한 logits
"""
# 1. 임베딩
# input_ids: (batch_size, seq_len)
# embedded: (batch_size, seq_len, embed_dim)
embedded = self.embedding(input_ids)
# 2. Conv1d 입력을 위해 차원 변경
# embedded: (batch_size, embed_dim, seq_len)
embedded = embedded.permute(0, 2, 1)
# 3. 컨볼루션 + ReLU
# conved: (batch_size, num_filters, new_seq_len)
conved = [F.relu(conv(embedded)) for conv in self.convs]
# 4. Max pooling
# F.max_pool1d(conv, conv.shape[2])는 (batch_size, num_filters, 1)을 반환
# .squeeze(2)를 통해 (batch_size, num_filters)로 만듦
# pooled: [ (batch_size, num_filters), (batch_size, num_filters), ... ]
pooled = [F.max_pool1d(conv, conv.shape[2]).squeeze(2) for conv in conved]
# 5. 피처 결합 (Concatenate)
# catted: (batch_size, num_filters * len(filter_sizes))
catted = torch.cat(pooled, dim=1)
# 6. 드롭아웃
dropped = self.dropout(catted)
# 7. 완전 연결 레이어 (분류)
# logits: (batch_size, num_classes)
logits = self.fc(dropped)
return logits
위의 아키텍처는 추후 Ablation Study를 통해 최종적으로 선택된 하이퍼파라미터를 적용한 모델이다.
구조를 보면 컨볼루션 레이어를 여러 개 쌓지 않고 병렬로 연결했는데, 그 이유는 서로 다른 n-gram 특징들을 동시에 추출한 뒤, 이 특징들을 하나로 합치기 위함이다.
이게 무슨 소리냐면, 만약 2-gram으로 훑은 정보를 연산하고, 다음 레이어(3-gram)에 전달하면 이것은 각 사이즈의 필터들이 존재하는 의미가 없어진다.
왜??
"예전 포스팅에서 CNN을 다룰 때 나왔던 사실: 각 필터들은 데이터에서 서로 다른 특징들을 훑고 피처 맵을 만든다."
이 피처 맵들이 모여서 한 번에 다음 단계로 가는 것처럼,
한 필터가 가공한 정보를 다음 필터가 받는 것이 아닌, 모든 필터가 동시에 정보를 추출해야 텍스트의 다양한 특징들을 학습할 수 있다는 점 때문에 여러 개의 conv 레이어를 쌓지 않은 것이다.
분량 상 이번 포스팅은 여기서 마치고 다음 포스팅에서 이어가겠다.
다음 포스팅:
반응형
'AI > NLP' 카테고리의 다른 글
| [11/13] 아이펠 리서치 15기 TIL | Going-Deeper 시작, 형태소 분석기 (1) | 2025.11.21 |
|---|---|
| [11/6~11/11] 아이펠 리서치 15기 TIL | DLthon: 감정 분류 해커톤 (2) (0) | 2025.11.15 |
| [11/05] 아이펠 리서치 15기 TIL | Transformer 구현 (0) | 2025.11.15 |
| [10/31] 아이펠 리서치 15기 TIL | Transformer (0) | 2025.11.05 |
| [10/29] 아이펠 리서치 15기 TIL | Attention (0) | 2025.11.01 |
