"""Section « Leviers d'amélioration » — Sprint 82 (A.I.9). Sprint 82 — A.I.9 du plan d'évolution 2026. Pourquoi ce module ------------------ Le moteur narratif (Sprint 19) émet des `Fact` qui décrivent **ce qui s'est passé** dans le benchmark : qui gagne, qui s'effondre, qui est fragile. Ce sprint répond à une question complémentaire : **sur quelle dimension le bénéfice attendu d'une amélioration serait-il le plus visible ?** Pas de prescription ------------------- Picarones est un **outil de recherche**, pas un atelier de production. Le module ne dit jamais *« faites X »* ni *« utilisez le moteur Y »* ; il agrège des **observations factuelles** déjà calculées dans d'autres modules (Sprints 75-81) et les présente comme un récapitulatif compact en bas du rapport. Le chercheur lit, juge et arbitre. Exemples de leviers émis ------------------------ - *« 65 % des erreurs de Tesseract sont de classe récupérable (case_error, ligature_error, abbreviation_error) — un post-processing trivial absorberait une partie. »* - *« 12 % de vos documents concentrent 78 % du CER total (Pareto-CER). »* - *« Le déficit projeté du moteur le plus fragile sur le corpus réel est de 4,2 points de CER (Sprint 81). »* - *« Le top-3 des tokens GT systématiquement modernisés est maistre, nostre, veoir (Sprint 80). »* Structure --------- Module parallèle au registre narratif Sprint 19 : `Lever` est la dataclass équivalente à `Fact`, `LeverImportance` reprend la sémantique de `FactImportance`, `@register_lever` indexe les détecteurs. Garde-fou anti-hallucination identique : chaque nombre rendu doit être présent dans le `payload` du `Lever`. Les détecteurs lisent **uniquement** des structures déjà construites par le pipeline du benchmark — ils ne calculent rien de nouveau, ils synthétisent. C'est pourquoi le module est résolument optionnel : si un benchmark n'expose pas `taxonomy_aggregated`, `inter_engine_analysis`, `corpus_difficulty`, `lexical_modernization` ou `robustness_projection`, le détecteur correspondant retourne tout simplement `[]`. """ from __future__ import annotations import logging import threading from dataclasses import dataclass from enum import Enum from typing import Callable logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────────── # Modèle # ────────────────────────────────────────────────────────────────────────── class LeverType(str, Enum): """Types de leviers détectés.""" DOMINANT_RECOVERABLE_CLASS = "dominant_recoverable_class" """Une part importante des erreurs d'un moteur est dans des classes catégorisées « récupérables » (Sprint 77).""" PARETO_CONCENTRATION = "pareto_concentration" """Une fraction minoritaire de documents concentre une fraction majoritaire du CER total — l'inspection ciblée est rentable.""" COMPLEMENTARITY_OBSERVATION = "complementarity_observation" """Le `complementarity_gap` (Sprint 35) entre l'oracle et le meilleur moteur seul est non négligeable — observation factuelle, aucune recommandation d'ensemble.""" LEXICAL_MODERNIZATION_OBSERVATION = "lexical_modernization_observation" """Top-N des tokens GT systématiquement modernisés (Sprint 80).""" ROBUSTNESS_PROJECTION_OBSERVATION = "robustness_projection_observation" """Déficit projeté global le plus important pour un moteur sur le corpus réel (Sprint 81).""" class LeverImportance(int, Enum): """Importance éditoriale d'un levier.""" HIGH = 70 MEDIUM = 40 LOW = 10 @dataclass class Lever: """Observation factuelle synthétisable en encart « Leviers ». Attributes ---------- type: Le type de levier (voir `LeverType`). importance: Score qui décide l'ordre d'affichage. payload: Données brutes — **tout chiffre rendu dans le HTML doit provenir d'ici**, jamais d'un calcul du renderer. engines_involved: Noms des moteurs concernés (peut être vide pour un levier corpus-wide). """ type: LeverType importance: LeverImportance payload: dict engines_involved: tuple[str, ...] = () def as_dict(self) -> dict: return { "type": self.type.value, "importance": int(self.importance), "payload": self.payload, "engines_involved": list(self.engines_involved), } # ────────────────────────────────────────────────────────────────────────── # Registre # ────────────────────────────────────────────────────────────────────────── LeverDetectorFn = Callable[[dict], list[Lever]] @dataclass(frozen=True) class LeverDetectorEntry: lever_type: LeverType fn: LeverDetectorFn priority: int _LEVER_REGISTRY: dict[LeverType, LeverDetectorEntry] = {} _LEVER_REGISTRY_LOCK = threading.Lock() def register_lever( lever_type: LeverType, *, priority: int, ) -> Callable[[LeverDetectorFn], LeverDetectorFn]: """Décorateur : enregistre un détecteur de levier. Une seule fonction par type — réenregistrer lève `ValueError`. """ def _decorator(fn: LeverDetectorFn) -> LeverDetectorFn: with _LEVER_REGISTRY_LOCK: if lever_type in _LEVER_REGISTRY: raise ValueError( f"Détecteur déjà enregistré pour {lever_type.value!r} : " f"{_LEVER_REGISTRY[lever_type].fn.__name__}." ) _LEVER_REGISTRY[lever_type] = LeverDetectorEntry( lever_type=lever_type, fn=fn, priority=int(priority), ) return fn return _decorator def unregister_lever(lever_type: LeverType) -> None: with _LEVER_REGISTRY_LOCK: _LEVER_REGISTRY.pop(lever_type, None) def iter_lever_detectors() -> list[LeverDetectorEntry]: with _LEVER_REGISTRY_LOCK: entries = list(_LEVER_REGISTRY.values()) entries.sort(key=lambda e: e.priority) return entries def detect_levers(benchmark_data: dict) -> list[Lever]: """Applique tous les détecteurs enregistrés et trie par importance décroissante puis priorité d'enregistrement croissante.""" levers: list[Lever] = [] for entry in iter_lever_detectors(): try: result = entry.fn(benchmark_data) except Exception as e: logger.warning( "[levers.detector.%s] fonctionnalité dégradée : %s", entry.lever_type.value, e, ) continue if result: levers.extend(result) # Tri stable : importance décroissante d'abord levers.sort(key=lambda lv: -int(lv.importance)) return levers # ────────────────────────────────────────────────────────────────────────── # Détecteurs # ────────────────────────────────────────────────────────────────────────── # Catégorisation reprise du Sprint 77 (taxonomy_comparison.py). # Volontairement dupliquée ici pour ne pas introduire d'import # circulaire — la sémantique est gelée. _RECOVERABILITY: dict[str, str] = { "case_error": "recoverable", "ligature_error": "recoverable", "abbreviation_error": "recoverable", "diacritic_error": "difficult", "visual_confusion": "difficult", "hapax": "difficult", "lacuna": "irrecoverable", "oov_character": "irrecoverable", "segmentation_error": "irrecoverable", } @register_lever(LeverType.DOMINANT_RECOVERABLE_CLASS, priority=10) def detect_dominant_recoverable_class( benchmark_data: dict, *, threshold: float = 0.30, ) -> list[Lever]: """Émet un levier si ≥ `threshold` des erreurs d'un moteur sont classifiées récupérables (catégorisation Sprint 77). Lit `benchmark_data["engines"][i]["aggregated_taxonomy"]` — structure produite par le runner historique. Si absent, retourne []. """ engines = benchmark_data.get("engines") or [] out: list[Lever] = [] for engine in engines: taxonomy = engine.get("aggregated_taxonomy") if not taxonomy: continue # `taxonomy` peut être {class_name: int} ou un dict avec une # sous-clé "counts" — on accepte les deux conventions. counts = taxonomy.get("counts") if isinstance(taxonomy, dict) and "counts" in taxonomy else taxonomy if not isinstance(counts, dict) or not counts: continue try: int_counts = {k: int(v) for k, v in counts.items() if isinstance(v, (int, float))} except (TypeError, ValueError): continue total = sum(int_counts.values()) if total <= 0: continue recoverable_total = sum( v for k, v in int_counts.items() if _RECOVERABILITY.get(k) == "recoverable" ) share = recoverable_total / total if share < threshold: continue # Classes récupérables non vides triées par count décroissant breakdown = sorted( ( (k, v) for k, v in int_counts.items() if _RECOVERABILITY.get(k) == "recoverable" and v > 0 ), key=lambda kv: -kv[1], ) importance = ( LeverImportance.HIGH if share >= 0.50 else LeverImportance.MEDIUM ) out.append(Lever( type=LeverType.DOMINANT_RECOVERABLE_CLASS, importance=importance, payload={ "engine": engine.get("name") or "?", "share_recoverable": share, "share_recoverable_pct": round(share * 100, 1), "n_recoverable": recoverable_total, "n_total_errors": total, "top_classes": [ {"class": k, "count": v} for k, v in breakdown[:3] ], }, engines_involved=(engine.get("name") or "?",), )) return out @register_lever(LeverType.PARETO_CONCENTRATION, priority=20) def detect_pareto_concentration( benchmark_data: dict, *, top_share: float = 0.20, cer_share_threshold: float = 0.50, ) -> list[Lever]: """Émet un levier si une fraction minoritaire de documents (`top_share`) concentre plus de `cer_share_threshold` du CER total cumulé sur le moteur leader. Lit `benchmark_data["per_doc_cer"][engine_name]` ou tente de reconstruire depuis `benchmark_data["engines"][...]["per_doc"]`. Si rien d'exploitable, retourne []. """ ranking = benchmark_data.get("ranking") or [] if not ranking: return [] leader = ranking[0] leader_name = leader.get("engine") if not leader_name: return [] per_doc_cer: list[float] = [] # Voie 1 : structure plate "per_doc_cer" flat = benchmark_data.get("per_doc_cer") or {} if isinstance(flat, dict) and leader_name in flat and isinstance(flat[leader_name], list): per_doc_cer = [float(x) for x in flat[leader_name] if isinstance(x, (int, float))] else: # Voie 2 : engine.per_doc liste de dicts {cer: float} for engine in benchmark_data.get("engines") or []: if engine.get("name") != leader_name: continue per_doc = engine.get("per_doc") or [] for entry in per_doc: if isinstance(entry, dict) and isinstance(entry.get("cer"), (int, float)): per_doc_cer.append(float(entry["cer"])) break if not per_doc_cer: return [] total_cer = sum(per_doc_cer) if total_cer <= 0: return [] sorted_cer = sorted(per_doc_cer, reverse=True) n = len(sorted_cer) n_top = max(1, int(round(top_share * n))) top_cer_sum = sum(sorted_cer[:n_top]) share_of_total = top_cer_sum / total_cer if share_of_total < cer_share_threshold: return [] importance = ( LeverImportance.HIGH if share_of_total >= 0.75 else LeverImportance.MEDIUM ) return [Lever( type=LeverType.PARETO_CONCENTRATION, importance=importance, payload={ "engine": leader_name, "n_docs": n, "n_docs_top": n_top, "top_share_pct": round((n_top / n) * 100, 1), "cer_share_of_total": share_of_total, "cer_share_pct": round(share_of_total * 100, 1), }, engines_involved=(leader_name,), )] @register_lever(LeverType.COMPLEMENTARITY_OBSERVATION, priority=30) def detect_complementarity_observation( benchmark_data: dict, *, min_relative_gap: float = 0.20, ) -> list[Lever]: """Reformule factuellement le `complementarity_gap` (Sprint 35). Lit `benchmark_data["inter_engine_analysis"]`. Garde-fou : ne déclenche que si `relative_gap` ≥ `min_relative_gap`. **Aucune recommandation d'ensemble** — le levier dit factuellement « X points séparent l'oracle du meilleur moteur », c'est tout. """ inter = benchmark_data.get("inter_engine_analysis") or {} cgap = inter.get("complementarity_gap") or {} relative_gap = cgap.get("relative_gap") absolute_gap = cgap.get("absolute_gap") if relative_gap is None or absolute_gap is None: return [] try: rg = float(relative_gap) ag = float(absolute_gap) except (TypeError, ValueError): return [] if rg < min_relative_gap: return [] importance = ( LeverImportance.HIGH if rg >= 0.50 else LeverImportance.MEDIUM ) payload: dict = { "absolute_gap": ag, "absolute_gap_pct": round(ag * 100, 1), "relative_gap": rg, "relative_gap_pct": round(rg * 100, 1), } best_engine = cgap.get("best_engine") or inter.get("best_engine") best_recall = cgap.get("best_recall") or inter.get("best_engine_recall") oracle_recall = cgap.get("oracle_recall") or inter.get("oracle_recall") engines_involved: tuple[str, ...] = () if best_engine: payload["best_engine"] = str(best_engine) engines_involved = (str(best_engine),) if isinstance(best_recall, (int, float)): payload["best_recall"] = float(best_recall) if isinstance(oracle_recall, (int, float)): payload["oracle_recall"] = float(oracle_recall) return [Lever( type=LeverType.COMPLEMENTARITY_OBSERVATION, importance=importance, payload=payload, engines_involved=engines_involved, )] @register_lever(LeverType.LEXICAL_MODERNIZATION_OBSERVATION, priority=40) def detect_lexical_modernization_observation( benchmark_data: dict, *, top_n: int = 3, min_total: int = 3, min_rate: float = 0.50, ) -> list[Lever]: """Pour chaque moteur disposant de `lexical_modernization`, émet un levier listant les `top_n` tokens GT les plus modernisés. Lit `benchmark_data["engines"][i]["lexical_modernization"]` qui suit la forme produite par `compute_lexical_modernization` du Sprint 80 (`{"n_gt_tokens": int, "tokens": dict}`). """ out: list[Lever] = [] for engine in benchmark_data.get("engines") or []: data = engine.get("lexical_modernization") if not isinstance(data, dict): continue tokens = data.get("tokens") or {} if not isinstance(tokens, dict) or not tokens: continue candidates: list[tuple[str, dict]] = [] for gt_token, slot in tokens.items(): if not isinstance(slot, dict): continue n_total = slot.get("n_total") rate = slot.get("rate_modernized") if not isinstance(n_total, (int, float)) or not isinstance(rate, (int, float)): continue if int(n_total) < min_total: continue if float(rate) < min_rate: continue candidates.append((gt_token, dict(slot))) if not candidates: continue candidates.sort( key=lambda kv: (-float(kv[1].get("rate_modernized", 0.0)), -int(kv[1].get("n_total", 0)), kv[0]), ) top = candidates[:top_n] engine_name = engine.get("name") or "?" max_rate = max(float(slot.get("rate_modernized", 0.0)) for _, slot in top) importance = ( LeverImportance.HIGH if max_rate >= 0.90 else LeverImportance.MEDIUM ) out.append(Lever( type=LeverType.LEXICAL_MODERNIZATION_OBSERVATION, importance=importance, payload={ "engine": engine_name, "top_tokens": [ { "gt_token": gt, "n_total": int(slot.get("n_total", 0)), "rate_modernized": float(slot.get("rate_modernized", 0.0)), "rate_modernized_pct": round( float(slot.get("rate_modernized", 0.0)) * 100, 1, ), } for gt, slot in top ], }, engines_involved=(engine_name,), )) return out @register_lever(LeverType.ROBUSTNESS_PROJECTION_OBSERVATION, priority=50) def detect_robustness_projection_observation( benchmark_data: dict, *, min_total_deficit: float = 0.02, ) -> list[Lever]: """Lit l'agrégation par moteur de la projection de robustesse (Sprint 81). Émet le levier pour le moteur dont `total_expected_deficit` est ≥ `min_total_deficit` (par défaut 2 points de CER). Lit `benchmark_data["robustness_projection_aggregated"]` — structure produite par `aggregate_projection_per_engine`. """ agg = benchmark_data.get("robustness_projection_aggregated") or {} if not isinstance(agg, dict) or not agg: return [] out: list[Lever] = [] for engine_name, info in agg.items(): if not isinstance(info, dict): continue total_deficit = info.get("total_expected_deficit") worst_type = info.get("worst_degradation_type") worst_deficit = info.get("worst_degradation_deficit") if not isinstance(total_deficit, (int, float)): continue if float(total_deficit) < min_total_deficit: continue importance = ( LeverImportance.HIGH if float(total_deficit) >= 0.05 else LeverImportance.MEDIUM ) payload: dict = { "engine": engine_name, "total_expected_deficit": float(total_deficit), "total_expected_deficit_pct": round(float(total_deficit) * 100, 1), "n_degradation_types": int(info.get("n_degradation_types") or 0), } if isinstance(worst_type, str): payload["worst_degradation_type"] = worst_type if isinstance(worst_deficit, (int, float)): payload["worst_degradation_deficit"] = float(worst_deficit) payload["worst_degradation_deficit_pct"] = round( float(worst_deficit) * 100, 1, ) out.append(Lever( type=LeverType.ROBUSTNESS_PROJECTION_OBSERVATION, importance=importance, payload=payload, engines_involved=(engine_name,), )) # Tri par déficit décroissant pour stabilité d'affichage. out.sort( key=lambda lv: -float(lv.payload.get("total_expected_deficit") or 0.0), ) return out __all__ = [ "Lever", "LeverImportance", "LeverType", "LeverDetectorEntry", "register_lever", "unregister_lever", "iter_lever_detectors", "detect_levers", "detect_dominant_recoverable_class", "detect_pareto_concentration", "detect_complementarity_observation", "detect_lexical_modernization_observation", "detect_robustness_projection_observation", ]