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