Spaces:
Runtime error
Runtime error
| """ | |
| Mesure le classement de la PAGE DE RECHERCHE, document par document. | |
| La page /search envoie une recherche en mode « text », filtree sur | |
| status=published, 20 resultats par page. Ce script rejoue exactement cette | |
| requete, sans cache et sans journaliser d'evenement de recherche (lecture | |
| seule), et mesure le rang de la loi attendue : | |
| - les 102 questions relues de tests/fixtures/eval/retrieval_eval_v1.json | |
| (loi attendue : expected_law_id) : hit@1, hit@3, hit@10, MRR@20 ; | |
| - des requetes TEMOINS, chacune avec ce qu'on attend d'elle : la requete qui | |
| a revele le defaut (« peine de mort au cameroun »), et celles qui fondent la | |
| passe floue sur les titres (fautes de frappe : « nominaton », « code miner »). | |
| Usage: | |
| python -m scripts.eval.evaluer_recherche | |
| python -m scripts.eval.evaluer_recherche --modes text hybrid --sortie data/eval_runs/recherche.json | |
| """ | |
| import argparse | |
| import asyncio | |
| import json | |
| import sys | |
| import unicodedata | |
| from pathlib import Path | |
| from typing import Any, Callable, Dict, List, Optional, Sequence | |
| RACINE = Path(__file__).resolve().parents[2] | |
| sys.path.insert(0, str(RACINE)) | |
| from app.core.database import AsyncSessionLocal | |
| from app.schemas.search import SearchFilters, SearchRequest | |
| from app.services.search_service import SearchService | |
| JEU = RACINE / "tests" / "fixtures" / "eval" / "retrieval_eval_v1.json" | |
| TAILLE_PAGE = 20 # TAILLE_PAGE_RECHERCHE du front (src/lib/pagination.ts) | |
| CODE_PENAL, JUSTICE_MILITAIRE = 2236, 2085 | |
| CODES_MINIERS = {2210, 2228} | |
| def _sans_accents(texte: str) -> str: | |
| return "".join(c for c in unicodedata.normalize("NFD", texte.lower()) | |
| if unicodedata.category(c) != "Mn") | |
| def _titres_contiennent(mot: str, n: int) -> Callable[[Dict[str, Any]], bool]: | |
| """Les n premiers titres contiennent tous `mot` (sans accents).""" | |
| def verifier(reponse: Dict[str, Any]) -> bool: | |
| titres = [_sans_accents(r["title"]) for r in reponse["results"][:n]] | |
| return len(titres) == n and all(mot in t for t in titres) | |
| return verifier | |
| def _rang_parmi(ids: set, n: int) -> Callable[[Dict[str, Any]], bool]: | |
| """Une des lois `ids` figure dans les n premiers resultats.""" | |
| def verifier(reponse: Dict[str, Any]) -> bool: | |
| return any(r["law_id"] in ids for r in reponse["results"][:n]) | |
| return verifier | |
| def _navigation_directe(ids: set) -> Callable[[Dict[str, Any]], bool]: | |
| def verifier(reponse: Dict[str, Any]) -> bool: | |
| return bool(reponse["direct_navigation"]) and reponse["target_law_id"] in ids | |
| return verifier | |
| # (requete, attente lisible, verification) | |
| TEMOINS: List[tuple] = [ | |
| ("peine de mort au cameroun", "Code penal ou justice militaire dans le top 3", | |
| _rang_parmi({CODE_PENAL, JUSTICE_MILITAIRE}, 3)), | |
| ("peine de mort", "Code penal ou justice militaire dans le top 3", | |
| _rang_parmi({CODE_PENAL, JUSTICE_MILITAIRE}, 3)), | |
| ("justice militaire", "Code de justice militaire premier", | |
| _rang_parmi({JUSTICE_MILITAIRE}, 1)), | |
| ("code pénal", "Code penal (loi 2236 ou decret 1769) dans le top 2", | |
| _rang_parmi({CODE_PENAL, 1769}, 2)), | |
| ("code minier", "un code minier dans le top 2", _rang_parmi(CODES_MINIERS, 2)), | |
| ("code miner", "faute de frappe : un code minier dans le top 3", _rang_parmi(CODES_MINIERS, 3)), | |
| ("cod minier", "faute de frappe : un code minier dans le top 3", _rang_parmi(CODES_MINIERS, 3)), | |
| ("nominaton", "faute de frappe : les 5 premiers titres parlent de nomination", | |
| _titres_contiennent("nomination", 5)), | |
| ("avancment", "faute de frappe : les 3 premiers titres parlent d'avancement", | |
| _titres_contiennent("avancement", 3)), | |
| ("article 35 du code minier", "navigation directe vers un code minier", | |
| _navigation_directe(CODES_MINIERS)), | |
| ] | |
| async def rejouer(service: SearchService, requete: str, mode: str) -> Dict[str, Any]: | |
| """La requete de la page, sans cache ni journal d'evenement.""" | |
| demande = SearchRequest( | |
| query=requete, mode=mode, limit=TAILLE_PAGE, offset=0, | |
| filters=SearchFilters(status="published"), | |
| ) | |
| chunks = await service._execute_search_by_mode(demande) | |
| reponse = await service._build_search_response(demande, chunks, 0) | |
| return { | |
| "results": [{"law_id": r.law_id, "title": r.title, "score": round(r.relevance_score, 3), | |
| "portee": r.match_scope} for r in reponse.results], | |
| "total": reponse.total, | |
| "direct_navigation": reponse.direct_navigation, | |
| "target_law_id": reponse.target_law_id, | |
| } | |
| def _rang(reponse: Dict[str, Any], law_id: int) -> Optional[int]: | |
| for position, r in enumerate(reponse["results"], start=1): | |
| if r["law_id"] == law_id: | |
| return position | |
| return None | |
| def synthese(rangs: Sequence[Optional[int]]) -> Dict[str, Any]: | |
| n = len(rangs) or 1 | |
| return { | |
| "n": len(rangs), | |
| "hit@1": round(sum(1 for r in rangs if r and r <= 1) / n, 3), | |
| "hit@3": round(sum(1 for r in rangs if r and r <= 3) / n, 3), | |
| "hit@10": round(sum(1 for r in rangs if r and r <= 10) / n, 3), | |
| "mrr@20": round(sum(1 / r for r in rangs if r) / n, 3), | |
| "aucun_resultat": None, | |
| } | |
| async def evaluer(modes: Sequence[str]) -> Dict[str, Any]: | |
| items = json.loads(JEU.read_text(encoding="utf-8"))["items"] | |
| sortie: Dict[str, Any] = {"modes": {}} | |
| async with AsyncSessionLocal() as session: | |
| service = SearchService(session, use_cache=False) | |
| for mode in modes: | |
| par_item, rangs, vides = [], {"keyword": [], "paraphrase": [], "tous": []}, 0 | |
| for item in items: | |
| reponse = await rejouer(service, item["question"], mode) | |
| rang = _rang(reponse, item["expected_law_id"]) | |
| vides += not reponse["results"] | |
| rangs["tous"].append(rang) | |
| rangs.setdefault(item.get("difficulty", "?"), []).append(rang) | |
| par_item.append({"question": item["question"], "attendu": item["expected_law_id"], | |
| "rang": rang, "total": reponse["total"]}) | |
| temoins = [] | |
| for requete, attente, verifier in TEMOINS: | |
| reponse = await rejouer(service, requete, mode) | |
| temoins.append({"requete": requete, "attente": attente, "ok": verifier(reponse), | |
| "top5": [f"{r['law_id']} {r['portee']} {r['score']} {r['title'][:70]}" | |
| for r in reponse["results"][:5]]}) | |
| resume = {cle: synthese(valeurs) for cle, valeurs in rangs.items() if valeurs} | |
| resume["tous"]["aucun_resultat"] = vides | |
| sortie["modes"][mode] = {"synthese": resume, "temoins": temoins, "items": par_item} | |
| return sortie | |
| def afficher(sortie: Dict[str, Any]) -> None: | |
| for mode, donnees in sortie["modes"].items(): | |
| print(f"\n=== mode {mode}") | |
| for cle, s in donnees["synthese"].items(): | |
| print(f" {cle:10} n={s['n']:3} hit@1 {s['hit@1']:.3f} hit@3 {s['hit@3']:.3f} " | |
| f"hit@10 {s['hit@10']:.3f} mrr@20 {s['mrr@20']:.3f}" | |
| + (f" sans resultat {s['aucun_resultat']}" if s["aucun_resultat"] is not None else "")) | |
| ok = sum(t["ok"] for t in donnees["temoins"]) | |
| print(f" temoins : {ok}/{len(donnees['temoins'])}") | |
| for t in donnees["temoins"]: | |
| print(f" [{'ok' if t['ok'] else 'KO'}] {t['requete']!r} : {t['attente']}") | |
| if not t["ok"]: | |
| for ligne in t["top5"]: | |
| print(f" {ligne}") | |
| def main(argv: Optional[Sequence[str]] = None) -> int: | |
| parseur = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) | |
| parseur.add_argument("--modes", nargs="+", default=["text"], choices=["text", "semantic", "hybrid"]) | |
| parseur.add_argument("--sortie", type=Path, help="Ecrit le detail en JSON") | |
| args = parseur.parse_args(argv) | |
| sortie = asyncio.run(evaluer(args.modes)) | |
| afficher(sortie) | |
| if args.sortie: | |
| args.sortie.parent.mkdir(parents=True, exist_ok=True) | |
| args.sortie.write_text(json.dumps(sortie, ensure_ascii=False, indent=1), encoding="utf-8") | |
| print(f"\nDetail : {args.sortie}") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |