tests/scripts/qwen_mcp_capability_assessment.py

"""
Évaluation complète des capacités MCP de Qwen2.5 0.5B
Teste la compréhension des outils MCP, la traduction d'instructions et la planification
"""

import json
import time
import sys
import os

# Ajouter le chemin pour les imports
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

def test_model_availability():
    """Vérifier si le modèle est disponible et fonctionnel"""
    print("🔍 Vérification de la disponibilité du modèle...")
    
    # Vérifier si le fichier modèle existe
    model_path = "qwen2.5-0.5b-instruct-q4_0.gguf"
    if os.path.exists(model_path):
        print(f"✅ Modèle trouvé: {model_path}")
        file_size = os.path.getsize(model_path) / (1024 * 1024)  # MB
        print(f"📏 Taille du modèle: {file_size:.1f} MB")
        return True
    else:
        print(f"❌ Modèle non trouvé: {model_path}")
        print("   Le modèle doit être téléchargé pour les tests")
        return False

def test_basic_mcp_comprehension():
    """Test de compréhension MCP de base"""
    
    print("\n🧪 TEST 1: Compréhension des outils MCP")
    print("-" * 50)
    
    tests = [
        {
            "name": "Outils simples",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils disponibles:
- get_game_state(): Obtenir l'état du jeu
- move_units(unit_ids, target_x, target_y): Déplacer des unités

Commande utilisateur: "Montre-moi l'état du jeu"

Réponds avec un objet JSON contenant l'appel d'outil à exécuter.
""",
            "expected_tool": "get_game_state"
        },
        {
            "name": "Extraction de paramètres",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils disponibles:
- move_units(unit_ids, target_x, target_y): Déplacer des unités

Commande utilisateur: "Déplace mes tanks vers la position 150, 75"

Réponds avec un objet JSON contenant l'appel d'outil avec les paramètres extraits.
""",
            "expected_tool": "move_units",
            "expected_params": ["unit_ids", "target_x", "target_y"]
        }
    ]
    
    return tests

def test_complex_planning():
    """Test de planification complexe"""
    
    print("\n🧪 TEST 2: Planification et stratégie")
    print("-" * 50)
    
    tests = [
        {
            "name": "Stratégie multi-étapes",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils disponibles:
- get_game_state(): Obtenir l'état du jeu
- move_units(): Déplacer des unités
- attack_unit(): Attaquer une unité ennemie
- build_building(): Construire un bâtiment

Commande utilisateur: "Construis une base près du minerai et défends-la"

Décompose cette stratégie en une séquence d'appels d'outils MCP.
Réponds avec un tableau JSON d'actions.
""",
            "expected_sequence": True
        },
        {
            "name": "Analyse tactique",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils disponibles:
- get_game_state(): Obtenir l'état du jeu
- get_ai_analysis(language): Obtenir une analyse tactique

Commande utilisateur: "Analyse le champ de bataille et suggère une stratégie"

Quels outils MCP utiliserais-tu et dans quel ordre?
""",
            "expected_analysis": True
        }
    ]
    
    return tests

def test_error_handling():
    """Test de gestion des erreurs et ambiguïtés"""
    
    print("\n🧪 TEST 3: Gestion des erreurs")
    print("-" * 50)
    
    tests = [
        {
            "name": "Instructions ambiguës",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils disponibles:
- move_units(unit_ids, target_x, target_y): Déplacer des unités

Commande utilisateur: "Déplace mes unités vers la base ennemie"

Comment gérerais-tu cette instruction ambiguë?
""",
            "expected_handling": "clarification"
        },
        {
            "name": "Paramètres manquants",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils disponibles:
- attack_unit(attacker_ids, target_id): Attaquer une unité ennemie

Commande utilisateur: "Attaque l'ennemi"

Que faire si les paramètres nécessaires ne sont pas fournis?
""",
            "expected_handling": "parameter_request"
        }
    ]
    
    return tests

def run_llm_test(prompt, test_name):
    """Exécuter un test avec le modèle LLM"""
    
    try:
        from ai_analysis import query_llm
        
        print(f"\n📋 Test: {test_name}")
        print(f"Prompt: {prompt[:100]}...")
        
        start_time = time.time()
        
        response = query_llm(
            prompt=prompt,
            max_tokens=300,
            temperature=0.1,
            system_message="Tu es un assistant IA spécialisé dans les jeux RTS et le protocole MCP."
        )
        
        response_time = time.time() - start_time
        
        print(f"⏱️  Temps de réponse: {response_time:.2f}s")
        print(f"📝 Réponse: {response[:200]}...")
        
        return {
            "success": True,
            "response": response,
            "response_time": response_time,
            "error": None
        }
        
    except Exception as e:
        print(f"❌ Erreur: {e}")
        return {
            "success": False,
            "response": None,
            "response_time": 0,
            "error": str(e)
        }

def analyze_response(response, test_config):
    """Analyser la réponse du modèle"""
    
    analysis = {
        "score": 0,
        "strengths": [],
        "weaknesses": [],
        "details": {}
    }
    
    if not response:
        analysis["weaknesses"].append("Réponse vide")
        return analysis
    
    # Vérifier la structure JSON
    try:
        json.loads(response)
        analysis["score"] += 3
        analysis["strengths"].append("JSON valide")
        analysis["details"]["json_valid"] = True
    except:
        analysis["details"]["json_valid"] = False
    
    # Vérifier la présence d'outils MCP
    mcp_tools = ["get_game_state", "move_units", "attack_unit", "build_building", "get_ai_analysis"]
    tools_found = [tool for tool in mcp_tools if tool in response]
    
    if tools_found:
        analysis["score"] += 2
        analysis["strengths"].append(f"Outils MCP identifiés: {', '.join(tools_found)}")
    
    # Vérifier la cohérence sémantique
    if "game" in response.lower() or "rts" in response.lower() or "mcp" in response.lower():
        analysis["score"] += 1
        analysis["strengths"].append("Contexte RTS/MCP compris")
    
    # Vérifier les paramètres
    if "unit_ids" in response or "target_x" in response or "target_y" in response:
        analysis["score"] += 1
        analysis["strengths"].append("Paramètres identifiés")
    
    analysis["score"] = min(analysis["score"], 10)
    
    return analysis

def generate_assessment_report(results):
    """Générer un rapport d'évaluation complet"""
    
    print("\n" + "="*70)
    print("📊 RAPPORT D'ÉVALUATION DES CAPACITÉS MCP")
    print("="*70)
    
    total_score = 0
    total_tests = 0
    
    for category, category_results in results.items():
        print(f"\n🔍 {category.upper()}:")
        
        category_score = 0
        category_count = 0
        
        for test_name, result in category_results.items():
            if result["success"]:
                analysis = result["analysis"]
                score = analysis["score"]
                category_score += score
                category_count += 1
                
                print(f"   {test_name}: {score}/10")
                if analysis["strengths"]:
                    print(f"     ✅ Points forts: {', '.join(analysis['strengths'][:2])}")
                if analysis["weaknesses"]:
                    print(f"     ⚠️  Points faibles: {', '.join(analysis['weaknesses'][:2])}")
            else:
                print(f"   {test_name}: ❌ ÉCHEC - {result['error']}")
        
        if category_count > 0:
            avg_score = category_score / category_count
            total_score += avg_score
            total_tests += 1
            print(f"   📊 Score moyen: {avg_score:.1f}/10")
    
    # Score global
    if total_tests > 0:
        overall_score = total_score / total_tests
        
        print(f"\n🎯 SCORE GLOBAL: {overall_score:.1f}/10")
        
        # Recommandations basées sur le score
        print("\n💡 RECOMMANDATIONS:")
        
        if overall_score >= 8:
            print("✅ EXCELLENT - Qwen2.5 0.5B est très capable pour les tâches MCP")
            print("   • Utilisez-le avec confiance pour la traduction MCP")
            print("   • Implémentez une validation légère pour la sécurité")
        elif overall_score >= 6:
            print("👍 BON - Qwen2.5 0.5B est capable avec quelques limitations")
            print("   • Utilisez-le pour les commandes simples")
            print("   • Ajoutez une validation robuste")
            print("   • Prévoyez des mécanismes de secours")
        elif overall_score >= 4:
            print("⚠️  MODÉRÉ - Limitations significatives")
            print("   • Utilisez-le uniquement pour des traductions très simples")
            print("   • Implémentez une validation stricte")
            print("   • Envisagez un modèle plus grand pour les tâches complexes")
        else:
            print("❌ FAIBLE - Pas adapté aux tâches MCP")
            print("   • Envisagez un modèle plus grand (1.5B+)")
            print("   • Utilisez des règles fixes pour les commandes simples")

def main():
    """Fonction principale"""
    
    print("🔍 ÉVALUATION DES CAPACITÉS MCP DE QWEN2.5 0.5B")
    print("=" * 70)
    
    # Vérifier la disponibilité du modèle
    if not test_model_availability():
        print("\n❌ Le modèle n'est pas disponible. Téléchargez-le d'abord.")
        return
    
    # Définir les tests
    basic_tests = test_basic_mcp_comprehension()
    planning_tests = test_complex_planning()
    error_tests = test_error_handling()
    
    # Exécuter les tests
    results = {}
    
    # Tests de base
    print("\n🚀 Exécution des tests de capacité MCP...")
    
    basic_results = {}
    for test in basic_tests:
        result = run_llm_test(test["prompt"], test["name"])
        if result["success"]:
            analysis = analyze_response(result["response"], test)
            result["analysis"] = analysis
        basic_results[test["name"]] = result
    
    results["compréhension_de_base"] = basic_results
    
    # Tests de planification
    planning_results = {}
    for test in planning_tests:
        result = run_llm_test(test["prompt"], test["name"])
        if result["success"]:
            analysis = analyze_response(result["response"], test)
            result["analysis"] = analysis
        planning_results[test["name"]] = result
    
    results["planification_complexe"] = planning_results
    
    # Tests de gestion d'erreurs
    error_results = {}
    for test in error_tests:
        result = run_llm_test(test["prompt"], test["name"])
        if result["success"]:
            analysis = analyze_response(result["response"], test)
            result["analysis"] = analysis
        error_results[test["name"]] = result
    
    results["gestion_des_erreurs"] = error_results
    
    # Générer le rapport
    generate_assessment_report(results)
    
    # Sauvegarder les résultats détaillés
    with open("qwen_mcp_capability_results.json", "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
    
    print(f"\n📄 Résultats détaillés sauvegardés dans: qwen_mcp_capability_results.json")

if __name__ == "__main__":
    main()