tests/test_by_classification.py

#!/usr/bin/env python3
"""
Enhanced GAIA Testing with Classification Filtering and Error Analysis
Test all questions by agent type with comprehensive error tracking and iterative improvement workflow.
"""

import json
import time
import argparse
import logging
import sys
from datetime import datetime
from typing import Dict, List, Optional
from collections import defaultdict
from pathlib import Path

# Add parent directory to path for imports
sys.path.append(str(Path(__file__).parent.parent))

from gaia_web_loader import GAIAQuestionLoaderWeb
from main import GAIASolver
from question_classifier import QuestionClassifier

class GAIAClassificationTester:
    """Enhanced GAIA testing with classification-based filtering and error analysis"""
    
    def __init__(self):
        self.loader = GAIAQuestionLoaderWeb()
        self.classifier = QuestionClassifier()
        self.solver = GAIASolver()
        self.results = []
        self.error_patterns = defaultdict(list)
        
        # Create logs directory if it doesn't exist
        Path("logs").mkdir(exist_ok=True)
        
        # Setup logging
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.log_file = f"logs/classification_test_{timestamp}.log"
        
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
        # Load validation answers after logger is set up
        self.validation_answers = self.load_validation_answers()
    
    def load_validation_answers(self):
        """Load correct answers from GAIA validation metadata"""
        answers = {}
        try:
            validation_path = Path(__file__).parent.parent / 'gaia_validation_metadata.jsonl'
            with open(validation_path, 'r') as f:
                for line in f:
                    if line.strip():
                        data = json.loads(line.strip())
                        task_id = data.get('task_id')
                        final_answer = data.get('Final answer')
                        if task_id and final_answer:
                            answers[task_id] = final_answer
            self.logger.info(f"📋 Loaded {len(answers)} validation answers")
        except Exception as e:
            self.logger.error(f"⚠️ Could not load validation data: {e}")
        return answers
    
    def validate_answer(self, task_id: str, our_answer: str):
        """Validate our answer against the correct answer with format normalization"""
        if task_id not in self.validation_answers:
            return {"status": "NO_VALIDATION", "expected": "N/A", "our": our_answer}
        
        expected = str(self.validation_answers[task_id]).strip()
        our_clean = str(our_answer).strip()
        
        # Exact match (case-insensitive)
        if our_clean.lower() == expected.lower():
            return {"status": "CORRECT", "expected": expected, "our": our_clean}
        
        # ENHANCED: Format normalization for comprehensive comparison
        def normalize_format(text):
            """Enhanced normalization for fair comparison"""
            import re
            text = str(text).lower().strip()
            
            # Remove currency symbols and normalize numbers
            text = re.sub(r'[$€£¥]', '', text)
            
            # Normalize spacing around commas and punctuation
            text = re.sub(r'\s*,\s*', ', ', text)  # "b,e" -> "b, e"
            text = re.sub(r'\s*;\s*', '; ', text)  # "a;b" -> "a; b"
            text = re.sub(r'\s*:\s*', ': ', text)  # "a:b" -> "a: b"
            
            # Remove extra whitespace
            text = re.sub(r'\s+', ' ', text).strip()
            
            # Normalize decimal places and numbers
            text = re.sub(r'(\d+)\.0+$', r'\1', text)  # "89706.00" -> "89706"
            text = re.sub(r'(\d+),(\d{3})', r'\1\2', text)  # "89,706" -> "89706"
            
            # Remove common formatting artifacts
            text = re.sub(r'["""''`]', '"', text)  # Normalize quotes
            text = re.sub(r'[–—]', '-', text)      # Normalize dashes
            text = re.sub(r'[^\w\s,.-]', '', text)  # Remove special characters
            
            # Handle common answer formats
            text = re.sub(r'^the answer is\s*', '', text)
            text = re.sub(r'^answer:\s*', '', text)
            text = re.sub(r'^final answer:\s*', '', text)
            
            return text
        
        normalized_expected = normalize_format(expected)
        normalized_our = normalize_format(our_clean)
        
        # Check normalized exact match
        if normalized_our == normalized_expected:
            return {"status": "CORRECT", "expected": expected, "our": our_clean}
        
        # For list-type answers, try element-wise comparison
        if ',' in expected and ',' in our_clean:
            expected_items = [item.strip().lower() for item in expected.split(',')]
            our_items = [item.strip().lower() for item in our_clean.split(',')]
            
            # Sort both lists for comparison (handles different ordering)
            if sorted(expected_items) == sorted(our_items):
                return {"status": "CORRECT", "expected": expected, "our": our_clean}
            
            # Check if most items match (partial credit)
            matching_items = set(expected_items) & set(our_items)
            if len(matching_items) >= len(expected_items) * 0.7:  # 70% match threshold
                return {"status": "PARTIAL", "expected": expected, "our": our_clean}
        
        # Check if our answer contains the expected answer (broader match)
        if normalized_expected in normalized_our or normalized_our in normalized_expected:
            return {"status": "PARTIAL", "expected": expected, "our": our_clean}
        
        # ENHANCED: Numeric equivalence checking
        import re
        expected_numbers = re.findall(r'\d+(?:\.\d+)?', expected)
        our_numbers = re.findall(r'\d+(?:\.\d+)?', our_clean)
        
        if expected_numbers and our_numbers:
            try:
                # Compare primary numbers
                expected_num = float(expected_numbers[0])
                our_num = float(our_numbers[0])
                
                # Allow small floating point differences
                if abs(expected_num - our_num) < 0.01:
                    return {"status": "CORRECT", "expected": expected, "our": our_clean}
                    
                # Check for percentage differences (e.g., rounding errors)
                if expected_num > 0:
                    percentage_diff = abs(expected_num - our_num) / expected_num
                    if percentage_diff < 0.01:  # 1% tolerance
                        return {"status": "CORRECT", "expected": expected, "our": our_clean}
            except (ValueError, IndexError):
                pass
        
        # ENHANCED: Fuzzy matching for near-correct answers
        def fuzzy_similarity(str1, str2):
            """Calculate simple character-based similarity"""
            if not str1 or not str2:
                return 0.0
            
            # Convert to character sets
            chars1 = set(str1.lower())
            chars2 = set(str2.lower())
            
            # Calculate Jaccard similarity
            intersection = len(chars1 & chars2)
            union = len(chars1 | chars2)
            
            return intersection / union if union > 0 else 0.0
        
        # Check fuzzy similarity for near matches
        similarity = fuzzy_similarity(normalized_expected, normalized_our)
        if similarity > 0.8:  # 80% character similarity
            return {"status": "PARTIAL", "expected": expected, "our": our_clean}
        
        # Final check: word-level matching
        expected_words = set(normalized_expected.split())
        our_words = set(normalized_our.split())
        
        if expected_words and our_words:
            word_overlap = len(expected_words & our_words) / len(expected_words)
            if word_overlap > 0.7:  # 70% word overlap
                return {"status": "PARTIAL", "expected": expected, "our": our_clean}
        
        return {"status": "INCORRECT", "expected": expected, "our": our_clean}
    
    def classify_all_questions(self) -> Dict[str, List[Dict]]:
        """Classify all questions and group by agent type"""
        
        self.logger.info("🧠 Classifying all GAIA questions...")
        
        questions_by_agent = defaultdict(list)
        classification_stats = defaultdict(int)
        
        for question_data in self.loader.questions:
            task_id = question_data.get('task_id', 'unknown')
            question_text = question_data.get('question', '')
            file_name = question_data.get('file_name', '')
            
            try:
                classification = self.classifier.classify_question(question_text, file_name)
                primary_agent = classification['primary_agent']
                
                # Add classification to question data
                question_data['classification'] = classification
                question_data['routing'] = self.classifier.get_routing_recommendation(classification)
                
                questions_by_agent[primary_agent].append(question_data)
                classification_stats[primary_agent] += 1
                
                self.logger.info(f"  {task_id[:8]}... → {primary_agent} (confidence: {classification['confidence']:.3f})")
                
            except Exception as e:
                self.logger.error(f"  ❌ Classification failed for {task_id[:8]}...: {e}")
                questions_by_agent['error'].append(question_data)
        
        # Print classification summary
        self.logger.info(f"\n📊 CLASSIFICATION SUMMARY:")
        total_questions = len(self.loader.questions)
        for agent_type, count in sorted(classification_stats.items()):
            percentage = (count / total_questions) * 100
            self.logger.info(f"  {agent_type}: {count} questions ({percentage:.1f}%)")
        
        return dict(questions_by_agent)
    
    def test_agent_type(self, agent_type: str, questions: List[Dict], test_all: bool = False) -> List[Dict]:
        """Test all questions for a specific agent type"""
        
        if not questions:
            self.logger.warning(f"No questions found for agent type: {agent_type}")
            return []
        
        self.logger.info(f"\n🤖 TESTING {agent_type.upper()} AGENT")
        self.logger.info(f"=" * 60)
        self.logger.info(f"Questions to test: {len(questions)}")
        
        agent_results = []
        success_count = 0
        
        for i, question_data in enumerate(questions, 1):
            task_id = question_data.get('task_id', 'unknown')
            question_text = question_data.get('question', '')
            file_name = question_data.get('file_name', '')
            
            self.logger.info(f"\n[{i}/{len(questions)}] Testing {task_id[:8]}...")
            self.logger.info(f"Question: {question_text[:100]}...")
            if file_name:
                self.logger.info(f"File: {file_name}")
            
            try:
                start_time = time.time()
                answer = self.solver.solve_question(question_data)
                solve_time = time.time() - start_time
                
                # Validate answer against expected result
                validation_result = self.validate_answer(task_id, answer)
                
                # Log results with validation
                self.logger.info(f"✅ Answer: {answer[:100]}...")
                self.logger.info(f"⏱️ Time: {solve_time:.1f}s")
                self.logger.info(f"🔍 Expected: {validation_result['expected']}")
                self.logger.info(f"📊 Validation: {validation_result['status']}")
                
                if validation_result['status'] == 'CORRECT':
                    self.logger.info(f"✅ PERFECT MATCH!")
                    actual_status = 'correct'
                elif validation_result['status'] == 'PARTIAL':
                    self.logger.info(f"🟡 PARTIAL MATCH - contains correct answer")
                    actual_status = 'partial'
                elif validation_result['status'] == 'INCORRECT':
                    self.logger.error(f"❌ INCORRECT - answers don't match")
                    actual_status = 'incorrect'
                else:
                    self.logger.warning(f"⚠️ NO VALIDATION DATA")
                    actual_status = 'no_validation'
                
                result = {
                    'question_id': task_id,
                    'question': question_text,
                    'file_name': file_name,
                    'agent_type': agent_type,
                    'classification': question_data.get('classification'),
                    'routing': question_data.get('routing'),
                    'answer': answer,
                    'solve_time': solve_time,
                    'status': 'completed',
                    'validation_status': validation_result['status'],
                    'expected_answer': validation_result['expected'],
                    'actual_status': actual_status,
                    'error_type': None,
                    'error_details': None
                }
                
                agent_results.append(result)
                if actual_status == 'correct':
                    success_count += 1
                
            except Exception as e:
                solve_time = time.time() - start_time
                error_type = self.categorize_error(str(e))
                
                self.logger.error(f"❌ Error: {e}")
                self.logger.error(f"Error Type: {error_type}")
                
                result = {
                    'question_id': task_id,
                    'question': question_text,
                    'file_name': file_name,
                    'agent_type': agent_type,
                    'classification': question_data.get('classification'),
                    'routing': question_data.get('routing'),
                    'answer': f"Error: {str(e)}",
                    'solve_time': solve_time,
                    'status': 'error',
                    'error_type': error_type,
                    'error_details': str(e)
                }
                
                agent_results.append(result)
                self.error_patterns[agent_type].append({
                    'question_id': task_id,
                    'error_type': error_type,
                    'error_details': str(e),
                    'question_preview': question_text[:100]
                })
            
            # Small delay to avoid overwhelming APIs
            time.sleep(1)
        
        # Agent type summary with accuracy metrics
        error_count = len([r for r in agent_results if r['status'] == 'error'])
        completed_count = len([r for r in agent_results if r['status'] == 'completed'])
        correct_count = len([r for r in agent_results if r.get('actual_status') == 'correct'])
        partial_count = len([r for r in agent_results if r.get('actual_status') == 'partial'])
        incorrect_count = len([r for r in agent_results if r.get('actual_status') == 'incorrect'])
        
        accuracy_rate = (correct_count / len(questions)) * 100 if questions else 0
        completion_rate = (completed_count / len(questions)) * 100 if questions else 0
        
        self.logger.info(f"\n📊 {agent_type.upper()} AGENT RESULTS:")
        self.logger.info(f"  Completed: {completed_count}/{len(questions)} ({completion_rate:.1f}%)")
        self.logger.info(f"  ✅ Correct: {correct_count}/{len(questions)} ({accuracy_rate:.1f}%)")
        self.logger.info(f"  🟡 Partial: {partial_count}/{len(questions)}")
        self.logger.info(f"  ❌ Incorrect: {incorrect_count}/{len(questions)}")
        self.logger.info(f"  💥 Errors: {error_count}/{len(questions)}")
        
        if agent_results:
            completed_results = [r for r in agent_results if r['status'] == 'completed']
            if completed_results:
                avg_time = sum(r['solve_time'] for r in completed_results) / len(completed_results)
                self.logger.info(f"  ⏱️ Average Solve Time: {avg_time:.1f}s")
        
        return agent_results
    
    def categorize_error(self, error_message: str) -> str:
        """Categorize error types for analysis"""
        
        error_message_lower = error_message.lower()
        
        if '503' in error_message or 'service unavailable' in error_message_lower:
            return 'API_OVERLOAD'
        elif 'timeout' in error_message_lower or 'time out' in error_message_lower:
            return 'TIMEOUT'
        elif 'api' in error_message_lower and ('key' in error_message_lower or 'auth' in error_message_lower):
            return 'AUTHENTICATION'
        elif 'wikipedia' in error_message_lower or 'wiki' in error_message_lower:
            return 'WIKIPEDIA_TOOL'
        elif 'chess' in error_message_lower or 'fen' in error_message_lower:
            return 'CHESS_TOOL'
        elif 'excel' in error_message_lower or 'xlsx' in error_message_lower:
            return 'EXCEL_TOOL'
        elif 'video' in error_message_lower or 'youtube' in error_message_lower:
            return 'VIDEO_TOOL'
        elif 'gemini' in error_message_lower:
            return 'GEMINI_API'
        elif 'download' in error_message_lower or 'file' in error_message_lower:
            return 'FILE_PROCESSING'
        elif 'hallucination' in error_message_lower or 'fabricat' in error_message_lower:
            return 'HALLUCINATION'
        elif 'parsing' in error_message_lower or 'extract' in error_message_lower:
            return 'PARSING_ERROR'
        else:
            return 'UNKNOWN'
    
    def analyze_errors_by_agent(self):
        """Analyze error patterns by agent type"""
        
        if not self.error_patterns:
            self.logger.info("🎉 No errors found across all agent types!")
            return
        
        self.logger.info(f"\n🔍 ERROR ANALYSIS BY AGENT TYPE")
        self.logger.info("=" * 60)
        
        for agent_type, errors in self.error_patterns.items():
            if not errors:
                continue
                
            self.logger.info(f"\n🚨 {agent_type.upper()} AGENT ERRORS ({len(errors)} total):")
            
            # Group errors by type
            error_type_counts = defaultdict(int)
            for error in errors:
                error_type_counts[error['error_type']] += 1
            
            for error_type, count in sorted(error_type_counts.items(), key=lambda x: x[1], reverse=True):
                percentage = (count / len(errors)) * 100
                self.logger.info(f"  {error_type}: {count} errors ({percentage:.1f}%)")
            
            # Show specific examples
            self.logger.info(f"  Examples:")
            for error in errors[:3]:  # Show first 3 errors
                self.logger.info(f"    - {error['question_id'][:8]}...: {error['error_type']} - {error['question_preview']}...")
    
    def generate_improvement_recommendations(self):
        """Generate specific recommendations for improving each agent type"""
        
        self.logger.info(f"\n💡 IMPROVEMENT RECOMMENDATIONS")
        self.logger.info("=" * 60)
        
        all_results = [r for agent_results in self.results for r in agent_results]
        
        # Calculate success rates by agent type
        agent_stats = defaultdict(lambda: {'total': 0, 'success': 0, 'errors': []})
        
        for result in all_results:
            agent_type = result['agent_type']
            agent_stats[agent_type]['total'] += 1
            
            if result['status'] == 'completed':
                agent_stats[agent_type]['success'] += 1
            else:
                agent_stats[agent_type]['errors'].append(result)
        
        # Generate recommendations for each agent type
        for agent_type, stats in agent_stats.items():
            success_rate = (stats['success'] / stats['total']) * 100 if stats['total'] > 0 else 0
            
            self.logger.info(f"\n🎯 {agent_type.upper()} AGENT (Success Rate: {success_rate:.1f}%):")
            
            if success_rate >= 90:
                self.logger.info(f"  ✅ Excellent performance! Minor optimizations only.")
            elif success_rate >= 75:
                self.logger.info(f"  ⚠️ Good performance with room for improvement.")
            elif success_rate >= 50:
                self.logger.info(f"  🔧 Moderate performance - needs attention.")
            else:
                self.logger.info(f"  🚨 Poor performance - requires major improvements.")
            
            # Analyze common error patterns for this agent
            error_types = defaultdict(int)
            for error in stats['errors']:
                if error['error_type']:
                    error_types[error['error_type']] += 1
            
            if error_types:
                self.logger.info(f"  Common Issues:")
                for error_type, count in sorted(error_types.items(), key=lambda x: x[1], reverse=True):
                    self.logger.info(f"    - {error_type}: {count} occurrences")
                    self.suggest_fix_for_error_type(error_type, agent_type)
    
    def suggest_fix_for_error_type(self, error_type: str, agent_type: str):
        """Suggest specific fixes for common error types"""
        
        suggestions = {
            'API_OVERLOAD': "Implement exponential backoff and retry logic",
            'TIMEOUT': "Increase timeout limits or optimize processing pipeline",
            'AUTHENTICATION': "Check API keys and authentication configuration",
            'WIKIPEDIA_TOOL': "Enhance Wikipedia search logic and error handling",
            'CHESS_TOOL': "Improve FEN parsing and chess engine integration",
            'EXCEL_TOOL': "Add better Excel format validation and error recovery",
            'VIDEO_TOOL': "Implement fallback mechanisms for video processing",
            'GEMINI_API': "Add Gemini API error handling and fallback models",
            'FILE_PROCESSING': "Improve file download and validation logic",
            'HALLUCINATION': "Strengthen anti-hallucination prompts and tool output validation",
            'PARSING_ERROR': "Enhance output parsing logic and format validation"
        }
        
        suggestion = suggestions.get(error_type, "Investigate error cause and implement appropriate fix")
        self.logger.info(f"      → Fix: {suggestion}")
    
    def save_comprehensive_results(self, questions_by_agent: Dict[str, List[Dict]]):
        """Save comprehensive test results with error analysis"""
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        results_file = f"gaia_classification_test_results_{timestamp}.json"
        
        # Flatten all results
        all_results = []
        for agent_results in self.results:
            all_results.extend(agent_results)
        
        # Create comprehensive results
        comprehensive_results = {
            'test_metadata': {
                'timestamp': timestamp,
                'total_questions': len(self.loader.questions),
                'questions_by_agent': {agent: len(questions) for agent, questions in questions_by_agent.items()},
                'log_file': self.log_file
            },
            'overall_stats': {
                'total_questions': len(all_results),
                'successful': len([r for r in all_results if r['status'] == 'completed']),
                'errors': len([r for r in all_results if r['status'] == 'error']),
                'success_rate': len([r for r in all_results if r['status'] == 'completed']) / len(all_results) * 100 if all_results else 0
            },
            'agent_performance': {},
            'error_patterns': dict(self.error_patterns),
            'detailed_results': all_results
        }
        
        # Calculate per-agent performance
        agent_stats = defaultdict(lambda: {'total': 0, 'success': 0, 'avg_time': 0})
        
        for result in all_results:
            agent_type = result['agent_type']
            agent_stats[agent_type]['total'] += 1
            
            if result['status'] == 'completed':
                agent_stats[agent_type]['success'] += 1
                agent_stats[agent_type]['avg_time'] += result['solve_time']
        
        for agent_type, stats in agent_stats.items():
            success_rate = (stats['success'] / stats['total']) * 100 if stats['total'] > 0 else 0
            avg_time = stats['avg_time'] / stats['success'] if stats['success'] > 0 else 0
            
            comprehensive_results['agent_performance'][agent_type] = {
                'total_questions': stats['total'],
                'successful': stats['success'],
                'success_rate': success_rate,
                'average_solve_time': avg_time
            }
        
        # Save results
        with open(results_file, 'w') as f:
            json.dump(comprehensive_results, f, indent=2, ensure_ascii=False)
        
        self.logger.info(f"\n💾 Comprehensive results saved to: {results_file}")
        return results_file
    
    def run_classification_test(self, agent_types: Optional[List[str]] = None, test_all: bool = True):
        """Run the complete classification-based testing workflow"""
        
        self.logger.info("🚀 GAIA CLASSIFICATION-BASED TESTING")
        self.logger.info("=" * 70)
        self.logger.info(f"Log file: {self.log_file}")
        
        # Step 1: Classify all questions
        questions_by_agent = self.classify_all_questions()
        
        # Step 2: Filter agent types to test
        if agent_types:
            agent_types_to_test = [agent for agent in agent_types if agent in questions_by_agent]
            if not agent_types_to_test:
                self.logger.error(f"No questions found for specified agent types: {agent_types}")
                return
        else:
            agent_types_to_test = list(questions_by_agent.keys())
        
        self.logger.info(f"\nTesting agent types: {agent_types_to_test}")
        
        # Step 3: Test each agent type
        for agent_type in agent_types_to_test:
            if agent_type == 'error':  # Skip classification errors for now
                continue
                
            questions = questions_by_agent[agent_type]
            agent_results = self.test_agent_type(agent_type, questions, test_all)
            self.results.append(agent_results)
        
        # Step 4: Comprehensive analysis
        self.analyze_errors_by_agent()
        self.generate_improvement_recommendations()
        
        # Step 5: Save results
        results_file = self.save_comprehensive_results(questions_by_agent)
        
        self.logger.info(f"\n✅ CLASSIFICATION TESTING COMPLETE!")
        self.logger.info(f"📊 Results saved to: {results_file}")
        self.logger.info(f"📋 Log file: {self.log_file}")

def main():
    """Main CLI interface for classification-based testing"""
    
    parser = argparse.ArgumentParser(description="GAIA Classification-Based Testing with Error Analysis")
    parser.add_argument(
        '--agent-types', 
        nargs='+', 
        choices=['multimedia', 'research', 'logic_math', 'file_processing', 'general'],
        help='Specific agent types to test (default: all)'
    )
    parser.add_argument(
        '--failed-only', 
        action='store_true',
        help='Test only questions that failed in previous runs'
    )
    parser.add_argument(
        '--quick-test', 
        action='store_true',
        help='Run a quick test with limited questions per agent type'
    )
    
    args = parser.parse_args()
    
    # Initialize and run tester
    tester = GAIAClassificationTester()
    
    print("🎯 Starting GAIA Classification-Based Testing...")
    if args.agent_types:
        print(f"📋 Testing specific agent types: {args.agent_types}")
    else:
        print("📋 Testing all agent types")
    
    tester.run_classification_test(
        agent_types=args.agent_types,
        test_all=not args.quick_test
    )

if __name__ == "__main__":
    main()