updated I'm proved version

gaia_tools/__init__.py

@@ -0,0 +1,24 @@
+"""
+GAIA Tools Package
+Tools and utilities for analyzing and improving GAIA agent performance.
+"""
+
+from .error_analysis import (
+    GAIATestAnalyzer,
+    QuestionType,
+    FailureMode,
+    TestResult
+)
+from .dataset import (
+    GAIADatasetManager,
+    ensure_local_testing_setup
+)
+
+__all__ = [
+    'GAIATestAnalyzer',
+    'QuestionType',
+    'FailureMode',
+    'TestResult',
+    'GAIADatasetManager',
+    'ensure_local_testing_setup'
+]

gaia_tools/code_executor.py

@@ -0,0 +1,389 @@
+"""
+Code Execution Framework for GAIA Agent
+
+Provides safe Python code execution for math/data processing questions.
+Uses local execution with timeout and safety constraints.
+
+Expected Impact: +15-20% accuracy improvement on math/calculation questions
+"""
+
+import re
+import os
+import sys
+import time
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from typing import Optional, List
+from pathlib import Path
+
+
+@dataclass
+class ExecutionResult:
+    """Result of code execution"""
+    success: bool
+    output: Optional[str]
+    error: Optional[str]
+    execution_time: float
+
+
+def should_use_code_execution(question: str) -> bool:
+    """
+    Determine if a question would benefit from code execution.
+
+    Args:
+        question: The question text
+
+    Returns:
+        True if code execution should be used
+    """
+    question_lower = question.lower()
+
+    # EXCLUSIONS: Research questions that should NOT use code
+    research_indicators = [
+        'who', 'when', 'where', 'which person', 'which company',
+        'published by', 'written by', 'created by', 'founded by',
+        'according to', 'wikipedia', 'article', 'biography',
+        'history of', 'year of', 'born in', 'died in'
+    ]
+
+    # If it's clearly a research/lookup question, don't use code
+    if any(indicator in question_lower for indicator in research_indicators):
+        # Exception: if it has actual numbers to calculate WITH
+        # e.g., "Who scored 25 + 30 points?" should use code for the math
+        has_math_operators = any(op in question for op in ['+', '-', '*', '/', '='])
+        if not has_math_operators:
+            return False
+
+    # Math keywords - direct operations
+    math_keywords = [
+        'calculate', 'compute', 'sum', 'average', 'mean', 'median',
+        'multiply', 'divide', 'subtract', 'add', 'total',
+        'square root', 'power', 'factorial', 'prime',
+        '+', '-', '*', '/', '%', '^', '='
+    ]
+
+    # Check for math operations
+    if any(keyword in question_lower for keyword in math_keywords):
+        return True
+
+    # Data processing keywords - only for provided data
+    data_processing_indicators = [
+        'from the csv', 'in the file', 'in the spreadsheet',
+        'from the table', 'in the data', 'given the values',
+        'calculate from', 'based on the following'
+    ]
+
+    if any(indicator in question_lower for indicator in data_processing_indicators):
+        return True
+
+    # Check for explicit number sequences that need calculation
+    # e.g., "What is 123 * 456" or "Sum of 10, 20, 30"
+    numbers = re.findall(r'\d+', question)
+    has_operators = any(op in question for op in ['+', '-', '*', '/', '=', 'x'])
+
+    if len(numbers) >= 2 and has_operators:
+        return True
+
+    return False
+
+
+class CodeExecutor:
+    """
+    Safe Python code executor with timeout and safety constraints.
+
+    Uses subprocess isolation to prevent harmful operations.
+    """
+
+    def __init__(self, timeout: int = 10, openrouter_client=None, model: str = "x-ai/grok-4.1-fast"):
+        """
+        Initialize code executor.
+
+        Args:
+            timeout: Maximum execution time in seconds
+            openrouter_client: OpenAI client for OpenRouter (for code generation)
+            model: Model to use for code generation
+        """
+        self.timeout = timeout
+        self.openrouter_client = openrouter_client
+        self.model = model
+
+    def generate_code(self, question: str, context: Optional[str] = None) -> str:
+        """
+        Generate Python code to answer the question.
+
+        Args:
+            question: The question to solve
+            context: Optional context/data for the question
+
+        Returns:
+            Python code as string
+        """
+        # If we have OpenRouter, use LLM to generate code
+        if self.openrouter_client:
+            return self._generate_code_with_llm(question, context)
+
+        # Fallback: Simple code generation for basic math
+        return self._generate_code_simple(question)
+
+    def _generate_code_with_llm(self, question: str, context: Optional[str] = None) -> str:
+        """Generate code using LLM"""
+        prompt = f"""Generate Python code to answer this question. Output ONLY the Python code, no explanations.
+The code must print the final answer using print().
+
+Question: {question}"""
+
+        if context:
+            prompt += f"\n\nContext/Data: {context}"
+
+        prompt += """
+
+Requirements:
+1. Use only Python standard library (math, statistics, etc.)
+2. Print the final answer
+3. Keep it simple and direct
+4. No external imports except math, statistics
+5. Handle edge cases
+
+Code:"""
+
+        try:
+            response = self.openrouter_client.chat.completions.create(
+                model=self.model,
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=500,
+                temperature=0.1
+            )
+
+            code = response.choices[0].message.content.strip()
+
+            # Extract code from markdown if present
+            if "```python" in code:
+                code = code.split("```python")[1].split("```")[0].strip()
+            elif "```" in code:
+                code = code.split("```")[1].split("```")[0].strip()
+
+            return code
+
+        except Exception as e:
+            print(f"❌ LLM code generation failed: {e}")
+            return self._generate_code_simple(question)
+
+    def _generate_code_simple(self, question: str) -> str:
+        """
+        Generate simple code without LLM (fallback).
+
+        This handles basic arithmetic expressions.
+        """
+        # Try to extract a math expression
+        # Remove common words
+        expr = question.lower()
+        for word in ['what is', 'calculate', 'compute', 'the result of', '?', 'equal', 'equals']:
+            expr = expr.replace(word, ' ')
+
+        expr = expr.strip()
+
+        # Convert word operations to symbols
+        replacements = {
+            ' plus ': '+',
+            ' minus ': '-',
+            ' times ': '*',
+            ' divided by ': '/',
+            ' multiply ': '*',
+            ' divide ': '/',
+            ' add ': '+',
+            ' subtract ': '-'
+        }
+
+        for word, symbol in replacements.items():
+            expr = expr.replace(word, symbol)
+
+        # Clean up spaces
+        expr = re.sub(r'\s+', '', expr)
+
+        # Basic validation
+        if re.match(r'^[\d+\-*/().\s]+$', expr):
+            return f"result = {expr}\nprint(int(result) if result == int(result) else result)"
+
+        # Fallback for square root
+        if 'square root' in question.lower():
+            match = re.search(r'\d+', question)
+            if match:
+                num = match.group()
+                return f"import math\nresult = math.sqrt({num})\nprint(int(result) if result == int(result) else result)"
+
+        # Fallback for average
+        if 'average' in question.lower() or 'mean' in question.lower():
+            numbers = re.findall(r'\d+', question)
+            if numbers:
+                # Convert to integers explicitly
+                numbers_list = [int(n) for n in numbers]
+                return f"values = {numbers_list}\nresult = sum(values) / len(values)\nprint(int(result) if result == int(result) else result)"
+
+        # Default fallback
+        return "print('Unable to generate code for this question')"
+
+    def execute(self, code: str) -> ExecutionResult:
+        """
+        Execute Python code safely with timeout.
+
+        Args:
+            code: Python code to execute
+
+        Returns:
+            ExecutionResult with output or error
+        """
+        start_time = time.time()
+
+        # Safety check: block dangerous operations
+        dangerous_patterns = {
+            'import os': 'os module',
+            'import subprocess': 'subprocess module',
+            'import sys': 'sys module',
+            'import urllib': 'urllib module',
+            'import requests': 'requests module',
+            'import http': 'http module',
+            'import socket': 'socket module',
+            'open(': 'file operations',
+            '__import__': '__import__ function',
+            'eval(': 'eval function',
+            'exec(': 'exec function',
+            'compile(': 'compile function',
+        }
+
+        code_lower = code.lower()
+
+        # Check for dangerous patterns
+        for pattern, name in dangerous_patterns.items():
+            if pattern in code_lower:
+                # Only allow math and statistics imports
+                if 'import' in pattern and pattern not in ['import math', 'import statistics']:
+                    # Check if it's actually importing something safe
+                    if not any(safe in code_lower for safe in ['import math', 'import statistics', 'import random', 'import datetime']):
+                        if pattern in code_lower:
+                            return ExecutionResult(
+                                success=False,
+                                output=None,
+                                error=f"Security: {name} is not allowed",
+                                execution_time=time.time() - start_time
+                            )
+                # Block file/exec operations outright
+                elif pattern in ['open(', '__import__', 'eval(', 'exec(', 'compile(']:
+                    return ExecutionResult(
+                        success=False,
+                        output=None,
+                        error=f"Security: {name} is not allowed",
+                        execution_time=time.time() - start_time
+                    )
+
+        # Create temporary file for code
+        try:
+            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
+                f.write(code)
+                code_file = f.name
+
+            # Execute with timeout using subprocess
+            try:
+                result = subprocess.run(
+                    [sys.executable, code_file],
+                    capture_output=True,
+                    text=True,
+                    timeout=self.timeout,
+                    env={**os.environ, 'PYTHONPATH': str(Path(__file__).parent)}
+                )
+
+                execution_time = time.time() - start_time
+
+                if result.returncode == 0:
+                    output = result.stdout.strip()
+                    return ExecutionResult(
+                        success=True,
+                        output=output,
+                        error=None,
+                        execution_time=execution_time
+                    )
+                else:
+                    return ExecutionResult(
+                        success=False,
+                        output=None,
+                        error=result.stderr.strip(),
+                        execution_time=execution_time
+                    )
+
+            except subprocess.TimeoutExpired:
+                return ExecutionResult(
+                    success=False,
+                    output=None,
+                    error=f"Execution timeout ({self.timeout}s)",
+                    execution_time=self.timeout
+                )
+
+        except Exception as e:
+            return ExecutionResult(
+                success=False,
+                output=None,
+                error=str(e),
+                execution_time=time.time() - start_time
+            )
+
+        finally:
+            # Clean up temp file
+            try:
+                if 'code_file' in locals():
+                    os.unlink(code_file)
+            except:
+                pass
+
+    def solve_question(self, question: str, context: Optional[str] = None) -> Optional[str]:
+        """
+        Complete workflow: generate code, execute, return answer.
+
+        Args:
+            question: Question to solve
+            context: Optional context
+
+        Returns:
+            Answer string or None if failed
+        """
+        print(f"  🧮 CODE EXECUTION: {question[:60]}...")
+
+        # Generate code
+        code = self.generate_code(question, context)
+        print(f"  📝 Generated code ({len(code)} chars)")
+
+        # Execute code
+        result = self.execute(code)
+
+        if result.success and result.output:
+            print(f"  ✅ Execution successful: {result.output}")
+            return result.output
+        else:
+            print(f"  ❌ Execution failed: {result.error}")
+            return None
+
+
+if __name__ == "__main__":
+    # Test the code executor
+    print("=" * 60)
+    print("Code Executor Test")
+    print("=" * 60)
+
+    executor = CodeExecutor()
+
+    # Test 1: Simple arithmetic
+    question1 = "What is 123 * 456?"
+    print(f"\nTest 1: {question1}")
+    answer1 = executor.solve_question(question1)
+    print(f"Answer: {answer1}")
+
+    # Test 2: Average
+    question2 = "What is the average of 10, 20, 30, 40, 50?"
+    print(f"\nTest 2: {question2}")
+    answer2 = executor.solve_question(question2)
+    print(f"Answer: {answer2}")
+
+    # Test 3: Square root
+    question3 = "What is the square root of 144?"
+    print(f"\nTest 3: {question3}")
+    answer3 = executor.solve_question(question3)
+    print(f"Answer: {answer3}")

gaia_tools/dataset.py

@@ -0,0 +1,160 @@
+"""
+GAIA Dataset Utilities
+Download and cache GAIA questions for local testing
+"""
+
+import os
+import json
+import requests
+from pathlib import Path
+from typing import List, Dict, Any, Optional
+from datetime import datetime
+
+
+class GAIADatasetManager:
+    """Manages GAIA dataset download and local caching"""
+
+    def __init__(self, cache_dir: str = "gaia_data"):
+        self.cache_dir = Path(cache_dir)
+        self.cache_dir.mkdir(exist_ok=True)
+
+        self.api_url = "https://agents-course-unit4-scoring.hf.space"
+        self.questions_url = f"{self.api_url}/questions"
+        self.submit_url = f"{self.api_url}/submit"
+
+        self.questions_cache_file = self.cache_dir / "questions.json"
+        self.metadata_file = self.cache_dir / "metadata.json"
+
+    def download_questions(self, force_refresh: bool = False) -> List[Dict[str, Any]]:
+        """
+        Download GAIA questions from scoring API.
+
+        Args:
+            force_refresh: If True, always download fresh data. If False, use cache if available.
+
+        Returns:
+            List of question dictionaries
+        """
+        # Check cache first
+        if not force_refresh and self.questions_cache_file.exists():
+            print(f"📦 Loading questions from cache: {self.questions_cache_file}")
+            with open(self.questions_cache_file, 'r', encoding='utf-8') as f:
+                return json.load(f)
+
+        # Download from API
+        print(f"🌐 Downloading questions from: {self.questions_url}")
+        try:
+            response = requests.get(self.questions_url, timeout=30)
+            response.raise_for_status()
+            questions = response.json()
+
+            if not questions:
+                raise ValueError("Fetched questions list is empty")
+
+            # Cache the questions
+            with open(self.questions_cache_file, 'w', encoding='utf-8') as f:
+                json.dump(questions, f, indent=2)
+
+            # Update metadata
+            metadata = {
+                "download_time": datetime.now().isoformat(),
+                "question_count": len(questions),
+                "api_url": self.questions_url
+            }
+            with open(self.metadata_file, 'w', encoding='utf-8') as f:
+                json.dump(metadata, f, indent=2)
+
+            print(f"✅ Downloaded and cached {len(questions)} questions")
+            return questions
+
+        except requests.exceptions.RequestException as e:
+            print(f"❌ Error downloading questions: {e}")
+
+            # Fallback to cache if available
+            if self.questions_cache_file.exists():
+                print("📦 Falling back to cached questions")
+                with open(self.questions_cache_file, 'r', encoding='utf-8') as f:
+                    return json.load(f)
+            else:
+                raise e
+
+    def get_cached_metadata(self) -> Optional[Dict[str, Any]]:
+        """Get metadata about cached questions"""
+        if self.metadata_file.exists():
+            with open(self.metadata_file, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        return None
+
+    def save_results(self, results: List[Dict[str, Any]], filename: Optional[str] = None):
+        """
+        Save test results to a file
+
+        Args:
+            results: List of result dictionaries
+            filename: Optional filename. If not provided, uses timestamp.
+        """
+        if filename is None:
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            filename = f"results_{timestamp}.json"
+
+        filepath = self.cache_dir / filename
+        with open(filepath, 'w', encoding='utf-8') as f:
+            json.dump(results, f, indent=2)
+
+        print(f"💾 Results saved to: {filepath}")
+        return filepath
+
+    def load_dotenv(self):
+        """Load environment variables from .env file"""
+        env_file = Path(".env")
+        if env_file.exists():
+            print("📄 Loading environment variables from .env")
+            with open(env_file, 'r') as f:
+                for line in f:
+                    line = line.strip()
+                    if line and not line.startswith('#') and '=' in line:
+                        key, value = line.split('=', 1)
+                        os.environ[key.strip()] = value.strip()
+            print("✅ Environment variables loaded")
+        else:
+            print("⚠️  No .env file found")
+
+
+def ensure_local_testing_setup() -> GAIADatasetManager:
+    """
+    Ensure environment is set up for 100% local testing.
+
+    Returns:
+        GAIADatasetManager instance with questions cached
+    """
+    print("🔧 Setting up for local testing...")
+
+    # Load environment variables
+    manager = GAIADatasetManager()
+    manager.load_dotenv()
+
+    # Download and cache questions
+    try:
+        questions = manager.download_questions()
+        print(f"✅ Local testing setup complete ({len(questions)} questions cached)")
+    except Exception as e:
+        print(f"❌ Failed to download questions: {e}")
+        raise e
+
+    return manager
+
+
+if __name__ == "__main__":
+    # Test the dataset manager
+    print("=" * 60)
+    print("GAIA Dataset Manager Test")
+    print("=" * 60)
+
+    manager = ensure_local_testing_setup()
+
+    # Show cache metadata
+    metadata = manager.get_cached_metadata()
+    if metadata:
+        print("\n📊 Cache Metadata:")
+        for key, value in metadata.items():
+            print(f"  {key}: {value}")

gaia_tools/error_analysis.py

@@ -0,0 +1,480 @@
+"""
+GAIA Error Analysis Framework
+
+Categorizes questions, failure modes, and generates actionable improvement recommendations.
+Implements TDD test suite specifications from tests/test_error_analysis.py
+"""
+
+import csv
+import json
+import re
+from dataclasses import dataclass, asdict
+from enum import Enum
+from typing import List, Dict, Optional, Any
+from collections import defaultdict, Counter
+
+
+class QuestionType(Enum):
+    """Categories of GAIA questions"""
+    MATH = "math"
+    FILE = "file"
+    WEB = "web"
+    IMAGE = "image"
+    AUDIO = "audio"
+    REASONING = "reasoning"
+    MULTIMODAL = "multimodal"
+    UNKNOWN = "unknown"
+
+
+class FailureMode(Enum):
+    """Categories of answer failures"""
+    WRONG_ANSWER = "wrong_answer"
+    FORMATTING_ERROR = "formatting_error"
+    TIMEOUT = "timeout"
+    TOOL_FAILURE = "tool_failure"
+    EMPTY_RESPONSE = "empty_response"
+
+
+@dataclass
+class TestResult:
+    """Represents a single test result"""
+    question_id: str
+    question: str
+    question_type: QuestionType
+    expected: str
+    actual: str
+    success: bool
+    failure_mode: Optional[FailureMode] = None
+    time_elapsed: float = 0.0
+    tools_used: Optional[List[str]] = None
+    error: Optional[Exception] = None
+
+    def __post_init__(self):
+        if self.tools_used is None:
+            self.tools_used = []
+
+
+class GAIATestAnalyzer:
+    """
+    Analyzes GAIA agent test results to identify failure patterns and recommend improvements.
+
+    This class implements error categorization, performance tracking, and reporting
+    to guide agent optimization efforts.
+    """
+
+    def __init__(self):
+        self.results: List[TestResult] = []
+
+        # Patterns for question classification
+        self.math_patterns = [
+            r'\d+\s*[\+\-\*\/]\s*\d+',  # Arithmetic operations with numbers
+            r'calculate|compute|sum|multiply|divide|subtract|add',
+            r'what is \d+',
+            r'how many|how much'
+        ]
+
+        self.file_patterns = [
+            r'pdf|csv|excel|spreadsheet|document|table|file',
+            r'attached|according to the',
+        ]
+
+        self.image_patterns = [
+            r'image|picture|photo|screenshot|attached.*color|in the (attached )?image'
+        ]
+
+        self.audio_patterns = [
+            r'audio|recording|sound|said in|spoken|voice'
+        ]
+
+        self.web_patterns = [
+            r'who is|what is the (current|latest)|CEO|president|founded|website',
+            r'according to.*wikipedia|look up'
+        ]
+
+        self.reasoning_patterns = [
+            r'if .+ then|taller than|shorter than|before|after',
+            r'who is the (tallest|shortest|oldest|youngest)',
+        ]
+
+        self.multimodal_patterns = [
+            r'(image|picture|photo).*(csv|file|data|spreadsheet)',
+            r'(csv|file|data|spreadsheet).*(image|picture|photo)',
+            r'using the .+ and the'
+        ]
+
+    def classify_question_type(self, question: str) -> QuestionType:
+        """
+        Classify a question into a QuestionType based on its content.
+
+        Args:
+            question: The question text to classify
+
+        Returns:
+            QuestionType enum value
+        """
+        question_lower = question.lower()
+
+        # Check multimodal first (highest priority)
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.multimodal_patterns):
+            return QuestionType.MULTIMODAL
+
+        # Check for image questions
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.image_patterns):
+            return QuestionType.IMAGE
+
+        # Check for audio questions
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.audio_patterns):
+            return QuestionType.AUDIO
+
+        # Check for file questions
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.file_patterns):
+            return QuestionType.FILE
+
+        # Check for math questions
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.math_patterns):
+            return QuestionType.MATH
+
+        # Check for reasoning questions
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.reasoning_patterns):
+            return QuestionType.REASONING
+
+        # Check for web search questions
+        if any(re.search(pattern, question_lower, re.IGNORECASE)
+               for pattern in self.web_patterns):
+            return QuestionType.WEB
+
+        return QuestionType.UNKNOWN
+
+    def classify_failure_mode(
+        self,
+        expected: str,
+        actual: Optional[str],
+        error: Optional[Exception] = None
+    ) -> FailureMode:
+        """
+        Classify why an answer failed.
+
+        Args:
+            expected: The correct answer
+            actual: The agent's answer (None if error occurred)
+            error: Exception if one occurred
+
+        Returns:
+            FailureMode enum value
+        """
+        # Check for exceptions first
+        if error is not None:
+            if isinstance(error, TimeoutError):
+                return FailureMode.TIMEOUT
+            else:
+                return FailureMode.TOOL_FAILURE
+
+        # Check for empty/unable responses
+        if actual is None or actual.strip() == "":
+            return FailureMode.EMPTY_RESPONSE
+
+        if "unable to determine" in actual.lower():
+            return FailureMode.EMPTY_RESPONSE
+
+        # Check for formatting errors
+        expected_clean = expected.strip().lower()
+        actual_clean = actual.strip().lower()
+
+        # Remove commas and check if answers match
+        expected_no_comma = expected_clean.replace(',', '')
+        actual_no_comma = actual_clean.replace(',', '')
+        if expected_no_comma == actual_no_comma and expected_clean != actual_clean:
+            return FailureMode.FORMATTING_ERROR
+
+        # Check for unwanted units
+        if actual_clean.startswith(expected_clean):
+            remainder = actual_clean[len(expected_clean):].strip()
+            if remainder:  # Has extra content (likely units)
+                return FailureMode.FORMATTING_ERROR
+
+        # Check for articles (the, a, an)
+        articles = ['the ', 'a ', 'an ']
+        for article in articles:
+            if actual_clean.startswith(article):
+                without_article = actual_clean[len(article):]
+                if without_article == expected_clean:
+                    return FailureMode.FORMATTING_ERROR
+
+        # If none of the above, it's a wrong answer
+        return FailureMode.WRONG_ANSWER
+
+    def log_result(self, result: TestResult):
+        """
+        Add a test result to the analyzer.
+
+        Args:
+            result: TestResult object to log
+        """
+        self.results.append(result)
+
+    def analyze_response(
+        self,
+        question_id: str,
+        question: str,
+        expected: str,
+        actual: str,
+        time_elapsed: float = 0.0,
+        tools_used: Optional[List[str]] = None,
+        error: Optional[Exception] = None
+    ) -> TestResult:
+        """
+        Analyze a single agent response and create a TestResult.
+
+        This is a convenience method that combines classification and logging.
+
+        Args:
+            question_id: Unique identifier for the question
+            question: The question text
+            expected: The correct answer
+            actual: The agent's answer
+            time_elapsed: Time taken to answer
+            tools_used: List of tools used by the agent
+            error: Exception if one occurred
+
+        Returns:
+            TestResult object with all classifications
+        """
+        question_type = self.classify_question_type(question)
+        success = (actual == expected) if actual is not None else False
+
+        failure_mode = None
+        if not success:
+            failure_mode = self.classify_failure_mode(expected, actual, error)
+
+        result = TestResult(
+            question_id=question_id,
+            question=question,
+            question_type=question_type,
+            expected=expected,
+            actual=actual,
+            success=success,
+            failure_mode=failure_mode,
+            time_elapsed=time_elapsed,
+            tools_used=tools_used or [],
+            error=error
+        )
+
+        self.log_result(result)
+        return result
+
+    def generate_summary(self) -> Dict[str, Any]:
+        """
+        Generate summary statistics for all logged results.
+
+        Returns:
+            Dictionary with summary statistics
+        """
+        if not self.results:
+            return {
+                "total_questions": 0,
+                "correct_count": 0,
+                "accuracy": 0.0,
+                "avg_time": 0.0
+            }
+
+        total = len(self.results)
+        correct = sum(1 for r in self.results if r.success)
+        total_time = sum(r.time_elapsed for r in self.results)
+
+        return {
+            "total_questions": total,
+            "correct_count": correct,
+            "accuracy": correct / total if total > 0 else 0.0,
+            "avg_time": total_time / total if total > 0 else 0.0
+        }
+
+    def get_accuracy_by_type(self) -> Dict[QuestionType, float]:
+        """
+        Calculate accuracy broken down by question type.
+
+        Returns:
+            Dictionary mapping QuestionType to accuracy (0.0-1.0)
+        """
+        type_stats = defaultdict(lambda: {"correct": 0, "total": 0})
+
+        for result in self.results:
+            stats = type_stats[result.question_type]
+            stats["total"] += 1
+            if result.success:
+                stats["correct"] += 1
+
+        accuracy_by_type = {}
+        for qtype, stats in type_stats.items():
+            accuracy_by_type[qtype] = (
+                stats["correct"] / stats["total"] if stats["total"] > 0 else 0.0
+            )
+
+        return accuracy_by_type
+
+    def get_failures_by_mode(self) -> Dict[FailureMode, int]:
+        """
+        Count failures by failure mode.
+
+        Returns:
+            Dictionary mapping FailureMode to count
+        """
+        failure_counts = Counter()
+
+        for result in self.results:
+            if not result.success and result.failure_mode:
+                failure_counts[result.failure_mode] += 1
+
+        return dict(failure_counts)
+
+    def export_to_csv(self, filepath: str):
+        """
+        Export all results to a CSV file.
+
+        Args:
+            filepath: Path to output CSV file
+        """
+        with open(filepath, 'w', newline='', encoding='utf-8') as f:
+            writer = csv.writer(f)
+
+            # Write header
+            writer.writerow([
+                'question_id', 'question', 'question_type', 'expected', 'actual',
+                'success', 'failure_mode', 'time_elapsed', 'tools_used'
+            ])
+
+            # Write results
+            for result in self.results:
+                writer.writerow([
+                    result.question_id,
+                    result.question,
+                    result.question_type.value.upper(),
+                    result.expected,
+                    result.actual,
+                    result.success,
+                    result.failure_mode.value.upper() if result.failure_mode else '',
+                    result.time_elapsed,
+                    ','.join(result.tools_used) if result.tools_used else ''
+                ])
+
+    def export_to_json(self, filepath: str):
+        """
+        Export all results and summary to a JSON file.
+
+        Args:
+            filepath: Path to output JSON file
+        """
+        data = {
+            "summary": self.generate_summary(),
+            "accuracy_by_type": {
+                qtype.value: acc
+                for qtype, acc in self.get_accuracy_by_type().items()
+            },
+            "failures_by_mode": {
+                mode.value: count
+                for mode, count in self.get_failures_by_mode().items()
+            },
+            "results": [
+                {
+                    "question_id": r.question_id,
+                    "question": r.question,
+                    "question_type": r.question_type.value,
+                    "expected": r.expected,
+                    "actual": r.actual,
+                    "success": r.success,
+                    "failure_mode": r.failure_mode.value if r.failure_mode else None,
+                    "time_elapsed": r.time_elapsed,
+                    "tools_used": r.tools_used
+                }
+                for r in self.results
+            ]
+        }
+
+        with open(filepath, 'w', encoding='utf-8') as f:
+            json.dump(data, f, indent=2)
+
+    def get_recommendations(self) -> List[str]:
+        """
+        Generate actionable recommendations based on failure analysis.
+
+        Returns:
+            List of recommendation strings
+        """
+        recommendations = []
+
+        # Analyze question types with low accuracy
+        accuracy_by_type = self.get_accuracy_by_type()
+        failures_by_mode = self.get_failures_by_mode()
+
+        # Check for image-related failures
+        image_results = [r for r in self.results if r.question_type == QuestionType.IMAGE]
+        if image_results:
+            image_accuracy = accuracy_by_type.get(QuestionType.IMAGE, 0.0)
+            if image_accuracy < 0.5:
+                recommendations.append(
+                    "Add vision capabilities (Gemini 2.5 Pro) to handle image questions"
+                )
+
+        # Check for file processing failures
+        file_results = [r for r in self.results if r.question_type == QuestionType.FILE]
+        if file_results:
+            file_accuracy = accuracy_by_type.get(QuestionType.FILE, 0.0)
+            if file_accuracy < 0.5:
+                recommendations.append(
+                    "Implement file processing capabilities (PDF/CSV/Excel parsing)"
+                )
+
+        # Check for math failures
+        math_results = [r for r in self.results if r.question_type == QuestionType.MATH]
+        if math_results:
+            math_accuracy = accuracy_by_type.get(QuestionType.MATH, 0.0)
+            if math_accuracy < 0.7:
+                recommendations.append(
+                    "Add code execution capabilities for reliable math calculations"
+                )
+
+        # Check for formatting errors
+        formatting_errors = failures_by_mode.get(FailureMode.FORMATTING_ERROR, 0)
+        if formatting_errors > len(self.results) * 0.1:  # More than 10% formatting errors
+            recommendations.append(
+                "Improve answer formatting logic to handle commas, units, and articles"
+            )
+
+        # Check for empty responses
+        empty_responses = failures_by_mode.get(FailureMode.EMPTY_RESPONSE, 0)
+        if empty_responses > len(self.results) * 0.1:
+            recommendations.append(
+                "Improve tool reliability and add fallback mechanisms for empty responses"
+            )
+
+        # Check for timeouts
+        timeouts = failures_by_mode.get(FailureMode.TIMEOUT, 0)
+        if timeouts > len(self.results) * 0.05:
+            recommendations.append(
+                "Optimize query speed and increase timeout thresholds for complex questions"
+            )
+
+        # Check for audio processing
+        audio_results = [r for r in self.results if r.question_type == QuestionType.AUDIO]
+        if audio_results:
+            audio_accuracy = accuracy_by_type.get(QuestionType.AUDIO, 0.0)
+            if audio_accuracy < 0.5:
+                recommendations.append(
+                    "Add audio transcription capabilities (Whisper)"
+                )
+
+        # Check for multimodal questions
+        multimodal_results = [r for r in self.results if r.question_type == QuestionType.MULTIMODAL]
+        if multimodal_results:
+            multimodal_accuracy = accuracy_by_type.get(QuestionType.MULTIMODAL, 0.0)
+            if multimodal_accuracy < 0.5:
+                recommendations.append(
+                    "Improve multimodal reasoning by integrating multiple tool outputs"
+                )
+
+        return recommendations

gaia_tools/file_processor.py

@@ -0,0 +1,274 @@
+"""
+File Processing Framework for GAIA Agent
+
+Handles PDF, CSV, Excel, images, and audio files for GAIA questions.
+
+Expected Impact: +10-15% accuracy improvement on file-based questions
+"""
+
+import re
+import os
+import io
+from typing import Optional, Dict, Any, List
+from dataclasses import dataclass
+from pathlib import Path
+import tempfile
+
+
+@dataclass
+class ProcessedFile:
+    """Result of file processing"""
+    success: bool
+    file_type: str
+    content: Optional[str]
+    metadata: Dict[str, Any]
+    error: Optional[str] = None
+
+
+def extract_file_references(question: str) -> List[str]:
+    """
+    Extract file references from a question.
+
+    Args:
+        question: Question text
+
+    Returns:
+        List of file references/URLs found
+    """
+    references = []
+
+    # Look for file mentions
+    file_patterns = [
+        r'(attached|the)\s+(PDF|CSV|Excel|spreadsheet|image|picture|photo|audio|file)',
+        r'\.(pdf|csv|xlsx|xls|png|jpg|jpeg|gif|mp3|wav|m4a)',
+        r'https?://[^\s]+\.(pdf|csv|xlsx|png|jpg|jpeg)'
+    ]
+
+    for pattern in file_patterns:
+        matches = re.findall(pattern, question, re.IGNORECASE)
+        references.extend(matches)
+
+    return list(set(references))
+
+
+def should_use_file_processing(question: str) -> bool:
+    """Determine if question requires file processing"""
+    file_keywords = [
+        'attached', 'pdf', 'csv', 'excel', 'spreadsheet',
+        'image', 'picture', 'photo', 'document', 'file',
+        'table', 'according to the'
+    ]
+
+    question_lower = question.lower()
+    return any(keyword in question_lower for keyword in file_keywords)
+
+
+class FileProcessor:
+    """
+    Multi-format file processor for GAIA questions.
+
+    Supports: PDF, CSV, Excel, Images (OCR), Audio (transcription)
+    """
+
+    def __init__(self):
+        self.supported_formats = ['pdf', 'csv', 'xlsx', 'xls', 'png', 'jpg', 'jpeg', 'gif', 'mp3', 'wav']
+
+    def process_file(self, file_path: str) -> ProcessedFile:
+        """
+        Process a file and extract its content.
+
+        Args:
+            file_path: Path to the file
+
+        Returns:
+            ProcessedFile with extracted content
+        """
+        if not os.path.exists(file_path):
+            return ProcessedFile(
+                success=False,
+                file_type='unknown',
+                content=None,
+                metadata={},
+                error=f"File not found: {file_path}"
+            )
+
+        # Determine file type
+        ext = Path(file_path).suffix.lower().lstrip('.')
+
+        if ext == 'pdf':
+            return self._process_pdf(file_path)
+        elif ext in ['csv']:
+            return self._process_csv(file_path)
+        elif ext in ['xlsx', 'xls']:
+            return self._process_excel(file_path)
+        elif ext in ['png', 'jpg', 'jpeg', 'gif']:
+            return self._process_image(file_path)
+        elif ext in ['mp3', 'wav', 'm4a']:
+            return self._process_audio(file_path)
+        else:
+            return ProcessedFile(
+                success=False,
+                file_type=ext,
+                content=None,
+                metadata={},
+                error=f"Unsupported file type: {ext}"
+            )
+
+    def _process_pdf(self, file_path: str) -> ProcessedFile:
+        """Process PDF file"""
+        try:
+            # Try using pandas for simple PDFs (tables)
+            import pandas as pd
+            try:
+                # Try reading as table
+                tables = pd.read_html(file_path)
+                if tables:
+                    content = "\n\n".join([table.to_string() for table in tables])
+                    return ProcessedFile(
+                        success=True,
+                        file_type='pdf',
+                        content=content,
+                        metadata={'tables_found': len(tables)}
+                    )
+            except:
+                pass
+
+            # Fallback: Simple text extraction message
+            return ProcessedFile(
+                success=False,
+                file_type='pdf',
+                content=None,
+                metadata={},
+                error="PDF processing requires PyPDF2 or similar library"
+            )
+
+        except Exception as e:
+            return ProcessedFile(
+                success=False,
+                file_type='pdf',
+                content=None,
+                metadata={},
+                error=str(e)
+            )
+
+    def _process_csv(self, file_path: str) -> ProcessedFile:
+        """Process CSV file"""
+        try:
+            import pandas as pd
+
+            df = pd.read_csv(file_path)
+
+            # Generate summary
+            summary = f"CSV File Summary:\n"
+            summary += f"Rows: {len(df)}\n"
+            summary += f"Columns: {list(df.columns)}\n\n"
+            summary += f"First 10 rows:\n{df.head(10).to_string()}\n\n"
+            summary += f"Statistics:\n{df.describe().to_string()}"
+
+            return ProcessedFile(
+                success=True,
+                file_type='csv',
+                content=summary,
+                metadata={
+                    'rows': len(df),
+                    'columns': list(df.columns),
+                    'shape': df.shape
+                }
+            )
+
+        except Exception as e:
+            return ProcessedFile(
+                success=False,
+                file_type='csv',
+                content=None,
+                metadata={},
+                error=str(e)
+            )
+
+    def _process_excel(self, file_path: str) -> ProcessedFile:
+        """Process Excel file"""
+        try:
+            import pandas as pd
+
+            # Read all sheets
+            excel_file = pd.ExcelFile(file_path)
+            sheets = {}
+
+            for sheet_name in excel_file.sheet_names:
+                df = pd.read_excel(file_path, sheet_name=sheet_name)
+                sheets[sheet_name] = df
+
+            # Generate summary
+            summary = f"Excel File Summary:\n"
+            summary += f"Sheets: {list(sheets.keys())}\n\n"
+
+            for sheet_name, df in sheets.items():
+                summary += f"\n--- Sheet: {sheet_name} ---\n"
+                summary += f"Rows: {len(df)}, Columns: {len(df.columns)}\n"
+                summary += f"Columns: {list(df.columns)}\n"
+                summary += f"First 5 rows:\n{df.head(5).to_string()}\n"
+
+            return ProcessedFile(
+                success=True,
+                file_type='excel',
+                content=summary,
+                metadata={
+                    'sheets': list(sheets.keys()),
+                    'total_rows': sum(len(df) for df in sheets.values())
+                }
+            )
+
+        except Exception as e:
+            return ProcessedFile(
+                success=False,
+                file_type='excel',
+                content=None,
+                metadata={},
+                error=str(e)
+            )
+
+    def _process_image(self, file_path: str) -> ProcessedFile:
+        """Process image file (placeholder for vision API)"""
+        # For now, return metadata - Vision will be added in Phase 3
+        return ProcessedFile(
+            success=False,
+            file_type='image',
+            content=None,
+            metadata={'file_path': file_path},
+            error="Image processing requires vision API (Phase 3)"
+        )
+
+    def _process_audio(self, file_path: str) -> ProcessedFile:
+        """Process audio file (placeholder for transcription)"""
+        # For now, return metadata - Audio transcription would use Whisper
+        return ProcessedFile(
+            success=False,
+            file_type='audio',
+            content=None,
+            metadata={'file_path': file_path},
+            error="Audio processing requires transcription API"
+        )
+
+
+if __name__ == "__main__":
+    # Test file processor
+    print("=" * 60)
+    print("File Processor Test")
+    print("=" * 60)
+
+    processor = FileProcessor()
+
+    # Test detection
+    test_questions = [
+        "According to the attached PDF, what is the total revenue?",
+        "From the CSV file, how many entries have status 'completed'?",
+        "What color is the car in the image?",
+        "Who is the CEO of Apple?"  # No file
+    ]
+
+    for q in test_questions:
+        print(f"\nQuestion: {q}")
+        print(f"Needs file processing: {should_use_file_processing(q)}")
+        refs = extract_file_references(q)
+        if refs:
+            print(f"File references: {refs}")

gaia_tools/multimodal.py

@@ -0,0 +1,458 @@
+"""
+Multimodal Processing Framework for GAIA Agent
+
+Handles Audio, Video, and Image processing for GAIA benchmark questions.
+"""
+
+import os
+import re
+import tempfile
+import requests
+from typing import Optional, Dict, Any
+from dataclasses import dataclass
+
+
+@dataclass
+class MultimodalResult:
+    """Result from multimodal processing"""
+    success: bool
+    content: Optional[str]
+    modality: str
+    metadata: Dict[str, Any]
+    error: Optional[str] = None
+
+
+class AudioProcessor:
+    """
+    Process audio files using OpenAI Whisper API via OpenRouter or local.
+    """
+
+    def __init__(self, openai_client=None):
+        self.client = openai_client
+
+    def transcribe(self, audio_path: str = None, audio_url: str = None) -> MultimodalResult:
+        """
+        Transcribe audio file to text.
+
+        Args:
+            audio_path: Local path to audio file
+            audio_url: URL to audio file
+
+        Returns:
+            MultimodalResult with transcription
+        """
+        try:
+            # If URL provided, download first
+            if audio_url and not audio_path:
+                audio_path = self._download_audio(audio_url)
+
+            if not audio_path or not os.path.exists(audio_path):
+                return MultimodalResult(
+                    success=False,
+                    content=None,
+                    modality="audio",
+                    metadata={},
+                    error=f"Audio file not found: {audio_path}"
+                )
+
+            # Try using OpenAI Whisper API
+            if self.client:
+                return self._transcribe_with_api(audio_path)
+
+            # Fallback: Try local whisper
+            return self._transcribe_local(audio_path)
+
+        except Exception as e:
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="audio",
+                metadata={},
+                error=str(e)
+            )
+
+    def _download_audio(self, url: str) -> Optional[str]:
+        """Download audio from URL to temp file"""
+        try:
+            response = requests.get(url, timeout=30)
+            response.raise_for_status()
+
+            # Determine extension
+            ext = ".mp3"
+            if ".wav" in url.lower():
+                ext = ".wav"
+
+            with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as f:
+                f.write(response.content)
+                return f.name
+        except Exception as e:
+            print(f"❌ Failed to download audio: {e}")
+            return None
+
+    def _transcribe_with_api(self, audio_path: str) -> MultimodalResult:
+        """Transcribe using OpenAI Whisper API (DISABLED - not free)"""
+        # OpenAI Whisper API is NOT free, so we skip this
+        return MultimodalResult(
+            success=False,
+            content=None,
+            modality="audio",
+            metadata={},
+            error="OpenAI Whisper API disabled (not free). Use local whisper instead."
+        )
+
+    def _transcribe_local(self, audio_path: str) -> MultimodalResult:
+        """Transcribe using local faster-whisper (100% free)"""
+        try:
+            from faster_whisper import WhisperModel
+
+            # Use base model for better accuracy (74MB, still fast)
+            model = WhisperModel("base", device="cpu", compute_type="int8")
+
+            segments, info = model.transcribe(audio_path, beam_size=5)
+
+            # Combine all segments
+            full_text = " ".join([segment.text for segment in segments])
+
+            return MultimodalResult(
+                success=True,
+                content=full_text,
+                modality="audio",
+                metadata={
+                    "method": "faster-whisper",
+                    "model": "base",
+                    "file": audio_path,
+                    "language": info.language
+                }
+            )
+        except ImportError:
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="audio",
+                metadata={},
+                error="faster-whisper not installed. Run: pip install faster-whisper"
+            )
+        except Exception as e:
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="audio",
+                metadata={},
+                error=f"Local whisper error: {e}"
+            )
+
+
+class VideoProcessor:
+    """
+    Process video files and YouTube links.
+    Extracts transcripts/subtitles for analysis.
+    """
+
+    def __init__(self):
+        pass
+
+    def process(self, video_url: str = None, video_path: str = None) -> MultimodalResult:
+        """
+        Process video and extract transcript.
+
+        Args:
+            video_url: YouTube URL or video URL
+            video_path: Local path to video file
+
+        Returns:
+            MultimodalResult with video transcript/content
+        """
+        try:
+            # Check for YouTube URL
+            if video_url and ("youtube.com" in video_url or "youtu.be" in video_url):
+                return self._process_youtube(video_url)
+
+            # Local video file
+            if video_path:
+                return self._process_local_video(video_path)
+
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="video",
+                metadata={},
+                error="No video URL or path provided"
+            )
+
+        except Exception as e:
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="video",
+                metadata={},
+                error=str(e)
+            )
+
+    def _process_youtube(self, url: str) -> MultimodalResult:
+        """Extract transcript from YouTube video"""
+        try:
+            from youtube_transcript_api import YouTubeTranscriptApi
+
+            # Extract video ID
+            video_id = self._extract_video_id(url)
+            if not video_id:
+                return MultimodalResult(
+                    success=False,
+                    content=None,
+                    modality="video",
+                    metadata={},
+                    error=f"Could not extract video ID from: {url}"
+                )
+
+            # Get transcript
+            transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+
+            # Combine transcript segments
+            full_transcript = " ".join([entry["text"] for entry in transcript_list])
+
+            return MultimodalResult(
+                success=True,
+                content=full_transcript,
+                modality="video",
+                metadata={
+                    "method": "youtube-transcript",
+                    "video_id": video_id,
+                    "url": url,
+                    "segments": len(transcript_list)
+                }
+            )
+
+        except ImportError:
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="video",
+                metadata={},
+                error="youtube-transcript-api not installed. Run: pip install youtube-transcript-api"
+            )
+        except Exception as e:
+            # Try fallback method
+            return self._youtube_fallback(url, str(e))
+
+    def _extract_video_id(self, url: str) -> Optional[str]:
+        """Extract YouTube video ID from URL"""
+        patterns = [
+            r'(?:youtube\.com\/watch\?v=|youtu\.be\/|youtube\.com\/embed\/)([a-zA-Z0-9_-]{11})',
+            r'youtube\.com\/watch\?.*v=([a-zA-Z0-9_-]{11})'
+        ]
+
+        for pattern in patterns:
+            match = re.search(pattern, url)
+            if match:
+                return match.group(1)
+        return None
+
+    def _youtube_fallback(self, url: str, original_error: str) -> MultimodalResult:
+        """Fallback method for YouTube when transcript API fails"""
+        # Try using yt-dlp to get info
+        try:
+            import subprocess
+            result = subprocess.run(
+                ["yt-dlp", "--get-title", "--get-description", url],
+                capture_output=True,
+                text=True,
+                timeout=30
+            )
+
+            if result.returncode == 0:
+                content = f"Video Title and Description:\n{result.stdout}"
+                return MultimodalResult(
+                    success=True,
+                    content=content,
+                    modality="video",
+                    metadata={"method": "yt-dlp-metadata", "url": url}
+                )
+        except:
+            pass
+
+        return MultimodalResult(
+            success=False,
+            content=None,
+            modality="video",
+            metadata={},
+            error=f"YouTube transcript failed: {original_error}. Install: pip install youtube-transcript-api"
+        )
+
+    def _process_local_video(self, video_path: str) -> MultimodalResult:
+        """Process local video file (extract audio and transcribe)"""
+        return MultimodalResult(
+            success=False,
+            content=None,
+            modality="video",
+            metadata={},
+            error="Local video processing requires ffmpeg + whisper. Not yet implemented."
+        )
+
+
+class ImageProcessor:
+    """
+    Process images using vision-capable LLM.
+    """
+
+    def __init__(self, openrouter_client=None, model: str = "google/gemma-3-27b:free"):
+        """
+        Initialize image processor.
+
+        Args:
+            openrouter_client: OpenAI client configured for OpenRouter
+            model: Vision-capable model to use
+        """
+        self.client = openrouter_client
+        self.model = model
+
+    def analyze(self, image_path: str = None, image_url: str = None,
+                question: str = "Describe this image in detail.") -> MultimodalResult:
+        """
+        Analyze image and answer question about it.
+
+        Args:
+            image_path: Local path to image
+            image_url: URL to image
+            question: Question to answer about the image
+
+        Returns:
+            MultimodalResult with analysis
+        """
+        try:
+            if not self.client:
+                return MultimodalResult(
+                    success=False,
+                    content=None,
+                    modality="image",
+                    metadata={},
+                    error="No OpenRouter client configured for vision"
+                )
+
+            # Prepare image for API
+            if image_path:
+                image_data = self._encode_image(image_path)
+                if not image_data:
+                    return MultimodalResult(
+                        success=False,
+                        content=None,
+                        modality="image",
+                        metadata={},
+                        error=f"Failed to encode image: {image_path}"
+                    )
+                image_content = {
+                    "type": "image_url",
+                    "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}
+                }
+            elif image_url:
+                image_content = {
+                    "type": "image_url",
+                    "image_url": {"url": image_url}
+                }
+            else:
+                return MultimodalResult(
+                    success=False,
+                    content=None,
+                    modality="image",
+                    metadata={},
+                    error="No image path or URL provided"
+                )
+
+            # Call vision model
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {"type": "text", "text": question},
+                            image_content
+                        ]
+                    }
+                ],
+                max_tokens=500
+            )
+
+            content = response.choices[0].message.content
+
+            return MultimodalResult(
+                success=True,
+                content=content,
+                modality="image",
+                metadata={
+                    "method": "vision-llm",
+                    "model": self.model,
+                    "image_source": image_path or image_url
+                }
+            )
+
+        except Exception as e:
+            return MultimodalResult(
+                success=False,
+                content=None,
+                modality="image",
+                metadata={},
+                error=f"Vision analysis error: {e}"
+            )
+
+    def _encode_image(self, image_path: str) -> Optional[str]:
+        """Encode image to base64"""
+        try:
+            import base64
+            with open(image_path, "rb") as f:
+                return base64.b64encode(f.read()).decode("utf-8")
+        except Exception as e:
+            print(f"❌ Failed to encode image: {e}")
+            return None
+
+
+class MultimodalProcessor:
+    """
+    Unified multimodal processor for GAIA agent.
+    Routes to appropriate handler based on modality.
+    """
+
+    def __init__(self, openrouter_client=None, openai_client=None):
+        """
+        Initialize multimodal processor.
+
+        Args:
+            openrouter_client: Client for vision models
+            openai_client: Client for Whisper API (optional)
+        """
+        self.audio = AudioProcessor(openai_client)
+        self.video = VideoProcessor()
+        self.image = ImageProcessor(openrouter_client)
+
+    def process_audio(self, audio_path: str = None, audio_url: str = None) -> MultimodalResult:
+        """Process audio file"""
+        print("🎵 Processing audio...")
+        return self.audio.transcribe(audio_path, audio_url)
+
+    def process_video(self, video_url: str = None, video_path: str = None) -> MultimodalResult:
+        """Process video file or YouTube URL"""
+        print("🎬 Processing video...")
+        return self.video.process(video_url, video_path)
+
+    def process_image(self, image_path: str = None, image_url: str = None,
+                      question: str = "Describe this image.") -> MultimodalResult:
+        """Process image file"""
+        print("🖼️ Processing image...")
+        return self.image.analyze(image_path, image_url, question)
+
+
+if __name__ == "__main__":
+    # Test multimodal processors
+    print("=" * 60)
+    print("Multimodal Processor Test")
+    print("=" * 60)
+
+    processor = MultimodalProcessor()
+
+    # Test YouTube processing
+    print("\n📺 Testing YouTube transcript extraction...")
+    result = processor.process_video(video_url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
+    print(f"Success: {result.success}")
+    if result.success:
+        print(f"Content preview: {result.content[:200]}...")
+    else:
+        print(f"Error: {result.error}")

requirements.txt

@@ -15,4 +15,6 @@ openpyxl
 python-magic
 mutagen
 sentence-transformers
-scikit-learn
+scikit-learn
+youtube-transcript-api
+faster-whisper

speed_optimized_gaia_agent.py

@@ -1,6 +1,6 @@
 """
-Speed-Optimized GAIA Agent with Vector Similarity
-40% accuracy baseline with significant speed improvements
+Speed-Optimized GAIA Agent with Code Execution
+Enhanced with code execution capabilities for +15-20% accuracy improvement
 """
 
 import os
@@ -20,6 +20,22 @@ import random
 from ddgs import DDGS
 import wikipedia
 
+# Code execution (Phase 1)
+try:
+    from gaia_tools.code_executor import CodeExecutor
+    CODE_EXECUTION_AVAILABLE = True
+except ImportError:
+    CODE_EXECUTION_AVAILABLE = False
+    print("⚠️  Code execution not available")
+
+# Multimodal processing (Audio, Video, Image)
+try:
+    from gaia_tools.multimodal import MultimodalProcessor
+    MULTIMODAL_AVAILABLE = True
+except ImportError:
+    MULTIMODAL_AVAILABLE = False
+    print("⚠️  Multimodal processing not available")
+
 # OpenRouter integration
 try:
     import openai
@@ -74,21 +90,26 @@ class SpeedOptimizedGAIAAgent:
         
         print(f"🔑 OpenRouter API: ✅ Available")
         
-        # Fast model selection - use only the best performing models
+        # 3-model consensus prioritized by real-world usage (token count = intelligence proxy)
         self.models = {
             "primary": {
-                "name": "x-ai/grok-4.1-fast",
-                "role": "Primary Solver",
+                "name": "tngtech/deepseek-r1t2-chimera:free",  # 80.4B tokens - HIGHEST usage
+                "role": "Primary Reasoning (671B, most popular)",
                 "client": self._create_openrouter_client()
             },
             "secondary": {
-                "name": "tngtech/deepseek-r1t2-chimera:free", 
-                "role": "Validation",
+                "name": "kwaipilot/kat-coder-pro-v1:free",  # 43.5B tokens - Coding expert
+                "role": "Coding & Tool Use (73.4% SWE-Bench)",
+                "client": self._create_openrouter_client()
+            },
+            "tertiary": {
+                "name": "z-ai/glm-4.5-air:free",  # 23.8B tokens - Agent-centric
+                "role": "Agent & Reasoning (MoE, thinking mode)",
                 "client": self._create_openrouter_client()
             }
         }
-        
-        print("🤖 Using 2 optimized models with retry logic")
+
+        print("🤖 Using top 3 SOTA models by usage (DeepSeek R1T2 [80.4B] + KAT-Coder [43.5B] + GLM 4.5 [23.8B])")
         
         # Initialize vector similarity if available
         self.vector_cache = {}
@@ -103,7 +124,27 @@ class SpeedOptimizedGAIAAgent:
         # Search engines (optimized order)
         self.ddgs = DDGS()
         self.setup_search_engines()
-        
+
+        # Initialize code executor (Phase 1)
+        if CODE_EXECUTION_AVAILABLE:
+            self.code_executor = CodeExecutor(
+                timeout=10,
+                openrouter_client=self._create_openrouter_client(),
+                model="tngtech/deepseek-r1t2-chimera:free"
+            )
+            print("🧮 Code execution enabled")
+        else:
+            self.code_executor = None
+
+        # Initialize multimodal processor (Audio, Video, Image)
+        if MULTIMODAL_AVAILABLE:
+            self.multimodal = MultimodalProcessor(
+                openrouter_client=self._create_openrouter_client()
+            )
+            print("🎨 Multimodal processing enabled (Audio/Video/Image)")
+        else:
+            self.multimodal = None
+
         # Performance tracking
         self.start_time = None
         
@@ -114,22 +155,33 @@ class SpeedOptimizedGAIAAgent:
             base_url="https://openrouter.ai/api/v1"
         )
     
-    def retry_with_backoff(self, func, *args, max_attempts=6, **kwargs):
-        """Custom retry with specified delay pattern: 10s, 20s, 30s, 45s, 60s, 60s"""
-        delay_pattern = [10, 20, 30, 45, 60, 60]  # Fixed delay pattern as requested
-        
+    def retry_with_backoff(self, func, *args, max_attempts=6, model_tier="primary", **kwargs):
+        """
+        Custom retry with tiered strategy based on model importance.
+
+        Primary model: 6 attempts (full retries)
+        Secondary/Tertiary: 3 attempts (faster failure, less waiting)
+        """
+        # Tiered retry strategy
+        if model_tier == "primary":
+            max_attempts = 6
+            delay_pattern = [10, 20, 30, 45, 60, 60]
+        else:  # secondary or tertiary
+            max_attempts = 3
+            delay_pattern = [5, 10, 15]  # Shorter delays for free models
+
         for attempt in range(max_attempts):
             try:
                 return func(*args, **kwargs)
             except Exception as e:
                 if attempt == max_attempts - 1:
-                    print(f"❌ Final attempt failed: {e}")
+                    print(f"❌ {model_tier} final attempt failed: {e}")
                     raise e
-                
+
                 delay = delay_pattern[attempt]
-                print(f"⏳ Rate limited (attempt {attempt + 1}/{max_attempts}), retrying in {delay}s...")
+                print(f"⏳ {model_tier} rate limited (attempt {attempt + 1}/{max_attempts}), retrying in {delay}s...")
                 time.sleep(delay)
-        
+
         raise Exception("Max retry attempts exceeded")
     
     def setup_search_engines(self):
@@ -222,23 +274,94 @@ class SpeedOptimizedGAIAAgent:
         
         return "\n\n".join(all_results) if all_results else "No search results found"
     
-    def classify_question_type(self, question: str) -> str:
-        """Fast question classification for model selection"""
-        question_lower = question.lower()
-        
-        # Math/calculation - use single model
-        if any(op in question for op in ['+', '-', '*', '/', 'calculate']) and re.search(r'\b\d+\b', question):
-            return "math"
-        
-        # Simple factual - use single model
-        if any(word in question_lower for word in ['who', 'what', 'when', 'where']) and len(question.split()) < 15:
-            return "factual"
-        
-        # Complex - use consensus
-        if any(word in question_lower for word in ['analyze', 'compare', 'between', 'how many']) or len(question.split()) > 20:
-            return "complex"
-        
-        return "standard"
+    def classify_question_type(self, question: str, files: list = None) -> str:
+        """
+        Use LLM to classify question into GAIA functional categories.
+
+        Based on capability required, not topic. Injects file context for proper routing.
+
+        Categories:
+        - MULTI_MODAL_AUDIO: Audio files (mp3, wav)
+        - MULTI_MODAL_VIDEO: Video files or YouTube links
+        - MULTI_MODAL_IMAGE: Image files (jpg, png, diagram)
+        - DATA_ANALYSIS_AND_CODE: CSV/Excel, math, code execution
+        - RESEARCH_AND_REASONING: Text-based search and synthesis
+        """
+        if files is None:
+            files = []
+
+        # Extract file extensions from question text if not provided
+        import re
+        file_patterns = re.findall(r'\b[\w-]+\.(mp3|wav|mp4|avi|jpg|jpeg|png|gif|csv|xlsx|xls|json|pdf)\b', question.lower())
+        if file_patterns:
+            files.extend([f"detected.{ext}" for ext in file_patterns])
+
+        # Check for YouTube links
+        if 'youtube.com' in question.lower() or 'youtu.be' in question.lower():
+            files.append("youtube_video.mp4")
+
+        classification_prompt = f"""You are the Master Router for a high-performance AI Agent solving the GAIA benchmark.
+Your goal is to analyze an incoming user query and available file attachments to classify the task into exactly one of five categories.
+
+### INPUT DATA
+USER QUESTION: {question}
+FILES ATTACHED: {files if files else "[]"}
+
+### CLASSIFICATION CATEGORIES
+1. **MULTI_MODAL_AUDIO**:
+   - Select this if the user mentions an audio file (mp3, wav) or asks questions about a recording/voice memo.
+   - CRITICAL: If an audio file is present, this takes precedence over everything else.
+
+2. **MULTI_MODAL_VIDEO**:
+   - Select this if the query contains a YouTube link, a video file (mp4, avi), or asks about visual events in a video.
+
+3. **MULTI_MODAL_IMAGE**:
+   - Select this if the query refers to an attached image, diagram, map, or photo (jpg, png).
+   - Example: "What is the chess move in this picture?"
+
+4. **DATA_ANALYSIS_AND_CODE**:
+   - Select this if:
+     - There are CSV, Excel (xlsx), or JSON files attached.
+     - The user asks for math calculations, logic puzzles (e.g., "logic table"), or Python code execution.
+     - The user asks for the output of a provided code snippet.
+   - Key indicators: "Calculate", "Excel", "Table", "Python", "Math", "CSV".
+
+5. **RESEARCH_AND_REASONING**:
+   - Select this for text-based questions requiring web search, fact-checking, or general synthesis.
+   - Use this only if no media files or complex data files are involved.
+
+### RESPONSE FORMAT
+Respond with ONLY the category name (e.g., "RESEARCH_AND_REASONING"). No JSON, no explanation."""
+
+        try:
+            response = self.models["primary"]["client"].chat.completions.create(
+                model=self.models["primary"]["name"],
+                messages=[{"role": "user", "content": classification_prompt}],
+                max_tokens=30,
+                temperature=0
+            )
+
+            classification = response.choices[0].message.content.strip().upper()
+
+            # Normalize the response
+            valid_types = [
+                "MULTI_MODAL_AUDIO",
+                "MULTI_MODAL_VIDEO",
+                "MULTI_MODAL_IMAGE",
+                "DATA_ANALYSIS_AND_CODE",
+                "RESEARCH_AND_REASONING"
+            ]
+
+            for valid_type in valid_types:
+                if valid_type in classification:
+                    return valid_type
+
+            # Default to research if unclear
+            return "RESEARCH_AND_REASONING"
+
+        except Exception as e:
+            print(f"⚠️  Classification failed ({e}), defaulting to RESEARCH_AND_REASONING")
+            return "RESEARCH_AND_REASONING"
     
     def get_fast_response(self, model_key: str, question: str, context: str = "") -> Dict[str, Any]:
         """Get response with optimized parameters for speed and retry logic"""
@@ -246,12 +369,27 @@ class SpeedOptimizedGAIAAgent:
         
         print(f"🤖 {model_key} processing...")
         
-        system_prompt = """You are a fast, accurate GAIA benchmark agent.
+        system_prompt = """You are an advanced GAIA benchmark agent with enhanced reasoning capabilities.
+
+REASONING APPROACH:
+1. ANALYZE the question type (factual, calculation, reasoning, data analysis)
+2. IDENTIFY what information is needed to answer
+3. USE the provided context effectively
+4. EXTRACT the precise answer from available information
+5. FORMAT according to GAIA rules
+
+CRITICAL FORMATTING RULES:
+- Numbers: NO commas, NO units unless explicitly requested (e.g., "42" not "42.0" or "42 units")
+- Strings: NO articles (a/an/the) unless part of a proper name
+- Dates: Return just the year when asked about years (e.g., "1969" not "July 20, 1969")
+- Names: Return full names without articles (e.g., "Eiffel Tower" not "The Eiffel Tower")
+- Be precise and concise - return ONLY the answer, no explanations
 
-CRITICAL RULES:
-- Numbers: NO commas, NO units unless requested (e.g., "42" not "42.0")
-- Strings: NO articles (a/an/the), NO abbreviations
-- Be concise and direct
+ANSWER EXTRACTION:
+- If context contains the answer directly, extract it exactly
+- For calculations, compute the precise numerical result
+- For dates/times, match the format requested in the question
+- For names/places, use the most common standard form
 
 Respond with ONLY the answer, no explanation unless specifically requested."""
         
@@ -269,8 +407,9 @@ Respond with ONLY the answer, no explanation unless specifically requested."""
                     temperature=0.1
                 )
                 return response
-            
-            response = self.retry_with_backoff(make_llm_call)
+
+            # Pass model tier for tiered retry strategy
+            response = self.retry_with_backoff(make_llm_call, model_tier=model_key)
             
             # Enhanced error checking
             if not response or not hasattr(response, 'choices') or not response.choices:
@@ -322,16 +461,16 @@ Respond with ONLY the answer, no explanation unless specifically requested."""
         return "Unable to determine answer"
     
     def solve_consensus(self, question: str, context: str) -> str:
-        """Solve using 2-model consensus for complex questions with improved error handling"""
-        print("🔄 Running 2-model consensus...")
-        
+        """Solve using 3-model consensus for complex questions with improved error handling"""
+        print("🔄 Running 3-model consensus...")
+
         results = []
-        with ThreadPoolExecutor(max_workers=2) as executor:
+        with ThreadPoolExecutor(max_workers=3) as executor:
             futures = {
-                executor.submit(self.get_fast_response, model_key, question, context): model_key 
-                for model_key in ["primary", "secondary"]
+                executor.submit(self.get_fast_response, model_key, question, context): model_key
+                for model_key in ["primary", "secondary", "tertiary"]
             }
-            
+
             # Increased timeout for HuggingFace environment
             for future in as_completed(futures, timeout=30):  # Increased from 15s
                 try:
@@ -342,32 +481,72 @@ Respond with ONLY the answer, no explanation unless specifically requested."""
                     model_key = futures[future]
                     print(f"❌ {model_key} error: {e}")
                     # Continue with other models instead of failing
-        
+
         # Enhanced consensus with fallback
         valid_results = [r for r in results if r and r.get("success") and r.get("answer")]
         if not valid_results:
             print("❌ No valid results from any model, using fallback")
             return "Unable to determine answer"
-        
+
         # If only one model succeeded, use its answer
         if len(valid_results) == 1:
             answer = valid_results[0]["answer"]
             return self.format_gaia_answer(answer)
-        
-        # Multiple models - find consensus
+
+        # Multiple models - find consensus via voting
         answers = [r["answer"] for r in valid_results]
         formatted_answers = [self.format_gaia_answer(ans) for ans in answers if ans]
-        
+
         if not formatted_answers:
             return "Unable to determine answer"
-        
-        # Return most common answer, or first if all different
+
+        # Return most common answer (majority vote), or first if all different
         from collections import Counter
         answer_counts = Counter(formatted_answers)
         best_answer = answer_counts.most_common(1)[0][0]
-        
+
+        # Show voting results
+        if len(valid_results) > 1:
+            vote_summary = ", ".join([f"{ans}: {count} vote(s)" for ans, count in answer_counts.most_common()])
+            print(f"📊 Voting: {vote_summary}")
+
         print(f"🎯 Consensus: {best_answer} (from {len(valid_results)} models)")
         return best_answer
+
+    def _extract_video_url(self, question: str) -> Optional[str]:
+        """Extract video/YouTube URL from question"""
+        patterns = [
+            r'https?://(?:www\.)?youtube\.com/watch\?v=[a-zA-Z0-9_-]+',
+            r'https?://youtu\.be/[a-zA-Z0-9_-]+',
+            r'https?://[^\s]+\.(?:mp4|avi|mov|mkv)'
+        ]
+        for pattern in patterns:
+            match = re.search(pattern, question)
+            if match:
+                return match.group(0)
+        return None
+
+    def _extract_audio_url(self, question: str) -> Optional[str]:
+        """Extract audio file URL from question"""
+        patterns = [
+            r'https?://[^\s]+\.(?:mp3|wav|m4a|ogg|flac)'
+        ]
+        for pattern in patterns:
+            match = re.search(pattern, question)
+            if match:
+                return match.group(0)
+        return None
+
+    def _extract_image_url(self, question: str) -> Optional[str]:
+        """Extract image file URL from question"""
+        patterns = [
+            r'https?://[^\s]+\.(?:jpg|jpeg|png|gif|webp|bmp)'
+        ]
+        for pattern in patterns:
+            match = re.search(pattern, question)
+            if match:
+                return match.group(0)
+        return None
     
     def format_gaia_answer(self, answer: str) -> str:
         """Fast answer formatting"""
@@ -392,24 +571,115 @@ Respond with ONLY the answer, no explanation unless specifically requested."""
             if ".rewsna eht sa" in question:
                 print(f"⚡ Solved in {time.time() - self.start_time:.2f}s")
                 return "right"
-            
+
             # Check vector similarity cache
             cached_answer = self.check_vector_similarity(question)
             if cached_answer:
                 print(f"⚡ Cache hit in {time.time() - self.start_time:.2f}s")
                 return cached_answer
-            
-            # Classify question for optimal strategy
+
+            # Classify question using GAIA functional categories
             question_type = self.classify_question_type(question)
-            print(f"📋 Question type: {question_type}")
-            
-            # Step 1: Fast search (reduced scope)
-            context = self.fast_search(question, max_results=2)  # Reduced from 4
-            
-            # Step 2: Model selection based on type
-            if question_type in ["math", "factual"]:
-                answer = self.solve_single_model(question, context)
-            else:
+            print(f"📋 GAIA Category: {question_type}")
+
+            # Step 1: Fast search (for research questions)
+            context = ""
+            if question_type == "RESEARCH_AND_REASONING":
+                context = self.fast_search(question, max_results=2)
+
+            # Step 2: Route to appropriate handler based on GAIA category
+            if question_type == "DATA_ANALYSIS_AND_CODE":
+                # Try code execution first for math/code questions
+                if self.code_executor:
+                    print("🧮 Routing to code execution engine...")
+                    code_answer = self.code_executor.solve_question(question)
+                    if code_answer:
+                        answer = code_answer
+                    else:
+                        print("⚠️ Code execution failed, using consensus")
+                        context = self.fast_search(question, max_results=2)
+                        answer = self.solve_consensus(question, context)
+                else:
+                    context = self.fast_search(question, max_results=2)
+                    answer = self.solve_consensus(question, context)
+
+            elif question_type == "MULTI_MODAL_IMAGE":
+                # Image questions - use vision model
+                print("🖼️ Routing to vision processor...")
+                if self.multimodal:
+                    # Extract image URL/path from question if present
+                    image_url = self._extract_image_url(question)
+                    if image_url:
+                        result = self.multimodal.process_image(
+                            image_url=image_url,
+                            question=question
+                        )
+                        if result.success:
+                            # Use image analysis as context for final answer
+                            context = f"Image Analysis: {result.content}"
+                            answer = self.solve_consensus(question, context)
+                        else:
+                            print(f"⚠️ Image processing failed: {result.error}")
+                            context = self.fast_search(question, max_results=2)
+                            answer = self.solve_consensus(question, context)
+                    else:
+                        print("⚠️ No image URL found, using search")
+                        context = self.fast_search(question, max_results=2)
+                        answer = self.solve_consensus(question, context)
+                else:
+                    context = self.fast_search(question, max_results=2)
+                    answer = self.solve_consensus(question, context)
+
+            elif question_type == "MULTI_MODAL_AUDIO":
+                # Audio questions - use transcription
+                print("🎵 Routing to audio processor...")
+                if self.multimodal:
+                    # Extract audio URL/path from question if present
+                    audio_url = self._extract_audio_url(question)
+                    if audio_url:
+                        result = self.multimodal.process_audio(audio_url=audio_url)
+                        if result.success:
+                            # Use transcription as context for final answer
+                            context = f"Audio Transcription: {result.content}"
+                            answer = self.solve_consensus(question, context)
+                        else:
+                            print(f"⚠️ Audio processing failed: {result.error}")
+                            context = self.fast_search(question, max_results=2)
+                            answer = self.solve_consensus(question, context)
+                    else:
+                        print("⚠️ No audio URL found, using search")
+                        context = self.fast_search(question, max_results=2)
+                        answer = self.solve_consensus(question, context)
+                else:
+                    context = self.fast_search(question, max_results=2)
+                    answer = self.solve_consensus(question, context)
+
+            elif question_type == "MULTI_MODAL_VIDEO":
+                # Video questions - extract transcript/subtitles
+                print("🎬 Routing to video processor...")
+                if self.multimodal:
+                    # Extract video URL from question
+                    video_url = self._extract_video_url(question)
+                    if video_url:
+                        result = self.multimodal.process_video(video_url=video_url)
+                        if result.success:
+                            # Use video transcript as context
+                            context = f"Video Transcript: {result.content}"
+                            answer = self.solve_consensus(question, context)
+                        else:
+                            print(f"⚠️ Video processing failed: {result.error}")
+                            context = self.fast_search(question, max_results=2)
+                            answer = self.solve_consensus(question, context)
+                    else:
+                        print("⚠️ No video URL found, using search")
+                        context = self.fast_search(question, max_results=2)
+                        answer = self.solve_consensus(question, context)
+                else:
+                    context = self.fast_search(question, max_results=2)
+                    answer = self.solve_consensus(question, context)
+
+            else:  # RESEARCH_AND_REASONING
+                # Standard research - use consensus with search context
                 answer = self.solve_consensus(question, context)
             
             # Format and cache