[AICE PROFESSIONAL 준비] 상품명으로 식품 카테고리 분류

AICE 시험을 준비하며 공식 홈페이지에 있는 활용하여 상품명으로 식품 카테고리를 분류하는 모델을 만들어봤다.

 

 

데이터 로드 및 확인

import pandas as pd
import numpy as np
import warnings

warnings.filterwarnings("ignore", category=UserWarning)

df = pd.read_csv("./data/02_train.csv")
df

 

df.isnull().sum()

 

df.duplicated().sum()

 

df[df.duplicated()].value_counts()

 

 

데이터 전처리

df_drop = df.drop_duplicates(subset=["상품명"])
df_drop.info()

 

df_drop["카테고리"].value_counts()

 

# 카테고리값을 라벨인코딩
from sklearn.preprocessing import LabelEncoder

le = LabelEncoder()

# 데이터 학습 및 변환 (fit_transform을 사용하면 한 번에 처리 가능)
encoded_labels = le.fit_transform(df_drop["카테고리"])
print(f"매핑된 클래스: {le.classes_}")

original_labels = le.inverse_transform(encoded_labels)
print(f"역변환 결과: {original_labels}")

NUM_CLASSES = len(le.classes_)
print(f"클래스 개수: {NUM_CLASSES}")

 

df_drop["카테고리"] = encoded_labels
df_drop

 

 

모델 학습 및 평가

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}")

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 TextDataset(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 = 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.long)

 

# PyTorch 내장 LSTM 기반 분류 모델
class ClassificationTextLSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_classes):
        super(ClassificationTextLSTM, 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)

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

    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.fc(combined_hidden)

 

# 가상의 데이터 로드 및 실행 파이프라인
# 데이터셋 분할
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 = TextDataset(train_df, vocab, MAX_LEN)
test_dataset = TextDataset(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 = ClassificationTextLSTM(VOCAB_SIZE, EMBEDDING_DIM, HIDDEN_DIM, NUM_CLASSES).to(DEVICE)
criterion = nn.CrossEntropyLoss() # 다중분류이므로 CrossEntropyLoss 사용
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)

        # 가장 확률이 높은 값 추출
        _, predicted = torch.max(outputs, 1)
        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 = torch.max(outputs, 1)
        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 = "text_classification.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를 사용하여 .pkl 파일로 저장
torch.save(save_data, MODEL_SAVE_PATH)
print(f"모델과 단어 사전이 성공적으로 저장되었습니다: {MODEL_SAVE_PATH}")

 

 

테스트데이터를 이용한 분류 예측

# 테스트데이터로 예측
target_file_path = "./data/02_test_x.csv"
df_new = pd.read_csv(target_file_path)
df_new

 

# 예측 결과를 담을 빈 리스트 생성
predicted_classes = [] # 라벨 이름 저장용

 

import torch
import torch.nn.functional as F

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

# 기울기 계산 비활성화 (메모리 절약 및 속도 향상)
with torch.no_grad():
    for sentence in df_new["상품명"]:
        # 결측치(NaN)나 문자열이 아닌 데이터 예외 처리
        if not isinstance(sentence, str):
            predicted_classes.append("Unknown")
            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)

        # 예측 수행
        logits = model(inputs)
        probs = F.softmax(logits, dim=1) # 확률값으로 변환

        # 가장 높은 확률의 인덱스 추출
        _, top_idx = torch.max(probs, 1)
        predicted_idx = top_idx.item()

        # 역변환 (인덱스 -> 카테고리 이름)
        predicted_label = le.inverse_transform([predicted_idx])[0]
        predicted_classes.append(predicted_label)

# 결과 저장
df_new["카테고리"] = predicted_classes
df_new.to_csv("./result.csv", index=False, encoding="utf-8-sig")
print("예측이 완료되었습니다. 결과가 result.csv에 저장되었습니다.")