P7-4.3 표현 정규화 연습¶

# 고객지원 문장의 표현 정규화 전후 coverage, OOV, 예측 팀 변화를 비교해 정규화 효과를 확인하는 예제입니다.
import csv
import numpy as np
import re
from pathlib import Path

data_path = Path("docs/assets/part-07/chapter-04/p7-4-support-routing-dataset.csv")
rows = list(csv.DictReader(data_path.open(encoding="utf-8")))
for row in rows:
    row["라벨"] = int(row["label"])

train_rows = [row for row in rows if row["split"] == "train"]
test_rows = [row for row in rows if row["split"] == "test"]

라벨_이름 = {0: "환불팀", 1: "배송팀"}

def tokenize(text):
    return text.split()

vocab = sorted({token for row in train_rows for token in tokenize(row["text"])})
token_to_index = {token: i for i, token in enumerate(vocab)}

normalization_map = {
    "캔슬": "취소",
    "스케줄": "일정",
    "하자": "불량",
}

def normalize_text(text):
    for raw_expression, normalized_expression in sorted(
        normalization_map.items(), key=lambda item: len(item[0]), reverse=True
    ):
        # 공백으로 나뉜 표현 경계에서만 바꿔 부분 문자열 치환을 피합니다.
        pattern = rf"(?<!\S){re.escape(raw_expression)}(?!\S)"
        text = re.sub(pattern, normalized_expression, text)
    return text

def vectorize(texts):
    X = np.zeros((len(texts), len(vocab)), dtype=float)
    token_lists = []
    oov_lists = []
    coverage_list = []

    for i, text in enumerate(texts):
        tokens = tokenize(text)
        token_lists.append(tokens)
        known = 0
        oov = []
        for token in tokens:
            if token in token_to_index:
                X[i, token_to_index[token]] += 1
                known += 1
            else:
                oov.append(token)
        oov_lists.append(oov)
        coverage_list.append(round(known / len(tokens), 3) if tokens else 0.0)

    return X, token_lists, oov_lists, coverage_list

X_train, _, _, _ = vectorize([row["text"] for row in train_rows])
y_train = np.array([row["라벨"] for row in train_rows])

class_profiles = np.vstack([
    X_train[y_train == 0].sum(axis=0),
    X_train[y_train == 1].sum(axis=0),
])

original_texts = [row["text"] for row in test_rows]
normalized_texts = [normalize_text(text) for text in original_texts]
y_test = np.array([row["라벨"] for row in test_rows])

X_original, _, original_oov, original_coverage = vectorize(original_texts)
X_normalized, _, normalized_oov, normalized_coverage = vectorize(normalized_texts)

def predict_labels(X):
    scores = X @ class_profiles.T
    predictions = np.argmax(scores, axis=1)
    has_signal = X.sum(axis=1) > 0
    has_single_winner = np.ptp(scores, axis=1) > 0
    # 근거가 없거나 팀 점수가 동률이면 첫 라벨을 정답처럼 내보내지 않습니다.
    return np.where(has_signal & has_single_winner, predictions, -1)

def display_prediction(prediction):
    return "판단 보류" if prediction == -1 else 라벨_이름[int(prediction)]

def evaluation_status(prediction, actual):
    if prediction == -1:
        return "판단 보류"
    return "정답" if prediction == actual else "오답"

def review_decision(original_prediction, normalized_prediction, actual, coverage_change):
    original_status = evaluation_status(original_prediction, actual)
    normalized_status = evaluation_status(normalized_prediction, actual)
    if original_status == "정답" and normalized_status != "정답":
        return "규칙 되돌리기 또는 재검토"
    if original_status != "정답" and normalized_status != "정답":
        return "규칙 또는 데이터 보강"
    if original_status != "정답" and normalized_status == "정답":
        return "우선 정규화"
    if coverage_change > 0:
        return "다음 정리 후보"
    return "추가 확인"

original_pred = predict_labels(X_original)
normalized_pred = predict_labels(X_normalized)

comparison_rows = []
for i, row in enumerate(test_rows):
    comparison_rows.append({
        "평가 샘플": row["sample_id"],
        "원문": original_texts[i],
        "정규화 후": normalized_texts[i],
        "원문 coverage": original_coverage[i],
        "정규화 후 coverage": normalized_coverage[i],
        "coverage 변화": round(normalized_coverage[i] - original_coverage[i], 3),
        "원문 OOV": original_oov[i],
        "남은 OOV": normalized_oov[i],
        "원문 예측": display_prediction(original_pred[i]),
        "정규화 후 예측": display_prediction(normalized_pred[i]),
        "예측 변화": (
            "변화 없음"
            if original_pred[i] == normalized_pred[i]
            else f"{display_prediction(original_pred[i])} -> {display_prediction(normalized_pred[i])}"
        ),
        "원문 정답 여부": evaluation_status(original_pred[i], y_test[i]),
        "정규화 후 정답 여부": evaluation_status(normalized_pred[i], y_test[i]),
        "실제 팀": 라벨_이름[y_test[i]],
        "검토 결정": review_decision(
            original_pred[i], normalized_pred[i], y_test[i],
            normalized_coverage[i] - original_coverage[i]
        ),
    })

summary = {
    "원문 정확도": round(float((original_pred == y_test).mean()), 3),
    "정규화 후 정확도": round(float((normalized_pred == y_test).mean()), 3),
    "coverage가 오른 샘플": [
        row["sample_id"]
        for i, row in enumerate(test_rows)
        if normalized_coverage[i] > original_coverage[i]
    ],
    "예측이 바뀐 샘플": [
        row["sample_id"]
        for i, row in enumerate(test_rows)
        if normalized_pred[i] != original_pred[i]
    ],
    "원문 판단 보류 샘플": [
        row["sample_id"]
        for i, row in enumerate(test_rows)
        if original_pred[i] == -1
    ],
    "정규화 후 판단 보류 샘플": [
        row["sample_id"]
        for i, row in enumerate(test_rows)
        if normalized_pred[i] == -1
    ],
}

print("정규화 비교 요약 =", summary)
print("핵심 샘플별 비교 =")
for row in comparison_rows:
    if row["평가 샘플"] in {"평가-05", "평가-07"}:
        print(row)

정규화 비교 요약 = {'원문 정확도': 0.857, '정규화 후 정확도': 1.0, 'coverage가 오른 샘플': ['평가-05', '평가-07'], '예측이 바뀐 샘플': ['평가-05'], '원문 판단 보류 샘플': [], '정규화 후 판단 보류 샘플': []}
핵심 샘플별 비교 =
{'평가 샘플': '평가-05', '원문': '캔슬 후 송장 남아 있어요', '정규화 후': '취소 후 송장 남아 있어요', '원문 coverage': 0.2, '정규화 후 coverage': 0.4, 'coverage 변화': 0.2, '원문 OOV': ['캔슬', '후', '남아', '있어요'], '남은 OOV': ['후', '남아', '있어요'], '원문 예측': '배송팀', '정규화 후 예측': '환불팀', '예측 변화': '배송팀 -> 환불팀', '원문 정답 여부': '오답', '정규화 후 정답 여부': '정답', '실제 팀': '환불팀', '검토 결정': '우선 정규화'}
{'평가 샘플': '평가-07', '원문': '하자 제품 환불 스케줄 알고 싶어요', '정규화 후': '불량 제품 환불 일정 알고 싶어요', '원문 coverage': 0.333, '정규화 후 coverage': 0.667, 'coverage 변화': 0.334, '원문 OOV': ['하자', '스케줄', '알고', '싶어요'], '남은 OOV': ['알고', '싶어요'], '원문 예측': '환불팀', '정규화 후 예측': '환불팀', '예측 변화': '변화 없음', '원문 정답 여부': '정답', '정규화 후 정답 여부': '정답', '실제 팀': '환불팀', '검토 결정': '다음 정리 후보'}