#!/usr/bin/env python3 """ Campagne de tests MCP - Phase 2 : test des RULE_MISS via EPHAISTOS (LLM reel) ============================================================================== Teste les requetes qui ne sont pas couvertes par _rule_based_detect en les passant directement au pipeline RAG + EPHAISTOS. Statuts: LLM_PASS : EPHAISTOS identifie le bon outil + args obligatoires corrects LLM_PARTIAL : EPHAISTOS identifie le bon outil + args optionnels manquants LLM_FAIL : EPHAISTOS retourne le mauvais outil ou args obligatoires absents Usage: python3 tests/test_campaign_llm.py """ import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from lyra.core.config import RAGConfig from lyra.core.pipeline import _resoudre_nom_outil # ============================================================ # Cas de test: les RULE_MISS attendus (TV, HUE complexe, CATT) # Format: (categorie, description, query, expected_tool, mandatory_args, optional_args) # ============================================================ TESTS_LLM = [ # ================================================================ # TV - power # ================================================================ ("TV/power", "allume TV", "allume la tv", "tv.power_on", {}, {}), ("TV/power", "eteins TV", "eteins la tele", "tv.power_off", {}, {}), ("TV/power", "TV en veille", "mets la tv en veille", "tv.power_off", {}, {}), # ================================================================ # TV - apps # ================================================================ ("TV/apps", "lance Netflix", "lance netflix sur la tv", "tv.launch_app", {}, {}), ("TV/apps", "ouvre YouTube", "ouvre youtube sur la tele", "tv.launch_app", {}, {}), ("TV/apps", "video YouTube", "joue cette video youtube sur la tv https://youtu.be/dQw4w9WgXcQ", "tv.youtube_video", {}, {}), # ================================================================ # TV - ambilight # ================================================================ ("TV/ambilight", "allume ambilight", "allume l ambilight", "tv.ambilight_on", {}, {}), ("TV/ambilight", "eteins ambilight", "eteins l ambilight", "tv.ambilight_off", {}, {}), ("TV/ambilight", "mode ambilight", "passe l ambilight en mode lounge", "tv.ambilight_mode", {}, {"mode": "lounge_light"}), # Couleur fixe : tv.ambilight_color existe depuis pylips-mcp 0.4.0 (le cas # avait ete retire le 2026-09-17 faute d'outil, rendu le 2026-09-27). ("TV/ambilight", "ambilight couleur", "mets l ambilight en bleu", "tv.ambilight_color", {"r": 0, "g": 0, "b": 255}, {}), # ================================================================ # HUE - lumiere individuelle # ================================================================ ("HUE/light", "allume lumiere chevet", "allume la lumiere chevet", "hue.turn_on_light", {}, {}), # ================================================================ # HUE - brightness / color # ================================================================ ("HUE/brightness", "luminosite 50", "mets la luminosite a 50 pour cent", "hue.set_group_brightness", {}, {}), ("HUE/brightness", "lumieres plus fortes", "mets les lumieres plus fortes", "hue.set_group_brightness", {}, {}), ("HUE/color", "couleur rouge", "mets les lumieres en rouge", "hue.set_color_rgb", {}, {}), ("HUE/color", "ambiance bleue", "mets une ambiance bleue", "hue.set_color_rgb", {}, {}), # ================================================================ # CATT - cast_youtube # ================================================================ ("CATT/youtube", "caste video youtube", "caste cette video youtube https://youtu.be/dQw4w9WgXcQ", "catt.cast_youtube", {}, {}), ("CATT/youtube", "diffuse video", "diffuse cette video sur la tv https://youtu.be/dQw4w9WgXcQ", "catt.cast_youtube", {}, {}), # ================================================================ # CATT - controles playback # ================================================================ ("CATT/control", "arrete cast", "arrete le cast", "catt.cast_stop", {}, {}), ("CATT/control", "pause cast", "mets le cast en pause", "catt.cast_pause", {}, {}), ("CATT/control", "reprends cast", "reprends le cast", "catt.cast_resume", {}, {}), ("CATT/control", "avance 30s", "avance de 30 secondes dans le cast", "catt.cast_seek", {}, {}), ("CATT/control", "baisse volume diffusion", "baisse le volume de la diffusion", "catt.cast_volume", {}, {}), ] # Outils acceptes a la place de l'attendu : meme effet pour l'utilisateur. # "mets les lumieres plus fortes" (pluriel) via set_group_brightness est au # moins aussi juste que set_brightness ; une video YouTube jouee sur la TV # passe par pylips ou par le Chromecast. Le score STRICT reste publie a cote # (cle `strict` de chaque resultat) pour rester comparable aux mesures # anterieures au 2026-09-17. EQUIVALENCES = { "tv.youtube_video": {"catt.cast_youtube"}, "catt.cast_youtube": {"tv.youtube_video"}, "hue.set_brightness": {"hue.set_group_brightness"}, "hue.set_color_rgb": {"hue.set_group_color_rgb", "hue.set_color_preset", "hue.set_group_color_preset"}, "hue.set_group_color_rgb": {"hue.set_group_color_preset"}, # "chambre en rouge" : preset rouge = rgb rouge "hue.turn_on_group": {"hue.turn_on_light"}, "hue.turn_off_group": {"hue.turn_off_light"}, } def tool_matches(result_tool, expected_tool): """Correspondance stricte (nom court).""" if expected_tool is None: return result_tool is None if result_tool is None: return False r = result_tool.split(".")[-1] if "." in result_tool else result_tool e = expected_tool.split(".")[-1] if "." in expected_tool else expected_tool return r == e def tool_equivalent(result_tool, expected_tool): """Correspondance stricte OU equivalence declaree dans EQUIVALENCES.""" if tool_matches(result_tool, expected_tool): return True if not result_tool or not expected_tool: return False return any(tool_matches(result_tool, eq) for eq in EQUIVALENCES.get(expected_tool, ())) def check_args(result_args, mandatory, optional): if result_args is None: result_args = {} missing_mandatory = [] for k, v in mandatory.items(): rv = result_args.get(k) if rv is None: missing_mandatory.append(k) elif v != "" and str(rv) != str(v): missing_mandatory.append(f"{k}={v}(got={rv})") missing_optional = [k for k in optional if k not in result_args] return missing_mandatory, missing_optional JEUX = {"modeles": "TESTS_LLM", "hors_regles": "TESTS_HORS_REGLES", "hors_regles_2": "TESTS_HORS_REGLES_2", "hors_regles_3": "TESTS_HORS_REGLES_3", "hors_regles_4": "TESTS_HORS_REGLES_4", "hors_regles_5": "TESTS_HORS_REGLES_5"} def cas_du_jeu(jeu: str) -> list: """Les cas d'un jeu : "modeles" (21, proches des regles) ou "hors_regles" (formulations inedites).""" if jeu == "hors_regles": from cases_hors_regles import TESTS_HORS_REGLES return TESTS_HORS_REGLES if jeu == "hors_regles_2": from cases_hors_regles_2 import TESTS_HORS_REGLES_2 return TESTS_HORS_REGLES_2 if jeu == "hors_regles_3": from cases_hors_regles_3 import TESTS_HORS_REGLES_3 return TESTS_HORS_REGLES_3 if jeu == "hors_regles_4": from cases_hors_regles_4 import TESTS_HORS_REGLES_4 return TESTS_HORS_REGLES_4 if jeu == "hors_regles_5": from cases_hors_regles_5 import TESTS_HORS_REGLES_5 return TESTS_HORS_REGLES_5 if jeu != "modeles": raise ValueError(f"jeu inconnu : {jeu} (attendu : {', '.join(JEUX)})") return TESTS_LLM def run_llm_tests(config_path=None, jeu: str = "modeles", reel: bool = False): G = "\033[32m" Y = "\033[33m" R = "\033[31m" C = "\033[36m" BOLD = "\033[1m" DIM = "\033[2m" RESET = "\033[0m" print(f"\n{BOLD}{'='*70}{RESET}") print(f"{BOLD} CAMPAGNE DE TESTS MCP - PHASE 2 : EPHAISTOS (LLM){RESET}") print(f"{BOLD}{'='*70}{RESET}") print(f" {DIM}Tests des RULE_MISS via RAG + Qwen 7B{RESET}\n") # Initialisation du pipeline (une seule fois) print(f"{C}[INIT]{RESET} Chargement du pipeline RAG (ChromaDB + modeles)...") t0 = time.time() chemin_config = Path(config_path) if config_path else Path(__file__).parent.parent / "config.yaml" config = RAGConfig.from_yaml(chemin_config) # EnhancedPipeline, et non Pipeline : c'est le mode de run.sh, et surtout # le seul branche sur le RAG 3-tier. Le Pipeline de base interroge l'index # v2, qui ne contient que les 19 outils fedora : aucune requete de cast n'y # trouvait de candidat, et le banc mesurait donc ce mauvais branchement # plutot que le modele (roadmap-github#72). import yaml as _yaml from lyra.rag_enhanced import EnhancedPipeline from lyra.rag_enhanced.config import RAGEnhancedConfig brut = _yaml.safe_load(chemin_config.read_text()) or {} enhanced = EnhancedPipeline( config=config, enhanced_config=RAGEnhancedConfig.from_dict(brut.get("rag_enhanced", {})), enabled=True, tts_mode=False, ) enhanced.initialize() pipeline = enhanced._pipeline_v2 # porte _ephaistos pipeline._rag_3tier = enhanced._rag_3tier print(f"{G}[INIT]{RESET} Pipeline pret ({time.time()-t0:.1f}s)\n") results = [] categories = {} erreurs_techniques: list[str] = [] cas = cas_du_jeu(jeu) if reel: print(f"{C}[MODE]{RESET} chemin reel : EnhancedPipeline.process (classificateur, contexte, LYRA) -- une session par cas\n") for i, (cat, desc, query, expected_tool, mandatory_args, optional_args) in enumerate(cas, 1): print(f"{DIM}[{i:02d}/{len(cas)}] {desc}: {query[:55]}...{RESET}" if len(query) > 55 else f"{DIM}[{i:02d}/{len(cas)}] {desc}: {query}{RESET}") t_start = time.time() refus_vm = False try: if reel: # Le chemin de production, de bout en bout jusqu'a la proposition # d'action (pas d'execution : le tool_call attend une confirmation # que personne ne donne). Une session neuve par cas. # process() ouvre lui-meme le scope de session_id ("default" sinon) : # sans session propre, la question laissee par un cas (choix, clarification) # etait relue comme la reponse du cas suivant, et tout partait en vm_clone. res = enhanced.process(query, session_id=f"bench-reel-{i}") result_tool = (res.tool_call or {}).get("name") result_args = (res.tool_call or {}).get("arguments") or {} # Le chemin reel verifie que la VM existe : "staging-03" (jeux 3) et # "preprod-01" (jeu 2) n'existent pas ici, la reponse "La VM n'existe pas" # est la bonne. Compte a part, hors du score. refus_vm = bool(res.error) and "n'existe pas" in (res.response or "") if not refus_vm and str(expected_tool).startswith("fedora.vm_"): from lyra.models import ephaistos_exp as _exp cites = _exp.noms_de_machines(query, inventaire=True) if cites and not any(n in _exp.inventaire_vm() for n in cites): refus_vm = True # workflow (clone) : la source n'existe pas, il questionne if result_tool is None and res.error is None and res.pending_args: result_tool = None # clarification demandee : compte comme echec sauf expected None else: # Recuperer les specs RAG resultats = pipeline._rag_3tier.cascade_search(query) # Meme construction que le pipeline enrichi (specs_pour_ephaistos) : # le banc et la production ne doivent plus pouvoir diverger. from lyra.rag_enhanced.rag_3tier import specs_pour_ephaistos specs = specs_pour_ephaistos(resultats) noms_outils = [s.split(":", 1)[0].strip() for s in specs] if not specs: result_tool = None result_args = {} else: # Appeler EPHAISTOS directement from lyra.utils.toon import toon_encode_specs ephaistos_model = pipeline.config.models.ephaistos.name # Les variantes LYRA_EXP agissent sur le chemin des specs # compactes ; avec TOON, analyze() les ignore. Pour comparer les # modeles a configuration egale, pas de TOON des qu'une variante # est active. from lyra.models import ephaistos_exp as _exp use_toon = "0.5b" not in ephaistos_model and not _exp.actives() specs_toon = toon_encode_specs(specs) if use_toon else None analysis = pipeline._ephaistos.analyze_with_retry( user_query=query, mcp_specs=specs, specs_toon=specs_toon ) result_tool = analysis.tool result_args = analysis.arguments or {} # Prefixe serveur : on reutilise la resolution du pipeline, # deja couverte par des tests, contre les noms du 3-tier. if result_tool: resolu = _resoudre_nom_outil(result_tool, [], noms_outils) if resolu: result_tool = resolu except Exception as e: # Une panne technique (API deplacee, RAG muet) n'est PAS un echec # du modele : elle etait comptee en LLM_FAIL et le banc publiait # un 0 % qui accusait le modele a tort (constate le 2026-09-17, # _retrieve_specs avait disparu du pipeline). erreurs_techniques.append(f"{desc}: {type(e).__name__}: {e}") result_tool = None result_args = {} print(f" {R}ERREUR: {e}{RESET}") elapsed = time.time() - t_start # Evaluer le resultat strict = tool_matches(result_tool, expected_tool) tool_ok = tool_equivalent(result_tool, expected_tool) missing_mand, missing_opt = check_args(result_args, mandatory_args, optional_args) if refus_vm: status = "LLM_REFUS" elif result_tool is None: status = "LLM_FAIL" elif not tool_ok: status = "LLM_FAIL" elif missing_mand: status = "LLM_FAIL" elif missing_opt: status = "LLM_PARTIAL" else: status = "LLM_PASS" color = G if status == "LLM_PASS" else (Y if status == "LLM_PARTIAL" else (DIM if status == "LLM_REFUS" else R)) tool_str = (result_tool or "NONE").split(".")[-1][:18].ljust(19) pad = desc[:35].ljust(36) args_str = "" if missing_mand: args_str = f" {R}ARGS_MANQUANTS={missing_mand}{RESET}" elif missing_opt: args_str = f" {Y}OPT_MISS={missing_opt}{RESET}" elif result_tool and not tool_ok: exp_short = (expected_tool or "").split(".")[-1] args_str = f" {R}attendu={exp_short}{RESET}" elif tool_ok and not strict: args_str = f" {Y}EQUIV de {(expected_tool or '').split('.')[-1]}{RESET}" print(f" {color}[{status:<10}]{RESET} {DIM}{pad}{RESET} -> {tool_str}{args_str} {DIM}({elapsed:.1f}s){RESET}") entry = { "cat": cat, "desc": desc, "query": query, "expected_tool": expected_tool, "result_tool": result_tool, "result_args": result_args, "mandatory_args": mandatory_args, "missing_mand": missing_mand, "missing_opt": missing_opt, "status": status, "strict": strict and status == "LLM_PASS", "elapsed": elapsed } results.append(entry) categories.setdefault(cat.split("/")[0], []).append(entry) # Score final print(f"\n{BOLD}{'='*70}{RESET}") print(f"{BOLD} SCORES PAR CATEGORIE (LLM){RESET}") print(f"{BOLD}{'='*70}{RESET}") total_pass = total_partial = total_fail = total_refus = 0 for cat_name, entries in sorted(categories.items()): refus = sum(1 for e in entries if e["status"] == "LLM_REFUS") total_refus += refus n = len(entries) - refus p = sum(1 for e in entries if e["status"] == "LLM_PASS") pa = sum(1 for e in entries if e["status"] == "LLM_PARTIAL") f = sum(1 for e in entries if e["status"] == "LLM_FAIL") total_pass += p total_partial += pa total_fail += f score_pct = round(100 * (p + 0.5 * pa) / n) if n else 0 bar = f"{G}{'#' * p}{Y}{'~' * pa}{R}{'-' * f}{RESET}" print(f" {BOLD}{cat_name:<8}{RESET} {bar} " f"{G}{p}P{RESET} {Y}{pa}~{RESET} {R}{f}F{RESET} / {n} [{score_pct}%]") total = len(results) - total_refus score_final = round(100 * (total_pass + 0.5 * total_partial) / total) if total else 0 if total_refus: print(f"\n {DIM}{total_refus} cas hors score : VM inexistante, refus correct du chemin reel{RESET}") print(f"\n{BOLD}{'='*70}{RESET}") print(f"{BOLD} SCORE LLM : {G}{total_pass}{RESET}{BOLD} LLM_PASS " f"{Y}{total_partial}{RESET}{BOLD} LLM_PARTIAL " f"{R}{total_fail}{RESET}{BOLD} LLM_FAIL / {total} tests{RESET}") print(f"{BOLD} SCORE LLM : {score_final}%{RESET}\n") # Sauvegarder rapport texte report_path = Path(__file__).parent / "test_campaign_llm_report.txt" with open(report_path, "w") as f: f.write("CAMPAGNE DE TESTS MCP - PHASE 2 : EPHAISTOS (LLM)\n") f.write("=" * 70 + "\n\n") f.write("Tests des requetes RULE_MISS via RAG + EPHAISTOS (Qwen 7B)\n\n") for r in results: f.write(f"[{r['status']:<10}] {r['desc']}\n") f.write(f" Requete : \"{r['query']}\"\n") f.write(f" Attendu : {r['expected_tool']}\n") f.write(f" Obtenu : {r['result_tool']}\n") if r['result_args']: f.write(f" Args : {r['result_args']}\n") if r['missing_mand']: f.write(f" MANQUANT : {r['missing_mand']}\n") f.write("\n") f.write("=" * 70 + "\n") f.write(f"LLM_PASS: {total_pass} LLM_PARTIAL: {total_partial} LLM_FAIL: {total_fail}\n") f.write(f"SCORE LLM: {score_final}%\n") print(f"Rapport sauvegarde : {report_path}") if erreurs_techniques: print(f"\n{R}{len(erreurs_techniques)} panne(s) technique(s) — " f"ces cas ne mesurent PAS le modele :{RESET}") for ligne in erreurs_techniques[:5]: print(f" {ligne}") return results, categories, score_final, erreurs_techniques def _config_derivee(ephaistos: str | None, lyra: str | None) -> str | None: """Ecrit une copie de config.yaml avec d'autres modeles. config.yaml n'est jamais modifie : c'est la configuration de production. """ if not ephaistos and not lyra: return None import tempfile import yaml as _yaml racine = Path(__file__).parent.parent cfg = _yaml.safe_load((racine / "config.yaml").read_text()) or {} cfg.setdefault("models", {}) if ephaistos: cfg["models"].setdefault("ephaistos", {})["name"] = ephaistos if lyra: cfg["models"].setdefault("lyra", {})["name"] = lyra tmp = tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) _yaml.safe_dump(cfg, tmp, allow_unicode=True) tmp.close() return tmp.name def main() -> None: import argparse from collections import Counter parser = argparse.ArgumentParser( description="Banc des requetes RULE_MISS : celles ou EPHAISTOS travaille vraiment.") parser.add_argument("--ephaistos", default=None, help="Modele EPHAISTOS a mesurer (ex: qwen2.5-coder:7b)") parser.add_argument("--lyra", default=None, help="Modele LYRA a mesurer") parser.add_argument("--json", action="store_true", help="Publie le resultat dans benchmarks/results/") parser.add_argument("--jeu", default="modeles", choices=sorted(JEUX), help="jeu de cas : modeles (21, proches des regles) ou hors_regles (inedites)") parser.add_argument("--reel", action="store_true", help="passer par EnhancedPipeline.process (le chemin du demon) au lieu d'appeler EPHAISTOS directement") args = parser.parse_args() chemin = _config_derivee(args.ephaistos, args.lyra) if chemin: print(f"Configuration derivee : ephaistos={args.ephaistos or '(inchange)'}") debut = time.time() results, categories, score_final, pannes = run_llm_tests(chemin, jeu=args.jeu, reel=args.reel) duree = time.time() - debut if args.json and pannes: print(f"\nPublication refusee : {len(pannes)} panne(s) technique(s). " "Un banc en panne ne publie pas de chiffre.") sys.exit(2) if args.json: sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) from bench_common import ecrire_resultat, modeles # noqa: E402 actifs = modeles() if args.ephaistos: actifs["ephaistos"] = args.ephaistos if args.lyra: actifs["lyra"] = args.lyra statuts = Counter(r["status"] for r in results) par_cat = {} for r in results: par_cat.setdefault(r["cat"].split("/")[0], Counter())[r["status"]] += 1 total = len(results) slug = actifs.get("ephaistos", "defaut").replace(":", "-") # Une mesure prise avec des variantes LYRA_EXP ne remplace pas la # mesure de reference du meme modele : fichier distinct, variantes # enregistrees dans le resultat. from lyra.models import ephaistos_exp as _exp variantes = sorted(_exp.actives()) if variantes: slug += "-exp" if args.jeu != "modeles": slug += f"-{args.jeu.replace('_', '')}" chemin_sortie = ecrire_resultat("modeles", { "jeu": args.jeu, "variantes": variantes, "banc": "RULE_MISS via EPHAISTOS (les regles ne couvrent pas ces requetes)", "cas": total, "statuts": dict(statuts), "taux_pass": round(statuts["LLM_PASS"] / total, 4) if total else 0.0, "score_pondere": score_final, "par_categorie": {c: dict(v) for c, v in sorted(par_cat.items())}, "modeles_mesures": actifs, "duree_s": round(duree, 1), "duree_inclut_initialisation": True, }, suffixe=slug) print(f"Resume publie : {chemin_sortie}") if __name__ == "__main__": main()