#!/usr/bin/env python3 """Harnais d'évaluation — jeu de tests de non-régression pour un agent LLM. Principe : un jeu de cas versionné (dataset.jsonl), deux modes de notation * regex : la réponse doit contenir une valeur vérifiable (déterministe, sans juge) * judge : un LLM note la réponse selon une grille écrite à l'avance (LLM-as-judge) On en sort un taux de réussite, une latence et un coût, comparables d'une version de prompt à l'autre. Le code de sortie vaut 1 sous le seuil : c'est ce qui bloque une CI. Trois pièges rencontrés en vrai, traités ici : * le fournisseur filtre le User-Agent de python-urllib (403 / error 1010) → on envoie un UA curl ; * un « je ne sais pas » n'est pas une réussite : une regex non trouvée est un échec ; * un juge qui n'a pas de grille explicite note au hasard → la grille est dans le cas, et le juge doit répondre par un JSON strict {score: 0|1, raison}. Usage : harness.py # exécute tout harness.py --ci # même chose + code de sortie 1 si le taux est sous le seuil harness.py --seuil 0.9 # seuil de non-régression (défaut 0.9) harness.py --limite 4 # n'exécuter que les 4 premiers cas (mise au point) """ import argparse import json import os import re import sys import time import urllib.error import urllib.request import uuid from pathlib import Path RACINE = Path(__file__).resolve().parent BASE_URL = os.environ.get("EVAL_BASE_URL", "https://opencode.ai/zen/go/v1") MODELE = os.environ.get("EVAL_MODEL", "deepseek-v4.1-flash") UA = "curl/8.5.0" # le WAF du fournisseur refuse le UA par défaut d'urllib SESSION = str(uuid.uuid4()) # opencode-go exige x-opencode-session, sinon HTTP 400 MissingSessionID SYSTEME_BASE = ( "Tu es un assistant technique francophone, factuel et concis. " "Tu réponds en 2 à 5 phrases maximum, avec les chiffres exacts quand ils existent. " "Si tu ne sais pas, tu dis que tu ne sais pas — tu n'inventes jamais un chiffre." ) SYSTEME = SYSTEME_BASE # complété par le contexte de référence au démarrage def cle_api(): """La clé n'est jamais écrite dans le code ni affichée.""" k = os.environ.get("OPENCODE_GO_API_KEY") or os.environ.get("EVAL_API_KEY") if k: return k env = Path.home() / ".hermes" / ".env" if env.exists(): for ligne in env.read_text(encoding="utf-8", errors="ignore").splitlines(): if ligne.startswith("OPENCODE_GO_API_KEY="): return ligne.split("=", 1)[1].strip().strip('"').strip("'") return "" def appelle(prompt, cle, max_tokens=800): corps = json.dumps({ "model": MODELE, "messages": [{"role": "system", "content": SYSTEME}, {"role": "user", "content": prompt}], "temperature": 0, "max_tokens": max_tokens, }).encode() req = urllib.request.Request(f"{BASE_URL}/chat/completions", data=corps, headers={"Content-Type": "application/json", "Authorization": f"Bearer {cle}", "User-Agent": UA, "x-opencode-session": SESSION}) debut = time.time() with urllib.request.urlopen(req, timeout=90) as r: d = json.loads(r.read().decode()) duree = time.time() - debut txt = d["choices"][0]["message"]["content"] usage = d.get("usage") or {} return txt, duree, usage def note_regex(cas, reponse): m = re.search(cas["expected"], reponse, re.I) return (1 if m else 0), ("valeur attendue trouvée" if m else f"valeur attendue absente (motif {cas['expected']})") def note_juge(cas, reponse, cle): prompt = ( "Tu es un correcteur. Tu notes une réponse UNIQUEMENT selon la grille fournie.\n\n" f"QUESTION POSÉE : {cas['question']}\n\n" f"GRILLE DE NOTATION : {cas['criteria']}\n\n" f"RÉPONSE À NOTER : {reponse}\n\n" "Réponds par un JSON strict, sans texte autour : {\"score\": 0 ou 1, \"raison\": \"...\"}" ) for tentative in (1, 2): try: txt, _, _ = appelle(prompt, cle, max_tokens=400) m = re.search(r"\{.*\}", txt, re.S) d = json.loads(m.group(0)) if m else {} raison = str(d.get("raison", "")).strip() if "score" in d and raison: return (1 if int(d["score"]) == 1 else 0), raison[:300], "ok" except Exception as e: raison = f"juge indisponible : {type(e).__name__}" if tentative == 1: continue # verdict inexploitable : on ne compte ni réussite ni échec du modèle return 0, (raison if 'raison' in dir() else "verdict du juge illisible"), "erreur_juge" def charge_contexte(chemin=None): """Le contexte de référence est la seule source autorisée : on évalue l'ancrage (la réponse doit venir du contexte) et l'abstention (ce qui n'y est pas doit être déclaré inconnu).""" f = Path(chemin) if chemin else (RACINE / "contexte.md") if not f.exists(): return "" return ("\n\n=== CONTEXTE DE RÉFÉRENCE (seule source autorisée) ===\n" + f.read_text(encoding="utf-8") + "\n=== FIN DU CONTEXTE ===\n" "Toute information absente de ce contexte doit être déclarée inconnue : " "tu ne complètes jamais avec tes connaissances générales.") def charge_dataset(limite=None): cas = [] with open(RACINE / "dataset.jsonl", encoding="utf-8") as f: for ligne in f: ligne = ligne.strip() if ligne: cas.append(json.loads(ligne)) return cas[:limite] if limite else cas def main(): ap = argparse.ArgumentParser() ap.add_argument("--ci", action="store_true", help="code de sortie 1 sous le seuil") ap.add_argument("--seuil", type=float, default=0.9) ap.add_argument("--limite", type=int, default=None) ap.add_argument("--sortie", default=str(RACINE / "rapport.jsonl")) ap.add_argument("--contexte", default=None, help="fichier de contexte de référence (défaut contexte.md)") a = ap.parse_args() global SYSTEME ctx = charge_contexte(a.contexte) SYSTEME = SYSTEME_BASE + ctx print(f" contexte de référence : {len(ctx)} caractères") cle = cle_api() if not cle: print("ERREUR : aucune clé API (OPENCODE_GO_API_KEY absent de l'environnement et de ~/.hermes/.env)") return 2 cas = charge_dataset(a.limite) print(f"═══ ÉVALUATION — {len(cas)} cas · modèle {MODELE}") resultats, tokens = [], 0 for c in cas: try: reponse, duree, usage = appelle(c["question"], cle) tokens += (usage.get("total_tokens") or 0) statut = "ok" # valeur par défaut : une erreur de l'outil n'est pas un échec du modèle if c["type"] == "regex": score, raison = note_regex(c, reponse) elif c["type"] == "abstention": mauvaise = re.search(c.get("piege", r"$^"), reponse, re.I) bonne = re.search(c["expected"], reponse, re.I) if mauvaise: score, raison = 0, "a inventé une réponse au lieu de s'abstenir" elif bonne: score, raison = 1, "s'est abstenu correctement" else: score, raison = 0, "ni abstention ni valeur piège détectée" else: score, raison, statut = note_juge(c, reponse, cle) etat = "⚠️" if statut != "ok" else ("✅" if score else "❌") print(f" {etat} {c['id']} · {duree:5.1f}s · {c['type']:5} · {raison[:80]}") resultats.append({"id": c["id"], "type": c["type"], "score": score, "raison": raison, "statut": statut, "duree_s": round(duree, 2), "reponse": reponse[:1200]}) except urllib.error.HTTPError as e: print(f" ❌ {c['id']} · HTTP {e.code} — {e.read()[:160].decode(errors='ignore')}") resultats.append({"id": c["id"], "type": c["type"], "score": 0, "raison": f"HTTP {e.code}", "duree_s": 0, "reponse": ""}) except Exception as e: print(f" ❌ {c['id']} · {type(e).__name__} : {e}") resultats.append({"id": c["id"], "type": c["type"], "score": 0, "raison": f"{type(e).__name__}", "duree_s": 0, "reponse": ""}) evalues = [r for r in resultats if r.get("statut", "ok") == "ok"] non_evalues = [r for r in resultats if r.get("statut", "ok") != "ok"] reussis = sum(r["score"] for r in evalues) taux = reussis / len(evalues) if evalues else 0 par_type = {} for r in resultats: t = par_type.setdefault(r["type"], [0, 0]) t[0] += r["score"]; t[1] += 1 with open(a.sortie, "w", encoding="utf-8") as f: for r in resultats: f.write(json.dumps(r, ensure_ascii=False) + "\n") print() print(f" RÉUSSITE : {reussis}/{len(evalues)} cas évalués = {100*taux:.1f} % (seuil {100*a.seuil:.0f} %)") if non_evalues: print(f" ⚠️ {len(non_evalues)} cas NON ÉVALUÉ(S) (outil de notation en défaut, exclus du taux) : " f"{', '.join(r['id'] for r in non_evalues)}") for t, (ok, n) in sorted(par_type.items()): print(f" {t:6} : {ok}/{n}") print(f" jetons consommés : {tokens}") print(f" rapport : {a.sortie}") echecs = [r["id"] for r in evalues if not r["score"]] if echecs: print(f" cas en échec : {', '.join(echecs)}") if taux < a.seuil: print(f"\n ⛔ SOUS LE SEUIL : {100*taux:.1f} % < {100*a.seuil:.0f} %" + (" — la CI échoue." if a.ci else " (relance avec --ci pour faire échouer la CI).")) return 1 if a.ci else 0 print(f"\n ✅ {100*taux:.1f} % ≥ {100*a.seuil:.0f} % : pas de régression détectée") return 0 if __name__ == "__main__": sys.exit(main())