Spaces:
Sleeping
Sleeping
File size: 3,451 Bytes
80c6417 9011070 80c6417 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """Détecteurs narratifs liés à l'*opportunité d'ensemble inter-moteurs* (chantier 5).
1 détecteur déplacé depuis ``narrative/detectors.py`` :
- :func:`detect_ensemble_opportunity` (Sprint 36)
"""
from __future__ import annotations
from typing import Optional
from picarones.domain.facts import Fact, FactImportance, FactType
from picarones.reports.narrative.registry import register_detector
@register_detector(
FactType.ENSEMBLE_OPPORTUNITY,
priority=130,
importance=FactImportance.MEDIUM,
)
def detect_ensemble_opportunity(benchmark_data: dict) -> list[Fact]:
"""Deux moteurs très complémentaires : un voting majoritaire entre eux
pourrait améliorer significativement le CER token-level.
Lit la structure ``inter_engine_analysis`` produite par le runner
(Sprint 35-36) et déclenche si la fraction d'erreurs du meilleur
moteur récupérable par un ensemble dépasse 25 %.
L'importance monte à ``HIGH`` quand le gap relatif dépasse 50 %
(ensemble franchement profitable) — sinon reste à ``MEDIUM``.
"""
iea = benchmark_data.get("inter_engine_analysis") or {}
comp = iea.get("complementarity") or {}
if not comp:
return []
relative_gap = float(comp.get("relative_gap") or 0.0)
if relative_gap < 0.25:
# En deçà de 25 %, l'ensemble n'apporterait quasi rien — on ne
# remonte pas le fait pour ne pas bruiter la synthèse.
return []
best_engine = comp.get("best_engine") or ""
if not best_engine:
return []
payload: dict = {
"best_engine": best_engine,
"best_recall_pct": round(float(comp.get("best_single_recall") or 0.0) * 100, 2),
"oracle_recall_pct": round(float(comp.get("oracle_recall") or 0.0) * 100, 2),
"absolute_gap_pct": round(float(comp.get("absolute_gap") or 0.0) * 100, 2),
"relative_gap_pct": round(relative_gap * 100, 1),
"doc_count": int(comp.get("doc_count") or 0),
}
# Paire la plus complémentaire — la divergence taxonomique, quand
# disponible, fournit deux moteurs « candidats naturels ». Sinon on
# tombe sur le best + le second-best en recall individuel.
div = iea.get("taxonomy_divergence") or {}
pair = div.get("max_pair") or []
pair_a = ""
pair_b = ""
divergence_value: Optional[float] = None
if pair and len(pair) >= 3 and isinstance(pair[2], (int, float)) and pair[2] > 0:
pair_a, pair_b, divergence_value = str(pair[0]), str(pair[1]), float(pair[2])
else:
# Fallback : best engine + second-best engine par recall individuel
per_engine = comp.get("per_engine_recall") or {}
if len(per_engine) >= 2:
ranked = sorted(per_engine.items(), key=lambda kv: kv[1], reverse=True)
pair_a, pair_b = ranked[0][0], ranked[1][0]
payload["pair_a"] = pair_a
payload["pair_b"] = pair_b
payload["divergence"] = round(divergence_value, 3) if divergence_value is not None else 0.0
payload["divergence_metric"] = (div.get("metric") or "js")
importance = (
FactImportance.HIGH if relative_gap >= 0.5 else FactImportance.MEDIUM
)
engines_involved: tuple[str, ...] = (
(pair_a, pair_b) if pair_a and pair_b else (best_engine,)
)
return [Fact(
type=FactType.ENSEMBLE_OPPORTUNITY,
importance=importance,
payload=payload,
engines_involved=engines_involved,
)]
|