[NLP] BERT - 식품 비식품 / 사계절 추론

 

 

BERT

• klue/bert-base (한국어 사전으로 학습된 모델) 모델 로드
• Device : Local CPU

 

Fine-Tunning

식품/비식품

• “text”와 “label(1=음식, 0=비음식)” 컬럼으로 구성으로 학습
• 식품 비식품 1031개 단어학습 중 - 1000개 학습 약 2분 10초 소요 (epoch 5)
• 추론 10000개 6분 30초 소요

 

테스트

• 추출 기간: 2024-06-01 00:00:00 ~ 2024-08-31 00:00:00
• 추출 키워드: 59,973개 (Type 에러 발생시키는 키워드 제거)
  • 결과
    • 식품 : 48,837개
    • 비식품: 11,136개

 

시즌 키워드

사계절 및 비시즌 파일로 추론 결과 생성

• 0: "비시즌" - nonseason_list.csv
• 1: "봄" - spring_list.csv
• 2: "여름" - summer_list.csv
• 3: "가을" - auturmn_list.csv
• 4: "겨울" - winter_list.csv

학습데이터 1123개 단어 학습 중 - 약 2분 32초 ( epoch 5)

 

테스트 (summer_list.csv 내용 확인) - top 20

• 11,137개 비식품 키워드 추론 시간 : 약 7분 5초

 

source code

식품 / 비식품 

 

import re
import random
import pandas as pd
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader            # Dataset 추가
from torch.optim import AdamW
from transformers import (
    BertTokenizer,
    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) Seed 고정 함수 정의
# ---------------------------------------------------
def set_seed(seed: int = 42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

# ---------------------------------------------------
# HuggingFace Dataset 래퍼 정의
# ---------------------------------------------------
class TextDataset(Dataset):
    def __init__(self, df: pd.DataFrame, tokenizer: BertTokenizer, 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

# ---------------------------------------------------
# 1) 전처리 함수 정의
# ---------------------------------------------------
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()

# ---------------------------------------------------
# 2) 데이터 로드 & 전처리 & 분할
# ---------------------------------------------------
def load_data(path: str, test_size: float = 0.2, seed: int = 42):
    df = pd.read_csv(path)
    df["label"] = df["label"].astype(str).str.strip().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)

# ---------------------------------------------------
# 3) compute_metrics 함수
# ---------------------------------------------------
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}

# ---------------------------------------------------
# 4) main() 안에서 호출
# ---------------------------------------------------
def main():
    # 시드 고정
    seed = 42
    set_seed(seed)

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

    # 로컬에 저장된 사전학습 모델·토크나이저 로드
    tokenizer = BertTokenizer.from_pretrained("./klue-bert-base", do_lower_case=False)
    model     = BertForSequenceClassification.from_pretrained(
        "./klue-bert-base", num_labels=2
    )
    model.to(device)

    # 데이터 준비
    train_df, test_df = load_data("./data/food_nonfood_data.csv", test_size=0.2, seed=seed)
    train_dataset = TextDataset(train_df, tokenizer)
    test_dataset  = TextDataset(test_df,  tokenizer)

    # DataLoader (optional, if not using Trainer directly)
    data_collator = DataCollatorWithPadding(tokenizer)
    train_loader  = DataLoader(train_dataset, batch_size=16, shuffle=True, collate_fn=data_collator)
    test_loader   = DataLoader(test_dataset,  batch_size=32, shuffle=False, collate_fn=data_collator)

    # 옵티마이저·스케줄러
    optimizer = AdamW(model.parameters(), lr=2e-5)
    total_steps = len(train_loader) * 5  # e.g., 5 epochs
    scheduler   = get_linear_schedule_with_warmup(
        optimizer, num_warmup_steps=0, num_training_steps=total_steps
    )

    # 1) TrainingArguments 선언
    training_args = TrainingArguments(
        output_dir="./food_classifier",
        num_train_epochs=5,
        save_strategy="epoch",
        per_device_train_batch_size=16,
        learning_rate=2e-5,
        logging_dir="./logs",
        logging_steps=100,
        seed=seed
    )

    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
    )

    # 학습
    trainer.train()

    # 수동으로 평가와 저장
    metrics = trainer.evaluate()
    print(metrics)
    trainer.save_model("./food_classifier")
    tokenizer.save_pretrained("./food_classifier")

if __name__ == "__main__":
    main()

 

import torch
from transformers import BertTokenizer, BertForSequenceClassification
import pandas as pd
from tqdm import tqdm  # 진행바용

# ──────────────────────────────
# 환경 설정
# ──────────────────────────────
MODEL_DIR      = "./food_classifier"
RESULT_FILE    = "../season_keyword/result/log_result_clean.txt"
OUTPUT_CSV     = "result/food/classification_results.csv"
OUTPUT_FOOD    = "result/food/food_list.csv"      # 식품만 저장할 파일
OUTPUT_NONFOOD = "result/food/nonfood_list.csv"   # 비식품만 저장할 파일

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

# ──────────────────────────────
# 모델·토크나이저 로드
# ──────────────────────────────
tokenizer = BertTokenizer.from_pretrained(MODEL_DIR)
model     = BertForSequenceClassification.from_pretrained(MODEL_DIR)
model.to(device)
model.eval()

# pandas에 tqdm 붙이기
tqdm.pandas(desc="분류중")

# ──────────────────────────────
# 원본 predict_text 함수
# ──────────────────────────────
def predict_text(text: str) -> str:
    inputs = tokenizer(
        text,
        truncation=True,
        padding="max_length",
        max_length=128,
        return_tensors="pt"
    ).to(device)
    with torch.no_grad():
        logits = model(**inputs).logits
    pred = logits.argmax(dim=-1).item()
    return "식품" if pred == 1 else "비식품"

# ──────────────────────────────
# 안전 래퍼: 에러 시 None 반환
# ──────────────────────────────
def safe_predict(text):
    try:
        # str 아닌 입력도 미리 걸러줄 수 있습니다
        if not isinstance(text, str):
            raise ValueError(f"non-str input: {text!r}")
        return predict_text(text)
    except Exception as e:
        # 문제 키워드 로그 (필요하면 파일로도 남기세요)
        print(f"[경고] 스킵된 키워드 {text!r}: {e}")
        return None

def main():
    # 1) 파일 읽기 (읽기 에러 라인은 skip)
    df = pd.read_csv(
        RESULT_FILE,
        sep=",",
        header=None,
        names=["keyword","count"],
        skipinitialspace=True,
        on_bad_lines='skip'
    )

    # 2) 예측 + 안전 처리 (None 은 나중에 drop)
    df["category"] = df["keyword"].progress_apply(safe_predict)

    # 3) None(에러)인 행은 제거
    bad_count = df["category"].isna().sum()
    if bad_count > 0:
        print(f"[정보] {bad_count}개의 행에서 에러 발생, 제거하고 진행합니다.")
        df = df.dropna(subset=["category"]).reset_index(drop=True)

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

    # 5) 식품/비식품으로 분리 & 저장
    food_df    = df[df["category"] == "식품"]
    nonfood_df = df[df["category"] == "비식품"]

    food_df.to_csv(
        OUTPUT_FOOD,
        columns=["keyword", "count"],
        index=False,
        encoding="utf-8-sig"
    )
    print(f"Saved food list to {OUTPUT_FOOD}")

    nonfood_df.to_csv(
        OUTPUT_NONFOOD,
        columns=["keyword", "count"],
        index=False,
        encoding="utf-8-sig"
    )
    print(f"Saved non-food list to {OUTPUT_NONFOOD}")

# ──────────────────────────────
# 디버그용 predict 함수
# ──────────────────────────────
def debug_predict(text: str):
    # 1) 원본 문자열 확인
    print(f"\n[DEBUG] input text: {text!r}")

    # 2) 토크나이저에 통과
    enc = tokenizer(
        text,
        truncation=True,
        padding="max_length",
        max_length=128,
        return_tensors="pt"
    )
    print("[DEBUG] raw encoding:", enc)

    # 3) input_ids → 토큰으로 복원
    tokens = tokenizer.convert_ids_to_tokens(enc["input_ids"][0])
    print("[DEBUG] tokens:", tokens)

    # 4) device 할당
    enc = {k: v.to(device) for k,v in enc.items()}

    # 5) 로짓 계산
    with torch.no_grad():
        outputs = model(**enc)
    logits = outputs.logits
    print("[DEBUG] logits:", logits)

    # 6) 확률로 변환
    probs = torch.softmax(logits, dim=-1)
    print("[DEBUG] probabilities:", probs)

    # 7) 최종 예측
    pred_idx = logits.argmax(dim=-1).item()
    label = "식품" if pred_idx == 1 else "비식품"
    print(f"[DEBUG] pred_idx={pred_idx} → {label}\n")

    return label



if __name__ == "__main__":
#     debug_predict("휴지")
    main()

 

사계절 

import re
import random
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 (
    BertTokenizer,
    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) Seed 고정 함수 정의
# ---------------------------------------------------
def set_seed(seed: int = 42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

# ---------------------------------------------------
# 1) HuggingFace Dataset 래퍼 정의
# ---------------------------------------------------
class TextDataset(Dataset):
    def __init__(self, df: pd.DataFrame, tokenizer: BertTokenizer, 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 함수 (multi-class)
# ---------------------------------------------------
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="macro", zero_division=0
    )
    acc = accuracy_score(labels, preds)
    return {"accuracy": acc, "precision": precision, "recall": recall, "f1": f1}

# ---------------------------------------------------
# 5) main() 안에서 호출
# ---------------------------------------------------
def main():
    # 1) 시드 고정
    seed = 42
    set_seed(seed)

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

    tokenizer = BertTokenizer.from_pretrained("./klue-bert-base", do_lower_case=False)
    model = BertForSequenceClassification.from_pretrained(
        "./klue-bert-base", num_labels=5
    )
    model.to(device)

    # 2) 데이터 준비
    train_df, test_df = load_data("./data/four_season_data.csv", test_size=0.2, seed=seed)
    train_dataset = TextDataset(train_df, tokenizer)
    test_dataset  = TextDataset(test_df,  tokenizer)

    # 3) DataLoader & 옵티마이저·스케줄러
    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
    )

    # 4) TrainingArguments 설정
    training_args = TrainingArguments(
        output_dir="./four_season_classifier",
        num_train_epochs=EPOCHS,
        save_strategy="epoch",
        per_device_train_batch_size=16,
        learning_rate=2e-5,
        logging_dir="./logs",
        logging_steps=100,
        seed=seed
    )

    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
    )

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

    trainer.save_model("./four_season_classifier")
    tokenizer.save_pretrained("./four_season_classifier")

if __name__ == "__main__":
    main()

 

import torch
from transformers import BertTokenizer, BertForSequenceClassification
import pandas as pd
from tqdm import tqdm  # 진행바용

# ──────────────────────────────
# 환경 설정
# ──────────────────────────────
MODEL_DIR        = "./four_season_classifier"
RESULT_FILE      = "result/food/nonfood_list.csv"

# 전체 결과
OUTPUT_ALL       = "result/five_class_results.csv"
# 클래스별 결과
OUTPUT_NONSEASON = "result/season/nonseason_list.csv"   # 0
OUTPUT_SPRING    = "result/season/spring_list.csv"      # 1
OUTPUT_SUMMER    = "result/season/summer_list.csv"      # 2
OUTPUT_AUTUMN    = "result/season/autumn_list.csv"      # 3
OUTPUT_WINTER    = "result/season/winter_list.csv"      # 4

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

# ──────────────────────────────
# 모델·토크나이저 로드
# ──────────────────────────────
tokenizer = BertTokenizer.from_pretrained(MODEL_DIR, local_files_only=True)
model     = BertForSequenceClassification.from_pretrained(MODEL_DIR, local_files_only=True)
model.to(device)
model.eval()

# pandas에 tqdm 붙이기
tqdm.pandas(desc="분류중")

# ──────────────────────────────
# 예측 함수
# ──────────────────────────────
# 학습 시 사용한 매핑과 동일하게 설정하세요.
id2label = {
    0: "시즌키워드아님",
    1: "봄",
    2: "여름",
    3: "가을",
    4: "겨울"
}

def predict_label(text: str) -> str:
    inputs = tokenizer(
        text,
        truncation=True,
        padding="max_length",
        max_length=128,
        return_tensors="pt"
    ).to(device)
    with torch.no_grad():
        logits = model(**inputs).logits
    idx = logits.argmax(dim=-1).item()
    return id2label[idx]

def main():
    # 1) 원본 키워드+카운트 읽기
    df = pd.read_csv(
        RESULT_FILE,
        header=None,
        names=["keyword","count"],
        skipinitialspace=True,
        # pandas ≥1.3:
        on_bad_lines='skip',      # 에러 나는 줄은 경고 후 건너뜀
        # pandas <1.3:
        # error_bad_lines=False,  # (deprecated) 동일 기능
        # warn_bad_lines=True      # (optional) 경고 메시지 출력
    )

    # 2) 5개 클래스로 분류
    df["label"] = df["keyword"].progress_apply(predict_label)

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

    # 4) 클래스별 분리 & 저장
    df[df["label"] == "시즌키워드아님"].to_csv(
        OUTPUT_NONSEASON, columns=["keyword","count"], index=False, encoding="utf-8-sig"
    )
    print(f"Saved non-season list to {OUTPUT_NONSEASON}")

    df[df["label"] == "봄"].to_csv(
        OUTPUT_SPRING, columns=["keyword","count"], index=False, encoding="utf-8-sig"
    )
    print(f"Saved spring list to {OUTPUT_SPRING}")

    df[df["label"] == "여름"].to_csv(
        OUTPUT_SUMMER, columns=["keyword","count"], index=False, encoding="utf-8-sig"
    )
    print(f"Saved summer list to {OUTPUT_SUMMER}")

    df[df["label"] == "가을"].to_csv(
        OUTPUT_AUTUMN, columns=["keyword","count"], index=False, encoding="utf-8-sig"
    )
    print(f"Saved autumn list to {OUTPUT_AUTUMN}")

    df[df["label"] == "겨울"].to_csv(
        OUTPUT_WINTER, columns=["keyword","count"], index=False, encoding="utf-8-sig"
    )
    print(f"Saved winter list to {OUTPUT_WINTER}")

if __name__ == "__main__":
    main()