tests/scripts/test_qwen_mcp_capabilities.py

"""
Test script to evaluate Qwen2.5 0.5B capabilities with MCP for RTS game
Tests basic MCP understanding and complex planning capabilities
"""

import json
import time
from typing import Dict, Any, List
from ai_analysis import query_llm

# Test prompts for different capability levels
TEST_PROMPTS = {
    "basic_mcp": [
        {
            "name": "Simple Tool Call",
            "prompt": """
You are an AI assistant controlling an RTS game through MCP (Model Context Protocol).
Available tools:
- get_game_state(): Get current game state
- move_units(unit_ids, target_x, target_y): Move units to position
- attack_unit(attacker_ids, target_id): Attack enemy unit

User command: "Show me the current game state"

Respond with a JSON object containing the tool call to execute.
""",
            "expected": {"tool": "get_game_state", "args": {}}
        },
        {
            "name": "Basic Action Translation",
            "prompt": """
You are an AI assistant controlling an RTS game through MCP.
Available tools:
- move_units(unit_ids, target_x, target_y): Move units to position
- attack_unit(attacker_ids, target_id): Attack enemy unit

User command: "Move my infantry to position 100, 200"

Respond with a JSON object containing the tool call to execute.
""",
            "expected": {"tool": "move_units", "args": {"unit_ids": ["infantry"], "target_x": 100, "target_y": 200}}
        }
    ],
    "complex_planning": [
        {
            "name": "Multi-step Strategy",
            "prompt": """
You are an AI assistant controlling an RTS game through MCP.
Available tools:
- get_game_state(): Get current game state
- move_units(unit_ids, target_x, target_y): Move units
- attack_unit(attacker_ids, target_id): Attack enemy
- build_building(building_type, position_x, position_y, player_id): Build building

User command: "I want to build a base near the ore field and defend it with turrets"

Break this down into a sequence of MCP tool calls. Respond with a JSON array of tool calls.
""",
            "expected": {"type": "sequence", "steps": ["get_game_state", "build_building", "build_building"]}
        },
        {
            "name": "Strategic Analysis",
            "prompt": """
You are an AI assistant controlling an RTS game through MCP.
Available tools:
- get_game_state(): Get current game state
- get_ai_analysis(language): Get tactical analysis

User command: "Analyze the battlefield and suggest the best strategy"

Respond with a JSON object containing the tool calls needed.
""",
            "expected": {"type": "analysis", "steps": ["get_game_state", "get_ai_analysis"]}
        }
    ],
    "advanced_mcp": [
        {
            "name": "Parameter Extraction",
            "prompt": """
You are an AI assistant controlling an RTS game through MCP.
Available tools:
- move_units(unit_ids, target_x, target_y): Move units

User command: "Move tanks 1, 3, and 7 to coordinates 150, 75"

Extract the parameters and respond with a JSON tool call.
""",
            "expected": {"tool": "move_units", "args": {"unit_ids": [1, 3, 7], "target_x": 150, "target_y": 75}}
        },
        {
            "name": "Error Handling",
            "prompt": """
You are an AI assistant controlling an RTS game through MCP.
Available tools:
- move_units(unit_ids, target_x, target_y): Move units

User command: "Move my units to the enemy base"

Since you don't know the exact coordinates, how would you handle this?
Respond with a JSON object showing your approach.
""",
            "expected": {"type": "needs_clarification", "message": "Need coordinates for enemy base"}
        }
    ]
}

def test_qwen_capabilities():
    """Run comprehensive tests on Qwen2.5 0.5B MCP capabilities"""
    
    print("=== Testing Qwen2.5 0.5B MCP Capabilities ===\n")
    
    results = {}
    
    for category, tests in TEST_PROMPTS.items():
        print(f"\n📊 Testing {category.replace('_', ' ').title()}:")
        print("-" * 50)
        
        category_results = []
        
        for test in tests:
            print(f"\n🧪 Test: {test['name']}")
            print(f"Prompt: {test['prompt'][:100]}...")
            
            try:
                # Query the LLM
                start_time = time.time()
                response = query_llm(
                    prompt=test['prompt'],
                    max_tokens=500,
                    temperature=0.1,
                    system_message="You are an AI assistant that responds with JSON objects for MCP tool calls."
                )
                response_time = time.time() - start_time
                
                # Parse and analyze response
                analysis = analyze_response(test, response, response_time)
                category_results.append(analysis)
                
                print(f"✅ Response time: {response_time:.2f}s")
                print(f"📝 Response: {response[:200]}...")
                print(f"📊 Analysis: {analysis['score']}/10")
                
            except Exception as e:
                print(f"❌ Error: {e}")
                category_results.append({
                    'test': test['name'],
                    'error': str(e),
                    'score': 0
                })
        
        results[category] = category_results
    
    # Generate summary report
    generate_summary_report(results)
    
    return results

def analyze_response(test: Dict[str, Any], response: str, response_time: float) -> Dict[str, Any]:
    """Analyze the LLM response and score its performance"""
    
    analysis = {
        'test': test['name'],
        'response': response,
        'response_time': response_time,
        'score': 0,
        'strengths': [],
        'weaknesses': [],
        'details': {}
    }
    
    # Basic response quality checks
    if not response or response.strip() == "":
        analysis['weaknesses'].append("Empty response")
        return analysis
    
    # Check for JSON structure
    try:
        # Try to parse as JSON
        parsed = json.loads(response)
        analysis['details']['json_valid'] = True
        analysis['strengths'].append("Valid JSON structure")
        
        # Check if it matches expected structure
        if 'expected' in test:
            expected = test['expected']
            if isinstance(expected, dict):
                if 'tool' in expected and 'tool' in parsed:
                    if parsed['tool'] == expected['tool']:
                        analysis['score'] += 4
                        analysis['strengths'].append("Correct tool selection")
                    else:
                        analysis['weaknesses'].append(f"Wrong tool: {parsed.get('tool')} vs {expected['tool']}")
                
                # Check arguments
                if 'args' in expected and 'args' in parsed:
                    arg_match = compare_arguments(parsed['args'], expected['args'])
                    analysis['score'] += arg_match * 3
                    if arg_match > 0.7:
                        analysis['strengths'].append("Good argument matching")
                    else:
                        analysis['weaknesses'].append("Poor argument matching")
        
    except json.JSONDecodeError:
        analysis['details']['json_valid'] = False
        analysis['weaknesses'].append("Invalid JSON format")
        
        # Check for tool-like patterns in text
        if 'get_game_state' in response:
            analysis['score'] += 2
            analysis['strengths'].append("Mentions correct tool")
        if 'move_units' in response or 'attack_unit' in response:
            analysis['score'] += 1
    
    # Response time scoring
    if response_time < 5.0:
        analysis['score'] += 1
        analysis['strengths'].append("Fast response")
    elif response_time > 15.0:
        analysis['weaknesses'].append("Slow response")
    
    # Content relevance scoring
    if any(keyword in response.lower() for keyword in ['game', 'state', 'move', 'attack', 'build']):
        analysis['score'] += 1
        analysis['strengths'].append("Relevant content")
    
    # Cap for score
    analysis['score'] = min(analysis['score'], 10)
    
    return analysis

def compare_arguments(actual: Dict, expected: Dict) -> float:
    """Compare argument dictionaries and return match percentage"""
    if not actual or not expected:
        return 0.0
    
    matches = 0
    total = len(expected)
    
    for key, expected_value in expected.items():
        if key in actual:
            actual_value = actual[key]
            if isinstance(expected_value, list) and isinstance(actual_value, list):
                # Compare lists
                if set(expected_value) == set(actual_value):
                    matches += 1
            elif expected_value == actual_value:
                matches += 1
    
    return matches / total if total > 0 else 0.0

def generate_summary_report(results: Dict[str, List[Dict]]):
    """Generate a comprehensive summary report"""
    
    print("\n" + "="*60)
    print("📊 QWEN2.5 0.5B MCP CAPABILITY ASSESSMENT REPORT")
    print("="*60)
    
    overall_scores = []
    
    for category, category_results in results.items():
        if not category_results:
            continue
            
        category_scores = [r.get('score', 0) for r in category_results if 'score' in r]
        avg_score = sum(category_scores) / len(category_scores) if category_scores else 0
        overall_scores.append(avg_score)
        
        print(f"\n🔍 {category.replace('_', ' ').title()}:")
        print(f"   Average Score: {avg_score:.1f}/10")
        
        for result in category_results:
            if 'error' in result:
                print(f"   ❌ {result['test']}: ERROR - {result['error']}")
            else:
                print(f"   {'✅' if result['score'] >= 6 else '⚠️'} {result['test']}: {result['score']}/10")
                if result['strengths']:
                    print(f"     Strengths: {', '.join(result['strengths'][:2])}")
                if result['weaknesses']:
                    print(f"     Weaknesses: {', '.join(result['weaknesses'][:2])}")
    
    # Overall assessment
    if overall_scores:
        overall_avg = sum(overall_scores) / len(overall_scores)
        print(f"\n🎯 OVERALL ASSESSMENT: {overall_avg:.1f}/10")
        
        if overall_avg >= 8:
            print("💪 EXCELLENT - Qwen2.5 0.5B is highly capable for MCP tasks")
        elif overall_avg >= 6:
            print("👍 GOOD - Qwen2.5 0.5B is capable with some limitations")
        elif overall_avg >= 4:
            print("⚠️  MODERATE - Qwen2.5 0.5B has significant limitations")
        else:
            print("❌ POOR - Qwen2.5 0.5B is not suitable for MCP tasks")
        
        # Recommendations
        print(f"\n💡 RECOMMENDATIONS:")
        if overall_avg >= 7:
            print("- Use Qwen2.5 0.5B for MCP translation with confidence")
            print("- Implement prompt engineering for complex tasks")
            print("- Add validation layer for safety")
        elif overall_avg >= 5:
            print("- Use Qwen2.5 0.5B for simple MCP tasks")
            print("- Implement strong validation and fallback mechanisms")
            print("- Consider using larger models for complex planning")
        else:
            print("- Consider upgrading to a larger model (1.5B+)")
            print("- Use Qwen2.5 0.5B only for very simple translations")
            print("- Implement extensive error handling")

if __name__ == "__main__":
    print("Starting Qwen2.5 0.5B MCP capability assessment...")
    print("This will test the model's ability to translate user commands to MCP tool calls.")
    print("Make sure the model is downloaded and available at: qwen2.5-0.5b-instruct-q4_0.gguf")
    
    try:
        results = test_qwen_capabilities()
        
        # Save detailed results to file
        with open("/home/luigi/rts/web/qwen_mcp_assessment.json", "w") as f:
            json.dump(results, f, indent=2)
        
        print("\n📄 Detailed results saved to: qwen_mcp_assessment.json")
        
    except Exception as e:
        print(f"❌ Assessment failed: {e}")
        print("Make sure the AI model is properly downloaded and configured.")