Day 41: MLflow, Experiment Tracking Và Model Registry

Model này được train bằng data version nào?
Code commit nào tạo ra model này?
Config và hyperparameters chính xác là gì?
Metric validation/test có đạt release gate không?
Artifact đánh giá nằm ở đâu?
Ai approve model version này?
Nếu production lỗi, rollback về version nào?

training code
  -> run metadata
  -> params
  -> metrics
  -> dataset lineage
  -> artifacts
  -> logged model
  -> registered model version
  -> alias candidate/champion
  -> deployment reads exact model URI

train_day41.py
  -> MLflow Tracking Server
      -> SQLite backend store
      -> local artifact directory
      -> Model Registry

training job / CI pipeline
  -> MLflow Tracking Server behind auth/TLS
      -> Postgres/MySQL backend store
      -> S3/GCS/Azure Blob artifact store
      -> Model Registry
  -> model validation job
  -> deployment system loads models:/<name>@champion
  -> monitoring writes drift/latency/error metrics

python3 -m venv .venv
source .venv/bin/activate
pip install -U mlflow scikit-learn pandas matplotlib

pip install -U torch transformers datasets peft accelerate

mkdir -p mlruns mlartifacts
mlflow server \
  --host 127.0.0.1 \
  --port 5000 \
  --backend-store-uri sqlite:///mlruns/mlflow.db \
  --default-artifact-root ./mlartifacts

http://127.0.0.1:5000

text,label
"Sản phẩm dùng ổn, giao hàng nhanh",positive
"Ứng dụng lỗi liên tục sau khi cập nhật",negative
"Dịch vụ bình thường, chưa có gì nổi bật",neutral

from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

import mlflow
import mlflow.data
import mlflow.sklearn
import pandas as pd
from mlflow import MlflowClient
from mlflow.models import infer_signature
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, classification_report, f1_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline


@dataclass(frozen=True)
class TrainConfig:
    tracking_uri: str
    experiment_name: str
    registered_model_name: str
    dataset_path: Path
    dataset_version: str
    artifact_dir: Path
    test_size: float = 0.2
    random_state: int = 42
    max_features: int = 30000
    ngram_min: int = 1
    ngram_max: int = 2
    classifier_c: float = 1.0
    min_macro_f1: float = 0.78
    max_p95_latency_ms: float = 30.0


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def git_commit() -> str:
    try:
        return subprocess.check_output(
            ["git", "rev-parse", "HEAD"],
            stderr=subprocess.DEVNULL,
            text=True,
        ).strip()
    except Exception:
        return "unknown"


def pip_freeze() -> str:
    try:
        return subprocess.check_output(
            [sys.executable, "-m", "pip", "freeze"],
            stderr=subprocess.DEVNULL,
            text=True,
        )
    except Exception:
        return "requirements unavailable\n"


def write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def load_dataset(path: Path) -> pd.DataFrame:
    df = pd.read_csv(path)
    required_columns = {"text", "label"}
    missing = required_columns - set(df.columns)
    if missing:
        raise ValueError(f"Dataset thiếu cột bắt buộc: {sorted(missing)}")

    df = df.dropna(subset=["text", "label"]).copy()
    df["text"] = df["text"].astype(str).str.strip()
    df["label"] = df["label"].astype(str).str.strip()
    df = df[(df["text"] != "") & (df["label"] != "")]
    if len(df) < 20:
        raise ValueError("Dataset quá nhỏ để đánh giá ổn định. Cần ít nhất 20 dòng.")
    return df


def build_model(cfg: TrainConfig) -> Pipeline:
    return Pipeline(
        steps=[
            (
                "tfidf",
                TfidfVectorizer(
                    lowercase=True,
                    max_features=cfg.max_features,
                    ngram_range=(cfg.ngram_min, cfg.ngram_max),
                ),
            ),
            (
                "classifier",
                LogisticRegression(
                    C=cfg.classifier_c,
                    max_iter=1000,
                    class_weight="balanced",
                    random_state=cfg.random_state,
                ),
            ),
        ]
    )


def estimate_latency_ms(model: Pipeline, samples: pd.Series, repeats: int = 30) -> dict[str, float]:
    durations = []
    batch = samples.head(min(len(samples), 32))
    for _ in range(repeats):
        start = time.perf_counter()
        model.predict(batch)
        durations.append((time.perf_counter() - start) * 1000)

    durations = sorted(durations)
    p95_index = max(0, int(len(durations) * 0.95) - 1)
    return {
        "mean_latency_ms": sum(durations) / len(durations),
        "p95_latency_ms": durations[p95_index],
    }


def train_and_log(cfg: TrainConfig) -> str:
    mlflow.set_tracking_uri(cfg.tracking_uri)
    mlflow.set_experiment(cfg.experiment_name)

    dataset_hash = sha256_file(cfg.dataset_path)
    commit = git_commit()
    df = load_dataset(cfg.dataset_path)

    stratify = df["label"] if df["label"].value_counts().min() >= 2 else None
    train_df, test_df = train_test_split(
        df,
        test_size=cfg.test_size,
        random_state=cfg.random_state,
        stratify=stratify,
    )

    model = build_model(cfg)
    run_name = f"{cfg.registered_model_name}-{cfg.dataset_version}-{commit[:8]}"

    with mlflow.start_run(run_name=run_name) as run:
        run_id = run.info.run_id

        mlflow.log_params(
            {
                "model_type": "tfidf_logistic_regression",
                "dataset_path": str(cfg.dataset_path),
                "dataset_version": cfg.dataset_version,
                "dataset_hash": dataset_hash,
                "test_size": cfg.test_size,
                "random_state": cfg.random_state,
                "max_features": cfg.max_features,
                "ngram_range": f"{cfg.ngram_min},{cfg.ngram_max}",
                "classifier_c": cfg.classifier_c,
            }
        )
        mlflow.set_tags(
            {
                "task": "vietnamese_sentiment",
                "owner": os.getenv("USER", "unknown"),
                "git_commit": commit,
                "training_script": "train_day41.py",
                "approval_status": "pending",
            }
        )

        train_input = mlflow.data.from_pandas(
            train_df,
            source=str(cfg.dataset_path),
            targets="label",
            name=f"{cfg.dataset_version}-train",
        )
        test_input = mlflow.data.from_pandas(
            test_df,
            source=str(cfg.dataset_path),
            targets="label",
            name=f"{cfg.dataset_version}-test",
        )
        mlflow.log_input(train_input, context="training")
        mlflow.log_input(test_input, context="evaluation")

        model.fit(train_df["text"], train_df["label"])
        predictions = model.predict(test_df["text"])

        report = classification_report(
            test_df["label"],
            predictions,
            output_dict=True,
            zero_division=0,
        )
        metrics = {
            "accuracy": accuracy_score(test_df["label"], predictions),
            "macro_f1": f1_score(test_df["label"], predictions, average="macro"),
            "weighted_f1": f1_score(test_df["label"], predictions, average="weighted"),
        }
        metrics.update(estimate_latency_ms(model, test_df["text"]))
        mlflow.log_metrics(metrics)

        cfg.artifact_dir.mkdir(parents=True, exist_ok=True)
        write_json(cfg.artifact_dir / "classification_report.json", report)
        write_json(
            cfg.artifact_dir / "eval_summary.json",
            {
                "run_id": run_id,
                "registered_model_name": cfg.registered_model_name,
                "metrics": metrics,
                "dataset_version": cfg.dataset_version,
                "dataset_hash": dataset_hash,
                "git_commit": commit,
            },
        )
        (cfg.artifact_dir / "model_card.md").write_text(
            "\n".join(
                [
                    f"# Model Card: {cfg.registered_model_name}",
                    "",
                    "## Intended use",
                    "Vietnamese sentiment classification for controlled text inputs.",
                    "",
                    "## Validation",
                    f"- Dataset version: `{cfg.dataset_version}`",
                    f"- Dataset hash: `{dataset_hash}`",
                    f"- Macro F1: `{metrics['macro_f1']:.4f}`",
                    f"- P95 latency ms: `{metrics['p95_latency_ms']:.2f}`",
                    "",
                    "## Known limitations",
                    "- Chưa kiểm thử drift theo thời gian.",
                    "- Chưa kiểm thử fairness theo domain/user segment.",
                    "- Không log raw PII; sample predictions phải được redact trước khi chia sẻ.",
                    "",
                "## Rollback",
                "Deployment nên load `models:/sentiment-classifier@champion`; rollback bằng cách trỏ alias `champion` về version trước đó.",
            ]
        ),
            encoding="utf-8",
        )
        (cfg.artifact_dir / "requirements-lock.txt").write_text(
            pip_freeze(),
            encoding="utf-8",
        )
        mlflow.log_artifacts(str(cfg.artifact_dir), artifact_path="reports")

        signature = infer_signature(test_df["text"].head(5), model.predict(test_df["text"].head(5)))
        model_info = mlflow.sklearn.log_model(
            sk_model=model,
            name="model",
            signature=signature,
            input_example=test_df["text"].head(2),
            registered_model_name=cfg.registered_model_name,
        )

        client = MlflowClient()
        version = str(model_info.registered_model_version)
        client.set_model_version_tag(
            name=cfg.registered_model_name,
            version=version,
            key="validation_status",
            value="candidate",
        )
        client.set_model_version_tag(
            name=cfg.registered_model_name,
            version=version,
            key="source_run_id",
            value=run_id,
        )

        passed_gate = (
            metrics["macro_f1"] >= cfg.min_macro_f1
            and metrics["p95_latency_ms"] <= cfg.max_p95_latency_ms
        )
        client.set_model_version_tag(
            name=cfg.registered_model_name,
            version=version,
            key="release_gate",
            value="passed" if passed_gate else "failed",
        )
        if passed_gate:
            client.set_registered_model_alias(
                name=cfg.registered_model_name,
                alias="candidate",
                version=version,
            )

        return run_id


def parse_args() -> TrainConfig:
    parser = argparse.ArgumentParser()
    parser.add_argument("--tracking-uri", default="http://127.0.0.1:5000")
    parser.add_argument("--experiment-name", default="day41-sentiment-classifier")
    parser.add_argument("--registered-model-name", default="sentiment-classifier")
    parser.add_argument("--dataset-path", type=Path, required=True)
    parser.add_argument("--dataset-version", required=True)
    parser.add_argument("--artifact-dir", type=Path, default=Path("artifacts/day41"))
    parser.add_argument("--classifier-c", type=float, default=1.0)
    args = parser.parse_args()
    return TrainConfig(**vars(args))


if __name__ == "__main__":
    config = parse_args()
    print(json.dumps(asdict(config), default=str, ensure_ascii=False, indent=2))
    print(f"Logged run_id={train_and_log(config)}")

python train_day41.py \
  --dataset-path data/sentiment_v1.csv \
  --dataset-version sentiment-v1 \
  --classifier-c 1.0

python train_day41.py --dataset-path data/sentiment_v1.csv --dataset-version sentiment-v1 --classifier-c 0.3
python train_day41.py --dataset-path data/sentiment_v1.csv --dataset-version sentiment-v1 --classifier-c 1.0
python train_day41.py --dataset-path data/sentiment_v1.csv --dataset-version sentiment-v1 --classifier-c 3.0

import argparse

import mlflow
from mlflow import MlflowClient


def promote(tracking_uri: str, model_name: str, source_alias: str, target_alias: str) -> None:
    mlflow.set_tracking_uri(tracking_uri)
    client = MlflowClient()

    candidate = client.get_model_version_by_alias(model_name, source_alias)
    if candidate.tags.get("release_gate") != "passed":
        raise RuntimeError(
            f"Model version {candidate.version} chưa qua release gate: "
            f"{candidate.tags.get('release_gate')}"
        )

    client.set_model_version_tag(
        name=model_name,
        version=candidate.version,
        key="approval_status",
        value="approved",
    )
    client.set_registered_model_alias(
        name=model_name,
        alias=target_alias,
        version=candidate.version,
    )
    print(f"Promoted {model_name} version {candidate.version} to @{target_alias}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--tracking-uri", default="http://127.0.0.1:5000")
    parser.add_argument("--model-name", default="sentiment-classifier")
    parser.add_argument("--source-alias", default="candidate")
    parser.add_argument("--target-alias", default="champion")
    args = parser.parse_args()
    promote(args.tracking_uri, args.model_name, args.source_alias, args.target_alias)

import mlflow.pyfunc

model = mlflow.pyfunc.load_model("models:/sentiment-classifier@champion")
predictions = model.predict(["Dịch vụ hỗ trợ rất nhanh"])

corpus_version
chunking_strategy
embedding_model
embedding_dimension
index_version
retriever_config
reranker_model
prompt_version
llm_model
eval_set_version

experiment: <phase>-<task>-<model-family>
run_name:   <dataset-version>-<model-config>-<git-short-sha>
model:      <business-task>-<model-role>
alias:      candidate | champion | shadow | rollback

experiment: day41-vietnamese-sentiment
run_name: sentiment-v1-tfidf-lr-a1b2c3d4
model: sentiment-classifier
alias: champion

artifacts/day41/
  classification_report.json
  confusion_matrix.png
  eval_summary.json
  model_card.md
  sample_predictions_redacted.jsonl
  requirements-lock.txt

{
  "run_id": "abc123",
  "registered_model_name": "sentiment-classifier",
  "model_version": "3",
  "dataset_version": "sentiment-v1",
  "dataset_hash": "sha256:...",
  "git_commit": "a1b2c3d4",
  "primary_metric": "macro_f1",
  "metrics": {
    "macro_f1": 0.84,
    "accuracy": 0.86,
    "p95_latency_ms": 18.5
  },
  "release_gate": {
    "macro_f1_min": 0.80,
    "p95_latency_ms_max": 30,
    "status": "passed"
  }
}

# Model Card: <registered_model_name>

## Intended Use
<Bài toán model được phép xử lý.>

## Not Intended Use
<Các tình huống không được dùng hoặc chưa được kiểm chứng.>

## Training Data
- Dataset version:
- Dataset hash:
- Split strategy:
- Data source:
- PII handling:

## Evaluation
- Primary metric:
- Secondary metrics:
- Latency:
- Cost estimate:
- Error analysis:

## Model Version
- Registered model:
- Version:
- Alias:
- Source run:
- Git commit:

## Limitations
- <Known limitation 1>
- <Known limitation 2>

## Production Conditions
- <Điều kiện hạ tầng, monitoring, security, fallback.>

## Rollback
- Previous champion version:
- Rollback command:

quality:
  macro_f1_min: 0.80
  negative_recall_min: 0.75
performance:
  p95_latency_ms_max: 30
security:
  pii_in_artifacts: false
reproducibility:
  requires_dataset_hash: true
  requires_git_commit: true
approval:
  reviewer: required

run training
  -> log params/metrics/artifacts/model
  -> register model version
  -> set version tag validation_status=candidate
  -> offline validation
  -> set alias candidate
  -> reviewer approves
  -> set alias champion
  -> deploy serving reads @champion
  -> monitor
  -> rollback alias if regression

Chỉ chọn run mới nếu:
- Macro F1 tăng ít nhất 1 điểm phần trăm hoặc sửa lỗi class quan trọng.
- Không làm p95 latency vượt gate.
- Không tăng cost vượt budget.
- Không có regression nghiêm trọng trong error analysis.

from mlflow import MlflowClient

client = MlflowClient()
model_name = "sentiment-classifier"
candidate = client.get_model_version_by_alias(model_name, "candidate")

client.set_model_version_tag(
    name=model_name,
    version=candidate.version,
    key="approval_status",
    value="approved",
)
client.set_registered_model_alias(
    name=model_name,
    alias="champion",
    version=candidate.version,
)

import mlflow.pyfunc

model = mlflow.pyfunc.load_model("models:/sentiment-classifier@champion")
result = model.predict(["Ứng dụng hoạt động ổn định sau bản cập nhật"])
print(result)

from mlflow import MlflowClient

client = MlflowClient()
client.set_registered_model_alias(
    name="sentiment-classifier",
    alias="champion",
    version="2",
)
client.set_model_version_tag(
    name="sentiment-classifier",
    version="3",
    key="incident_status",
    value="rolled_back",
)

## Dùng được trong production không?

Có/Chưa.

### Nếu có, điều kiện bắt buộc
- Tracking server:
- Backend store:
- Artifact store:
- Auth/TLS:
- Dataset/code lineage:
- Release gate:
- Monitoring:
- Rollback:

### Rủi ro còn lại
- Performance:
- Cost:
- Security:
- Reproducibility:
- Model quality:

### Quyết định release
- Champion hiện tại:
- Candidate:
- Metric thay đổi:
- Người approve:
- Rollback version:

day41-mlflow-lab/
  data/
    sentiment_v1.csv
  artifacts/
  train_day41.py
  promote_model.py
  decision-note.md
  README.md

python3 -m venv .venv
source .venv/bin/activate
pip install -U mlflow scikit-learn pandas matplotlib

mkdir -p mlruns mlartifacts
mlflow server \
  --host 127.0.0.1 \
  --port 5000 \
  --backend-store-uri sqlite:///mlruns/mlflow.db \
  --default-artifact-root ./mlartifacts

text,label
"Giao hàng nhanh và đóng gói cẩn thận",positive
"Ứng dụng bị lỗi khi đăng nhập",negative
"Dịch vụ ở mức chấp nhận được",neutral

python train_day41.py \
  --dataset-path data/sentiment_v1.csv \
  --dataset-version sentiment-v1 \
  --classifier-c 0.3

python train_day41.py \
  --dataset-path data/sentiment_v1.csv \
  --dataset-version sentiment-v1 \
  --classifier-c 1.0

python train_day41.py \
  --dataset-path data/sentiment_v1.csv \
  --dataset-version sentiment-v1 \
  --classifier-c 3.0

Candidate được chọn nếu:
- macro_f1 cao nhất hoặc tốt hơn champion hiện tại ít nhất 1 điểm phần trăm.
- Không class quan trọng nào có recall quá thấp.
- p95_latency_ms không vượt 30 ms với batch nhỏ local.
- Artifact đầy đủ và không chứa PII.
- Có dataset_hash và git_commit.

# Day 41 Release Decision

## Candidate
- Registered model:
- Version:
- Source run:
- Dataset version:
- Dataset hash:
- Git commit:

## Metrics
- Macro F1:
- Accuracy:
- Weighted F1:
- P95 latency:

## So sánh với baseline/champion
- Metric tăng/giảm:
- Latency tăng/giảm:
- Cost/storage impact:

## Quyết định
Promote/Không promote.

## Lý do
<Giải thích ngắn, dựa trên metric và limitation.>

## Rollback
- Previous champion version:
- Rollback command:

from mlflow import MlflowClient

client = MlflowClient()
client.set_registered_model_alias(
    name="sentiment-classifier",
    alias="candidate",
    version="1",
)

from mlflow import MlflowClient

client = MlflowClient()
candidate = client.get_model_version_by_alias("sentiment-classifier", "candidate")
client.set_registered_model_alias(
    name="sentiment-classifier",
    alias="champion",
    version=candidate.version,
)

import mlflow.pyfunc

model = mlflow.pyfunc.load_model("models:/sentiment-classifier@champion")
print(model.predict(["Sản phẩm tốt, hỗ trợ nhanh"]))

from mlflow import MlflowClient

client = MlflowClient()
client.set_registered_model_alias(
    name="sentiment-classifier",
    alias="champion",
    version="1",
)
client.set_model_version_tag(
    name="sentiment-classifier",
    version="2",
    key="incident_status",
    value="rolled_back_in_lab",
)

Adapter không đủ để reproduce nếu thiếu base model id/revision, tokenizer revision,
prompt/chat template version và serving config.

## Dùng được trong production không?

Chưa, nếu chỉ chạy local bằng SQLite và local artifact store.

Có thể dùng trong production nếu:
- MLflow Tracking Server được deploy sau auth/TLS.
- Backend store dùng Postgres/MySQL có backup.
- Artifact store dùng S3/GCS/Azure Blob có encryption và lifecycle policy.
- Training pipeline bắt buộc log dataset hash, code commit, params, metrics, artifacts và model signature.
- Promotion qua release gate và reviewer approval.
- Serving load model bằng `models:/sentiment-classifier@champion`.
- Có monitoring cho drift, latency, error rate và business metric.
- Có rollback alias về previous champion.

Rủi ro còn lại:
- Dataset nhỏ nên metric chưa ổn định.
- Chưa có drift monitoring.
- Chưa có security review đầy đủ cho artifact.
- Chưa benchmark trên traffic thật.