classifier.py

"""Cœur de classification zero-shot multilingue.

Partagé par l'app Gradio (app.py) et l'évaluation (evaluate.py) pour garantir
que le Space déployé et la mesure de précision utilisent exactement la même logique.
"""

from __future__ import annotations

from functools import lru_cache

MODEL_ID = "MoritzLaurer/mDeBERTa-v3-base-mnli-xnli"

# 7 thèmes canoniques — source de vérité (cf. veille-analytics/scripts/reclassify.js
# et veille-dashboard/shared/utils/themes.ts).
THEMES = [
    "IA/ML",
    "DevOps/Infrastructure",
    "Architecture",
    "Sécurité",
    "Développement",
    "Pratiques/Qualité",
    "Productivité/Outils",
]

# Libellés canoniques -> hypothèses descriptives FR. Le NLI zero-shot est nettement
# plus précis avec des phrases explicites qu'avec des labels bruts (« IA/ML »).
LABEL_MAP = {
    "IA/ML": "intelligence artificielle et machine learning",
    "DevOps/Infrastructure": "DevOps, cloud et infrastructure",
    "Architecture": "architecture logicielle et conception de systèmes",
    "Sécurité": "sécurité informatique et cybersécurité",
    "Développement": "développement logiciel et programmation",
    "Pratiques/Qualité": "pratiques d'ingénierie, tests et qualité logicielle",
    "Productivité/Outils": "productivité et outils pour développeurs",
}

_LABEL_INVERSE = {v: k for k, v in LABEL_MAP.items()}
CANDIDATE_LABELS = list(LABEL_MAP.values())
HYPOTHESIS_TEMPLATE = "Cet article parle de {}."


@lru_cache(maxsize=1)
def get_pipeline():
    """Charge le pipeline zero-shot une seule fois (coûteux : ~280M params)."""
    from transformers import pipeline

    return pipeline("zero-shot-classification", model=MODEL_ID)


def classify(titre: str, resume: str = "") -> dict[str, float]:
    """Classe un article dans les 7 thèmes. Retourne {theme: score} trié desc.

    multi_label=True -> un score sigmoïde indépendant par thème (les scores ne
    somment pas à 1), adapté au multi-étiquetage : un article peut relever de
    plusieurs thèmes.
    """
    text = f"{titre}. {resume}".strip()
    out = get_pipeline()(
        text,
        candidate_labels=CANDIDATE_LABELS,
        hypothesis_template=HYPOTHESIS_TEMPLATE,
        multi_label=True,
    )
    scores = {
        _LABEL_INVERSE[label]: float(score)
        for label, score in zip(out["labels"], out["scores"], strict=True)
    }
    return dict(sorted(scores.items(), key=lambda kv: kv[1], reverse=True))