Spaces:
Running
Running
File size: 10,863 Bytes
93de262 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 |
#!/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() |