batchCooking/services/tech-step-intent-service/intent_service/locale_pipeline.py

"""Pipeline spaCy pour UNE locale — l'équivalent Python de ce que
`node-nlp`'s `NlpManager` faisait pour cette locale dans
`TechStepClassifierService` (`apps/api/src/lib/recipe-matching/tech-step-matcher.ts`) :
NER par entités enum (ici un `PhraseMatcher`) + classification d'intention
(ici un `textcat`), les deux entraînés à partir du corpus possédé par ce
service lui-même (`training_data.TECH_STEP_TRAINING_DATA` — plus poussé par
`apps/api` via HTTP, voir `pipeline_registry.py`).

Le modèle de base spaCy (tokenizer + vecteurs + le composant
`diacritics_normalizer` défini plus bas) est chargé une seule fois
(`preload()`, appelé au démarrage du process — voir `main.py` — pas
paresseusement au premier `train()`, pour que `GET /health` ne devienne
`200` qu'une fois ce coût payé) puis réutilisé à chaque `train()` : seul le
`textcat` (retiré puis rajouté à neuf) et le `PhraseMatcher` (remplacé) sont
reconstruits à chaque appel, jamais le tokenizer/les vecteurs. Rien n'est
jamais persisté sur disque — `training_data.py` reste l'unique source de
vérité, reconstruite en mémoire depuis zéro à chaque démarrage du process.
"""

from __future__ import annotations

import logging
import random
from dataclasses import dataclass, field

import spacy
from spacy.language import Language
from spacy.matcher import PhraseMatcher
from spacy.tokens import Doc, Span
from spacy.training import Example
from spacy.util import filter_spans, fix_random_seed, minibatch

from .text_normalization import normalize_text

logger = logging.getLogger(__name__)

# Modèle spaCy de base par locale — voir pyproject.toml pour la version
# pinnée exacte. `md` (pas `sm`) : conserve les vecteurs de mots, inutilisés
# par le pipeline v1 (textcat bag-of-words) mais retenus pour l'ambition
# future de similarité sémantique (voir le README de ce service).
SUPPORTED_LOCALES = {
    "fr": "fr_core_news_md",
    "en": "en_core_web_md",
}

# Composants du modèle de base non utilisés par ce pipeline (on ne s'appuie
# ni sur le NER générique de spaCy, ni sur l'analyse syntaxique/morphologique
# — seuls le tokenizer et les vecteurs de mots restent nécessaires) : les
# exclure au chargement évite le coût mémoire/CPU de composants qui ne
# tourneraient jamais.
_EXCLUDED_COMPONENTS = ["parser", "ner", "tagger", "morphologizer", "attribute_ruler", "lemmatizer"]

_TEXTCAT_PIPE_NAME = "textcat"

# Nombre d'itérations d'entraînement du textcat et taille de minibatch —
# calibrés empiriquement contre le corpus réel (`training_data.py`), pas
# seulement contre les petits corpus jouets des tests de ce fichier. Trop
# peu d'itérations laisse des clauses correctement classifiées (bon argmax)
# mais avec une confiance dérisoire — bien en dessous de tout seuil
# raisonnable pour `CONFIDENCE_THRESHOLD` (`tech-step-matcher.ts`).
#
# Trois passes de calibration successives, toutes mesurées contre le
# corpus réel (74 techniques) :
# 1. `150` itérations (calibré pour le corpus original, ~26 techniques) ne
#    passe plus à l'échelle une fois élargi : `150` sur 74 classes
#    dépassait 17 minutes pour une seule locale, constaté en CI.
# 2. `40` itérations, `examples` limité aux `utterances` (pas les
#    `synonyms`) : ~200s/locale, mais confiance faible sur les clauses
#    ancrées sans paraphrase entraînée (`simmer`/`cook`/`bake` ~0.25-0.34).
# 3. **Configuration actuelle** : les `synonyms` de chaque technique sont
#    désormais aussi des exemples d'entraînement du textcat (voir plus bas
#    dans `train()`) — un signal "mot-clé isolé -> sa propre technique"
#    qui manquait complètement avant. À `_TRAINING_ITERATIONS` inchangé
#    (40), le nombre d'exemples par époque grimpe de ~286 à ~749 et le
#    temps d'entraînement suit (~535s/locale) ; réduire à `25` retrouve un
#    temps proche de l'étape 2 (~336s/locale, ~670s pour fr+en combinés)
#    tout en gardant l'essentiel du gain de confiance apporté par les
#    synonymes : melt ~0.89, preheat ~0.77, compote ~0.78, julienne ~0.76,
#    zest ~0.66, bake ~0.62, cook ~0.38, simmer ~0.31 — le plus faible
#    observé, mais désormais nettement au-dessus du seuil de confiance
#    (contre ~0.25, sous le seuil d'alors, à l'étape 2). Bruit
#    hors-vocabulaire toujours négligeable (anglais via le classifieur
#    français : `~0.02`). Une vraie repasse de
#    `calibrate-tech-step-threshold.ts` contre `TECH_STEP_EVAL_DATASET`
#    reste nécessaire pour confirmer/affiner ces valeurs (voir
#    `CONFIDENCE_THRESHOLD`'s propre commentaire, `tech-step-matcher.ts`)
#    — ce qui précède est une mesure manuelle ponctuelle, pas un
#    remplacement de cette calibration.
_TRAINING_ITERATIONS = 25
_TRAINING_BATCH_SIZE = 16
# Arrêt anticipé : `_TRAINING_ITERATIONS` reste le plafond (le pire cas ne
# change pas), un corpus/locale qui converge plus vite n'a pas à payer les
# itérations restantes pour rien. Une époque compte comme "sans progrès"
# quand sa perte totale ne descend pas d'au moins `_EARLY_STOPPING_MIN_DELTA`
# sous la meilleure perte vue jusqu'ici ; `_EARLY_STOPPING_PATIENCE` époques
# consécutives sans progrès arrêtent l'entraînement.
#
# Mesuré contre le corpus réel (74 techniques, budget de 40 itérations,
# avant le passage à 25) : ne s'est jamais déclenché — la perte continuait
# de baisser significativement sur toute la plage (cohérent avec la
# confiance qui grimpait encore nettement entre 15 et 40 itérations, voir
# le commentaire de `_TRAINING_ITERATIONS`). Ce n'est donc pas un gain de
# temps aujourd'hui, mais un filet de sécurité peu coûteux pour la suite : si
# `_TRAINING_ITERATIONS` est un jour augmenté pour une meilleure confiance,
# ceci évite de payer des itérations supplémentaires une fois la
# convergence réellement atteinte, sans qu'il faille retrouver le bon
# plafond à la main à chaque changement du corpus.
_EARLY_STOPPING_PATIENCE = 3
_EARLY_STOPPING_MIN_DELTA = 0.001
# Abaissé de `0.2` avec le reste de cette recalibration — `0.1` régularise
# encore contre la petite taille du corpus par technique tout en laissant
# plus de signal passer à chaque pas, ce qui a mesurablement aidé la
# confiance finale sans signe de sur-ajustement (le bruit hors-vocabulaire
# reste aussi bas qu'avant, voir ci-dessus).
_TRAINING_DROPOUT = 0.1
# Seed fixe — un warm-up reproductible d'un redémarrage à l'autre (même
# corpus en entrée) est préférable à un score qui varie légèrement à chaque
# déploiement pour la même donnée, en particulier pendant la calibration du
# seuil de confiance côté apps/api.
_TRAINING_SEED = 0


class _DiacriticsNormalizer:
    """Composant de pipeline réécrivant `token.norm_` avec `normalize_text()`
    (le port Python de `normalizeText()` côté `apps/api`) pour chaque token.

    Point clé : ce composant tourne aussi bien sur les `Doc` construits pour
    les *patterns* du `PhraseMatcher` (voir `LocalePipeline.train`) que sur
    le *texte cible* passé à `process()` — les deux passent donc par
    exactement la même normalisation, ce qui garantit qu'un synonyme comme
    "mijoter" matche indifféremment "MIJOTER"/"mijoté"/"Mijotée" dans le
    texte, reproduisant le comportement `ner.threshold: 1` (exact après
    normalisation, sans tolérance floue Levenshtein) de l'ancien `NlpManager`.
    Indépendant des `entries` entraînées — ajouté une seule fois par
    `preload()`, jamais retiré/rajouté par `train()`.

    Opère token par token, sur du texte déjà tokenisé — `normalize_text()`
    ne fait que réécrire la forme d'un token existant (minuscule, sans
    diacritique), jamais fusionner/scinder des tokens : les patterns
    (`nlp.make_doc(synonym)` + ce composant appliqué à la main, voir
    `LocalePipeline.train`) et le texte cible (`nlp(text)`, pipeline
    complet) passent donc toujours par le *même* découpage en tokens que
    le tokenizer du modèle de base leur donne, avant que ce composant n'y
    touche — pas de risque de désalignement entre les deux.
    """

    def __call__(self, doc: Doc) -> Doc:
        for token in doc:
            token.norm_ = normalize_text(token.text)
        return doc


@Language.factory("diacritics_normalizer")
def _create_diacritics_normalizer(nlp: Language, name: str) -> _DiacriticsNormalizer:
    return _DiacriticsNormalizer()


@dataclass
class TrainEntry:
    """Une technique à entraîner pour une locale — construit par
    `PipelineRegistry.initialize()` depuis `training_data.entries_for_locale`."""

    uid: str
    synonyms: list[str] = field(default_factory=list)
    utterances: list[str] = field(default_factory=list)


@dataclass
class Entity:
    """Une mention candidate trouvée par le `PhraseMatcher` — offsets
    caractère `[start, end)`, miroir de `EntityPayload` (`schemas.py`)."""

    uid: str
    start: int
    end: int


@dataclass
class ProcessResult:
    """Résultat complet d'un `process()` — miroir de `ProcessResponse`
    (`schemas.py`)."""

    entities: list[Entity]
    intent: str | None
    score: float


class UnsupportedLocaleError(ValueError):
    """`locale` ne correspond à aucun modèle spaCy connu (voir
    `SUPPORTED_LOCALES`) — distinct d'une locale simplement "pas encore
    entraînée" (`LocalePipeline.is_trained is False`), qui n'est pas une
    erreur (voir `process()`)."""


class LocalePipeline:
    """Pipeline spaCy (NER par phrases + textcat) pour une locale donnée.
    Un `PipelineRegistry` (voir `pipeline_registry.py`) en détient une
    instance par locale supportée.
    """

    def __init__(self, locale: str) -> None:
        if locale not in SUPPORTED_LOCALES:
            raise UnsupportedLocaleError(f"Unsupported locale: {locale!r}")
        self._locale = locale
        self._model_name = SUPPORTED_LOCALES[locale]
        # `None` tant que `preload()` n'a pas tourné.
        self._base_nlp: Language | None = None
        # `None` tant qu'aucun `train()` n'a réussi — `process()` traite ça
        # comme "rien à trouver" plutôt qu'une erreur, exactement le
        # comportement testé côté `apps/api` pour "une locale jamais
        # entraînée".
        self._matcher: PhraseMatcher | None = None
        self._trained = False

    @property
    def is_trained(self) -> bool:
        return self._trained

    def preload(self) -> None:
        """Charge le modèle spaCy de base (tokenizer + vecteurs) et le
        composant `diacritics_normalizer` — idempotent, sans effet si déjà
        chargé. Appelé au démarrage du process pour les deux locales
        connues (voir `main.py`), pas paresseusement au premier `train()`.
        """
        if self._base_nlp is not None:
            return
        nlp = spacy.load(self._model_name, exclude=_EXCLUDED_COMPONENTS)
        nlp.add_pipe("diacritics_normalizer", first=True)
        self._base_nlp = nlp

    def train(self, entries: list[TrainEntry]) -> tuple[int, int, int]:
        """Reconstruit le `textcat` et le `PhraseMatcher` de ce pipeline à
        partir de `entries` (le tokenizer/les vecteurs restent ceux chargés
        par `preload()`). Retourne `(label_count, example_count,
        synonym_count)` pour la journalisation (`pipeline_registry.py`) —
        `example_count` est le nombre réel d'exemples donnés au `textcat`
        (`utterances` *et* `synonyms` combinés, voir plus bas), pas
        seulement `entry.utterances`.

        `entries` vide retombe à `is_trained == False` plutôt que de lever —
        un appelant qui n'a rien à entraîner pour cette locale obtient le
        même comportement que "jamais entraîné", pas une erreur 500.
        """
        self.preload()
        assert self._base_nlp is not None  # garanti par preload() ci-dessus

        if _TEXTCAT_PIPE_NAME in self._base_nlp.pipe_names:
            self._base_nlp.remove_pipe(_TEXTCAT_PIPE_NAME)

        if not entries:
            self._matcher = None
            self._trained = False
            return (0, 0, 0)

        nlp = self._base_nlp
        # `nlp.make_doc()` ne fait tourner *que* le tokenizer, pas les
        # composants du pipeline — le `diacritics_normalizer` ajouté par
        # `preload()` ne tournerait donc jamais sur les `Doc` de patterns
        # s'ils n'étaient construits qu'avec `make_doc()`, alors que
        # `process()` appelle `nlp(text)` (le pipeline complet) sur le texte
        # cible. Sans ce correctif, un synonyme accentué comme "préchauffer"
        # n'aurait jamais matché "PRÉCHAUFFER"/"Préchauffer" : trouvé en
        # calibrant contre les cas exacts de `tech-step-matcher.test.ts`
        # (fr, la locale la plus concernée par les accents) — un synonyme
        # sans diacritique comme "faire fondre" masquait le bug en semblant
        # fonctionner par coïncidence. Appliquer explicitement le même
        # composant aux deux côtés garantit qu'ils passent par la même
        # normalisation.
        diacritics_normalizer = nlp.get_pipe("diacritics_normalizer")

        matcher = PhraseMatcher(nlp.vocab, attr="NORM")
        synonym_count = 0
        for entry in entries:
            if not entry.synonyms:
                continue
            patterns = [diacritics_normalizer(nlp.make_doc(synonym)) for synonym in entry.synonyms]
            matcher.add(entry.uid, patterns)
            synonym_count += len(entry.synonyms)

        # `textcat` (exclusive_classes) exige au moins deux labels (voir
        # spaCy's error E867) — jamais un problème avec le vrai corpus
        # (`TECH_STEP_TRAINING_DATA` a ~74 techniques), mais un `entries` à
        # un seul élément resterait structurellement valide pour le NER
        # seul : ne pas planter, juste ne pas construire de textcat du tout
        # (`process()` retombe alors sur `intent: null` via son garde
        # `if not cats`, exactement comme "rien à classifier"). Journalisé
        # explicitement — sans ça, "pourquoi cette locale ne classifie
        # jamais rien" ne serait visible qu'en déduisant `labelCount < 2`
        # de la ligne "tech-step NLP pipeline trained" (`pipeline_registry.py`).
        examples: list[Example] = []
        if len(entries) < 2:
            logger.warning(
                "tech-step NLP textcat skipped: fewer than 2 labels, intent classification disabled for this locale",
                extra={"locale": self._locale, "labelCount": len(entries)},
            )
        else:
            textcat = nlp.add_pipe(
                _TEXTCAT_PIPE_NAME,
                config={
                    "model": {
                        "@architectures": "spacy.TextCatBOW.v3",
                        "exclusive_classes": True,
                        "ngram_size": 1,
                        "no_output_layer": False,
                    },
                },
            )
            for entry in entries:
                textcat.add_label(entry.uid)

            for entry in entries:
                cats = {other.uid: 0.0 for other in entries}
                cats[entry.uid] = 1.0
                # `synonyms` (déjà utilisés pour le `PhraseMatcher` ci-dessus)
                # sont aussi de bonnes phrases d'entraînement pour le
                # `textcat` — un texte réduit au mot-clé lui-même ("fondre",
                # "faire fondre") est le cas le plus net qui soit pour sa
                # propre technique, et n'était auparavant vu par le textcat
                # que noyé dans le contexte plus riche des `utterances`.
                for text in (*entry.synonyms, *entry.utterances):
                    doc = nlp.make_doc(text)
                    examples.append(Example.from_dict(doc, {"cats": cats}))

            # Graine le RNG Python *et* celui de numpy/thinc sous-jacent à
            # `nlp.update()` (initialisation des poids, masque de dropout) —
            # `random.Random(_TRAINING_SEED)` ci-dessous ne couvre que l'ordre
            # de mélange des exemples choisi par ce module, pas ce que spaCy
            # fait en interne à chaque pas de gradient.
            fix_random_seed(_TRAINING_SEED)
            rng = random.Random(_TRAINING_SEED)
            if examples:
                optimizer = nlp.initialize(lambda: examples)
                best_loss = float("inf")
                epochs_without_improvement = 0
                for iteration in range(_TRAINING_ITERATIONS):
                    rng.shuffle(examples)
                    losses: dict[str, float] = {}
                    for batch in minibatch(examples, size=_TRAINING_BATCH_SIZE):
                        nlp.update(batch, sgd=optimizer, drop=_TRAINING_DROPOUT, losses=losses)
                    epoch_loss = losses.get(_TEXTCAT_PIPE_NAME, 0.0)
                    # Arrêt anticipé — voir `_EARLY_STOPPING_PATIENCE`'s propre
                    # commentaire. `_TRAINING_ITERATIONS` reste le plafond
                    # (pire cas inchangé), ceci ne fait que raccourcir les
                    # cas qui convergent plus vite.
                    if epoch_loss < best_loss - _EARLY_STOPPING_MIN_DELTA:
                        best_loss = epoch_loss
                        epochs_without_improvement = 0
                    else:
                        epochs_without_improvement += 1
                        if epochs_without_improvement >= _EARLY_STOPPING_PATIENCE:
                            logger.info(
                                "tech-step NLP textcat training stopped early",
                                extra={
                                    "locale": self._locale,
                                    "iteration": iteration + 1,
                                    "maxIterations": _TRAINING_ITERATIONS,
                                    "finalLoss": epoch_loss,
                                },
                            )
                            break
            else:
                # Des `entries` avec des `uid` mais aucune `utterance` nulle
                # part (corpus incomplet) : le textcat a des labels mais rien
                # pour apprendre à les distinguer — toujours initialisé pour
                # rester un pipeline valide ; `process()` renverra alors un
                # score ~uniforme entre labels. Ce n'est pas ce module qui doit
                # juger la qualité du corpus reçu (voir `tech-step-eval-runner.ts`
                # côté apps/api pour ce rôle).
                nlp.initialize()

        self._matcher = matcher
        self._trained = True
        return (len(entries), len(examples), synonym_count)

    def process(self, text: str) -> ProcessResult:
        """Reproduit la forme de `NlpManager.process(locale, text)` : les
        entités candidates (NER) et le verdict du classifieur d'intention
        sur `text` tel quel — que ce soit la description complète ou une
        clause déjà découpée côté `apps/api`, ce module ne le sait pas et ne
        s'en soucie pas, exactement comme l'ancien `NlpManager`.

        `intent` vaut `None` dans deux cas distincts, tous deux silencieux
        côté retour (voir le log d'avertissement de `train()` pour repérer
        le second en amont) : `text` vide/blanc, ou `doc.cats` vide parce
        que `train()` a reçu moins de deux labels pour cette locale (le
        textcat n'a alors jamais été construit — voir son propre
        commentaire).
        """
        if not self._trained or self._base_nlp is None or self._matcher is None or not text.strip():
            return ProcessResult(entities=[], intent=None, score=0.0)

        doc = self._base_nlp(text)

        # A technique's own synonym list can legitimately contain one phrase
        # nested inside another (`melt`'s "fondre" is a literal substring of
        # its own "faire fondre") — the `PhraseMatcher` reports *both* as
        # separate matches at overlapping positions, which without
        # resolution would hand `splitIntoClauses` (apps/api) two candidates
        # for what a human reads as one mention, producing the same
        # techStepId twice in the final result. `filter_spans` keeps only
        # the longest match at each position (so "faire fondre" wins over
        # the "fondre" it contains) — found by a real regression in
        # `tech-step-matcher.test.ts`'s "detects several distinct
        # techniques..." case once this service replaced node-nlp (which
        # apparently resolved this internally; nothing here recreates that
        # by choice, `filter_spans` is spaCy's own documented tool for
        # exactly this "one span per position" problem, e.g. as used for
        # NER-style outputs).
        matched_spans = [
            Span(doc, start, end, label=match_id) for match_id, start, end in self._matcher(doc)
        ]
        entities = sorted(
            (
                Entity(uid=self._base_nlp.vocab.strings[span.label], start=span.start_char, end=span.end_char)
                for span in filter_spans(matched_spans)
            ),
            key=lambda entity: entity.start,
        )

        cats = doc.cats
        if not cats:
            return ProcessResult(entities=entities, intent=None, score=0.0)
        intent = max(cats, key=cats.get)
        return ProcessResult(entities=entities, intent=intent, score=cats[intent])