Spaces:
Sleeping
Sleeping
File size: 3,793 Bytes
06b6d63 0033bf8 06b6d63 | 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 91 92 93 94 95 96 | import json
import requests
import time
from pathlib import Path
# Configuration de l'endpoint local de ton app.py
API_URL = "http://localhost:7860/query"
JSON_TEST_PATH = Path(__file__).parent / "test_qcm.json"
def load_test_dataset():
if not JSON_TEST_PATH.exists():
raise FileNotFoundError(f"Impossible de trouver le fichier {JSON_TEST_PATH}")
with open(JSON_TEST_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def run_evaluation():
try:
dataset = load_test_dataset()
except Exception as e:
print(f"❌ Erreur de chargement du dataset : {e}")
return
print("=" * 60)
print(f"🚀 LANCEMENT DE L'ÉVALUATION AUTOMATIQUE ({len(dataset)} QCM)")
print("=" * 60)
metrics = {
"correct": 0,
"total": len(dataset),
"total_tokens": 0,
"prompt_tokens": 0,
"completion_tokens": 0,
"total_co2_grams": 0.0,
"total_time_ms": 0.0
}
for idx, item in enumerate(dataset, start=1):
# Formatage de la question et des options pour l'envoi au RAG
options_text = "\n".join([f"{k}) {v}" for k, v in item["options"].items()])
query_text = f"{item['question']}\n\nOptions :\n{options_text}"
payload = {
"query": query_text,
"top_k": 3 # Configurable selon tes tests d'optimisation
}
try:
start_perf = time.perf_counter()
response = requests.post(API_URL, json=payload, timeout=60)
response.raise_for_status()
res_data = response.json()
elapsed_ms = round((time.perf_counter() - start_perf) * 1000, 2)
# Extraction de l'option prédite par le LLM
answer_llm = res_data.get("answer", "").strip().upper()
expected_answer = item["correct_answer"].upper()
# Vérification de la lettre (ex: si "B - Le modèle..." commence par "B")
is_correct = answer_llm.startswith(expected_answer)
if is_correct:
metrics["correct"] += 1
status = "✅ CORRECT"
else:
status = f"❌ FAUX (Attendu: {expected_answer} | Reçu: {answer_llm[:5]})"
# Accumulation des métriques de performance et de sobriété
metrics["total_tokens"] += res_data.get("total_token", 0)
metrics["prompt_tokens"] += res_data.get("prompt_tokens", 0)
metrics["completion_tokens"] += res_data.get("completion_tokens", 0)
metrics["total_time_ms"] += res_data.get("run_time_in_ms", elapsed_ms)
co2 = res_data.get("co2_grams")
if isinstance(co2, (int, float)):
metrics["total_co2_grams"] += co2
print(f"[{idx}/{len(dataset)}] Q{item['id']} : {status} | Temps: {elapsed_ms}ms | Tokens: {res_data.get('total_token', 0)}")
except Exception as e:
print(f"[{idx}/{len(dataset)}] Q{item['id']} : 💥 ERREUR API : {e}")
# Calcul des scores finaux
accuracy = (metrics["correct"] / metrics["total"]) * 100 if metrics["total"] > 0 else 0
avg_time = metrics["total_time_ms"] / metrics["total"] if metrics["total"] > 0 else 0
print("\n" + "=" * 60)
print("📊 BILAN GLOBAL DE PERFORMANCE DU RAG")
print("=" * 60)
print(f"🎯 ACCURACY GLOBALE : {accuracy:.2f}% ({metrics['correct']}/{metrics['total']})")
print(f"🪙 TOTAL TOKENS CONSOMMÉS : {metrics['total_tokens']} (Prompt: {metrics['prompt_tokens']} | Gen: {metrics['completion_tokens']})")
print(f"🍃 EMPREINTE CARBONE : {metrics['total_co2_grams']:.4f} g CO2")
print(f"⚡ TEMPS MOYEN PAR REQUÊTE : {avg_time:.2f} ms")
print("=" * 60)
if __name__ == "__main__":
run_evaluation() |