evaluate.py

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()