backend/tests/test_lighteval.py

#!/usr/bin/env python3
"""
Script minimal pour tester directement lighteval avec la tâche yourbench
"""
import os
import sys
import subprocess
import json
import time
from pathlib import Path
import logging

# Assurez-vous que l'environnement est correctement configuré
from dotenv import load_dotenv
load_dotenv()

# Importer le module de définition de tâche yourbench
sys.path.append(os.getcwd())
from tasks.yourbench_lighteval_task import create_yourbench_task

def run_lighteval_test():
    """
    Exécuter un test minimal avec lighteval
    """
    # Parameters
    dataset_name = "yourbench_a"
    organization = "yourbench"
    model_name = "Qwen/Qwen2.5-72B-Instruct"
    provider = "novita"
    output_dir = f"uploaded_files/test_{provider}/lighteval_results"
    
    # Créer le répertoire de sortie
    os.makedirs(output_dir, exist_ok=True)
    
    # Définir le chemin d'accès complet au dataset
    dataset_path = f"{organization}/{dataset_name}"
    print(f"Dataset à évaluer: {dataset_path}")
    
    # Créer un fichier temporaire
    import tempfile
    temp_file_path = tempfile.mktemp(suffix=".py")
    print(f"Création du fichier temporaire: {temp_file_path}")
    
    with open(temp_file_path, 'w') as temp_file:
        # Écrire le contenu du fichier
        temp_file.write(f"""
import os
import sys
import logging
sys.path.append("{os.getcwd()}")

from tasks.yourbench_lighteval_task import create_yourbench_task

# Configurer le logging
logging.basicConfig(level=logging.INFO)

# Créer la tâche yourbench
yourbench = create_yourbench_task("{dataset_path}", "lighteval")

# Définir la variable TASKS_TABLE dont lighteval a besoin
TASKS_TABLE = [yourbench]
""")
    
    # Construire la commande lighteval
    cmd = [
        "lighteval",
        "endpoint", 
        "inference-providers",
        f"model={model_name},provider={provider}",
        "custom|yourbench|0|0",
        "--custom-tasks",
        temp_file_path,
        "--max-samples", "5",  # Seulement 1 échantillon
        "--output-dir", output_dir,
        "--save-details",
        "--no-push-to-hub"  # Pas de push pour gagner du temps
    ]
    
    # Afficher la commande
    print(f"Exécution de la commande: {' '.join(cmd)}")
    print(f"Heure de début: {time.strftime('%H:%M:%S')}")
    
    # Exécuter la commande
    try:
        # Exécuter avec capture des sorties
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        # Afficher les résultats
        print(f"Code de retour: {result.returncode}")
        print("--- SORTIE STANDARD ---")
        print(result.stdout)
        print("--- ERREUR STANDARD ---")
        print(result.stderr)
        
        # Vérifier si des résultats ont été générés
        results_dir = Path(output_dir) / "results"
        if results_dir.exists():
            print(f"Dossier de résultats créé: {results_dir}")
            # Lister les fichiers de résultats
            result_files = list(results_dir.glob("**/*.json"))
            if result_files:
                print(f"Fichiers de résultats trouvés: {result_files}")
                # Trier les fichiers par date de modification pour prendre le plus récent
                result_files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
                latest_result = result_files[0]
                print(f"Fichier de résultats le plus récent: {latest_result}")
                
                # Lire le fichier de résultats
                with open(latest_result, 'r') as f:
                    results = json.load(f)
                    print("Contenu du fichier de résultats:")
                    print(json.dumps(results, indent=2))
                    
                    # Analyse des résultats
                    print("\n==== ANALYSE DES RÉSULTATS ====")
                    if "results" in results:
                        for task_name, task_results in results["results"].items():
                            print(f"Tâche: {task_name}")
                            for metric_name, metric_value in task_results.items():
                                print(f"  {metric_name}: {metric_value}")
                    else:
                        print("Aucun résultat trouvé dans le fichier JSON")
                
                # Vérifier les détails
                details_dir = Path(output_dir) / "details"
                if details_dir.exists():
                    print(f"\nDossier de détails trouvé: {details_dir}")
                    model_details_dirs = list(details_dir.glob("**/*"))
                    if model_details_dirs:
                        print(f"Dossiers de détails par modèle: {model_details_dirs}")
            else:
                print("Aucun fichier de résultats trouvé.")
        else:
            print(f"Aucun dossier de résultats créé.")
    
    except subprocess.CalledProcessError as e:
        print(f"Erreur lors de l'exécution de la commande: {e}")
    except Exception as e:
        print(f"Exception: {e}")
    finally:
        # Supprimer le fichier temporaire
        try:
            os.unlink(temp_file_path)
            print(f"Fichier temporaire supprimé: {temp_file_path}")
        except:
            pass
    
    print(f"Heure de fin: {time.strftime('%H:%M:%S')}")

if __name__ == "__main__":
    run_lighteval_test()