tests/scripts/direct_qwen_mcp_test.py

"""
Test direct des capacités MCP de Qwen2.5 0.5B
Évalue la compréhension des outils MCP et la traduction d'instructions
"""

import sys
import os
import json
import time

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

def test_qwen_mcp_capabilities():
    """Test direct des capacités MCP de Qwen2.5 0.5B"""
    
    print("🔍 ÉVALUATION DES CAPACITÉS MCP DE QWEN2.5 0.5B")
    print("=" * 70)
    
    # Vérifier que le modèle existe
    model_path = "qwen2.5-0.5b-instruct-q4_0.gguf"
    if not os.path.exists(model_path):
        print("❌ Modèle non trouvé. Téléchargez-le d'abord.")
        return
    
    print(f"✅ Modèle trouvé: {model_path}")
    print(f"📏 Taille: {os.path.getsize(model_path) / (1024*1024):.1f} MB\n")
    
    # Importer la classe AIAnalyzer
    try:
        from ai_analysis import AIAnalyzer
        
        # Initialiser l'analyseur
        print("🔄 Initialisation de l'analyseur AI...")
        analyzer = AIAnalyzer(model_path)
        
        if not analyzer.model_available:
            print("❌ Impossible de charger le modèle")
            return
        
        print("✅ Analyseur AI initialisé avec succès\n")
        
    except Exception as e:
        print(f"❌ Erreur lors de l'import: {e}")
        return
    
    # Tests de capacité MCP
    tests = [
        {
            "name": "Compréhension MCP de base",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP (Model Context Protocol).

Outils MCP disponibles:
- get_game_state(): Obtenir l'état actuel du jeu
- move_units(unit_ids, target_x, target_y): Déplacer des unités vers une position
- attack_unit(attacker_ids, target_id): Attaquer une unité ennemie

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

Réponds avec un objet JSON contenant l'appel d'outil MCP à exécuter.
""",
            "expected": "get_game_state"
        },
        {
            "name": "Traduction d'action simple",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

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

Commande utilisateur: "Déplace mon infanterie vers la position 150, 200"

Réponds avec un objet JSON contenant l'appel d'outil avec les paramètres extraits.
""",
            "expected": "move_units"
        },
        {
            "name": "Extraction de paramètres complexes",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

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

Commande utilisateur: "Déplace les tanks 1, 3 et 7 vers les coordonnées 120, 80"

Extrais les paramètres et réponds avec un objet JSON d'appel d'outil.
""",
            "expected": "move_units avec paramètres"
        },
        {
            "name": "Planification stratégique",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

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

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

Décompose cette stratégie en une séquence d'actions MCP.
""",
            "expected": "séquence d'actions"
        },
        {
            "name": "Gestion d'ambiguïté",
            "prompt": """
Tu es un assistant IA qui contrôle un jeu RTS via MCP.

Outils MCP 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": "demande de clarification"
        }
    ]
    
    results = []
    
    for i, test in enumerate(tests, 1):
        print(f"\n🧪 TEST {i}: {test['name']}")
        print("-" * 50)
        
        try:
            # Utiliser la méthode d'analyse de l'AIAnalyzer
            start_time = time.time()
            
            # Pour ce test, nous allons utiliser une approche directe
            # en appelant la fonction interne de l'analyseur
            
            # Créer un prompt formaté pour l'analyseur
            formatted_prompt = f"""
{test['prompt']}

Réponds uniquement avec un objet JSON valide.
"""
            
            print("📤 Envoi de la requête au modèle...")
            
            # Utiliser la méthode d'analyse de l'AIAnalyzer
            # Note: Nous devons adapter cela à l'interface existante
            
            # Pour l'instant, utilisons une approche simplifiée
            from ai_analysis import _llama_worker
            import multiprocessing as mp
            
            # Créer une queue pour les résultats
            result_queue = mp.Queue()
            
            # Lancer le worker
            worker = mp.Process(
                target=_llama_worker,
                args=(result_queue, model_path, formatted_prompt, [], 300, 0.1)
            )
            worker.start()
            worker.join(timeout=30)  # Timeout de 30 secondes
            
            if worker.is_alive():
                worker.terminate()
                response = "TIMEOUT"
            else:
                result = result_queue.get()
                if result['status'] == 'ok':
                    response_data = result.get('data', {})
                    if 'raw' in response_data:
                        response = response_data['raw']
                    else:
                        response = str(response_data)
                else:
                    response = f"ERROR: {result.get('message', 'Unknown error')}"
            
            response_time = time.time() - start_time
            
            print(f"⏱️  Temps de réponse: {response_time:.2f}s")
            print(f"📝 Réponse: {response[:200]}...")
            
            # Analyser la réponse
            score = analyze_response(response, test)
            
            results.append({
                'test': test['name'],
                'response': response,
                'response_time': response_time,
                'score': score,
                'success': score >= 5
            })
            
            print(f"📊 Score: {score}/10")
            
        except Exception as e:
            print(f"❌ Erreur: {e}")
            results.append({
                'test': test['name'],
                'error': str(e),
                'score': 0,
                'success': False
            })
    
    # Générer le rapport final
    generate_final_report(results)

def analyze_response(response, test_config):
    """Analyser et noter la réponse du modèle"""
    
    if not response or response == "TIMEOUT":
        return 0
    
    score = 0
    
    # Vérifier la structure JSON
    try:
        json_data = json.loads(response)
        score += 3  # JSON valide
        
        # Vérifier la présence d'outils MCP
        if 'tool' in json_data:
            score += 2
        if 'args' in json_data:
            score += 1
            
    except json.JSONDecodeError:
        # Vérifier les patterns dans le texte
        if 'get_game_state' in response:
            score += 2
        if 'move_units' in response:
            score += 2
        if any(param in response for param in ['unit_ids', 'target_x', 'target_y']):
            score += 1
    
    # Vérifier la cohérence sémantique
    if 'game' in response.lower() or 'rts' in response.lower():
        score += 1
    
    # Vérifier la pertinence par rapport au test
    expected = test_config['expected']
    if expected in response.lower():
        score += 1
    
    return min(score, 10)

def generate_final_report(results):
    """Générer un rapport final d'évaluation"""
    
    print("\n" + "="*70)
    print("📊 RAPPORT FINAL D'ÉVALUATION MCP")
    print("="*70)
    
    successful_tests = [r for r in results if r.get('success', False)]
    total_score = sum(r.get('score', 0) for r in results)
    avg_score = total_score / len(results) if results else 0
    
    print(f"\n📈 STATISTIQUES:")
    print(f"   Tests réussis: {len(successful_tests)}/{len(results)}")
    print(f"   Score moyen: {avg_score:.1f}/10")
    
    print(f"\n🔍 DÉTAILS PAR TEST:")
    for result in results:
        status = "✅" if result.get('success', False) else "❌"
        print(f"   {status} {result['test']}: {result.get('score', 0)}/10")
        if 'response_time' in result:
            print(f"     ⏱️  Temps: {result['response_time']:.2f}s")
    
    print(f"\n🎯 ÉVALUATION FINALE:")
    
    if avg_score >= 8:
        print("💪 EXCELLENT - Qwen2.5 0.5B est très capable pour les tâches MCP")
        print("   • Utilisation recommandée pour la traduction MCP")
        print("   • Bonne compréhension des outils et paramètres")
    elif avg_score >= 6:
        print("👍 BON - Capable avec quelques limitations")
        print("   • Adapté pour les commandes simples")
        print("   • Validation recommandée pour les actions complexes")
    elif avg_score >= 4:
        print("⚠️  MODÉRÉ - Limitations significatives")
        print("   • Utilisation limitée aux commandes très simples")
        print("   • Validation stricte nécessaire")
    else:
        print("❌ FAIBLE - Pas adapté aux tâches MCP")
        print("   • Envisagez un modèle plus grand")
        print("   • Utilisez des règles fixes à la place")
    
    print(f"\n💡 RECOMMANDATIONS:")
    print("1. Commencez par des commandes simples")
    print("2. Ajoutez une validation des actions")
    print("3. Utilisez des prompts structurés")
    print("4. Testez avec différents types de commandes")

if __name__ == "__main__":
    test_qwen_mcp_capabilities()