Spaces:

rahim237cm
/

jurix-api

Runtime error

App Files Files Community

jurix-api / app /services /intent_classifier.py

rahim237cm

Claude Opus 5.5

fix(intention): budget de sortie suffisant, nominations reconnues comme du droit

c84fadf about 1 month ago

Raw History Blame Contribute Delete

16 kB

"""
Aiguillage d'un message utilisateur : question de droit, ou conversation ?
POURQUOI CE MODULE EXISTE. `RAGService.ask()` envoyait TOUTE question a la
recherche hybride, sans exception. Une salutation remonte quand meme des
chunks — le plein texte et le vectoriel trouvent toujours quelque chose — si
bien que le garde-fou « aucun resultat » ne se declenchait jamais. Mesure sur
une vraie session : « comment vas tu ? » a produit un pave a rubriques
(« **Source legale** L'information demandee n'est pas presente dans les
documents fournis, notamment la Loi N°2023/014 portant Code Minier ») et un
bloc « Sources » citant un article du Code Minier. Le cout n'etait pas
qu'esthetique : chaque salutation payait un appel d'embedding Gemini, occupait
`embedding_cache` sept jours et laissait une ligne dans `search_events`.
CE MODULE NE CLASSE PAS DES DOCUMENTS. `legal_domain_classifier.py` range un
texte du corpus dans un domaine juridique a l'ingestion ; ici on lit la
question de l'utilisateur, ce qui n'a ni les memes entrees ni le meme verdict.
LA MECANIQUE EST CELLE DE `reranker.rerank_with_llm` : l'appel est borne par
`asyncio.wait_for`, et AUCUNE exception ne sort. Le repli d'un composant
optionnel, c'est l'absence du composant : « juridique ».
DEUX MODES (INTENT_CLASSIFIER). « groq », le defaut : consignes courtes,
schema strict, verdicts en cache, et « juridique » des que Groq ne peut pas
repondre dans les temps (quota, attente, panne). « llm » : le modele du chat,
passe en parametre. Le classement local par embeddings a ete retire.
Author: JuriX Team
"""

import asyncio
import json
import logging
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Any, List, Optional, Tuple

from app.core.config import settings
from app.services.prompts import (
    CLASSIFICATION_SYSTEM,
    CONSIGNES_INTENTION,
    ECHANGES_POUR_L_INTENTION,
    LONGUEUR_ECHANGE_INTENTION,
    construire_prompt_de_classification,
    format_conversation_history,
    message_d_intention,
)
from app.services.reranker import _article_number_in
from app.services.text_features import fold_accents

logger = logging.getLogger(__name__)


# Ordre explicite, jamais l'ordre d'insertion d'un dictionnaire.
INTENTS: Tuple[str, ...] = ("juridique", "smalltalk", "meta", "hors_sujet")

# Le seul qui emprunte le RAG. Les trois autres partagent le chemin
# conversationnel, avec un prompt different chacun.
INTENT_JURIDIQUE = INTENTS[0]

# Messages dont la TOTALITE est une formule de politesse. Le rapprochement se
# fait sur le message ENTIER normalise, jamais par sous-chaine : « bonjour »
# court-circuite, « bonjour, puis-je divorcer sans avocat ? » non — c'est
# precisement le piege qu'un filtre lexical naif tend, et il classerait une
# vraie question de droit en salutation.
#
# Le gain n'est pas theorique : sur le palier gratuit, chaque message coute
# desormais deux appels au modele, et « bonjour » ouvre une conversation sur
# deux. Ceux-la n'en coutent plus aucun.
_SALUTATIONS_EXACTES = frozenset({
    "bonjour", "bonsoir", "salut", "coucou", "hello", "hi", "hey",
    "merci", "merci beaucoup", "merci bien", "thanks", "thank you",
    "au revoir", "bye", "a bientot", "bonne journee", "bonne soiree",
    "ok", "d'accord", "tres bien", "parfait",
})

# Ponctuation et accents retires avant comparaison : « Bonjour ! », « bonjour »
# et « BONJOUR. » sont le meme message.
_PONCTUATION = str.maketrans("", "", "!?.,;:…\"'«»()")


def _est_une_salutation_pure(question: str) -> bool:
    """Le message ENTIER est-il une formule de politesse, et rien d'autre ?"""
    from app.services.text_features import fold_accents

    nettoye = fold_accents((question or "").strip().lower()).translate(_PONCTUATION)
    return " ".join(nettoye.split()) in _SALUTATIONS_EXACTES


@dataclass(frozen=True)
class IntentResult:
    """
    Verdict de routage.
    `intent` vaut TOUJOURS l'un de INTENTS, jamais None : l'appelant n'a pas de
    branche « je ne sais pas » a ecrire. `rule` nomme ce qui a decide, ce qui
    rend chaque erreur diagnosticable dans les journaux sans rejouer la
    requete — un `defaut-timeout` massif et un `defaut-json` massif appellent
    des corrections opposees.
    """

    intent: str
    confidence: float
    rule: str


# Schema impose a la sortie du modele. `gemini_service.generate` expose
# `response_schema` explicitement, precisement pour ce genre d'appel.
_INTENT_SCHEMA = {
    "type": "object",
    "properties": {
        "intention": {"type": "string", "enum": list(INTENTS)},
        "confiance": {"type": "number"},
    },
    "required": ["intention"],
}


def _parse_intent(raw: Any) -> Optional[Tuple[str, float]]:
    """
    Lit le verdict du modele. Rend None sur tout ce qui n'est pas exploitable.
    Fonction PURE : elle ne journalise pas et ne leve pas, ce qui la rend
    testable sans doublure.
    """
    if not raw or not isinstance(raw, str):
        return None

    try:
        charge = json.loads(raw)
    except (ValueError, TypeError):
        return None

    return _verdict_de(charge)


def _verdict_de(charge: Any) -> Optional[Tuple[str, float]]:
    """Le verdict d'un objet JSON deja decode, ou None. Pure, comme _parse_intent."""
    if not isinstance(charge, dict):
        return None

    intention = charge.get("intention")
    if intention not in INTENTS:
        return None

    try:
        confiance = float(charge.get("confiance", 0.5))
    except (TypeError, ValueError):
        confiance = 0.5

    return intention, max(0.0, min(1.0, confiance))


# ==================== CLASSEMENT PAR GROQ ====================

# Schema STRICT : la sortie ne peut etre que cet objet, rien d'autre.
_SCHEMA_GROQ = {
    "type": "object",
    "properties": {
        "intention": {"type": "string", "enum": list(INTENTS)},
        "confiance": {"type": "number"},
    },
    "required": ["intention", "confiance"],
    "additionalProperties": False,
}

# Un verdict tient en une quinzaine de jetons, mais le modele en depense
# d'autres avant, invisibles, meme sans reflexion : mesure sur
# qwen3.8-27b, 128 jetons factures pour un verdict de 15. A 40, deux
# messages sur soixante revenaient VIDES (HTTP 400 json_validate_failed),
# donc en « juridique » par defaut. Le quota compte l'usage reel, pas ce
# plafond.
_JETONS_DE_SORTIE = 256

# Attente acceptee devant le limiteur. Au-dela, la question part tout de suite
# en « juridique » : faire patienter l'utilisateur pour savoir s'il dit
# bonjour couterait plus qu'une recherche inutile.
ATTENTE_MAX_INTENTION_S = 0.5


class _CacheDeVerdicts:
    """
    Les derniers verdicts, par message et echanges precedents.
    « bonjour », « merci » ou « que dit le code du travail ? » reviennent
    souvent a l'identique : chacun coutait une requete sur les 1 000 du jour.
    Taille et duree bornees ; seuls les verdicts du modele y entrent, jamais
    un defaut.
    """

    def __init__(self, taille: int = 2000, duree_s: float = 86_400.0, horloge=time.monotonic):
        self.taille = taille
        self.duree_s = duree_s
        self._horloge = horloge
        self._entrees: "OrderedDict[Tuple[str, str], Tuple[float, IntentResult]]" = OrderedDict()

    def lire(self, cle: Tuple[str, str]) -> Optional[IntentResult]:
        entree = self._entrees.get(cle)
        if entree is None:
            return None
        if self._horloge() - entree[0] > self.duree_s:
            del self._entrees[cle]
            return None
        self._entrees.move_to_end(cle)
        return entree[1]

    def ecrire(self, cle: Tuple[str, str], verdict: IntentResult) -> None:
        self._entrees[cle] = (self._horloge(), verdict)
        self._entrees.move_to_end(cle)
        while len(self._entrees) > self.taille:
            self._entrees.popitem(last=False)

    def vider(self) -> None:
        self._entrees.clear()

    def __len__(self) -> int:
        return len(self._entrees)


cache_des_verdicts = _CacheDeVerdicts()


def _normaliser(texte: str) -> str:
    return " ".join(fold_accents((texte or "").lower()).split())


def _echanges(history: Optional[List[Any]]) -> List[Tuple[str, str]]:
    """Les derniers echanges (role, texte tronque), du plus ancien au plus recent."""
    echanges = []
    for message in (history or [])[-ECHANGES_POUR_L_INTENTION:]:
        role = "utilisateur" if getattr(message, "role", "") == "user" else "assistant"
        texte = " ".join((getattr(message, "content", "") or "").split())
        echanges.append((role, texte[:LONGUEUR_ECHANGE_INTENTION]))
    return echanges


async def _classer_par_groq(
    question: str, history: Optional[List[Any]], timeout: Optional[float]
) -> IntentResult:
    """
    Classement par Groq. Ne leve jamais : toute impossibilite rend
    « juridique », avec une regle `defaut-groq-<raison>` qui dit laquelle.
    """
    from app.services.groq_service import (
        GroqLimiteError,
        GroqQuotaError,
        GroqReponseInvalideError,
        get_groq_service,
    )

    if not settings.GROQ_API_KEY:
        return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-groq-cle")

    echanges = _echanges(history)
    cle = (_normaliser(question), _normaliser(" | ".join(t for _, t in echanges)))
    en_cache = cache_des_verdicts.lire(cle)
    if en_cache is not None:
        return IntentResult(en_cache.intent, en_cache.confidence, "groq-cache")

    delai = timeout if timeout is not None else settings.GROQ_INTENTION_TIMEOUT_S
    raison = None
    try:
        reponse = await asyncio.wait_for(
            get_groq_service().completer_json(
                systeme=CONSIGNES_INTENTION,
                message=message_d_intention(question, echanges),
                schema=_SCHEMA_GROQ,
                nom_schema="intention",
                max_jetons=_JETONS_DE_SORTIE,
                attente_max=ATTENTE_MAX_INTENTION_S,
                timeout=delai,
            ),
            timeout=delai,
        )
        verdict = _verdict_de(reponse.donnees)
        if verdict is None:
            raison = "json"
    except asyncio.TimeoutError:
        raison = "timeout"
    except GroqQuotaError:
        raison = "quota"
    except GroqLimiteError:
        raison = "limite"
    except GroqReponseInvalideError:
        raison = "json"
    except Exception as exc:
        # Panne, cle refusee, schema refuse : tout est avale. Une defaillance
        # du routeur ne doit jamais faire echouer la question.
        logger.warning("🧭 Groq en echec (%r)", exc)
        raison = "erreur"

    if raison is not None:
        logger.warning("🧭 Classement d'intention impossible (%s) : repli sur juridique", raison)
        return IntentResult(INTENT_JURIDIQUE, 0.0, f"defaut-groq-{raison}")

    intention, confiance = verdict
    resultat = IntentResult(intention, confiance, "groq")
    cache_des_verdicts.ecrire(cle, resultat)
    return resultat


async def classify_intent(
    question: str,
    *,
    llm: Optional[Any] = None,
    history: Optional[List[Any]] = None,
    timeout: Optional[float] = None,
    max_tokens: Optional[int] = None,
) -> IntentResult:
    """
    Classe un message. Ne leve jamais.
    Args:
        question: le message de l'utilisateur, tel qu'il l'a ecrit
        llm: service de generation du mode « llm ». PARAMETRE et non
            `get_llm_service()` en dur, meme raison que `rerank_with_llm` : un
            test injecte une doublure, le harnais d'evaluation peut changer de
            modele. Le mode « groq » ne s'en sert pas.
        history: les derniers messages, pour les questions de suivi. « et
            l'article 12 ? » n'est juridique que par ce qui precede.
    Returns:
        IntentResult, toujours. Sur echec : `juridique`, ce qui redonne le
        comportement anterieur.
    POURQUOI LE DEFAUT EST « juridique ». L'erreur n'est pas symetrique.
    Classer une salutation en juridique coute un pave disgracieux — le bug
    d'hier, donc une degradation vers l'etat anterieur. Classer « quelles sont
    les conditions du permis de recherche ? » en conversation coute une reponse
    INVENTEE et SANS SOURCE sur une question de droit, et l'utilisateur n'a
    aucun moyen de s'en apercevoir : on lui a justement retire le bloc Sources
    qui l'aurait alerte. S'y ajoute que les pannes sont correlees a la charge,
    pas au contenu : un defaut conversationnel ferait basculer TOUTE la
    plateforme en mode bavardage pendant une saturation du fournisseur.
    """
    if not settings.INTENT_ROUTING_ENABLED:
        return IntentResult(INTENT_JURIDIQUE, 1.0, "desactive")

    # Court-circuit deterministe. « que dit l'article 33 du Code Minier ? » n'a
    # pas besoin d'un modele pour etre reconnue comme juridique, et le
    # re-ranking documente cette forme comme dominante sur ce produit. La regex
    # est celle du re-ranking, pas une seconde : deux auraient diverge.
    #
    # Pas de court-circuit lexical symetrique cote salutation (une liste
    # « bonjour / merci / salut ») : il classerait « Bonjour, puis-je
    # divorcer ? » en smalltalk. L'asymetrie est voulue — on ne court-circuite
    # que dans le sens qui degrade vers le comportement actuel.
    if _article_number_in(question or ""):
        return IntentResult(INTENT_JURIDIQUE, 1.0, "court-circuit-article")

    # Symetrique, et sur une egalite STRICTE du message entier : c'est ce qui
    # le rend sur. Voir _SALUTATIONS_EXACTES.
    if _est_une_salutation_pure(question):
        return IntentResult("smalltalk", 1.0, "court-circuit-salutation")

    if settings.INTENT_CLASSIFIER == "groq":
        return await _classer_par_groq(question, history, timeout)

    # « llm » : le modele du chat, passe en parametre.
    prompt = construire_prompt_de_classification(
        question, format_conversation_history(history or [])
    )

    try:
        reponse = await asyncio.wait_for(
            llm.generate(
                prompt=prompt,
                # Systeme court et neutre, OBLIGATOIRE : `generate` retombe sur
                # `GeminiService.SYSTEM_INSTRUCTION` quand `system` est absent,
                # et cette instruction-la impose de citer des articles et de
                # terminer par « Sources: ». Sur une tache de classification,
                # elle produirait de la prose au lieu du JSON demande.
                system=CLASSIFICATION_SYSTEM,
                temperature=0.0,
                max_tokens=max_tokens or settings.INTENT_MAX_TOKENS,
                reflexion=settings.GEMINI_REFLEXION_CLASSIFICATION,
                response_mime_type="application/json",
                response_schema=_INTENT_SCHEMA,
            ),
            timeout=timeout if timeout is not None else settings.INTENT_TIMEOUT_S,
        )
    except asyncio.TimeoutError:
        logger.warning("🧭 Classification expiree : repli sur juridique")
        return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-timeout")
    except Exception as exc:
        # Quota, saturation, panne reseau : tout est avale. Une defaillance du
        # routeur ne doit jamais faire echouer la question de l'utilisateur.
        logger.warning("🧭 Classification en echec (%s) : repli sur juridique", exc)
        return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-erreur")

    # Extraction TOTALE, jamais `reponse["response"]`. `generate` rend un dict,
    # mais ce module promet de ne jamais lever : une doublure ou une future
    # variante rendant une chaine nue ne doit pas transformer le routeur en
    # panne. Tout ce qui n'est pas exploitable retombe sur le defaut.
    brut = reponse.get("response") if isinstance(reponse, dict) else reponse
    verdict = _parse_intent(brut)
    if verdict is None:
        logger.warning("🧭 Verdict illisible : repli sur juridique")
        return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-json")

    intention, confiance = verdict
    return IntentResult(intention, confiance, "llm")

	"""
	Aiguillage d'un message utilisateur : question de droit, ou conversation ?

	POURQUOI CE MODULE EXISTE. `RAGService.ask()` envoyait TOUTE question a la
	recherche hybride, sans exception. Une salutation remonte quand meme des
	chunks — le plein texte et le vectoriel trouvent toujours quelque chose — si
	bien que le garde-fou « aucun resultat » ne se declenchait jamais. Mesure sur
	une vraie session : « comment vas tu ? » a produit un pave a rubriques
	(« Source legale L'information demandee n'est pas presente dans les
	documents fournis, notamment la Loi N°2023/014 portant Code Minier ») et un
	bloc « Sources » citant un article du Code Minier. Le cout n'etait pas
	qu'esthetique : chaque salutation payait un appel d'embedding Gemini, occupait
	`embedding_cache` sept jours et laissait une ligne dans `search_events`.

	CE MODULE NE CLASSE PAS DES DOCUMENTS. `legal_domain_classifier.py` range un
	texte du corpus dans un domaine juridique a l'ingestion ; ici on lit la
	question de l'utilisateur, ce qui n'a ni les memes entrees ni le meme verdict.

	LA MECANIQUE EST CELLE DE `reranker.rerank_with_llm` : l'appel est borne par
	`asyncio.wait_for`, et AUCUNE exception ne sort. Le repli d'un composant
	optionnel, c'est l'absence du composant : « juridique ».

	DEUX MODES (INTENT_CLASSIFIER). « groq », le defaut : consignes courtes,
	schema strict, verdicts en cache, et « juridique » des que Groq ne peut pas
	repondre dans les temps (quota, attente, panne). « llm » : le modele du chat,
	passe en parametre. Le classement local par embeddings a ete retire.

	Author: JuriX Team
	"""

	import asyncio
	import json
	import logging
	import time
	from collections import OrderedDict
	from dataclasses import dataclass
	from typing import Any, List, Optional, Tuple

	from app.core.config import settings
	from app.services.prompts import (
	CLASSIFICATION_SYSTEM,
	CONSIGNES_INTENTION,
	ECHANGES_POUR_L_INTENTION,
	LONGUEUR_ECHANGE_INTENTION,
	construire_prompt_de_classification,
	format_conversation_history,
	message_d_intention,
	)
	from app.services.reranker import _article_number_in
	from app.services.text_features import fold_accents

	logger = logging.getLogger(__name__)


	# Ordre explicite, jamais l'ordre d'insertion d'un dictionnaire.
	INTENTS: Tuple[str, ...] = ("juridique", "smalltalk", "meta", "hors_sujet")

	# Le seul qui emprunte le RAG. Les trois autres partagent le chemin
	# conversationnel, avec un prompt different chacun.
	INTENT_JURIDIQUE = INTENTS[0]

	# Messages dont la TOTALITE est une formule de politesse. Le rapprochement se
	# fait sur le message ENTIER normalise, jamais par sous-chaine : « bonjour »
	# court-circuite, « bonjour, puis-je divorcer sans avocat ? » non — c'est
	# precisement le piege qu'un filtre lexical naif tend, et il classerait une
	# vraie question de droit en salutation.
	#
	# Le gain n'est pas theorique : sur le palier gratuit, chaque message coute
	# desormais deux appels au modele, et « bonjour » ouvre une conversation sur
	# deux. Ceux-la n'en coutent plus aucun.
	_SALUTATIONS_EXACTES = frozenset({
	"bonjour", "bonsoir", "salut", "coucou", "hello", "hi", "hey",
	"merci", "merci beaucoup", "merci bien", "thanks", "thank you",
	"au revoir", "bye", "a bientot", "bonne journee", "bonne soiree",
	"ok", "d'accord", "tres bien", "parfait",
	})

	# Ponctuation et accents retires avant comparaison : « Bonjour ! », « bonjour »
	# et « BONJOUR. » sont le meme message.
	_PONCTUATION = str.maketrans("", "", "!?.,;:…\"'«»()")


	def _est_une_salutation_pure(question: str) -> bool:
	"""Le message ENTIER est-il une formule de politesse, et rien d'autre ?"""
	from app.services.text_features import fold_accents

	nettoye = fold_accents((question or "").strip().lower()).translate(_PONCTUATION)
	return " ".join(nettoye.split()) in _SALUTATIONS_EXACTES


	@dataclass(frozen=True)
	class IntentResult:
	"""
	Verdict de routage.

	`intent` vaut TOUJOURS l'un de INTENTS, jamais None : l'appelant n'a pas de
	branche « je ne sais pas » a ecrire. `rule` nomme ce qui a decide, ce qui
	rend chaque erreur diagnosticable dans les journaux sans rejouer la
	requete — un `defaut-timeout` massif et un `defaut-json` massif appellent
	des corrections opposees.
	"""

	intent: str
	confidence: float
	rule: str


	# Schema impose a la sortie du modele. `gemini_service.generate` expose
	# `response_schema` explicitement, precisement pour ce genre d'appel.
	_INTENT_SCHEMA = {
	"type": "object",
	"properties": {
	"intention": {"type": "string", "enum": list(INTENTS)},
	"confiance": {"type": "number"},
	},
	"required": ["intention"],
	}


	def _parse_intent(raw: Any) -> Optional[Tuple[str, float]]:
	"""
	Lit le verdict du modele. Rend None sur tout ce qui n'est pas exploitable.

	Fonction PURE : elle ne journalise pas et ne leve pas, ce qui la rend
	testable sans doublure.
	"""
	if not raw or not isinstance(raw, str):
	return None

	try:
	charge = json.loads(raw)
	except (ValueError, TypeError):
	return None

	return _verdict_de(charge)


	def _verdict_de(charge: Any) -> Optional[Tuple[str, float]]:
	"""Le verdict d'un objet JSON deja decode, ou None. Pure, comme _parse_intent."""
	if not isinstance(charge, dict):
	return None

	intention = charge.get("intention")
	if intention not in INTENTS:
	return None

	try:
	confiance = float(charge.get("confiance", 0.5))
	except (TypeError, ValueError):
	confiance = 0.5

	return intention, max(0.0, min(1.0, confiance))


	# ==================== CLASSEMENT PAR GROQ ====================

	# Schema STRICT : la sortie ne peut etre que cet objet, rien d'autre.
	_SCHEMA_GROQ = {
	"type": "object",
	"properties": {
	"intention": {"type": "string", "enum": list(INTENTS)},
	"confiance": {"type": "number"},
	},
	"required": ["intention", "confiance"],
	"additionalProperties": False,
	}

	# Un verdict tient en une quinzaine de jetons, mais le modele en depense
	# d'autres avant, invisibles, meme sans reflexion : mesure sur
	# qwen3.8-27b, 128 jetons factures pour un verdict de 15. A 40, deux
	# messages sur soixante revenaient VIDES (HTTP 400 json_validate_failed),
	# donc en « juridique » par defaut. Le quota compte l'usage reel, pas ce
	# plafond.
	_JETONS_DE_SORTIE = 256

	# Attente acceptee devant le limiteur. Au-dela, la question part tout de suite
	# en « juridique » : faire patienter l'utilisateur pour savoir s'il dit
	# bonjour couterait plus qu'une recherche inutile.
	ATTENTE_MAX_INTENTION_S = 0.5


	class _CacheDeVerdicts:
	"""
	Les derniers verdicts, par message et echanges precedents.

	« bonjour », « merci » ou « que dit le code du travail ? » reviennent
	souvent a l'identique : chacun coutait une requete sur les 1 000 du jour.
	Taille et duree bornees ; seuls les verdicts du modele y entrent, jamais
	un defaut.
	"""

	def __init__(self, taille: int = 2000, duree_s: float = 86_400.0, horloge=time.monotonic):
	self.taille = taille
	self.duree_s = duree_s
	self._horloge = horloge
	self._entrees: "OrderedDict[Tuple[str, str], Tuple[float, IntentResult]]" = OrderedDict()

	def lire(self, cle: Tuple[str, str]) -> Optional[IntentResult]:
	entree = self._entrees.get(cle)
	if entree is None:
	return None
	if self._horloge() - entree[0] > self.duree_s:
	del self._entrees[cle]
	return None
	self._entrees.move_to_end(cle)
	return entree[1]

	def ecrire(self, cle: Tuple[str, str], verdict: IntentResult) -> None:
	self._entrees[cle] = (self._horloge(), verdict)
	self._entrees.move_to_end(cle)
	while len(self._entrees) > self.taille:
	self._entrees.popitem(last=False)

	def vider(self) -> None:
	self._entrees.clear()

	def __len__(self) -> int:
	return len(self._entrees)


	cache_des_verdicts = _CacheDeVerdicts()


	def _normaliser(texte: str) -> str:
	return " ".join(fold_accents((texte or "").lower()).split())


	def _echanges(history: Optional[List[Any]]) -> List[Tuple[str, str]]:
	"""Les derniers echanges (role, texte tronque), du plus ancien au plus recent."""
	echanges = []
	for message in (history or [])[-ECHANGES_POUR_L_INTENTION:]:
	role = "utilisateur" if getattr(message, "role", "") == "user" else "assistant"
	texte = " ".join((getattr(message, "content", "") or "").split())
	echanges.append((role, texte[:LONGUEUR_ECHANGE_INTENTION]))
	return echanges


	async def _classer_par_groq(
	question: str, history: Optional[List[Any]], timeout: Optional[float]
	) -> IntentResult:
	"""
	Classement par Groq. Ne leve jamais : toute impossibilite rend
	« juridique », avec une regle `defaut-groq-<raison>` qui dit laquelle.
	"""
	from app.services.groq_service import (
	GroqLimiteError,
	GroqQuotaError,
	GroqReponseInvalideError,
	get_groq_service,
	)

	if not settings.GROQ_API_KEY:
	return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-groq-cle")

	echanges = _echanges(history)
	cle = (_normaliser(question), _normaliser(" \| ".join(t for _, t in echanges)))
	en_cache = cache_des_verdicts.lire(cle)
	if en_cache is not None:
	return IntentResult(en_cache.intent, en_cache.confidence, "groq-cache")

	delai = timeout if timeout is not None else settings.GROQ_INTENTION_TIMEOUT_S
	raison = None
	try:
	reponse = await asyncio.wait_for(
	get_groq_service().completer_json(
	systeme=CONSIGNES_INTENTION,
	message=message_d_intention(question, echanges),
	schema=_SCHEMA_GROQ,
	nom_schema="intention",
	max_jetons=_JETONS_DE_SORTIE,
	attente_max=ATTENTE_MAX_INTENTION_S,
	timeout=delai,
	),
	timeout=delai,
	)
	verdict = _verdict_de(reponse.donnees)
	if verdict is None:
	raison = "json"
	except asyncio.TimeoutError:
	raison = "timeout"
	except GroqQuotaError:
	raison = "quota"
	except GroqLimiteError:
	raison = "limite"
	except GroqReponseInvalideError:
	raison = "json"
	except Exception as exc:
	# Panne, cle refusee, schema refuse : tout est avale. Une defaillance
	# du routeur ne doit jamais faire echouer la question.
	logger.warning("🧭 Groq en echec (%r)", exc)
	raison = "erreur"

	if raison is not None:
	logger.warning("🧭 Classement d'intention impossible (%s) : repli sur juridique", raison)
	return IntentResult(INTENT_JURIDIQUE, 0.0, f"defaut-groq-{raison}")

	intention, confiance = verdict
	resultat = IntentResult(intention, confiance, "groq")
	cache_des_verdicts.ecrire(cle, resultat)
	return resultat


	async def classify_intent(
	question: str,
	*,
	llm: Optional[Any] = None,
	history: Optional[List[Any]] = None,
	timeout: Optional[float] = None,
	max_tokens: Optional[int] = None,
	) -> IntentResult:
	"""
	Classe un message. Ne leve jamais.

	Args:
	question: le message de l'utilisateur, tel qu'il l'a ecrit
	llm: service de generation du mode « llm ». PARAMETRE et non
	`get_llm_service()` en dur, meme raison que `rerank_with_llm` : un
	test injecte une doublure, le harnais d'evaluation peut changer de
	modele. Le mode « groq » ne s'en sert pas.
	history: les derniers messages, pour les questions de suivi. « et
	l'article 12 ? » n'est juridique que par ce qui precede.

	Returns:
	IntentResult, toujours. Sur echec : `juridique`, ce qui redonne le
	comportement anterieur.

	POURQUOI LE DEFAUT EST « juridique ». L'erreur n'est pas symetrique.
	Classer une salutation en juridique coute un pave disgracieux — le bug
	d'hier, donc une degradation vers l'etat anterieur. Classer « quelles sont
	les conditions du permis de recherche ? » en conversation coute une reponse
	INVENTEE et SANS SOURCE sur une question de droit, et l'utilisateur n'a
	aucun moyen de s'en apercevoir : on lui a justement retire le bloc Sources
	qui l'aurait alerte. S'y ajoute que les pannes sont correlees a la charge,
	pas au contenu : un defaut conversationnel ferait basculer TOUTE la
	plateforme en mode bavardage pendant une saturation du fournisseur.
	"""
	if not settings.INTENT_ROUTING_ENABLED:
	return IntentResult(INTENT_JURIDIQUE, 1.0, "desactive")

	# Court-circuit deterministe. « que dit l'article 33 du Code Minier ? » n'a
	# pas besoin d'un modele pour etre reconnue comme juridique, et le
	# re-ranking documente cette forme comme dominante sur ce produit. La regex
	# est celle du re-ranking, pas une seconde : deux auraient diverge.
	#
	# Pas de court-circuit lexical symetrique cote salutation (une liste
	# « bonjour / merci / salut ») : il classerait « Bonjour, puis-je
	# divorcer ? » en smalltalk. L'asymetrie est voulue — on ne court-circuite
	# que dans le sens qui degrade vers le comportement actuel.
	if _article_number_in(question or ""):
	return IntentResult(INTENT_JURIDIQUE, 1.0, "court-circuit-article")

	# Symetrique, et sur une egalite STRICTE du message entier : c'est ce qui
	# le rend sur. Voir _SALUTATIONS_EXACTES.
	if _est_une_salutation_pure(question):
	return IntentResult("smalltalk", 1.0, "court-circuit-salutation")

	if settings.INTENT_CLASSIFIER == "groq":
	return await _classer_par_groq(question, history, timeout)

	# « llm » : le modele du chat, passe en parametre.
	prompt = construire_prompt_de_classification(
	question, format_conversation_history(history or [])
	)

	try:
	reponse = await asyncio.wait_for(
	llm.generate(
	prompt=prompt,
	# Systeme court et neutre, OBLIGATOIRE : `generate` retombe sur
	# `GeminiService.SYSTEM_INSTRUCTION` quand `system` est absent,
	# et cette instruction-la impose de citer des articles et de
	# terminer par « Sources: ». Sur une tache de classification,
	# elle produirait de la prose au lieu du JSON demande.
	system=CLASSIFICATION_SYSTEM,
	temperature=0.0,
	max_tokens=max_tokens or settings.INTENT_MAX_TOKENS,
	reflexion=settings.GEMINI_REFLEXION_CLASSIFICATION,
	response_mime_type="application/json",
	response_schema=_INTENT_SCHEMA,
	),
	timeout=timeout if timeout is not None else settings.INTENT_TIMEOUT_S,
	)
	except asyncio.TimeoutError:
	logger.warning("🧭 Classification expiree : repli sur juridique")
	return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-timeout")
	except Exception as exc:
	# Quota, saturation, panne reseau : tout est avale. Une defaillance du
	# routeur ne doit jamais faire echouer la question de l'utilisateur.
	logger.warning("🧭 Classification en echec (%s) : repli sur juridique", exc)
	return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-erreur")

	# Extraction TOTALE, jamais `reponse["response"]`. `generate` rend un dict,
	# mais ce module promet de ne jamais lever : une doublure ou une future
	# variante rendant une chaine nue ne doit pas transformer le routeur en
	# panne. Tout ce qui n'est pas exploitable retombe sur le defaut.
	brut = reponse.get("response") if isinstance(reponse, dict) else reponse
	verdict = _parse_intent(brut)
	if verdict is None:
	logger.warning("🧭 Verdict illisible : repli sur juridique")
	return IntentResult(INTENT_JURIDIQUE, 0.0, "defaut-json")

	intention, confiance = verdict
	return IntentResult(intention, confiance, "llm")