실습 6 — 학습과 배포 파이프라인

import json
import shutil
import sys
from pathlib import Path

import joblib
import mlflow
import mlflow.sklearn
import pandas as pd
from mlflow import MlflowClient
from sentence_transformers import SentenceTransformer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import f1_score, recall_score

sys.path.insert(0, ".")                        # 실습 1의 out_flow.py
from out_flow import Stop, run_flow

mlflow.set_tracking_uri("sqlite:///out_mlflow.db")
mlflow.set_experiment("문의분류")
client = MlflowClient()
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
shutil.rmtree("out_serving", ignore_errors=True)   # 다시 돌려도 같게: 지난 배포 파일을 지운다


def embed(texts):
    return emb.encode(["query: " + t for t in texts], normalize_embeddings=True)


test = pd.read_csv("inquiries.csv")
E_test = embed(test["문의"])
LABELS = sorted(test["분류"].unique())
mv = client.get_model_version_by_alias("inquiry-clf", "champion")
print("시작할 때 챔피언 버전", mv.version, "· 학습에 넣은 묶음", client.get_run(mv.run_id).data.params["batches"])

def pipeline(batch_file):
    """새 묶음 하나로 파이프라인을 돌린다. 학습 데이터 = 기본 + 챔피언이 받아들인 묶음 + 새 묶음"""
    def check(ctx):
        df = pd.read_csv(batch_file)
        problems = []
        if list(df.columns) != ["문의", "분류"]:
            problems.append(f"열 이름 {list(df.columns)}")
        if df["문의"].isna().any() or (df["문의"].str.strip() == "").any():
            problems.append("빈 문의")
        if not set(df["분류"]).issubset(LABELS):
            problems.append(f"모르는 분류 {sorted(set(df['분류']) - set(LABELS))}")
        if df["분류"].value_counts().min() < 5:
            problems.append("분류 하나가 5개 미만")
        if problems:
            raise Stop("; ".join(problems))
        return df

    def train(ctx):
        champ = client.get_model_version_by_alias("inquiry-clf", "champion")
        params = client.get_run(champ.run_id).data.params
        batches = params["batches"].split(",") + [batch_file]      # 챔피언이 받아들인 묶음을 이어 받는다
        data = pd.concat([pd.read_csv("inquiry_train.csv")] + [pd.read_csv(b) for b in batches], ignore_index=True)
        model = LogisticRegression(C=float(params["C"]), max_iter=1000).fit(embed(data["문의"]), data["분류"])
        return {"모델": model, "C": float(params["C"]), "묶음": ",".join(batches), "행": len(data), "챔피언": champ.version}

    def evaluate(ctx):
        t = ctx["학습"]
        pred = t["모델"].predict(E_test)
        champ_pred = mlflow.sklearn.load_model("models:/inquiry-clf@champion").predict(E_test)
        result = {"accuracy": (pred == test["분류"]).mean(), "macro_f1": f1_score(test["분류"], pred, average="macro"),
                  "champion_accuracy": (champ_pred == test["분류"]).mean(),
                  "min_recall": recall_score(test["분류"], pred, labels=LABELS, average=None).min()}
        # 후보도 실행으로 기록하고 버전으로 등록한다. 게이트에서 떨어져도 기록은 남긴다
        with mlflow.start_run(run_name=f"pipeline-{Path(batch_file).stem}"):
            mlflow.log_params({"features": "e5", "C": t["C"], "train_rows": t["행"], "batches": t["묶음"]})
            mlflow.log_metrics(result)
            info = mlflow.sklearn.log_model(t["모델"], name="model", registered_model_name="inquiry-clf",
                                            pip_requirements=["scikit-learn==1.9.1"])
        return dict(result, version=info.registered_model_version)

    def gate(ctx):
        r = ctx["평가"]
        reasons = []
        if r["accuracy"] < r["champion_accuracy"]:
            reasons.append(f"정확도 {r['accuracy']:.3f} < 챔피언 {r['champion_accuracy']:.3f}")
        if r["min_recall"] < 0.25:
            reasons.append(f"가장 낮은 재현율 {r['min_recall']:.3f} < 0.25")
        client.set_model_version_tag("inquiry-clf", str(r["version"]), "gate", "탈락" if reasons else "통과")
        if reasons:
            raise Stop("; ".join(reasons))
        return "통과"

    def deploy(ctx):
        r, t = ctx["평가"], ctx["학습"]
        client.set_registered_model_alias("inquiry-clf", "previous", t["챔피언"])
        client.set_registered_model_alias("inquiry-clf", "champion", r["version"])
        # 서비스가 읽을 파일: 모델과 판 정보. 새 파일을 다 쓴 뒤 이름을 바꾼다(실습 3과 같은 방법)
        Path("out_serving").mkdir(exist_ok=True)
        joblib.dump(t["모델"], "out_serving/model.tmp")
        Path("out_serving/model.tmp").replace("out_serving/inquiry_model.joblib")
        release = {"version": r["version"], "accuracy": round(r["accuracy"], 3), "batches": t["묶음"],
                   "embedding": "intfloat/multilingual-e5-small"}
        Path("out_serving/release.json").write_text(json.dumps(release, ensure_ascii=False), encoding="utf-8")
        return release

    tasks = {"검사": (check, []), "학습": (train, ["검사"]), "평가": (evaluate, ["학습"]),
             "게이트": (gate, ["평가"]), "배포": (deploy, ["게이트"])}
    return run_flow(tasks, log=lambda m: print("  " + m))


print("파이프라인 준비 끝")

records, ctx = pipeline("inquiry_nov.csv")
r = ctx.get("평가", {})
# 결과: 후보와 챔피언의 정확도, 그리고 지금 챔피언
print(f"후보 버전 {r.get('version')} · 정확도 {r.get('accuracy', 0):.3f} · 챔피언 {r.get('champion_accuracy', 0):.3f}")
print("지금 챔피언 버전", client.get_model_version_by_alias("inquiry-clf", "champion").version)

champion = mlflow.sklearn.load_model("models:/inquiry-clf@champion")
nov = pd.read_csv("inquiry_nov.csv")
# 챔피언의 예측과 11월 묶음의 라벨을 엇갈려 센다. 대각선에서 벗어난 칸이 몰려 있으면 라벨이 수상하다
cross = pd.crosstab(nov["분류"].rename("11월 라벨"), pd.Series(champion.predict(embed(nov["문의"])), name="챔피언 예측"))
print(cross)
print(f"챔피언과 라벨이 같은 비율: {(champion.predict(embed(nov['문의'])) == nov['분류']).mean():.3f}")

# 위 표를 보고 라벨 도구가 분류를 한 칸씩 밀어 적었다고 판단했다. 거꾸로 돌려 바로잡은 파일을 따로 쓴다
fix = {"교환반품": "배송", "결제": "교환반품", "회원": "결제", "기타": "회원", "배송": "기타"}
nov.assign(분류=nov["분류"].map(fix)).to_csv("out_inquiry_nov_fixed.csv", index=False, encoding="utf-8-sig")
records, ctx = pipeline("out_inquiry_nov_fixed.csv")
# 결과: 배포한 판 정보와 별칭
print(json.loads(Path("out_serving/release.json").read_text(encoding="utf-8")) if Path("out_serving/release.json").exists() else "배포 안 됨")
print("별칭:", client.get_registered_model("inquiry-clf").aliases)
versions = client.search_model_versions("name='inquiry-clf'")
print("게이트 꼬리표:", {v.version: v.tags.get("gate", "-") for v in sorted(versions, key=lambda v: int(v.version))})