Update app.py

app.py

@@ -1,186 +1,384 @@
-"""
-Minimal Gradio interface for a simple AI assistant without smolagents
-
-This is a standalone version that uses only Hugging Face Inference API directly.
-It creates a simple Gradio interface for text generation.
-"""
-
 import os
-import sys
-import json
-import requests
 import gradio as gr
+import requests
+import inspect
+import pandas as pd
+import json
+import re
+from typing import List, Dict, Any, Optional
 
-# Check if running in Hugging Face Spaces
-IS_HF_SPACES = os.environ.get("SPACE_ID") is not None
-
-class MinimalAIAssistant:
-    """
-    Minimal AI Assistant using Hugging Face Inference API directly
-    """
-    def __init__(self, api_key=None, model_id="mistralai/Mixtral-8x7B-Instruct-v0.1"):
-        """
-        Initialize the minimal AI assistant
-        
-        Args:
-            api_key: Hugging Face API key
-            model_id: Model ID to use
-        """
-        self.api_key = api_key or os.environ.get("HF_API_KEY", "")
-        self.model_id = model_id
-        self.api_url = f"https://api-inference.huggingface.co/models/{model_id}"
-        self.headers = {"Authorization": f"Bearer {self.api_key}"}
-        
-        # System prompt
-        self.system_prompt = """
-You are an advanced AI assistant designed to help with various tasks.
-You can answer questions, provide information, and assist with problem-solving.
-Always be helpful, accurate, and concise in your responses.
-"""
+# (Keep Constants as is)
+# --- Constants ---
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-    def query(self, prompt):
-        """
-        Query the model with a prompt
+# --- Enhanced GAIA Agent Definition ---
+# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
+class EnhancedGAIAAgent:
+    def __init__(self):
+        print("EnhancedGAIAAgent initialized.")
+        self.tools = {
+            "web_search": self._web_search,
+            "calculator": self._calculator,
+            "image_analysis": self._image_analysis,
+            "text_analysis": self._text_analysis,
+            "code_execution": self._code_execution
+        }
+        
+        # Tracking for reasoning steps
+        self.reasoning_steps = []
+        self.max_reasoning_steps = 5
+        
+    def _web_search(self, query: str) -> str:
+        """Simulates web search functionality"""
+        print(f"Performing web search for: {query}")
+        
+        # Simulate search results based on query keywords
+        if "population" in query.lower():
+            return "The population of the queried location is approximately X million people as of 2023."
+        elif "capital" in query.lower():
+            return "The capital city of the queried location is X, with a population of Y million."
+        elif "president" in query.lower() or "prime minister" in query.lower() or "leader" in query.lower():
+            return "The current leader of the queried location is X, who has been in office since Y."
+        elif "tallest" in query.lower() or "highest" in query.lower():
+            return "The tallest structure in the queried location is X, with a height of Y meters."
+        elif "founded" in query.lower() or "established" in query.lower() or "history" in query.lower():
+            return "The queried entity was established/founded in X year. Its history includes Y and Z significant events."
+        elif "weather" in query.lower() or "temperature" in query.lower() or "climate" in query.lower():
+            return "The current weather/climate in the queried location is X with temperatures ranging from Y to Z degrees."
+        else:
+            return f"Search results for '{query}' include various websites and information sources that may contain relevant information."
+    
+    def _calculator(self, expression: str) -> str:
+        """Performs mathematical calculations"""
+        print(f"Calculating: {expression}")
+        
+        # Clean the expression
+        cleaned_expr = expression.replace('×', '*').replace('÷', '/')
+        cleaned_expr = re.sub(r'[^0-9+\-*/().^ ]', '', cleaned_expr)
         
-        Args:
-            prompt: User prompt
-            
-        Returns:
-            Model response
-        """
         try:
-            # Format the prompt with system message
-            formatted_prompt = f"{self.system_prompt}\n\nUser: {prompt}\n\nAssistant:"
+            # Handle exponentiation separately
+            if '^' in cleaned_expr:
+                cleaned_expr = cleaned_expr.replace('^', '**')
             
-            # Prepare the payload
-            payload = {
-                "inputs": formatted_prompt,
-                "parameters": {
-                    "max_new_tokens": 1024,
-                    "temperature": 0.7,
-                    "top_p": 0.95,
-                    "do_sample": True
-                }
-            }
+            # Safely evaluate the expression
+            result = eval(cleaned_expr)
+            return f"The result of {expression} is {result}"
+        except Exception as e:
+            return f"Error calculating {expression}: {str(e)}"
+    
+    def _image_analysis(self, image_description: str) -> str:
+        """Simulates image analysis functionality"""
+        print(f"Analyzing image: {image_description}")
+        
+        # Simulate image analysis based on description keywords
+        if "person" in image_description.lower() or "people" in image_description.lower() or "human" in image_description.lower():
+            return "The image contains one or more people. They appear to be [activity/pose/expression]."
+        elif "animal" in image_description.lower() or "dog" in image_description.lower() or "cat" in image_description.lower():
+            return "The image shows an animal, likely a [specific animal]. It appears to be [activity/state]."
+        elif "building" in image_description.lower() or "architecture" in image_description.lower():
+            return "The image depicts a building or architectural structure. It appears to be [style/type] architecture."
+        elif "landscape" in image_description.lower() or "nature" in image_description.lower():
+            return "The image shows a natural landscape featuring [elements like mountains, rivers, forests, etc.]."
+        elif "chart" in image_description.lower() or "graph" in image_description.lower() or "diagram" in image_description.lower():
+            return "The image contains a chart/graph showing data about [topic]. The trend appears to be [increasing/decreasing/stable]."
+        else:
+            return f"The image appears to show {image_description}. Key elements include [objects/subjects] and [notable features]."
+    
+    def _text_analysis(self, text: str) -> str:
+        """Analyzes text for sentiment, entities, and key information"""
+        print(f"Analyzing text (first 50 chars): {text[:50]}...")
+        
+        # Count words and sentences
+        word_count = len(text.split())
+        sentence_count = len(re.split(r'[.!?]+', text))
+        
+        # Simple sentiment analysis
+        positive_words = ['good', 'great', 'excellent', 'positive', 'happy', 'best', 'love', 'wonderful', 'fantastic']
+        negative_words = ['bad', 'poor', 'negative', 'terrible', 'worst', 'hate', 'awful', 'horrible', 'disappointing']
+        
+        positive_count = sum(1 for word in text.lower().split() if word in positive_words)
+        negative_count = sum(1 for word in text.lower().split() if word in negative_words)
+        
+        if positive_count > negative_count:
+            sentiment = "positive"
+        elif negative_count > positive_count:
+            sentiment = "negative"
+        else:
+            sentiment = "neutral"
+        
+        return f"Text analysis: {word_count} words, {sentence_count} sentences. The sentiment appears to be {sentiment}."
+    
+    def _code_execution(self, code: str) -> str:
+        """Simulates code execution and analysis"""
+        print(f"Analyzing code (first 50 chars): {code[:50]}...")
+        
+        # Identify language
+        language = "unknown"
+        if "def " in code or "import " in code or "print(" in code:
+            language = "Python"
+        elif "function " in code or "var " in code or "const " in code or "let " in code:
+            language = "JavaScript"
+        elif "public class " in code or "System.out.println" in code:
+            language = "Java"
+        elif "#include" in code or "int main" in code:
+            language = "C/C++"
+        
+        # Simple code analysis
+        lines = code.count('\n') + 1
+        
+        return f"Code analysis: {lines} lines of {language} code. The code appears to [purpose/functionality]."
+    
+    def _reason(self, question: str) -> List[str]:
+        """Performs step-by-step reasoning about the question"""
+        reasoning = []
+        
+        # Initial analysis
+        reasoning.append(f"Question: '{question}'")
+        reasoning.append("Let me analyze what this question is asking for.")
+        
+        # Identify question type
+        if any(keyword in question.lower() for keyword in ["calculate", "compute", "sum", "difference", "product", "divide"]):
+            reasoning.append("This appears to be a calculation question.")
             
-            # Make the API request
-            response = requests.post(self.api_url, headers=self.headers, json=payload)
+            # Extract mathematical expression
+            expression = re.search(r'calculate\s+(.+?)(?:\?|$)', question.lower())
+            if expression:
+                reasoning.append(f"I need to calculate: {expression.group(1)}")
+                reasoning.append(f"Using the calculator tool to compute this.")
+            else:
+                reasoning.append("I need to identify the mathematical operation required.")
+                
+        elif any(keyword in question.lower() for keyword in ["image", "picture", "photo", "graph", "chart"]):
+            reasoning.append("This question involves analyzing an image or visual content.")
+            reasoning.append("I should use image analysis to identify key elements in the image.")
             
-            # Check for errors
-            if response.status_code != 200:
-                return f"Error: API returned status code {response.status_code}. {response.text}"
+        elif any(keyword in question.lower() for keyword in ["population", "capital", "country", "city", "president", "leader"]):
+            reasoning.append("This is a factual question about geography, demographics, or leadership.")
+            reasoning.append("I should search for the most up-to-date information.")
             
-            # Parse the response
-            result = response.json()
+        elif any(keyword in question.lower() for keyword in ["code", "function", "program", "algorithm"]):
+            reasoning.append("This question involves code analysis or programming.")
+            reasoning.append("I should examine the code structure and functionality.")
             
-            # Extract the generated text
-            if isinstance(result, list) and len(result) > 0:
-                generated_text = result[0].get("generated_text", "")
-                # Remove the prompt from the response
-                if generated_text.startswith(formatted_prompt):
-                    generated_text = generated_text[len(formatted_prompt):].strip()
-                return generated_text
-            else:
-                return "Error: Unexpected response format from API"
+        else:
+            reasoning.append("This appears to be a general knowledge question.")
+            reasoning.append("I'll need to search for relevant information and synthesize an answer.")
         
-        except Exception as e:
-            return f"Error querying model: {str(e)}"
-
-
-def create_gradio_interface():
-    """
-    Create a Gradio interface for the minimal AI assistant
+        return reasoning
     
-    Returns:
-        Gradio interface
-    """
-    # Initialize the assistant
-    assistant = MinimalAIAssistant()
-    
-    def process_query(query, api_key=""):
-        """
-        Process a user query
+    def __call__(self, question: str) -> str:
+        """Main method to process questions and generate answers"""
+        print(f"Agent received question: {question}")
         
-        Args:
-            query: User query
-            api_key: Hugging Face API key (optional)
-            
-        Returns:
-            Assistant's response
-        """
-        # Update API key if provided
-        if api_key:
-            assistant.api_key = api_key
-            assistant.headers = {"Authorization": f"Bearer {api_key}"}
-        
-        # Check if API key is set
-        if not assistant.api_key:
-            return "Error: No API key provided. Please enter your Hugging Face API key."
-        
-        # Process the query
-        return assistant.query(query)
-    
-    # Create the interface
-    with gr.Blocks(title="Minimal AI Assistant") as interface:
-        gr.Markdown("# Minimal AI Assistant")
-        gr.Markdown("""
-        This is a minimal AI assistant using the Hugging Face Inference API.
-        Enter your query below and the assistant will respond.
-        """)
-        
-        api_key_input = gr.Textbox(
-            label="Hugging Face API Key",
-            placeholder="Enter your Hugging Face API key here...",
-            type="password"
-        )
+        # Step 1: Reasoning
+        self.reasoning_steps = self._reason(question)
         
-        query_input = gr.Textbox(
-            label="Your Query",
-            placeholder="Enter your query here...",
-            lines=3
-        )
+        # Step 2: Determine approach and tools to use
+        answer = ""
         
-        submit_button = gr.Button("Submit")
+        # Handle calculation questions
+        if any(keyword in question.lower() for keyword in ["calculate", "compute", "sum", "difference", "product", "divide"]):
+            # Extract mathematical expression
+            expression_match = re.search(r'calculate\s+(.+?)(?:\?|$)', question.lower())
+            if expression_match:
+                expression = expression_match.group(1).strip()
+                answer = self._calculator(expression)
+            else:
+                # Try to extract numbers and operations
+                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)
+                        answer = 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])
+                        answer = 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])
+                        answer = 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])
+                            answer = f"The result of dividing {numbers[0]} by {numbers[1]} is {result}"
+                        else:
+                            answer = "Cannot divide by zero"
+                else:
+                    answer = "I couldn't identify a clear calculation to perform."
         
-        response_output = gr.Textbox(
-            label="Assistant Response",
-            lines=15
-        )
+        # Handle image analysis questions
+        elif any(keyword in question.lower() for keyword in ["image", "picture", "photo", "graph", "chart"]):
+            # Extract image description if available
+            image_desc = question
+            answer = self._image_analysis(image_desc)
         
-        # Sample queries
-        gr.Markdown("### Sample Queries")
-        sample_queries = [
-            "What is the capital of France?",
-            "Explain the concept of machine learning in simple terms.",
-            "Write a short poem about artificial intelligence."
-        ]
-        
-        for query in sample_queries:
-            sample_button = gr.Button(f"Try: {query}")
-            sample_button.click(
-                fn=lambda q=query: q,
-                outputs=query_input
-            )
-        
-        # Set up event handlers
-        submit_button.click(
-            fn=process_query,
-            inputs=[query_input, api_key_input],
-            outputs=response_output
-        )
+        # Handle factual questions
+        elif any(keyword in question.lower() for keyword in ["who", "what", "where", "when", "why", "how"]):
+            search_query = question.replace("?", "")
+            search_results = self._web_search(search_query)
+            
+            # Process and synthesize search results
+            answer = f"Based on available information: {search_results}"
+            
+            # Add specific details for common question types
+            if "who" in question.lower():
+                answer += " The individual mentioned is known for their contributions to the field."
+            elif "when" in question.lower():
+                answer += " This occurred during a significant period in history."
+            elif "where" in question.lower():
+                answer += " The location is notable for its geographical and cultural significance."
         
-        # Add examples
-        gr.Examples(
-            examples=sample_queries,
-            inputs=query_input
-        )
-    
-    return interface
+        # Handle code questions
+        elif any(keyword in question.lower() for keyword in ["code", "function", "program", "algorithm"]):
+            # Extract code if present or use the question itself
+            code_sample = question
+            answer = self._code_execution(code_sample)
+        
+        # General knowledge questions
+        else:
+            # Combine web search and text analysis
+            search_results = self._web_search(question)
+            text_analysis = self._text_analysis(question)
+            
+            answer = f"To answer your question: {search_results}"
+        
+        # Add reasoning steps if available
+        if self.reasoning_steps:
+            reasoning_summary = " ".join(self.reasoning_steps[-2:])  # Use last two reasoning steps
+            answer = f"{answer}\n\nReasoning: {reasoning_summary}"
+        
+        return answer
+
+def run_and_submit_all(profile: gr.OAuthProfile | None):
+    """
+    Fetches all questions, runs the BasicAgent on them, submits all answers, and displays the results.
+    """
+    # --- 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"
 
-# Create and launch the interface
-interface = create_gradio_interface()
+    # 1. Instantiate Agent ( modify this part to create your agent)
+    try:
+        agent = EnhancedGAIAAgent()
+    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
+
+        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}"})
+
+    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)
+
+    # 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}")
+    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)
+    except Exception as e:
+        error_msg = f"An unexpected error occurred during submission: {e}"
+        print(error_msg)
+        return error_msg, pd.DataFrame(results_log)
+
+# --- Gradio Interface ---
+with gr.Blocks() as demo:
+    gr.Markdown("# Basic 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("---")
+    
+    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.")
+    
+    with gr.Row():
+        login_button = gr.LoginButton(value="Sign in with Hugging Face")
+    
+    with gr.Row():
+        submit_button = gr.Button("Run Evaluation & Submit All Answers")
+    
+    with gr.Row():
+        with gr.Column():
+            output_status = gr.Textbox(label="Run Status / 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])
 
-# For Hugging Face Spaces
 if __name__ == "__main__":
-    interface.launch()
+    demo.launch()