실습 4 — MLflow 로 실험 기록하기

import hashlib
import shutil
from pathlib import Path

import mlflow
import mlflow.sklearn
import pandas as pd
from mlflow import MlflowClient

# 다시 돌려도 같은 결과가 나오게 지난 기록을 지운다
Path("out_mlflow.db").unlink(missing_ok=True)
shutil.rmtree("out_mlruns", ignore_errors=True)
# 기록은 SQLite 파일 하나에, 모델 같은 파일(산출물)은 out_mlruns 폴더에 둔다
mlflow.set_tracking_uri("sqlite:///out_mlflow.db")
mlflow.create_experiment("문의분류", artifact_location=Path("out_mlruns").resolve().as_uri())
mlflow.set_experiment("문의분류")

train = pd.read_csv("inquiry_train.csv")
test = pd.read_csv("inquiries.csv")
# 데이터의 지문: 같은 파일로 학습했는지 나중에 확인할 수 있게 기록에 남긴다
data_fp = hashlib.sha256(Path("inquiry_train.csv").read_bytes()).hexdigest()[:12]
# 결과: 학습·시험 행 수와 지문
print(len(train), len(test), data_fp)

from sentence_transformers import SentenceTransformer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import f1_score
from sklearn.pipeline import make_pipeline

emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
# e5 는 문장 앞에 "query: " 를 붙여 임베딩한다(7차시). 학습·시험 문장을 한 번씩만 임베딩해 둔다
E_train = emb.encode(["query: " + q for q in train["문의"]], normalize_embeddings=True)
E_test = emb.encode(["query: " + q for q in test["문의"]], normalize_embeddings=True)

models = {}
for features in ["tfidf", "e5"]:
    for C in [0.3, 1, 3, 10]:
        name = f"{features}-C{C}"
        # start_run: 실행 하나를 연다. with 블록이 끝나면 실행이 닫힌다
        with mlflow.start_run(run_name=name):
            if features == "tfidf":
                model = make_pipeline(TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 3)),
                                      LogisticRegression(C=C, max_iter=1000)).fit(train["문의"], train["분류"])
                pred = model.predict(test["문의"])
            else:
                model = LogisticRegression(C=C, max_iter=1000).fit(E_train, train["분류"])
                pred = model.predict(E_test)
            # 설정값(파라미터), 점수(지표), 메모(태그)를 남긴다
            mlflow.log_params({"features": features, "C": C, "train_rows": len(train)})
            mlflow.log_metrics({"accuracy": (pred == test["분류"]).mean(),
                                "macro_f1": f1_score(test["분류"], pred, average="macro")})
            mlflow.set_tag("data_fp", data_fp)
        models[name] = model
print("기록한 실행:", len(models))

# search_runs: 실험의 실행을 표(DataFrame)로 돌려준다. 정확도, 같으면 macro F1 이 높은 차례
runs = mlflow.search_runs(experiment_names=["문의분류"], order_by=["metrics.accuracy DESC", "metrics.macro_f1 DESC"])
table = runs[["tags.mlflow.runName", "params.features", "params.C", "metrics.accuracy", "metrics.macro_f1"]]
table.columns = ["실행", "특성", "C", "정확도", "macro_f1"]
# 결과: 여덟 실행의 점수(소수 셋째 자리)
print(table.round(3).to_string(index=False))

best = runs.iloc[0]
best_name = best["tags.mlflow.runName"]
# 고른 실행에 모델 파일을 남기고, 레지스트리의 'inquiry-clf' 에 새 버전으로 등록한다
with mlflow.start_run(run_id=best["run_id"]):
    info = mlflow.sklearn.log_model(models[best_name], name="model", registered_model_name="inquiry-clf",
                                    pip_requirements=["scikit-learn==1.9.1"])
client = MlflowClient()
version = info.registered_model_version        # 새로 붙은 버전 번호
# 서비스가 부를 이름표: champion. 이 모델은 e5 임베딩 위의 분류기라 임베딩 모델 이름도 버전에 적는다
client.set_registered_model_alias("inquiry-clf", "champion", version)
client.set_model_version_tag("inquiry-clf", str(version), "embedding", "intfloat/multilingual-e5-small")
mv = client.get_model_version_by_alias("inquiry-clf", "champion")
# 결과: 등록한 실행, 버전, 별칭
print(best_name, "→ inquiry-clf 버전", mv.version, "· 별칭", mv.aliases, f"· 정확도 {best['metrics.accuracy']:.3f}")