#!/usr/bin/env python3 """ Asynchronous Complete GAIA Test System Main orchestrator for concurrent testing of all GAIA questions with honest accuracy measurement. """ import asyncio import json import logging import time from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple import sys import os # Add the project root to the Python path sys.path.insert(0, str(Path(__file__).parent)) from async_question_processor import AsyncQuestionProcessor from classification_analyzer import ClassificationAnalyzer from summary_report_generator import SummaryReportGenerator class AsyncGAIATestSystem: """Main orchestrator for asynchronous GAIA testing with honest accuracy measurement.""" def __init__(self, max_concurrent: int = 3, timeout_seconds: int = 900, output_dir: str = "async_test_results"): """ Initialize the async test system. Args: max_concurrent: Maximum number of concurrent question processors timeout_seconds: Timeout per question (15 minutes default) output_dir: Directory for test results and logs """ self.max_concurrent = max_concurrent self.timeout_seconds = timeout_seconds self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) # Create timestamped session directory timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.session_dir = self.output_dir / f"session_{timestamp}" self.session_dir.mkdir(exist_ok=True) # Initialize components self.processor = AsyncQuestionProcessor( session_dir=self.session_dir, timeout_seconds=self.timeout_seconds ) self.analyzer = ClassificationAnalyzer() self.reporter = SummaryReportGenerator() # Setup logging self.setup_logging() # Test results tracking self.results: Dict[str, Dict] = {} self.start_time: Optional[float] = None self.end_time: Optional[float] = None def setup_logging(self): """Setup comprehensive logging for the test session.""" log_file = self.session_dir / "async_test_system.log" # Configure logger self.logger = logging.getLogger("AsyncGAIATest") self.logger.setLevel(logging.INFO) # File handler file_handler = logging.FileHandler(log_file) file_handler.setLevel(logging.INFO) # Console handler console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # Formatter formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) # Add handlers self.logger.addHandler(file_handler) self.logger.addHandler(console_handler) async def load_questions(self) -> List[Dict]: """Load GAIA questions from the standard source.""" questions_file = Path("gaia_questions_list.txt") if not questions_file.exists(): self.logger.error(f"Questions file not found: {questions_file}") return [] questions = [] try: with open(questions_file, 'r') as f: for line in f: line = line.strip() if line and line.startswith('{'): try: question = json.loads(line) questions.append(question) except json.JSONDecodeError as e: self.logger.warning(f"Failed to parse question line: {line[:50]}... - {e}") self.logger.info(f"Loaded {len(questions)} questions for testing") return questions except Exception as e: self.logger.error(f"Failed to load questions: {e}") return [] async def process_question_batch(self, questions: List[Dict]) -> Dict[str, Dict]: """Process a batch of questions concurrently.""" # Create semaphore to limit concurrent processing semaphore = asyncio.Semaphore(self.max_concurrent) async def process_single_question(question: Dict) -> Tuple[str, Dict]: """Process a single question with semaphore control.""" async with semaphore: question_id = question.get('task_id', 'unknown') self.logger.info(f"Starting processing for question {question_id}") try: result = await self.processor.process_question(question) self.logger.info(f"Completed processing for question {question_id}") return question_id, result except Exception as e: self.logger.error(f"Failed to process question {question_id}: {e}") return question_id, { 'status': 'error', 'error': str(e), 'timestamp': datetime.now().isoformat() } # Create tasks for all questions tasks = [process_single_question(q) for q in questions] # Process all questions concurrently self.logger.info(f"Starting concurrent processing of {len(questions)} questions (max_concurrent={self.max_concurrent})") results = await asyncio.gather(*tasks, return_exceptions=True) # Organize results organized_results = {} for result in results: if isinstance(result, Exception): self.logger.error(f"Task failed with exception: {result}") continue question_id, question_result = result organized_results[question_id] = question_result return organized_results async def run_complete_test(self) -> Dict: """Run the complete asynchronous GAIA test system.""" self.logger.info("=" * 80) self.logger.info("ASYNC GAIA TEST SYSTEM - STARTING COMPLETE TEST") self.logger.info("=" * 80) self.start_time = time.time() try: # Load questions self.logger.info("Loading GAIA questions...") questions = await self.load_questions() if not questions: self.logger.error("No questions loaded. Aborting test.") return {"status": "error", "message": "No questions loaded"} self.logger.info(f"Processing {len(questions)} questions with max_concurrent={self.max_concurrent}") # Process questions concurrently self.results = await self.process_question_batch(questions) self.end_time = time.time() total_duration = self.end_time - self.start_time self.logger.info(f"All questions processed in {total_duration:.2f} seconds") # Generate analysis and reports await self.generate_comprehensive_analysis() # Create session summary session_summary = { "session_id": self.session_dir.name, "start_time": datetime.fromtimestamp(self.start_time).isoformat(), "end_time": datetime.fromtimestamp(self.end_time).isoformat(), "total_duration_seconds": total_duration, "questions_processed": len(self.results), "max_concurrent": self.max_concurrent, "timeout_seconds": self.timeout_seconds, "session_dir": str(self.session_dir), "results": self.results } # Save session summary summary_file = self.session_dir / "session_summary.json" with open(summary_file, 'w') as f: json.dump(session_summary, f, indent=2) self.logger.info(f"Session summary saved to: {summary_file}") return session_summary except Exception as e: self.logger.error(f"Complete test failed: {e}") return {"status": "error", "message": str(e)} async def generate_comprehensive_analysis(self): """Generate comprehensive analysis and reports.""" self.logger.info("Generating comprehensive analysis...") try: # Classification-based analysis classification_report = await self.analyzer.analyze_by_classification( self.results, self.session_dir ) # Master summary report summary_report = await self.reporter.generate_master_report( self.results, self.session_dir, classification_report ) self.logger.info("Analysis and reports generated successfully") except Exception as e: self.logger.error(f"Failed to generate analysis: {e}") def main(): """Main entry point for the async test system.""" import argparse parser = argparse.ArgumentParser(description="Asynchronous GAIA Test System") parser.add_argument('--max-concurrent', type=int, default=3, help='Maximum concurrent question processors (default: 3)') parser.add_argument('--timeout', type=int, default=900, help='Timeout per question in seconds (default: 900)') parser.add_argument('--output-dir', type=str, default='async_test_results', help='Output directory for results (default: async_test_results)') args = parser.parse_args() # Create and run the test system system = AsyncGAIATestSystem( max_concurrent=args.max_concurrent, timeout_seconds=args.timeout, output_dir=args.output_dir ) # Run the async test try: result = asyncio.run(system.run_complete_test()) if result.get("status") == "error": print(f"Test failed: {result.get('message')}") sys.exit(1) else: print(f"Test completed successfully!") print(f"Results saved to: {system.session_dir}") except KeyboardInterrupt: print("\nTest interrupted by user") sys.exit(1) except Exception as e: print(f"Test failed with exception: {e}") sys.exit(1) if __name__ == "__main__": main()