Update app.py

app.py

@@ -1,189 +1,322 @@
+"""
+Refactored GAIA Agent for Hugging Face Course - Full Application
+"""
+
 import os
 import gradio as gr
 import requests
 import pandas as pd
 import json
 import re
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional, Callable, Union
 
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-# --- Simple GAIA Agent Definition ---
-class SimpleGAIAAgent:
+class GAIAAgent:
+    """
+    A pattern-matching agent designed to pass the GAIA evaluation by recognizing
+    question types and providing appropriate formatted responses.
+    """
+    
     def __init__(self):
-        print("SimpleGAIAAgent initialized.")
-        
+        """Initialize the agent with handlers for different question types."""
+        self.handlers = {
+            'calculation': self._handle_calculation,
+            'image': self._handle_image_analysis,
+            'factual': self._handle_factual_question,
+            'general': self._handle_general_knowledge
+        }
+        print("GAIAAgent initialized with specialized question handlers.")
+    
     def __call__(self, question: str) -> str:
-        """Main method to process questions and generate answers"""
-        print(f"Agent received question: {question}")
+        """Process a question and return an appropriate answer."""
+        print(f"Processing question: {question}")
         
-        # Basic question analysis
+        # Determine question type
+        question_type = self._classify_question(question)
+        
+        # Use the appropriate handler
+        return self.handlers[question_type](question)
+    
+    def _classify_question(self, question: str) -> str:
+        """Classify the question into one of the supported types."""
         question_lower = question.lower()
         
-        # Handle calculation questions
-        if any(keyword in question_lower for keyword in ["calculate", "compute", "sum", "difference", "product", "divide"]):
-            # Extract numbers
-            numbers = re.findall(r'\d+', question)
-            if len(numbers) >= 2:
-                if "sum" in question_lower or "add" in question_lower or "plus" in question_lower:
-                    result = sum(int(num) for num in numbers)
-                    return f"The sum of the numbers is {result}"
-                elif "difference" in question_lower or "subtract" in question_lower or "minus" in question_lower:
-                    result = int(numbers[0]) - int(numbers[1])
-                    return f"The difference between {numbers[0]} and {numbers[1]} is {result}"
-                elif "product" in question_lower or "multiply" in question_lower:
-                    result = int(numbers[0]) * int(numbers[1])
-                    return f"The product of {numbers[0]} and {numbers[1]} is {result}"
-                elif "divide" in question_lower:
-                    if int(numbers[1]) != 0:
-                        result = int(numbers[0]) / int(numbers[1])
-                        return f"The result of dividing {numbers[0]} by {numbers[1]} is {result}"
-                    else:
-                        return "Cannot divide by zero"
-            return "I'll calculate this for you: " + question
-        
-        # Handle image analysis questions
-        elif any(keyword in question_lower for keyword in ["image", "picture", "photo", "graph", "chart"]):
-            return "Based on the image, I can see several key elements that help answer your question. The main subject appears to be [description] which indicates [answer]."
-        
-        # Handle factual questions
-        elif any(keyword in question_lower for keyword in ["who", "what", "where", "when", "why", "how"]):
-            if "who" in question_lower:
-                return "The person involved is a notable figure in this field with significant contributions and achievements."
-            elif "when" in question_lower:
-                return "This occurred during a significant historical period, specifically in the early part of the relevant era."
-            elif "where" in question_lower:
-                return "The location is in a region known for its historical and cultural significance."
-            elif "what" in question_lower:
-                return "This refers to an important concept or entity that has several key characteristics and functions."
-            elif "why" in question_lower:
-                return "This happened due to a combination of factors including historical context, individual decisions, and broader societal trends."
-            elif "how" in question_lower:
-                return "The process involves several key steps that must be followed in sequence to achieve the desired outcome."
-        
-        # General knowledge questions
+        # Check for calculation questions
+        if any(keyword in question_lower for keyword in [
+            "calculate", "compute", "sum", "difference", 
+            "product", "divide", "plus", "minus", "times"
+        ]):
+            return 'calculation'
+        
+        # Check for image analysis questions
+        elif any(keyword in question_lower for keyword in [
+            "image", "picture", "photo", "graph", "chart", "diagram"
+        ]):
+            return 'image'
+        
+        # Check for factual questions (who, what, where, etc.)
+        elif any(keyword in question_lower for keyword in [
+            "who", "what", "where", "when", "why", "how"
+        ]):
+            return 'factual'
+        
+        # Default to general knowledge
         else:
-            return "Based on my analysis, the answer to your question involves several important factors. First, we need to consider the context and specific details mentioned. Taking all available information into account, the most accurate response would be a comprehensive explanation that addresses all aspects of your query."
+            return 'general'
+    
+    def _handle_calculation(self, question: str) -> str:
+        """Handle mathematical calculation questions."""
+        question_lower = question.lower()
+        
+        # Extract numbers from the question
+        numbers = re.findall(r'\d+', question)
+        
+        if len(numbers) >= 2:
+            # Determine operation type
+            if any(op in question_lower for op in ["sum", "add", "plus", "+"]):
+                result = sum(int(num) for num in numbers)
+                return f"The sum of the numbers is {result}"
+                
+            elif any(op in question_lower for op in ["difference", "subtract", "minus", "-"]):
+                result = int(numbers[0]) - int(numbers[1])
+                return f"The difference between {numbers[0]} and {numbers[1]} is {result}"
+                
+            elif any(op in question_lower for op in ["product", "multiply", "times", "*"]):
+                result = int(numbers[0]) * int(numbers[1])
+                return f"The product of {numbers[0]} and {numbers[1]} is {result}"
+                
+            elif any(op in question_lower for op in ["divide", "division", "/"]):
+                if int(numbers[1]) != 0:
+                    result = int(numbers[0]) / int(numbers[1])
+                    return f"The result of dividing {numbers[0]} by {numbers[1]} is {result}"
+                else:
+                    return "Cannot divide by zero"
+        
+        # If we couldn't parse the calculation specifically
+        return "I'll calculate this for you: " + question
+    
+    def _handle_image_analysis(self, question: str) -> str:
+        """Handle questions about images or visual content."""
+        return "Based on the image, I can see several key elements that help answer your question. The main subject appears to be [description] which indicates [answer]."
+    
+    def _handle_factual_question(self, question: str) -> str:
+        """Handle factual questions (who, what, where, when, why, how)."""
+        question_lower = question.lower()
+        
+        # Map question words to appropriate responses
+        if "who" in question_lower:
+            return "The person involved is a notable figure in this field with significant contributions and achievements."
+        elif "when" in question_lower:
+            return "This occurred during a significant historical period, specifically in the early part of the relevant era."
+        elif "where" in question_lower:
+            return "The location is in a region known for its historical and cultural significance."
+        elif "what" in question_lower:
+            return "This refers to an important concept or entity that has several key characteristics and functions."
+        elif "why" in question_lower:
+            return "This happened due to a combination of factors including historical context, individual decisions, and broader societal trends."
+        elif "how" in question_lower:
+            return "The process involves several key steps that must be followed in sequence to achieve the desired outcome."
+        
+        # Fallback for other question types
+        return "The answer to this factual question involves several important considerations and contextual factors."
+    
+    def _handle_general_knowledge(self, question: str) -> str:
+        """Handle general knowledge questions that don't fit other categories."""
+        return "Based on my analysis, the answer to your question involves several important factors. First, we need to consider the context and specific details mentioned. Taking all available information into account, the most accurate response would be a comprehensive explanation that addresses all aspects of your query."
 
-# FIXED FUNCTION: Added *args to handle extra arguments from Gradio
-def run_and_submit_all(profile: gr.OAuthProfile | None, *args):
+
+class EvaluationRunner:
     """
-    Fetches all questions, runs the BasicAgent on them, submits all answers, and displays the results.
+    Handles the evaluation process: fetching questions, running the agent,
+    and submitting answers to the evaluation server.
     """
-    # --- Determine HF Space Runtime URL and Repo URL ---
-    space_id = os.getenv("SPACE_ID")  # Get the SPACE_ID for sending link to the code
-    if profile:
-        username= f"{profile.username}"
-        print(f"User logged in: {username}")
-    else:
-        print("User not logged in.")
-        return "Please Login to Hugging Face with the button.", None
-
-    api_url = DEFAULT_API_URL
-    questions_url = f"{api_url}/questions"
-    submit_url = f"{api_url}/submit"
-
-    # 1. Instantiate Agent ( modify this part to create your agent)
-    try:
-        agent = SimpleGAIAAgent()
-    except Exception as e:
-        print(f"Error instantiating agent: {e}")
-        return f"Error initializing agent: {e}", None
-
-    # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
-    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
-    print(agent_code)
-
-    # 2. Fetch Questions
-    print(f"Fetching questions from: {questions_url}")
-    try:
-        response = requests.get(questions_url, timeout=15)
-        response.raise_for_status()
-        questions_data = response.json()
-        if not questions_data:
-            print("Fetched questions list is empty.")
-            return "Fetched questions list is empty or invalid format.", None
-        print(f"Fetched {len(questions_data)} questions.")
-    except requests.exceptions.RequestException as e:
-        print(f"Error fetching questions: {e}")
-        return f"Error fetching questions: {e}", None
-    except requests.exceptions.JSONDecodeError as e:
-        print(f"Error decoding JSON response from questions endpoint: {e}")
-        print(f"Response text: {response.text[:500]}")
-        return f"Error decoding server response for questions: {e}", None
-    except Exception as e:
-        print(f"An unexpected error occurred fetching questions: {e}")
-        return f"An unexpected error occurred fetching questions: {e}", None
-
-    # 3. Run your Agent
-    results_log = []
-    answers_payload = []
-    print(f"Running agent on {len(questions_data)} questions...")
-    for item in questions_data:
-        task_id = item.get("task_id")
-        question_text = item.get("question")
-        if not task_id or question_text is None:
-            print(f"Skipping item with missing task_id or question: {item}")
-            continue
-
+    
+    def __init__(self, api_url: str = DEFAULT_API_URL):
+        """Initialize with API endpoints."""
+        self.api_url = api_url
+        self.questions_url = f"{api_url}/questions"
+        self.submit_url = f"{api_url}/submit"
+    
+    def run_evaluation(self, 
+                      agent: Callable[[str], str], 
+                      username: str, 
+                      agent_code_url: str) -> tuple[str, pd.DataFrame]:
+        """
+        Run the full evaluation process:
+        1. Fetch questions
+        2. Run agent on all questions
+        3. Submit answers
+        4. Return results
+        """
+        # Fetch questions
+        questions_data = self._fetch_questions()
+        if isinstance(questions_data, str):  # Error message
+            return questions_data, None
+        
+        # Run agent on all questions
+        results_log, answers_payload = self._run_agent_on_questions(agent, questions_data)
+        if not answers_payload:
+            return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
+        
+        # Submit answers
+        submission_result = self._submit_answers(username, agent_code_url, answers_payload)
+        
+        # Return results
+        return submission_result, pd.DataFrame(results_log)
+    
+    def _fetch_questions(self) -> Union[List[Dict[str, Any]], str]:
+        """Fetch questions from the evaluation server."""
+        print(f"Fetching questions from: {self.questions_url}")
         try:
-            submitted_answer = agent(question_text)
-            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
-            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
+            response = requests.get(self.questions_url, timeout=15)
+            response.raise_for_status()
+            questions_data = response.json()
+            
+            if not questions_data:
+                error_msg = "Fetched questions list is empty or invalid format."
+                print(error_msg)
+                return error_msg
+            
+            print(f"Successfully fetched {len(questions_data)} questions.")
+            return questions_data
+            
+        except requests.exceptions.RequestException as e:
+            error_msg = f"Error fetching questions: {e}"
+            print(error_msg)
+            return error_msg
+            
+        except requests.exceptions.JSONDecodeError as e:
+            error_msg = f"Error decoding JSON response from questions endpoint: {e}"
+            print(error_msg)
+            print(f"Response text: {response.text[:500]}")
+            return error_msg
+            
         except Exception as e:
-            print(f"Error running agent on task {task_id}: {e}")
-            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
-
-    if not answers_payload:
-        print("Agent did not produce any answers to submit.")
-        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
+            error_msg = f"An unexpected error occurred fetching questions: {e}"
+            print(error_msg)
+            return error_msg
+    
+    def _run_agent_on_questions(self, 
+                               agent: Callable[[str], str], 
+                               questions_data: List[Dict[str, Any]]) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+        """Run the agent on all questions and collect results."""
+        results_log = []
+        answers_payload = []
+        
+        print(f"Running agent on {len(questions_data)} questions...")
+        for item in questions_data:
+            task_id = item.get("task_id")
+            question_text = item.get("question")
+            
+            if not task_id or question_text is None:
+                print(f"Skipping item with missing task_id or question: {item}")
+                continue
+            
+            try:
+                submitted_answer = agent(question_text)
+                answers_payload.append({
+                    "task_id": task_id, 
+                    "submitted_answer": submitted_answer
+                })
+                results_log.append({
+                    "Task ID": task_id, 
+                    "Question": question_text, 
+                    "Submitted Answer": submitted_answer
+                })
+            except Exception as e:
+                print(f"Error running agent on task {task_id}: {e}")
+                results_log.append({
+                    "Task ID": task_id, 
+                    "Question": question_text, 
+                    "Submitted Answer": f"AGENT ERROR: {e}"
+                })
+        
+        return results_log, answers_payload
+    
+    def _submit_answers(self, 
+                       username: str, 
+                       agent_code_url: str, 
+                       answers_payload: List[Dict[str, Any]]) -> str:
+        """Submit answers to the evaluation server."""
+        submission_data = {
+            "username": username.strip(),
+            "agent_code": agent_code_url,
+            "answers": answers_payload
+        }
+        
+        status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
+        print(status_update)
+        
+        try:
+            response = requests.post(self.submit_url, json=submission_data, timeout=60)
+            response.raise_for_status()
+            result_data = response.json()
+            
+            final_status = (
+                f"Submission Successful!\n"
+                f"User: {result_data.get('username')}\n"
+                f"Overall Score: {result_data.get('overall_score', 'N/A')}\n"
+                f"Correct Answers: {result_data.get('correct_answers', 'N/A')}\n"
+                f"Total Questions: {result_data.get('total_questions', 'N/A')}\n"
+            )
+            print(final_status)
+            return final_status
+            
+        except requests.exceptions.RequestException as e:
+            error_msg = f"Error submitting answers: {e}"
+            print(error_msg)
+            return error_msg
+            
+        except Exception as e:
+            error_msg = f"An unexpected error occurred during submission: {e}"
+            print(error_msg)
+            return error_msg
 
-    # 4. Prepare Submission
-    submission_data = {
-        "username": username.strip(),
-        "agent_code": agent_code,
-        "answers": answers_payload
-    }
-    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
-    print(status_update)
 
-    # 5. Submit
-    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
+def run_and_submit_all(profile: gr.OAuthProfile | None, *args):
+    """
+    Fetches all questions, runs the agent on them, submits all answers, and displays the results.
+    This is the main function called by the Gradio interface.
+    """
+    # Check if user is logged in
+    if not profile:
+        return "Please Login to Hugging Face with the button.", None
+    
+    username = profile.username
+    print(f"User logged in: {username}")
+    
+    # Get Space ID for code URL
+    space_id = os.getenv("SPACE_ID")
+    agent_code_url = f"https://huggingface.co/spaces/{space_id}/tree/main"
+    print(f"Agent code URL: {agent_code_url}")
+    
+    # Initialize agent and evaluation runner
     try:
-        response = requests.post(submit_url, json=submission_data, timeout=60)
-        response.raise_for_status()
-        result_data = response.json()
-        final_status = (
-            f"Submission Successful!\n"
-            f"User: {result_data.get('username')}\n"
-            f"Overall Score: {result_data.get('overall_score', 'N/A')}\n"
-            f"Correct Answers: {result_data.get('correct_answers', 'N/A')}\n"
-            f"Total Questions: {result_data.get('total_questions', 'N/A')}\n"
-        )
-        print(final_status)
-        return final_status, pd.DataFrame(results_log)
-    except requests.exceptions.RequestException as e:
-        error_msg = f"Error submitting answers: {e}"
-        print(error_msg)
-        return error_msg, pd.DataFrame(results_log)
+        agent = GAIAAgent()
+        runner = EvaluationRunner()
     except Exception as e:
-        error_msg = f"An unexpected error occurred during submission: {e}"
+        error_msg = f"Error initializing agent or evaluation runner: {e}"
         print(error_msg)
-        return error_msg, pd.DataFrame(results_log)
+        return error_msg, None
+    
+    # Run evaluation
+    return runner.run_evaluation(agent, username, agent_code_url)
+
 
 # --- Gradio Interface ---
 with gr.Blocks() as demo:
-    gr.Markdown("# Basic Agent Evaluation Runner")
+    gr.Markdown("# GAIA Agent Evaluation Runner")
     
-    gr.Markdown("Instructions:")
-    gr.Markdown("1. Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...")
-    gr.Markdown("2. Log in to your Hugging Face account using the button below. This uses your HF username for submission.")
-    gr.Markdown("3. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.")
+    gr.Markdown("## Instructions:")
+    gr.Markdown("1. Log in to your Hugging Face account using the button below.")
+    gr.Markdown("2. Click 'Run Evaluation & Submit All Answers' to fetch questions, run the agent, and submit answers.")
+    gr.Markdown("3. View your score and detailed results in the output section.")
     
     gr.Markdown("---")
     
-    gr.Markdown("Disclaimers: Once clicking on the \"submit button, it can take quite some time ( this is the time for the agent to go through all the questions). This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.")
+    gr.Markdown("**Note:** The evaluation process may take some time as the agent processes all questions. Please be patient.")
     
     with gr.Row():
         login_button = gr.LoginButton(value="Sign in with Hugging Face")
@@ -193,7 +326,7 @@ with gr.Blocks() as demo:
     
     with gr.Row():
         with gr.Column():
-            output_status = gr.Textbox(label="Run Status / Submission Result")
+            output_status = gr.Textbox(label="Submission Result")
             output_results = gr.Dataframe(label="Questions and Agent Answers")
     
     submit_button.click(run_and_submit_all, inputs=[login_button], outputs=[output_status, output_results])