실습 5 — 드리프트 감지하기

from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from scipy.stats import ks_2samp
from sentence_transformers import SentenceTransformer

plt.rcParams["font.family"] = "Malgun Gothic"        # 한글 글꼴(4단계의 그래프)
plt.rcParams["axes.unicode_minus"] = False

log = pd.read_csv("chat_log.csv", parse_dates=["시각"])
log["주"] = (log["시각"] - pd.Timestamp("2026-09-07")).dt.days // 7 + 1     # 2026-09-07(월)부터 1주
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
chunks, docs = [], []
for f in sorted(Path("library/day4").glob("*.txt")):
    paras = [p.strip() for p in f.read_text(encoding="utf-8").split("\n\n") if p.strip()]
    chunks += [f"{f.stem}: {p}" for p in paras[1:]]
    docs += [f.stem] * (len(paras) - 1)
V = emb.encode(["passage: " + c for c in chunks], normalize_embeddings=True)
# 같은 질문이 여러 번 들어오므로 서로 다른 질문만 한 번씩 임베딩한다
uniq = log["질문"].unique()
Q = emb.encode(["query: " + q for q in uniq], normalize_embeddings=True)
S = Q @ V.T
log["검색점수"] = log["질문"].map(dict(zip(uniq, S.max(axis=1))))
log["주제"] = log["질문"].map(dict(zip(uniq, [docs[i] for i in S.argmax(axis=1)])))
# 결과: 줄 수, 서로 다른 질문 수, 주마다의 질문 수
print(len(log), "줄 · 서로 다른 질문", len(uniq), "개")
print(log.groupby("주").size().to_dict())

def psi(base, now, eps=1e-4):
    keys = sorted(set(base.index) | set(now.index))
    b = base.reindex(keys, fill_value=0).to_numpy() + eps
    n = now.reindex(keys, fill_value=0).to_numpy() + eps
    return float(((n - b) * np.log(n / b)).sum())


base = log[log["주"] <= 4]                     # 기준 기간: 1~4주
p0 = base["주제"].value_counts(normalize=True)
weekly = pd.DataFrame({"질문": log.groupby("주").size(),
                       "PSI": log.groupby("주")["주제"].apply(lambda s: psi(p0, s.value_counts(normalize=True)))})
# 결과: 주마다의 PSI(1~4주는 기준 기간 안이라 작다)
print(weekly.round(3).T)

# KS 검정: 그 주의 검색 점수 분포가 기준 기간과 같은지. p 값이 작으면 다르다
weekly["KS p"] = [ks_2samp(base["검색점수"], g["검색점수"]).pvalue for _, g in log.groupby("주")]
# 처음 보는 질문: 기준 기간의 서로 다른 질문 가운데 가장 가까운 것과의 유사도가 0.95 아래
base_q = Q[np.isin(uniq, base["질문"].unique())]
nearest = dict(zip(uniq, (Q @ base_q.T).max(axis=1)))
log["처음"] = log["질문"].map(nearest) < 0.95
weekly["처음 보는 비율"] = log.groupby("주")["처음"].mean()
weekly["모름"] = log.groupby("주")["모름"].mean()
weekly["싫어요"] = log.groupby("주")["피드백"].apply(lambda s: (s == "싫어요").mean())
pd.set_option("display.float_format", lambda v: f"{v:.3g}" if v < 0.001 else f"{v:.3f}")
print(weekly)
pd.reset_option("display.float_format")       # 표시 설정을 되돌린다
# 결과: 처음 보는 질문 가운데 자주 나온 것
print(log[log["처음"]]["질문"].value_counts().head(8))

fig, axes = plt.subplots(1, 3, figsize=(11, 3.2))
weekly["PSI"].plot(ax=axes[0], marker="o", title="주제 분포의 PSI")
axes[0].axhline(0.1, linestyle="--", color="gray")
weekly["처음 보는 비율"].plot(ax=axes[1], marker="o", title="처음 보는 질문의 비율")
weekly[["모름", "싫어요"]].plot(ax=axes[2], marker="o", title="모름과 싫어요 비율")
for ax in axes:
    ax.set_xlabel("주")
plt.tight_layout()
plt.show()