[BERT] 외래어 구분

koBert 를 사용

 

 

loanword_classifier 생성 

 

 

 

process

1. 로그 추출 후 필터링
   1. 공백기준 1단어이상 조합, 숫자로만 이루어진 단어, 특수문자 들어간 단어 제외
   2. /Users/doo/doo_py/homeplus/season_keyword/log_extrect_clean.py (로그추출 스크립트)
      1. result/log_result_clean.txt (로그 파일)
2. 외래어 분류
   1. /Users/doo/doo_py/homeplus/new_nlp/loanword_inference.py (외래어 분류)
      1. /result/loanword/
         1. loanword_list.csv (외래어)
         2. native_list.csv (일반어)
3. API 조회
   1. /Users/doo/doo_py/homeplus/homeplus_api/search.py (api 조회 - 중계)
      1. result = 0
         1. /Users/doo/doo_py/homeplus/homeplus_api/result/result_zero.txt

 

 

 

 

train.py

import os
import random
import re
import pandas as pd
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader
from torch.optim import AdamW
from transformers import (
    AutoTokenizer,
    BertForSequenceClassification,
    DataCollatorWithPadding,
    get_linear_schedule_with_warmup,
    Trainer,
    TrainingArguments
)
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_recall_fscore_support

# ---------------------------------------------------
# 0) 시드 고정 함수 정의
# ---------------------------------------------------
def set_seed(seed: int = 42):
    """
    고정된 시드를 설정하여 재현 가능한 학습 환경을 만듭니다.
    """
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    # GPU 연산의 결정적(deterministic) 설정
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False
    # 환경변수 설정으로 CUBLAS 워크스페이스 결정성 강화
    os.environ['CUBLAS_WORKSPACE_CONFIG'] = ':16:8'

# ---------------------------------------------------
# 1) HuggingFace Dataset 래퍼 정의
# ---------------------------------------------------
class TextDataset(Dataset):
    def __init__(self, df: pd.DataFrame, tokenizer: AutoTokenizer, max_len: int = 128):
        self.texts  = df["text"].tolist()
        self.labels = df["label"].tolist()
        self.tokenizer = tokenizer
        self.max_len = max_len

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

    def __getitem__(self, idx):
        tokenized = self.tokenizer(
            self.texts[idx],
            truncation=True,
            padding="max_length",
            max_length=self.max_len,
            return_tensors="pt"
        )
        item = {k: v.squeeze(0) for k, v in tokenized.items()}
        item["labels"] = torch.tensor(self.labels[idx], dtype=torch.long)
        return item

# ---------------------------------------------------
# 2) 전처리 함수 정의
# ---------------------------------------------------
def clean_text(s: str) -> str:
    s = re.sub(r"\s+", " ", s)
    s = re.sub(r"[^가-힣a-zA-Z0-9\s]", "", s)
    return s.strip()

# ---------------------------------------------------
# 3) 데이터 로드 & 전처리 & 분할
# ---------------------------------------------------
def load_data(path: str, test_size: float = 0.2, seed: int = 42):
    df = pd.read_csv(path)
    df["label"] = df["label"].astype(int)
    df["text"]  = df["text"].astype(str).apply(clean_text)
    train_df, test_df = train_test_split(
        df,
        test_size=test_size,
        random_state=seed,
        stratify=df["label"]
    )
    return train_df.reset_index(drop=True), test_df.reset_index(drop=True)

# ---------------------------------------------------
# 4) compute_metrics 함수 (binary classification)
# ---------------------------------------------------
def compute_metrics(eval_pred):
    logits, labels = eval_pred
    preds = np.argmax(logits, axis=-1)
    precision, recall, f1, _ = precision_recall_fscore_support(
        labels, preds, average="binary", zero_division=0
    )
    acc = accuracy_score(labels, preds)
    return {"accuracy": acc, "precision": precision, "recall": recall, "f1": f1}

# ---------------------------------------------------
# 5) main() 안에서 호출
# ---------------------------------------------------
def main():
    # ** 재현성을 위한 시드 고정 **
    set_seed(42)

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

    # (1) 로컬 KoBERT 모델·토크나이저 경로
    LOCAL_MODEL_PATH = "./kobert"

    tokenizer = AutoTokenizer.from_pretrained(
        LOCAL_MODEL_PATH,
        local_files_only=True,
        trust_remote_code=True
    )
    # KoBertTokenizer.save_vocabulary 시그니처 패치
    _orig_save_vocab = tokenizer.save_vocabulary
    def _patched_save_vocab(save_directory, filename_prefix=None):
        return _orig_save_vocab(save_directory)
    tokenizer.save_vocabulary = _patched_save_vocab

    model = BertForSequenceClassification.from_pretrained(
        LOCAL_MODEL_PATH,
        num_labels=2,
        local_files_only=True
    )
    model.to(device)

    # (2) 외래어 분류용 데이터 준비
    train_df, test_df = load_data("./data/loanword_data.csv", test_size=0.2, seed=42)
    train_dataset = TextDataset(train_df, tokenizer)
    test_dataset  = TextDataset(test_df, tokenizer)

    # (3) DataCollator & DataLoader & Scheduler
    data_collator = DataCollatorWithPadding(tokenizer)
    train_loader  = DataLoader(
        train_dataset,
        batch_size=16,
        shuffle=True,
        collate_fn=data_collator
    )
    total_steps = len(train_loader) * EPOCHS
    optimizer = AdamW(model.parameters(), lr=2e-5)
    scheduler = get_linear_schedule_with_warmup(
        optimizer,
        num_warmup_steps=0,
        num_training_steps=total_steps
    )

    training_args = TrainingArguments(
        output_dir="./loanword_classifier",
        num_train_epochs=EPOCHS,
        save_strategy="epoch",
        per_device_train_batch_size=16,
        learning_rate=2e-5,
        logging_dir="./logs",
        logging_steps=100
    )

    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=test_dataset,
        tokenizer=tokenizer,
        data_collator=data_collator,
        compute_metrics=compute_metrics
    )

    # (6) 학습 & 평가 & 저장
    trainer.train()
    metrics = trainer.evaluate()
    print(metrics)

    model.config.id2label = {"0": "고유어", "1": "외래어"}
    model.config.label2id = {"고유어": 0, "외래어": 1}
    trainer.save_model("./loanword_classifier")
    tokenizer.save_pretrained("./loanword_classifier")

if __name__ == "__main__":
    main()

 

inference.py

import os
import re
import torch
import pandas as pd
from tqdm import tqdm
import importlib.util
from transformers import BertForSequenceClassification, AutoConfig

# ──────────────────────────────
# 환경 설정
# ──────────────────────────────
MODEL_DIR       = "./loanword_classifier"            # monologg/kobert 레포 전체를 clone 해 둔 폴더
INPUT_FILE      = "../season_keyword/result/log_result_clean.txt"

OUTPUT_ALL      = "result/loanword/loanword_classification_results.csv"
OUTPUT_NATIVE   = "result/loanword/native_list.csv"      # 0: 고유어
OUTPUT_LOANWORD = "result/loanword/loanword_list.csv"    # 1: 외래어

# 디렉터리 생성
for path in (OUTPUT_ALL, OUTPUT_NATIVE, OUTPUT_LOANWORD):
    os.makedirs(os.path.dirname(path), exist_ok=True)

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

# ──────────────────────────────
# 전처리 함수 정의 (학습 파이프라인과 동일)
# ──────────────────────────────
def clean_text(s) -> str:
    # 문자열이 아닌 경우 빈 문자열로 처리
    if not isinstance(s, str):
        return ""
    # 중복 공백 축소
    s = re.sub(r"\s+", " ", s)
    # 한글/영문/숫자/공백 외 모든 문자 제거
    s = re.sub(r"[^가-힣a-zA-Z0-9\s]", "", s)
    return s.strip()

# ──────────────────────────────
# KoBERT 토크나이저 동적 로드
# ──────────────────────────────
spec = importlib.util.spec_from_file_location(
    "tokenization_kobert",
    os.path.join(MODEL_DIR, "tokenization_kobert.py")
)
token_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(token_module)
KoBertTokenizer = token_module.KoBertTokenizer

# ──────────────────────────────
# 토크나이저·모델·매핑 로드
# ──────────────────────────────
tokenizer = KoBertTokenizer(
    vocab_file=os.path.join(MODEL_DIR, "tokenizer_78b3253a26.model"),
    vocab_txt=os.path.join(MODEL_DIR, "vocab.txt"),
    do_lower_case=False
)
model = BertForSequenceClassification.from_pretrained(
    MODEL_DIR,
    local_files_only=True
)
model.to(device)
model.eval()

# 학습 시 config에 저장된 매핑을 그대로 읽어옵니다
cfg = AutoConfig.from_pretrained(MODEL_DIR, local_files_only=True)
id2label = {int(k): v for k, v in cfg.id2label.items()}

# 진행바 설정
tqdm.pandas(desc="분류중")

# ──────────────────────────────
# 디버깅용 함수
# ──────────────────────────────
def debug_predict(text: str) -> str:
    cleaned = clean_text(text)
    tokens = tokenizer.tokenize(cleaned)
    print(f"[DEBUG] Tokens for {text!r}: {tokens}")

    enc = tokenizer(
        cleaned,
        max_length=128,
        padding="max_length",
        truncation=True,
        return_tensors="pt"
    ).to(device)
    with torch.no_grad():
        logits = model(**enc).logits[0].detach().cpu().tolist()
    print(f"[DEBUG] Logits: {logits}")

    pred_idx = int(torch.tensor(logits).argmax().item())
    pred_label = id2label[pred_idx]
    print(f"[DEBUG] Prediction idx={pred_idx}, label={pred_label!r}")

    return pred_label

# ──────────────────────────────
# 예측 함수
# ──────────────────────────────
def predict_label(text: str) -> str:
    text = clean_text(text)
    enc = tokenizer(
        text,
        max_length=128,
        padding="max_length",
        truncation=True,
        return_tensors="pt"
    ).to(device)
    with torch.no_grad():
        logits = model(**enc).logits[0].detach().cpu().tolist()
    pred_idx = int(torch.tensor(logits).argmax().item())
    return id2label[pred_idx]


def main():
    # 1) 키워드+카운트 읽기
    df = pd.read_csv(
        INPUT_FILE,
        sep=",",
        header=None,
        names=["keyword", "count"],
        skipinitialspace=True,
        on_bad_lines="skip",
        dtype={"count": int}
    )

    # 2) NaN 처리 및 전처리
    df["keyword"] = df["keyword"].fillna("").apply(clean_text)

    # 3) 분류
    df["label"] = df["keyword"].progress_apply(predict_label)

    # 4) 전체 결과 저장
    df.to_csv(
        OUTPUT_ALL,
        columns=["keyword", "count", "label"],
        index=False,
        encoding="utf-8-sig"
    )
    print(f"Saved all to {OUTPUT_ALL}")

    # 5) 클래스별 저장
    df[df["label"] == id2label[0]].to_csv(
        OUTPUT_NATIVE,
        columns=["keyword", "count"],
        index=False,
        encoding="utf-8-sig"
    )
    print(f"Saved native to {OUTPUT_NATIVE}")

    df[df["label"] == id2label[1]].to_csv(
        OUTPUT_LOANWORD,
        columns=["keyword", "count"],
        index=False,
        encoding="utf-8-sig"
    )
    print(f"Saved loanword to {OUTPUT_LOANWORD}")

    # 디버깅 예시
    print(debug_predict("삼겹살"))

if __name__ == "__main__":
    main()