capstone-ai / models /classifier.py
wjsdbqls's picture
Claude Opus 5.5
refactor: KoBERT ์ด๋ฆ„์„ ์ •๋ฆฌํ•˜๊ณ  ๋ฏธ์‚ฌ์šฉ ๋ฆฌ๋žญ์ปค ๊ฒฝ๋กœ ์ œ๊ฑฐ
1f204c9
Raw History Blame Contribute Delete
4.3 kB
"""
KLUE BERT ๊ธฐ๋ฐ˜ ๋ฌธ์˜ ์นดํ…Œ๊ณ ๋ฆฌ ๋ถ„๋ฅ˜ ๋ชจ๋ธ.
๊ตฌ์กฐ:
- klue/bert-base โ†’ [CLS] ํ† ํฐ โ†’ Dropout โ†’ Linear โ†’ ์นดํ…Œ๊ณ ๋ฆฌ ํ™•๋ฅ 
- ์นดํ…Œ๊ณ ๋ฆฌ: ์ˆ˜๊ฐ•์‹ ์ฒญ(0), ์„ฑ์ (1), ์กธ์—…(2), ์žฅํ•™๊ธˆ(3), ํœด๋ณตํ•™(4),
๋“ฑ๋ก๊ธˆ(5), ๊ธฐ์ˆ™์‚ฌ(6), ๊ณต๊ฒฐ_์ถœ์„(7), ์ฆ๋ช…์„œ(8), ๊ธฐํƒ€(9)
attention ๊ตฌํ˜„:
์ตœ์‹  transformers์˜ ๊ธฐ๋ณธ SDPA ๊ตฌํ˜„์€ attention ๊ฐ€์ค‘์น˜๋ฅผ ๋ฐ˜ํ™˜ํ•˜์ง€ ์•Š์œผ๋ฏ€๋กœ
ํ•˜์ด๋ผ์ดํŒ…์„ ์œ„ํ•ด eager ๊ตฌํ˜„์„ ๋ช…์‹œํ•œ๋‹ค.
"""
import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer, AutoTokenizer
# ์นดํ…Œ๊ณ ๋ฆฌ ๋งคํ•‘ (generate_qa.py์˜ CATEGORIES์™€ ๋™์ผํ•˜๊ฒŒ ์œ ์ง€)
CATEGORY_LABELS = [
"์ˆ˜๊ฐ•์‹ ์ฒญ", "์„ฑ์ ", "์กธ์—…", "์žฅํ•™๊ธˆ", "ํœด๋ณตํ•™",
"๋“ฑ๋ก๊ธˆ", "๊ธฐ์ˆ™์‚ฌ", "๊ณต๊ฒฐ_์ถœ์„", "์ฆ๋ช…์„œ", "๊ธฐํƒ€"
]
NUM_CLASSES = len(CATEGORY_LABELS)
MODEL_NAME = "klue/bert-base"
class CategoryClassifier(nn.Module):
"""KLUE BERT ํŒŒ์ธํŠœ๋‹ ๋ถ„๋ฅ˜ ๋ชจ๋ธ."""
def __init__(self, num_classes: int = NUM_CLASSES, dropout_prob: float = 0.3):
super().__init__()
# eager attention: output_attentions ์ง€์›์„ ์œ„ํ•ด ๋ช…์‹œ
self.bert = BertModel.from_pretrained(
MODEL_NAME,
attn_implementation="eager",
)
hidden_size = self.bert.config.hidden_size
self.dropout = nn.Dropout(dropout_prob)
self.classifier = nn.Linear(hidden_size, num_classes)
def forward(
self,
input_ids: torch.Tensor,
attention_mask: torch.Tensor,
token_type_ids: torch.Tensor | None = None,
) -> torch.Tensor:
outputs = self.bert(
input_ids=input_ids,
attention_mask=attention_mask,
token_type_ids=token_type_ids,
output_attentions=True, # ํ•˜์ด๋ผ์ดํŒ…์„ ์œ„ํ•ด attention ๋ฐ˜ํ™˜
)
# [CLS] ํ† ํฐ์˜ pooled output ์‚ฌ์šฉ
pooled = outputs.pooler_output # (batch, hidden)
pooled = self.dropout(pooled)
logits = self.classifier(pooled) # (batch, num_classes)
# attentions: tuple of (batch, heads, seq, seq) โ€” ๋งˆ์ง€๋ง‰ ๋ ˆ์ด์–ด๋งŒ ๋ฐ˜ํ™˜
self._last_attentions = outputs.attentions # inference ์‹œ ํ•˜์ด๋ผ์ดํŒ…์— ์‚ฌ์šฉ
return logits
def get_attention_weights(self) -> torch.Tensor:
"""๋งˆ์ง€๋ง‰ forward์—์„œ์˜ ๋งˆ์ง€๋ง‰ ๋ ˆ์ด์–ด attention ๋ฐ˜ํ™˜. highlight.py์—์„œ ์‚ฌ์šฉ."""
if not hasattr(self, "_last_attentions"):
raise RuntimeError("forward()๋ฅผ ๋จผ์ € ํ˜ธ์ถœํ•˜์„ธ์š”.")
# ๋งˆ์ง€๋ง‰ ๋ ˆ์ด์–ด, ๋ชจ๋“  ํ—ค๋“œ ํ‰๊ท  โ†’ (batch, seq, seq)
last_layer_attn = self._last_attentions[-1] # (batch, heads, seq, seq)
return last_layer_attn.mean(dim=1) # (batch, seq, seq)
def load_tokenizer() -> AutoTokenizer:
return AutoTokenizer.from_pretrained(MODEL_NAME)
def load_model(checkpoint_path: str | None = None, device: str = "cpu") -> CategoryClassifier:
"""
๋ชจ๋ธ ๋กœ๋“œ. checkpoint_path๊ฐ€ ์žˆ์œผ๋ฉด ํŒŒ์ธํŠœ๋‹๋œ ๊ฐ€์ค‘์น˜ ๋กœ๋“œ.
"""
model = CategoryClassifier()
if checkpoint_path:
state = torch.load(checkpoint_path, map_location=device)
model.load_state_dict(state)
model.to(device)
return model
class InquiryDataset(torch.utils.data.Dataset):
"""ํ•™์Šต/๊ฒ€์ฆ ๋ฐ์ดํ„ฐ์…‹."""
def __init__(self, data: list[dict], tokenizer: BertTokenizer, max_length: int = 128):
self.tokenizer = tokenizer
self.max_length = max_length
self.texts = [item["question"] for item in data]
self.labels = [item["label"] for item in data]
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
encoding = self.tokenizer(
self.texts[idx],
max_length=self.max_length,
padding="max_length",
truncation=True,
return_tensors="pt",
)
return {
"input_ids": encoding["input_ids"].squeeze(0),
"attention_mask": encoding["attention_mask"].squeeze(0),
"token_type_ids": encoding.get("token_type_ids", torch.zeros(self.max_length, dtype=torch.long)).squeeze(0),
"label": torch.tensor(self.labels[idx], dtype=torch.long),
}