admesh/agentic-intent-classifier

1from __future__ import annotations2 3import json4import sys5from pathlib import Path6 7import numpy as np8import torch9from datasets import Dataset10from sklearn.metrics import accuracy_score, f1_score11 12BASE_DIR = Path(__file__).resolve().parent.parent13if str(BASE_DIR) not in sys.path:14    sys.path.insert(0, str(BASE_DIR))15 16 17def load_labeled_rows(path: Path, label_field: str, label2id: dict[str, int]) -> list[dict]:18    rows = []19    with path.open("r", encoding="utf-8") as handle:20        for line in handle:21            item = json.loads(line)22            rows.append({"text": item["text"], "label": label2id[item[label_field]]})23    return rows24 25 26def load_labeled_rows_from_paths(paths: list[Path], label_field: str, label2id: dict[str, int]) -> list[dict]:27    rows = []28    for path in paths:29        if not path.exists():30            continue31        rows.extend(load_labeled_rows(path, label_field, label2id))32    return rows33 34 35def prepare_dataset(rows: list[dict], tokenizer, max_length: int) -> Dataset:36    dataset = Dataset.from_list(rows)37 38    def tokenize(batch):39        return tokenizer(batch["text"], truncation=True, padding="max_length", max_length=max_length)40 41    dataset = dataset.map(tokenize, batched=True)42    dataset = dataset.remove_columns(["text"])43    dataset.set_format("torch")44    return dataset45 46 47def build_balanced_class_weights(rows: list[dict], num_labels: int) -> torch.Tensor:48    counts = np.zeros(num_labels, dtype=np.float32)49    for row in rows:50        counts[row["label"]] += 1.051 52    nonzero = counts > 053    if not np.any(nonzero):54        return torch.ones(num_labels, dtype=torch.float32)55 56    total = float(counts.sum())57    active_labels = float(np.count_nonzero(nonzero))58    weights = np.ones(num_labels, dtype=np.float32)59    weights[nonzero] = total / (active_labels * counts[nonzero])60    return torch.tensor(weights, dtype=torch.float32)61 62 63def build_label_weight_tensor(labels: tuple[str, ...], weight_map: dict[str, float]) -> torch.Tensor:64    return torch.tensor(65        [float(weight_map.get(label, 1.0)) for label in labels],66        dtype=torch.float32,67    )68 69 70def compute_classification_metrics(eval_pred):71    logits, labels = eval_pred72    preds = np.argmax(logits, axis=-1)73    return {74        "accuracy": accuracy_score(labels, preds),75        "macro_f1": f1_score(labels, preds, average="macro"),76    }77 78 79def write_json(path: Path, payload: dict) -> None:80    path.parent.mkdir(parents=True, exist_ok=True)81    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")82