classifier/train.py

from classifier.utils import CHECKPOINT_PATH, DATETIME_FORMAT, get_models, CATEGORIES, DEVICE, CLASSIFIER_NAME
from classifier.config import HF_TOKEN
from huggingface_hub import HfApi
from jinja2 import Template

import argparse
from datetime import datetime
import datasets as ds
import matplotlib.pyplot as plt
import numpy as np
import os
import pandas as pd
import torch
from torch.utils.data import DataLoader

def even_split(prefix: str, target: int, splits: int, total: int) -> str:
    result = ""
    target_amount_per_split = int(target / splits)
    total_amount_per_split = int(total / splits)

    for i in range(splits):
        left = total_amount_per_split*i
        right = left + target_amount_per_split
        result += f"{prefix}[{int(left)}:{int(right)}]"

        if i != splits - 1:
            result += "+"

    return result

def get_model_train_test():
    # Login using e.g. `huggingface-cli login` to access this dataset

    def add_static_label(row, column_name, label):
        row[column_name] = label
        return row

    # Miriad
    train_split = even_split("train", 50000, 100, 4470000)
    miriad = ds.load_dataset("tomaarsen/miriad-4.4M-split", split={"train":train_split, "test": "test", "validation": "eval"})
    miriad = miriad.rename_column("question", "text")
    miriad = miriad.remove_columns("passage_text")
    miriad = miriad.map(add_static_label, fn_kwargs={"column_name": "label", "label": "medical"})
    # print(miriad)

    # Insurance
    train_split = even_split("train", 5000, 20, 21300)
    insurance = ds.load_dataset("deccan-ai/insuranceQA-v2", split={"train":train_split, "test":"test", "validation":"validation"})
    insurance = insurance.rename_column("input", "text")
    insurance = insurance.remove_columns(["output"])
    insurance = insurance.map(add_static_label, fn_kwargs={"column_name": "label", "label": "insurance"})
    # print(insurance)

    # Interleave datasets (mix the datasets into one randomly)
    train = ds.interleave_datasets([miriad["train"], insurance["train"]], stopping_strategy="all_exhausted")
    _ , unique_indices = np.unique(train["text"], return_index=True, axis=0)
    train = train.select(unique_indices.tolist())
    test = ds.interleave_datasets([miriad["test"], insurance["test"]], stopping_strategy="all_exhausted")
    _ , unique_indices = np.unique(test["text"], return_index=True, axis=0)
    test = test.select(unique_indices.tolist())
    validation = ds.interleave_datasets([miriad["validation"], insurance["validation"]], stopping_strategy="all_exhausted")
    _ , unique_indices = np.unique(validation["text"], return_index=True, axis=0)
    validation = validation.select(unique_indices.tolist())

    print(f"train: {len(train)}, validation: {len(validation)}, test: {len(test)}")
    
    # Get models
    embedding_model, classifier = get_models()
    
    return embedding_model, classifier, train, test, validation, CATEGORIES

def test_loop(dataloader, model, loss_fn):
    # Set the model to evaluation mode - important for batch normalization and dropout layers
    # Unnecessary in this situation but added for best practices
    model.eval()
    size = len(dataloader.dataset)
    num_batches = len(dataloader)
    test_loss, correct = 0, 0

    # Evaluating the model with torch.no_grad() ensures that no gradients are computed during test mode
    # also serves to reduce unnecessary gradient computations and memory usage for tensors with requires_grad=True
    with torch.no_grad():
        for batch in dataloader:
            pred = model(batch)['logits']
            test_loss += loss_fn(pred, batch['label']).item()
            correct += (pred.argmax(1) == batch['label']).type(torch.float).sum().item()

    avg_loss = test_loss / num_batches
    accuracy = correct / size

    return avg_loss, accuracy

def train_loop(dataloader, model, loss_fn, optimizer, batch_size = 64, epochs = 10):
    size = len(dataloader.dataset)
    total_loss = 0
    batch_losses = []

    # Set models to training mode
    model.train()

    for iteration, batch in enumerate(dataloader):
        # --- 1. Zero Gradients ---
        # Only zero gradients for the parameters you want to update (the classifier head)
        optimizer.zero_grad()
        
        # --- 3. Forward Pass: Embeddings -> Logits ---
        # The classifier head takes the embeddings from the body
        pred = model(batch)['logits']
        
        # --- 4. Calculate Loss ---
        loss = loss_fn(pred, batch['label'])

        # --- 5. Backward Pass & Update ---
        loss.backward()
        optimizer.step()

        cur_loss = loss.item()
        batch_losses.append(cur_loss)
        total_loss += cur_loss

        if iteration % 100 == 0:
            current = iteration * batch_size + len(batch['label'])
            print(f"loss: {cur_loss:>7f}  [{current:>5d}/{size:>5d}]")
        
    return total_loss, batch_losses

def generate_model_card(save_dir: str, accuracy: float, loss: float, epoch: int):
    with open("classifier/modelcard_template.md", "r") as f:
        template_content = f.read()
    
    template = Template(template_content)
    
    card_content = template.render(
        model_id=CLASSIFIER_NAME,
        model_summary="A simple medical query triage classifier.",
        model_description="This model classifies queries into 'medical' or 'insurance' categories. It uses EmbeddingGemma-300M as a backbone.",
        developers="David Gray",
        model_type="Text Classification",
        language="en",
        license="mit",
        base_model="sentence-transformers/embeddinggemma-300m-medical",
        repo=f"https://huggingface.co/{CLASSIFIER_NAME}",
        results_summary=f"Epoch: {epoch+1}\nValidation Accuracy: {accuracy*100:.2f}%\nValidation Loss: {loss:.4f}",
        training_data="Miriad (medical) and InsuranceQA (insurance) datasets.",
        testing_metrics="Accuracy, Loss",
        results=f"Accuracy: {accuracy:.4f}, Loss: {loss:.4f}"
    )
    
    with open(f"{save_dir}/README.md", "w") as f:
        f.write(card_content)

def push_model_card(save_dir: str, repo_id: str, token: str = None):
    api = HfApi(token=token)
    api.upload_file(
        path_or_fileobj=f"{save_dir}/README.md",
        path_in_repo="README.md",
        repo_id=repo_id,
        repo_type="model"
    )

def label_to_int(embedding_model, label_names: list):
    """Creates a dictionary mapping label strings to their integer IDs."""
    label_map = {name: i for i, name in enumerate(label_names)}
    
    def collate_fn(batch):
        # 1. Extract texts and labels from the batch (list of dictionaries)
        texts = [item['text'] for item in batch]
        labels = [item['label'] for item in batch]
        
        # 2. Tokenize the texts using the embedding model's tokenizer
        # The tokenizer is attached to the embedding_model
        with torch.no_grad():
            tokenized_text = embedding_model.encode(
                texts,
                convert_to_tensor=True,
                device=DEVICE
            ).clone().detach()
        
        # 3. Convert string labels to integers
        int_labels = [label_map[l] for l in labels]
        tokenized_labels = torch.tensor(int_labels, dtype=torch.long)
        
        # 4. Add the labels as a PyTorch tensor
        tokenized_batch = {'sentence_embedding': tokenized_text.to(DEVICE), 'label': tokenized_labels.to(DEVICE)}
        
        return tokenized_batch
    
    return collate_fn

def train(push_to_hub: bool = False):
    start_datetime = datetime.now()

    save_dir = f'{CHECKPOINT_PATH}/{start_datetime.strftime(DATETIME_FORMAT)}'
    os.makedirs(save_dir, exist_ok=True)

    embedding_model, model, train_ds, test_ds, validation_ds, labels = get_model_train_test()
    batch_size = 64
    custom_collate_fn = label_to_int(embedding_model, labels)

    train_dataloader = DataLoader(
        train_ds, 
        batch_size=batch_size, 
        shuffle=True,
        collate_fn=custom_collate_fn
    )
    test_dataloader = DataLoader(
        test_ds, 
        batch_size=batch_size, 
        shuffle=True,
        collate_fn=custom_collate_fn
    )
    validation_dataloader = DataLoader(
        validation_ds, 
        batch_size=batch_size, 
        shuffle=True,
        collate_fn=custom_collate_fn
    )

    loss_fn = model.get_loss_fn()
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5)
    save_per_epoch = 1
    epochs = 1
    patience = 1
    min_val_loss = float('inf')
    patience_counter = 0
    history = {
        'train_loss_epoch': [],
        'train_loss_batch': [],
        'validation_accuracy': [],
        'validation_loss_epoch': [],
        'test_accuracy': [],
        'test_loss': []
    }

    for epoch in range(epochs):
        print(f"Epoch {epoch+1}:\n-------------------------------")

        # Train
        total_loss, batch_losses = train_loop(train_dataloader, model, loss_fn, optimizer)
        avg_epoch_loss = total_loss / len(train_dataloader)
        history['train_loss_epoch'].append(avg_epoch_loss)
        history['train_loss_batch'].extend(batch_losses)

        summary = f"Epoch {epoch+1}:"

        # Validate
        val_loss_avg, val_accuracy = test_loop(validation_dataloader, model, loss_fn)
        history['validation_accuracy'].append(val_accuracy)
        history['validation_loss_epoch'].append(val_loss_avg)

        summary += f" - loss: {avg_epoch_loss}\n"
        summary += f" - training loss: {avg_epoch_loss}\n"
        summary += f" - validation loss: {val_loss_avg:>8f}\n"
        summary += f" - validation accuracy: {(100*val_accuracy):>0.1f}%\n"

        # Save checkpoint
        if epoch % save_per_epoch == 0:
            # Save model
            model.save_pretrained(save_dir)
            
            # Generate and push model card
            # generate_model_card(save_dir, val_accuracy, val_loss_avg, epoch)
            # push_model_card(save_dir, CLASSIFIER_NAME, token=HF_TOKEN)
            
            summary += f" -- {save_dir}\n"

            history_df = pd.DataFrame.from_dict(history, orient='index').transpose()
            history_df.to_csv(f"{save_dir}/history.csv", index=False)

            # Push model to Hugging Face
            if push_to_hub:
                model.push_to_hub(CLASSIFIER_NAME, token=HF_TOKEN)
        else:
            summary += "\n"

        print(summary)

        if val_loss_avg < min_val_loss:
            min_val_loss = val_loss_avg
            patience_counter = 0
        else:
            patience_counter += 1
            if patience_counter >= patience:
                print("Early stopping triggered due to no improvement in validation loss.")
                break

    # Evaluate on test dataset
    test_loss_avg, test_accuracy = test_loop(test_dataloader, model, loss_fn)
    history['test_accuracy'].append(test_accuracy)
    history['test_loss'].append(test_loss_avg)
    print(f"Test: Accuracy: {(100*test_accuracy):>0.1f}%, Avg loss: {test_loss_avg:>8f}")

    # Save the final model
    model.save_pretrained(save_dir)
    
    # generate_model_card(save_dir, test_accuracy, test_loss_avg, epochs-1)
    # push_model_card(save_dir, CLASSIFIER_NAME, token=HF_TOKEN)

    # Save loss history
    history_df = pd.DataFrame.from_dict(history, orient='index').transpose()
    history_df.to_csv(f"{save_dir}/history.csv", index=False)

    # Plot training loss per batch
    fig, ax = plt.subplots()
    ax.plot(history['train_loss_batch'])
    ax.set_title('Training Loss per Batch')
    ax.set_xlabel('Batch')
    ax.set_ylabel('Loss')
    fig.savefig(f"{save_dir}/loss.png")

    if push_to_hub:
        model.push_to_hub(CLASSIFIER_NAME, token=HF_TOKEN)

if __name__ == "__main__":
    ap = argparse.ArgumentParser(
        description="Train a classifier for triaging health queries"
    )
    ap.add_argument(
        "--push", action="store_true",
        help="Push model to Hugging Face"
    )
    args = ap.parse_args()

    train(push_to_hub=args.push)