src/training/train_transformer.py

import os

import mlflow
import numpy as np
import torch
from torch.utils.data import DataLoader
from transformers import EarlyStoppingCallback, Trainer, TrainingArguments

from src.data.loader import load_clinc150, load_splits, save_splits
from src.data.preprocessor import preprocess
from src.evaluation.metrics import (
    compute_classification_metrics,
    compute_latency,
    get_classification_report,
)
from src.models.transformer import IntentDatasetHF, TransformerModel
from src.storage.s3 import upload_artifact
from src.utils.config import load_config
from src.utils.mlflow_utils import (
    get_or_create_experiment,
    log_confusion_matrix,
    log_metrics,
)
from src.utils.settings import settings

DATA_DIR = "data/raw"
MODEL_DIR = "artifacts/models/distilbert"


def load_or_download_data(config: dict) -> tuple:
    train_path = os.path.join(DATA_DIR, "train.csv")
    if os.path.exists(train_path):
        print("loading data from disk...")
        splits = load_splits(DATA_DIR)
    else:
        print("downloading CLINC150...")
        splits = load_clinc150(config["data"]["subset"])
        save_splits(splits, DATA_DIR)

    processed, label_map = preprocess(splits)
    return processed, label_map


def compute_metrics_hf(eval_pred) -> dict:
    logits, labels = eval_pred
    preds = np.argmax(logits, axis=1)
    metrics = compute_classification_metrics(labels, preds)
    return {
        "accuracy": metrics["accuracy"],
        "macro_f1": metrics["macro_f1"],
    }


def predict_all(
    transformer: TransformerModel,
    dataset: IntentDatasetHF,
    batch_size: int,
    device: torch.device,
) -> tuple[np.ndarray, np.ndarray]:
    transformer.model.eval()
    transformer.model.to(device)

    loader = DataLoader(dataset, batch_size=batch_size, shuffle=False)
    all_preds = []
    all_labels = []

    with torch.no_grad():
        for batch in loader:
            input_ids = batch["input_ids"].to(device)
            attention_mask = batch["attention_mask"].to(device)
            labels = batch["labels"]

            outputs = transformer.model(input_ids=input_ids, attention_mask=attention_mask)
            preds = outputs.logits.argmax(dim=1).cpu().numpy()
            all_preds.extend(preds)
            all_labels.extend(labels.numpy())

    return np.array(all_preds), np.array(all_labels)


def main():
    config = load_config("transformer")
    torch.manual_seed(config["training"]["random_state"])

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"device: {device}")

    processed, label_map = load_or_download_data(config)

    model_cfg = config["model"]
    training_cfg = config["training"]
    max_length = config["data"]["max_length"]

    print(f"loading {model_cfg['name']}...")
    transformer = TransformerModel(
        model_name=model_cfg["name"],
        num_labels=model_cfg["num_labels"],
        dropout=model_cfg["dropout"],
    )

    print("tokenizing datasets...")
    train_dataset = IntentDatasetHF(
        processed["train"]["text"].tolist(),
        processed["train"]["label_id"].tolist(),
        transformer.tokenizer,
        max_length,
    )
    val_dataset = IntentDatasetHF(
        processed["validation"]["text"].tolist(),
        processed["validation"]["label_id"].tolist(),
        transformer.tokenizer,
        max_length,
    )
    test_dataset = IntentDatasetHF(
        processed["test"]["text"].tolist(),
        processed["test"]["label_id"].tolist(),
        transformer.tokenizer,
        max_length,
    )

    mlflow.set_tracking_uri(settings.mlflow_tracking_uri)
    experiment_id = get_or_create_experiment(config["mlflow"]["experiment_name"])

    with mlflow.start_run(
        experiment_id=experiment_id,
        run_name=config["mlflow"]["run_name"],
    ) as run:
        mlflow.log_params(
            {
                "model.name": model_cfg["name"],
                "model.num_labels": model_cfg["num_labels"],
                "model.dropout": model_cfg["dropout"],
                "training.epochs": training_cfg["epochs"],
                "training.batch_size": training_cfg["batch_size"],
                "training.learning_rate": training_cfg["learning_rate"],
                "training.warmup_steps": training_cfg["warmup_steps"],
                "training.weight_decay": training_cfg["weight_decay"],
                "data.max_length": max_length,
            }
        )

        training_args = TrainingArguments(
            output_dir=MODEL_DIR,
            num_train_epochs=training_cfg["epochs"],
            per_device_train_batch_size=training_cfg["batch_size"],
            per_device_eval_batch_size=training_cfg["batch_size"],
            learning_rate=training_cfg["learning_rate"],
            warmup_steps=training_cfg["warmup_steps"],
            weight_decay=training_cfg["weight_decay"],
            eval_strategy="epoch",
            save_strategy="epoch",
            load_best_model_at_end=True,
            metric_for_best_model="macro_f1",
            greater_is_better=True,
            logging_steps=50,
            report_to="none",
            seed=training_cfg["random_state"],
        )

        trainer = Trainer(
            model=transformer.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=val_dataset,
            compute_metrics=compute_metrics_hf,
            callbacks=[EarlyStoppingCallback(early_stopping_patience=2)],
        )

        print("\nfine-tuning distilbert...")
        trainer.train()

        print("\nloading best checkpoint...")
        transformer.save(MODEL_DIR)

        id_to_label = {v: k for k, v in label_map.items()}
        label_names = [id_to_label[i] for i in range(len(label_map))]

        print("evaluating on test set...")
        test_preds, test_labels = predict_all(transformer, test_dataset, training_cfg["batch_size"], device)

        test_metrics = compute_classification_metrics(test_labels, test_preds)
        test_metrics_logged = {f"test_{k}": v for k, v in test_metrics.items()}
        log_metrics(test_metrics_logged)

        def predict_fn(dataset):
            predict_all(transformer, dataset, training_cfg["batch_size"], device)

        latency = compute_latency(predict_fn, test_dataset, n_runs=20)
        log_metrics(latency)

        report = get_classification_report(test_labels, test_preds, label_names)
        report_path = "artifacts/distilbert_report.txt"
        with open(report_path, "w") as f:
            f.write(report)
        mlflow.log_artifact(report_path)

        log_confusion_matrix(
            test_labels,
            test_preds,
            label_names,
            save_path="artifacts/distilbert_confusion_matrix.png",
        )

        mlflow.log_artifact(MODEL_DIR)

        print("uploading to S3...")
        for fname in os.listdir(MODEL_DIR):
            local_path = os.path.join(MODEL_DIR, fname)
            if os.path.isfile(local_path):
                upload_artifact(local_path, f"{config['s3']['prefix']}/distilbert/{fname}")

        print(f"\ntest accuracy : {test_metrics['accuracy']}")
        print(f"test macro_f1 : {test_metrics['macro_f1']}")
        print(f"latency p50 : {latency['latency_p50_ms']}ms")
        print(f"run id : {run.info.run_id}")


if __name__ == "__main__":
    main()