2025-05-13

README.md

@@ -1,14 +1,15 @@
 ---
-title: Final Assignment
-emoji: 😻
-colorFrom: blue
-colorTo: gray
+title: Template Final Assignment
+emoji: 🕵🏻‍♂️
+colorFrom: indigo
+colorTo: indigo
 sdk: gradio
-sdk_version: 5.29.0
+sdk_version: 5.25.2
 app_file: app.py
 pinned: false
-license: mit
-short_description: HuggigFace Agentic AI course
+hf_oauth: true
+# optional, default duration is 8 hours/480 minutes. Max duration is 30 days/43200 minutes.
+hf_oauth_expiration_minutes: 480
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

agent.py

@@ -0,0 +1,528 @@
+from smolagents import (
+    CodeAgent,
+    DuckDuckGoSearchTool,
+    HfApiModel,
+    LiteLLMModel,
+    OpenAIServerModel,
+    PythonInterpreterTool,
+    tool,
+    InferenceClientModel,
+)
+from typing import List, Dict, Any, Optional
+import os
+import tempfile
+import re
+import json
+import requests
+from urllib.parse import urlparse
+
+
+@tool
+def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
+    """
+    Save content to a temporary file and return the path.
+    Useful for processing files from the GAIA API.
+
+    Args:
+        content: The content to save to the file
+        filename: Optional filename, will generate a random name if not provided
+
+    Returns:
+        Path to the saved file
+    """
+    temp_dir = tempfile.gettempdir()
+    if filename is None:
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        filepath = temp_file.name
+    else:
+        filepath = os.path.join(temp_dir, filename)
+
+    # Write content to the file
+    with open(filepath, "w") as f:
+        f.write(content)
+
+    return f"File saved to {filepath}. You can read this file to process its contents."
+
+
+@tool
+def download_file_from_url(url: str, filename: Optional[str] = None) -> str:
+    """
+    Download a file from a URL and save it to a temporary location.
+
+    Args:
+        url: The URL to download from
+        filename: Optional filename, will generate one based on URL if not provided
+
+    Returns:
+        Path to the downloaded file
+    """
+    try:
+        # Parse URL to get filename if not provided
+        if not filename:
+            path = urlparse(url).path
+            filename = os.path.basename(path)
+            if not filename:
+                # Generate a random name if we couldn't extract one
+                import uuid
+
+                filename = f"downloaded_{uuid.uuid4().hex[:8]}"
+
+        # Create temporary file
+        temp_dir = tempfile.gettempdir()
+        filepath = os.path.join(temp_dir, filename)
+
+        # Download the file
+        response = requests.get(url, stream=True)
+        response.raise_for_status()
+
+        # Save the file
+        with open(filepath, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+
+        return f"File downloaded to {filepath}. You can now process this file."
+    except Exception as e:
+        return f"Error downloading file: {str(e)}"
+
+
+@tool
+def extract_text_from_image(image_path: str) -> str:
+    """
+    Extract text from an image using pytesseract (if available).
+
+    Args:
+        image_path: Path to the image file
+
+    Returns:
+        Extracted text or error message
+    """
+    try:
+        # Try to import pytesseract
+        import pytesseract
+        from PIL import Image
+
+        # Open the image
+        image = Image.open(image_path)
+
+        # Extract text
+        text = pytesseract.image_to_string(image)
+
+        return f"Extracted text from image:\n\n{text}"
+    except ImportError:
+        return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
+    except Exception as e:
+        return f"Error extracting text from image: {str(e)}"
+
+
+@tool
+def analyze_csv_file(file_path: str, query: str) -> str:
+    """
+    Analyze a CSV file using pandas and answer a question about it.
+
+    Args:
+        file_path: Path to the CSV file
+        query: Question about the data
+
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+
+        # Read the CSV file
+        df = pd.read_csv(file_path)
+
+        # Run various analyses based on the query
+        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+
+        return result
+    except ImportError:
+        return "Error: pandas is not installed. Please install it with 'pip install pandas'."
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+
+@tool
+def analyze_excel_file(file_path: str, query: str) -> str:
+    """
+    Analyze an Excel file using pandas and answer a question about it.
+
+    Args:
+        file_path: Path to the Excel file
+        query: Question about the data
+
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+
+        # Read the Excel file
+        df = pd.read_excel(file_path)
+
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+
+        return result
+    except ImportError:
+        return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"
+
+
+class GAIAAgent:
+    def __init__(
+        self,
+        model_type: str = "HfApiModel",
+        model_id: Optional[str] = None,
+        api_key: Optional[str] = None,
+        api_base: Optional[str] = None,
+        temperature: float = 0.2,
+        executor_type: str = "local",  # Changed from use_e2b to executor_type
+        additional_imports: List[str] = None,
+        additional_tools: List[Any] = None,
+        system_prompt: Optional[
+            str
+        ] = None,  # We'll still accept this parameter but not use it directly
+        verbose: bool = False,
+        provider: Optional[str] = None,  # Add provider for InferenceClientModel
+        timeout: Optional[int] = None,  # Add timeout for InferenceClientModel
+    ):
+        """
+        Initialize a GAIAAgent with specified configuration
+
+        Args:
+            model_type: Type of model to use (HfApiModel, LiteLLMModel, OpenAIServerModel, InferenceClientModel)
+            model_id: ID of the model to use
+            api_key: API key for the model provider
+            api_base: Base URL for API calls
+            temperature: Temperature for text generation
+            executor_type: Type of executor for code execution ('local' or 'e2b')
+            additional_imports: Additional Python modules to allow importing
+            additional_tools: Additional tools to provide to the agent
+            system_prompt: Custom system prompt to use (not directly used, kept for backward compatibility)
+            verbose: Enable verbose logging
+            provider: Provider for InferenceClientModel (e.g., "hf-inference")
+            timeout: Timeout in seconds for API calls
+        """
+        # Set verbosity
+        self.verbose = verbose
+        self.system_prompt = system_prompt  # Store for potential future use
+
+        # Initialize model based on configuration
+        if model_type == "HfApiModel":
+            if api_key is None:
+                api_key = os.getenv("HUGGINGFACEHUB_API_TOKEN")
+                if not api_key:
+                    raise ValueError(
+                        "No Hugging Face token provided. Please set HUGGINGFACEHUB_API_TOKEN environment variable or pass api_key parameter."
+                    )
+
+            if self.verbose:
+                print(f"Using Hugging Face token: {api_key[:5]}...")
+
+            self.model = HfApiModel(
+                model_id=model_id or "meta-llama/Llama-3-70B-Instruct",
+                token=api_key,
+                temperature=temperature,
+            )
+        elif model_type == "InferenceClientModel":
+            if api_key is None:
+                api_key = os.getenv("HUGGINGFACEHUB_API_TOKEN")
+                if not api_key:
+                    raise ValueError(
+                        "No Hugging Face token provided. Please set HUGGINGFACEHUB_API_TOKEN environment variable or pass api_key parameter."
+                    )
+
+            if self.verbose:
+                print(f"Using Hugging Face token: {api_key[:5]}...")
+
+            self.model = InferenceClientModel(
+                model_id=model_id or "meta-llama/Llama-3-70B-Instruct",
+                provider=provider or "hf-inference",
+                token=api_key,
+                timeout=timeout or 120,
+                temperature=temperature,
+            )
+        elif model_type == "LiteLLMModel":
+            from smolagents import LiteLLMModel
+
+            self.model = LiteLLMModel(
+                model_id=model_id or "gpt-4o",
+                api_key=api_key or os.getenv("OPENAI_API_KEY"),
+                temperature=temperature,
+            )
+        elif model_type == "OpenAIServerModel":
+            # Check for xAI API key and base URL first
+            xai_api_key = os.getenv("XAI_API_KEY")
+            xai_api_base = os.getenv("XAI_API_BASE")
+
+            # If xAI credentials are available, use them
+            if xai_api_key and api_key is None:
+                api_key = xai_api_key
+                if self.verbose:
+                    print(f"Using xAI API key: {api_key[:5]}...")
+
+            # If no API key specified, fall back to OPENAI_API_KEY
+            if api_key is None:
+                api_key = os.getenv("OPENAI_API_KEY")
+                if not api_key:
+                    raise ValueError(
+                        "No OpenAI API key provided. Please set OPENAI_API_KEY or XAI_API_KEY environment variable or pass api_key parameter."
+                    )
+
+            # If xAI API base is available and no api_base is provided, use it
+            if xai_api_base and api_base is None:
+                api_base = xai_api_base
+                if self.verbose:
+                    print(f"Using xAI API base URL: {api_base}")
+
+            # If no API base specified but environment variable available, use it
+            if api_base is None:
+                api_base = os.getenv("AGENT_API_BASE")
+                if api_base and self.verbose:
+                    print(f"Using API base from AGENT_API_BASE: {api_base}")
+
+            self.model = OpenAIServerModel(
+                model_id=model_id or "gpt-4o",
+                api_key=api_key,
+                api_base=api_base,
+                temperature=temperature,
+            )
+        else:
+            raise ValueError(f"Unknown model type: {model_type}")
+
+        if self.verbose:
+            print(f"Initialized model: {model_type} - {model_id}")
+
+        # Initialize default tools
+        self.tools = [
+            DuckDuckGoSearchTool(),
+            PythonInterpreterTool(),
+            save_and_read_file,
+            download_file_from_url,
+            analyze_csv_file,
+            analyze_excel_file,
+        ]
+
+        # Add extract_text_from_image if PIL and pytesseract are available
+        try:
+            import pytesseract
+            from PIL import Image
+
+            self.tools.append(extract_text_from_image)
+            if self.verbose:
+                print("Added image processing tool")
+        except ImportError:
+            if self.verbose:
+                print("Image processing libraries not available")
+
+        # Add any additional tools
+        if additional_tools:
+            self.tools.extend(additional_tools)
+
+        if self.verbose:
+            print(f"Initialized with {len(self.tools)} tools")
+
+        # Setup imports allowed
+        self.imports = [
+            "pandas",
+            "numpy",
+            "datetime",
+            "json",
+            "re",
+            "math",
+            "os",
+            "requests",
+            "csv",
+            "urllib",
+        ]
+        if additional_imports:
+            self.imports.extend(additional_imports)
+
+        # Initialize the CodeAgent
+        executor_kwargs = {}
+        if executor_type == "e2b":
+            try:
+                # Try to import e2b dependencies to check if they're available
+                from e2b_code_interpreter import Sandbox
+
+                if self.verbose:
+                    print("Using e2b executor")
+            except ImportError:
+                if self.verbose:
+                    print("e2b dependencies not found, falling back to local executor")
+                executor_type = "local"  # Fallback to local if e2b is not available
+
+        self.agent = CodeAgent(
+            tools=self.tools,
+            model=self.model,
+            additional_authorized_imports=self.imports,
+            executor_type=executor_type,
+            executor_kwargs=executor_kwargs,
+            verbosity_level=2 if self.verbose else 0,
+        )
+
+        if self.verbose:
+            print("Agent initialized and ready")
+
+    def answer_question(
+        self, question: str, task_file_path: Optional[str] = None
+    ) -> str:
+        """
+        Process a GAIA benchmark question and return the answer
+
+        Args:
+            question: The question to answer
+            task_file_path: Optional path to a file associated with the question
+
+        Returns:
+            The answer to the question
+        """
+        try:
+            if self.verbose:
+                print(f"Processing question: {question}")
+                if task_file_path:
+                    print(f"With associated file: {task_file_path}")
+
+            # Create a context with file information if available
+            context = question
+            file_content = None
+
+            # If there's a file, read it and include its content in the context
+            if task_file_path:
+                try:
+                    with open(task_file_path, "r") as f:
+                        file_content = f.read()
+
+                    # Determine file type from extension
+                    import os
+
+                    file_ext = os.path.splitext(task_file_path)[1].lower()
+
+                    context = f"""
+Question: {question}
+
+This question has an associated file. Here is the file content:
+
+```{file_ext}
+{file_content}
+```
+
+Analyze the file content above to answer the question.
+"""
+                except Exception as file_e:
+                    context = f"""
+Question: {question}
+
+This question has an associated file at path: {task_file_path}
+However, there was an error reading the file: {file_e}
+You can still try to answer the question based on the information provided.
+"""
+
+            # Check for special cases that need specific formatting
+            # Reversed text questions
+            if question.startswith(".") or ".rewsna eht sa" in question:
+                context = f"""
+This question appears to be in reversed text. Here's the reversed version:
+{question[::-1]}
+
+Now answer the question above. Remember to format your answer exactly as requested.
+"""
+
+            # Add a prompt to ensure precise answers
+            full_prompt = f"""{context}
+
+When answering, provide ONLY the precise answer requested.
+Do not include explanations, steps, reasoning, or additional text.
+Be direct and specific. GAIA benchmark requires exact matching answers.
+For example, if asked "What is the capital of France?", respond simply with "Paris".
+"""
+
+            # Run the agent with the question
+            answer = self.agent.run(full_prompt)
+
+            # Clean up the answer to ensure it's in the expected format
+            # Remove common prefixes that models often add
+            answer = self._clean_answer(answer)
+
+            if self.verbose:
+                print(f"Generated answer: {answer}")
+
+            return answer
+        except Exception as e:
+            error_msg = f"Error answering question: {e}"
+            if self.verbose:
+                print(error_msg)
+            return error_msg
+
+    def _clean_answer(self, answer: any) -> str:
+        """
+        Clean up the answer to remove common prefixes and formatting
+        that models often add but that can cause exact match failures.
+
+        Args:
+            answer: The raw answer from the model
+
+        Returns:
+            The cleaned answer as a string
+        """
+        # Convert non-string types to strings
+        if not isinstance(answer, str):
+            # Handle numeric types (float, int)
+            if isinstance(answer, float):
+                # Format floating point numbers properly
+                # Check if it's an integer value in float form (e.g., 12.0)
+                if answer.is_integer():
+                    formatted_answer = str(int(answer))
+                else:
+                    # For currency values that might need formatting
+                    if abs(answer) >= 1000:
+                        formatted_answer = f"${answer:,.2f}"
+                    else:
+                        formatted_answer = str(answer)
+                return formatted_answer
+            elif isinstance(answer, int):
+                return str(answer)
+            else:
+                # For any other type
+                return str(answer)
+
+        # Now we know answer is a string, so we can safely use string methods
+        # Normalize whitespace
+        answer = answer.strip()
+
+        # Remove common prefixes and formatting that models add
+        prefixes_to_remove = [
+            "The answer is ",
+            "Answer: ",
+            "Final answer: ",
+            "The result is ",
+            "To answer this question: ",
+            "Based on the information provided, ",
+            "According to the information: ",
+        ]
+
+        for prefix in prefixes_to_remove:
+            if answer.startswith(prefix):
+                answer = answer[len(prefix) :].strip()
+
+        # Remove quotes if they wrap the entire answer
+        if (answer.startswith('"') and answer.endswith('"')) or (
+            answer.startswith("'") and answer.endswith("'")
+        ):
+            answer = answer[1:-1].strip()
+
+        return answer

agent_gemini.py

@@ -0,0 +1,700 @@
+import os
+import tempfile
+import time
+import re
+import json
+from typing import List, Optional, Dict, Any
+from urllib.parse import urlparse
+import requests
+import yt_dlp
+from bs4 import BeautifulSoup
+from difflib import SequenceMatcher
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_community.utilities import (
+    DuckDuckGoSearchAPIWrapper,
+    WikipediaAPIWrapper,
+)
+from langchain.agents import (
+    Tool,
+    AgentExecutor,
+    ConversationalAgent,
+    initialize_agent,
+    AgentType,
+)
+from langchain.memory import ConversationBufferMemory
+from langchain.prompts import MessagesPlaceholder
+from langchain.tools import BaseTool, Tool, tool
+from google.generativeai.types import HarmCategory, HarmBlockThreshold
+from PIL import Image
+import google.generativeai as genai
+from pydantic import Field
+
+from smolagents import WikipediaSearchTool
+
+
+class SmolagentToolWrapper(BaseTool):
+    """Wrapper for smolagents tools to make them compatible with LangChain."""
+
+    wrapped_tool: object = Field(description="The wrapped smolagents tool")
+
+    def __init__(self, tool):
+        """Initialize the wrapper with a smolagents tool."""
+        super().__init__(
+            name=tool.name,
+            description=tool.description,
+            return_direct=False,
+            wrapped_tool=tool,
+        )
+
+    def _run(self, query: str) -> str:
+        """Use the wrapped tool to execute the query."""
+        try:
+            # For WikipediaSearchTool
+            if hasattr(self.wrapped_tool, "search"):
+                return self.wrapped_tool.search(query)
+            # For DuckDuckGoSearchTool and others
+            return self.wrapped_tool(query)
+        except Exception as e:
+            return f"Error using tool: {str(e)}"
+
+    def _arun(self, query: str) -> str:
+        """Async version - just calls sync version since smolagents tools don't support async."""
+        return self._run(query)
+
+
+class WebSearchTool:
+    def __init__(self):
+        self.last_request_time = 0
+        self.min_request_interval = 2.0  # Minimum time between requests in seconds
+        self.max_retries = 10
+
+    def search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform web search with rate limiting and retries."""
+        for attempt in range(self.max_retries):
+            # Implement rate limiting
+            current_time = time.time()
+            time_since_last = current_time - self.last_request_time
+            if time_since_last < self.min_request_interval:
+                time.sleep(self.min_request_interval - time_since_last)
+
+            try:
+                # Make the search request
+                results = self._do_search(query, domain)
+                self.last_request_time = time.time()
+                return results
+            except Exception as e:
+                if "202 Ratelimit" in str(e):
+                    if attempt < self.max_retries - 1:
+                        # Exponential backoff
+                        wait_time = (2**attempt) * self.min_request_interval
+                        time.sleep(wait_time)
+                        continue
+                return f"Search failed after {self.max_retries} attempts: {str(e)}"
+
+        return "Search failed due to rate limiting"
+
+    def _do_search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform the actual search request."""
+        try:
+            # Construct search URL
+            base_url = "https://html.duckduckgo.com/html"
+            params = {"q": query}
+            if domain:
+                params["q"] += f" site:{domain}"
+
+            # Make request with increased timeout
+            response = requests.get(base_url, params=params, timeout=10)
+            response.raise_for_status()
+
+            if response.status_code == 202:
+                raise Exception("202 Ratelimit")
+
+            # Extract search results
+            results = []
+            soup = BeautifulSoup(response.text, "html.parser")
+            for result in soup.find_all("div", {"class": "result"}):
+                title = result.find("a", {"class": "result__a"})
+                snippet = result.find("a", {"class": "result__snippet"})
+                if title and snippet:
+                    results.append(
+                        {
+                            "title": title.get_text(),
+                            "snippet": snippet.get_text(),
+                            "url": title.get("href"),
+                        }
+                    )
+
+            # Format results
+            formatted_results = []
+            for r in results[:10]:  # Limit to top 5 results
+                formatted_results.append(
+                    f"[{r['title']}]({r['url']})\n{r['snippet']}\n"
+                )
+
+            return "## Search Results\n\n" + "\n".join(formatted_results)
+
+        except requests.RequestException as e:
+            raise Exception(f"Search request failed: {str(e)}")
+
+
+def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
+    """
+    Save content to a temporary file and return the path.
+    Useful for processing files from the GAIA API.
+
+    Args:
+        content: The content to save to the file
+        filename: Optional filename, will generate a random name if not provided
+
+    Returns:
+        Path to the saved file
+    """
+    temp_dir = tempfile.gettempdir()
+    if filename is None:
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        filepath = temp_file.name
+    else:
+        filepath = os.path.join(temp_dir, filename)
+
+    # Write content to the file
+    with open(filepath, "w") as f:
+        f.write(content)
+
+    return f"File saved to {filepath}. You can read this file to process its contents."
+
+
+def download_file_from_url(url: str, filename: Optional[str] = None) -> str:
+    """
+    Download a file from a URL and save it to a temporary location.
+
+    Args:
+        url: The URL to download from
+        filename: Optional filename, will generate one based on URL if not provided
+
+    Returns:
+        Path to the downloaded file
+    """
+    try:
+        # Parse URL to get filename if not provided
+        if not filename:
+            path = urlparse(url).path
+            filename = os.path.basename(path)
+            if not filename:
+                # Generate a random name if we couldn't extract one
+                import uuid
+
+                filename = f"downloaded_{uuid.uuid4().hex[:8]}"
+
+        # Create temporary file
+        temp_dir = tempfile.gettempdir()
+        filepath = os.path.join(temp_dir, filename)
+
+        # Download the file
+        response = requests.get(url, stream=True)
+        response.raise_for_status()
+
+        # Save the file
+        with open(filepath, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+
+        return f"File downloaded to {filepath}. You can now process this file."
+    except Exception as e:
+        return f"Error downloading file: {str(e)}"
+
+
+def extract_text_from_image(image_path: str) -> str:
+    """
+    Extract text from an image using pytesseract (if available).
+
+    Args:
+        image_path: Path to the image file
+
+    Returns:
+        Extracted text or error message
+    """
+    try:
+        # Try to import pytesseract
+        import pytesseract
+        from PIL import Image
+
+        # Open the image
+        image = Image.open(image_path)
+
+        # Extract text
+        text = pytesseract.image_to_string(image)
+
+        return f"Extracted text from image:\n\n{text}"
+    except ImportError:
+        return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
+    except Exception as e:
+        return f"Error extracting text from image: {str(e)}"
+
+
+def analyze_csv_file(file_path: str, query: str) -> str:
+    """
+    Analyze a CSV file using pandas and answer a question about it.
+
+    Args:
+        file_path: Path to the CSV file
+        query: Question about the data
+
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+
+        # Read the CSV file
+        df = pd.read_csv(file_path)
+
+        # Run various analyses based on the query
+        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+
+        return result
+    except ImportError:
+        return "Error: pandas is not installed. Please install it with 'pip install pandas'."
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+
+@tool
+def analyze_excel_file(file_path: str, query: str) -> str:
+    """
+    Analyze an Excel file using pandas and answer a question about it.
+
+    Args:
+        file_path: Path to the Excel file
+        query: Question about the data
+
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+
+        # Read the Excel file
+        df = pd.read_excel(file_path)
+
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+
+        return result
+    except ImportError:
+        return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"
+
+
+class GeminiAgent:
+    def __init__(self, api_key: str, model_name: str = "gemini-2.0-flash"):
+        # Suppress warnings
+        import warnings
+
+        warnings.filterwarnings("ignore", category=UserWarning)
+        warnings.filterwarnings("ignore", category=DeprecationWarning)
+        warnings.filterwarnings("ignore", message=".*will be deprecated.*")
+        warnings.filterwarnings("ignore", "LangChain.*")
+
+        self.api_key = api_key
+        self.model_name = model_name
+
+        # Configure Gemini
+        genai.configure(api_key=api_key)
+
+        # Initialize the LLM
+        self.llm = self._setup_llm()
+
+        # Setup tools
+        self.tools = [
+            SmolagentToolWrapper(WikipediaSearchTool()),
+            Tool(
+                name="analyze_video",
+                func=self._analyze_video,
+                description="Analyze YouTube video content directly",
+            ),
+            Tool(
+                name="analyze_image",
+                func=self._analyze_image,
+                description="Analyze image content",
+            ),
+            Tool(
+                name="analyze_table",
+                func=self._analyze_table,
+                description="Analyze table or matrix data",
+            ),
+            Tool(
+                name="analyze_list",
+                func=self._analyze_list,
+                description="Analyze and categorize list items",
+            ),
+            Tool(
+                name="web_search",
+                func=self._web_search,
+                description="Search the web for information",
+            ),
+        ]
+
+        # Setup memory
+        self.memory = ConversationBufferMemory(
+            memory_key="chat_history", return_messages=True
+        )
+
+        # Initialize agent
+        self.agent = self._setup_agent()
+
+    def run(self, query: str) -> str:
+        """Run the agent on a query with incremental retries."""
+        max_retries = 3
+        base_sleep = 1  # Start with 1 second sleep
+
+        for attempt in range(max_retries):
+            try:
+                # If no match found in answer bank, use the agent
+                response = self.agent.run(query)
+                return response
+
+            except Exception as e:
+                sleep_time = base_sleep * (attempt + 1)  # Incremental sleep: 1s, 2s, 3s
+                if attempt < max_retries - 1:
+                    print(
+                        f"Attempt {attempt + 1} failed. Retrying in {sleep_time} seconds..."
+                    )
+                    time.sleep(sleep_time)
+                    continue
+                return f"Error processing query after {max_retries} attempts: {str(e)}"
+
+        print("Agent processed all queries!")
+
+    def _clean_response(self, response: str) -> str:
+        """Clean up the response from the agent."""
+        # Remove any tool invocation artifacts
+        cleaned = re.sub(
+            r"> Entering new AgentExecutor chain...|> Finished chain.", "", response
+        )
+        cleaned = re.sub(
+            r"Thought:.*?Action:.*?Action Input:.*?Observation:.*?\n",
+            "",
+            cleaned,
+            flags=re.DOTALL,
+        )
+        return cleaned.strip()
+
+    def run_interactive(self):
+        print("AI Assistant Ready! (Type 'exit' to quit)")
+
+        while True:
+            query = input("You: ").strip()
+            if query.lower() == "exit":
+                print("Goodbye!")
+                break
+
+            print("Assistant:", self.run(query))
+
+    def _web_search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform web search with rate limiting and retries."""
+        try:
+            # Use DuckDuckGo API wrapper for more reliable results
+            search = DuckDuckGoSearchAPIWrapper(max_results=5)
+            results = search.run(f"{query} {f'site:{domain}' if domain else ''}")
+
+            if not results or results.strip() == "":
+                return "No search results found."
+
+            return results
+
+        except Exception as e:
+            return f"Search error: {str(e)}"
+
+    def _analyze_video(self, url: str) -> str:
+        """Analyze video content using Gemini's video understanding capabilities."""
+        try:
+            # Validate URL
+            parsed_url = urlparse(url)
+            if not all([parsed_url.scheme, parsed_url.netloc]):
+                return (
+                    "Please provide a valid video URL with http:// or https:// prefix."
+                )
+
+            # Check if it's a YouTube URL
+            if "youtube.com" not in url and "youtu.be" not in url:
+                return "Only YouTube videos are supported at this time."
+
+            try:
+                # Configure yt-dlp with minimal extraction
+                ydl_opts = {
+                    "quiet": True,
+                    "no_warnings": True,
+                    "extract_flat": True,
+                    "no_playlist": True,
+                    "youtube_include_dash_manifest": False,
+                }
+
+                with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                    try:
+                        # Try basic info extraction
+                        info = ydl.extract_info(url, download=False, process=False)
+                        if not info:
+                            return "Could not extract video information."
+
+                        title = info.get("title", "Unknown")
+                        description = info.get("description", "")
+
+                        # Create a detailed prompt with available metadata
+                        prompt = f"""Please analyze this YouTube video:
+Title: {title}
+URL: {url}
+Description: {description}
+
+Please provide a detailed analysis focusing on:
+1. Main topic and key points from the title and description
+2. Expected visual elements and scenes
+3. Overall message or purpose
+4. Target audience"""
+
+                        # Use the LLM with proper message format
+                        messages = [HumanMessage(content=prompt)]
+                        response = self.llm.invoke(messages)
+                        return (
+                            response.content
+                            if hasattr(response, "content")
+                            else str(response)
+                        )
+
+                    except Exception as e:
+                        if "Sign in to confirm" in str(e):
+                            return "This video requires age verification or sign-in. Please provide a different video URL."
+                        return f"Error accessing video: {str(e)}"
+
+            except Exception as e:
+                return f"Error extracting video info: {str(e)}"
+
+        except Exception as e:
+            return f"Error analyzing video: {str(e)}"
+
+    def _analyze_table(self, table_data: str) -> str:
+        """Analyze table or matrix data."""
+        try:
+            if not table_data or not isinstance(table_data, str):
+                return "Please provide valid table data for analysis."
+
+            prompt = f"""Please analyze this table:
+
+{table_data}
+
+Provide a detailed analysis including:
+1. Structure and format
+2. Key patterns or relationships
+3. Notable findings
+4. Any mathematical properties (if applicable)"""
+
+            messages = [HumanMessage(content=prompt)]
+            response = self.llm.invoke(messages)
+            return response.content if hasattr(response, "content") else str(response)
+
+        except Exception as e:
+            return f"Error analyzing table: {str(e)}"
+
+    def _analyze_image(self, image_data: str) -> str:
+        """Analyze image content."""
+        try:
+            if not image_data or not isinstance(image_data, str):
+                return "Please provide a valid image for analysis."
+
+            prompt = f"""Please analyze this image:
+
+{image_data}
+
+Focus on:
+1. Visual elements and objects
+2. Colors and composition
+3. Text or numbers (if present)
+4. Overall context and meaning"""
+
+            messages = [HumanMessage(content=prompt)]
+            response = self.llm.invoke(messages)
+            return response.content if hasattr(response, "content") else str(response)
+
+        except Exception as e:
+            return f"Error analyzing image: {str(e)}"
+
+    def _analyze_list(self, list_data: str) -> str:
+        """Analyze and categorize list items."""
+        if not list_data:
+            return "No list data provided."
+        try:
+            items = [x.strip() for x in list_data.split(",")]
+            if not items:
+                return "Please provide a comma-separated list of items."
+            # Add list analysis logic here
+            return "Please provide the list items for analysis."
+        except Exception as e:
+            return f"Error analyzing list: {str(e)}"
+
+    def _setup_llm(self):
+        """Set up the language model."""
+        # Set up model with video capabilities
+        generation_config = {
+            "temperature": 0.0,
+            "max_output_tokens": 2000,
+            "candidate_count": 1,
+        }
+
+        safety_settings = {
+            HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+            HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+            HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+            HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+        }
+
+        return ChatGoogleGenerativeAI(
+            model="gemini-2.0-flash",
+            google_api_key=self.api_key,
+            temperature=0,
+            max_output_tokens=2000,
+            generation_config=generation_config,
+            safety_settings=safety_settings,
+            system_message=SystemMessage(
+                content=(
+                    "You are a precise AI assistant that helps users find information and analyze content. "
+                    "You can directly understand and analyze YouTube videos, images, and other content. "
+                    "When analyzing videos, focus on relevant details like dialogue, text, and key visual elements. "
+                    "For lists, tables, and structured data, ensure proper formatting and organization. "
+                    "If you need additional context, clearly explain what is needed."
+                )
+            ),
+        )
+
+    def _setup_agent(self) -> AgentExecutor:
+        """Set up the agent with tools and system message."""
+
+        # Define the system message template
+        PREFIX = """You are a helpful AI assistant that can use various tools to answer questions and analyze content. You have access to tools for web search, Wikipedia lookup, and multimedia analysis.
+
+TOOLS:
+------
+You have access to the following tools:"""
+
+        FORMAT_INSTRUCTIONS = """To use a tool, use the following format:
+
+Thought: Do I need to use a tool? Yes
+Action: the action to take, should be one of [{tool_names}]
+Action Input: the input to the action
+Observation: the result of the action
+
+When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:
+
+Thought: Do I need to use a tool? No
+Final Answer: [your response here]
+
+Begin! Remember to ALWAYS include 'Thought:', 'Action:', 'Action Input:', and 'Final Answer:' in your responses."""
+
+        SUFFIX = """Previous conversation history:
+{chat_history}
+
+New question: {input}
+{agent_scratchpad}"""
+
+        # Create the base agent
+        agent = ConversationalAgent.from_llm_and_tools(
+            llm=self.llm,
+            tools=self.tools,
+            prefix=PREFIX,
+            format_instructions=FORMAT_INSTRUCTIONS,
+            suffix=SUFFIX,
+            input_variables=["input", "chat_history", "agent_scratchpad", "tool_names"],
+            handle_parsing_errors=True,
+        )
+
+        # Initialize agent executor with custom output handling
+        return AgentExecutor.from_agent_and_tools(
+            agent=agent,
+            tools=self.tools,
+            memory=self.memory,
+            max_iterations=5,
+            verbose=True,
+            handle_parsing_errors=True,
+            return_only_outputs=True,  # This ensures we only get the final output
+        )
+
+
+@tool
+def analyze_csv_file(file_path: str, query: str) -> str:
+    """
+    Analyze a CSV file using pandas and answer a question about it.
+
+    Args:
+        file_path: Path to the CSV file
+        query: Question about the data
+
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+
+        # Read the CSV file
+        df = pd.read_csv(file_path)
+
+        # Run various analyses based on the query
+        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+
+        return result
+    except ImportError:
+        return "Error: pandas is not installed. Please install it with 'pip install pandas'."
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+
+@tool
+def analyze_excel_file(file_path: str, query: str) -> str:
+    """
+    Analyze an Excel file using pandas and answer a question about it.
+
+    Args:
+        file_path: Path to the Excel file
+        query: Question about the data
+
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+
+        # Read the Excel file
+        df = pd.read_excel(file_path)
+
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+
+        return result
+    except ImportError:
+        return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"

app.py

@@ -0,0 +1,227 @@
+import os
+import gradio as gr
+import requests
+import pandas as pd
+from dotenv import load_dotenv
+from agent_gemini import GeminiAgent
+
+# Constants
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+
+class BasicAgent:
+    def __init__(self):
+        print("Initializing the BasicAgent")
+
+        # Get Gemini API key
+        api_key = os.getenv("GOOGLE_API_KEY")
+        if not api_key:
+            raise ValueError("GOOGLE_API_KEY environment variable not set.")
+
+        # Initialize GeminiAgent
+        self.agent = GeminiAgent(api_key=api_key)
+        print("GeminiAgent initialized successfully")
+
+    def __call__(self, question: str) -> str:
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        final_answer = self.agent.run(question)
+        print(f"Agent returning fixed answer: {final_answer}")
+        return final_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"
+
+    # 1. Instantiate Agent ( modify this part to create your agent)
+    try:
+        agent = BasicAgent()
+    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('score', 'N/A')}% "
+            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+            f"Message: {result_data.get('message', 'No message received.')}"
+        )
+        print("Submission successful.")
+        results_df = pd.DataFrame(results_log)
+        return final_status, results_df
+    except requests.exceptions.HTTPError as e:
+        error_detail = f"Server responded with status {e.response.status_code}."
+        try:
+            error_json = e.response.json()
+            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
+        except requests.exceptions.JSONDecodeError:
+            error_detail += f" Response: {e.response.text[:500]}"
+        status_message = f"Submission Failed: {error_detail}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.Timeout:
+        status_message = "Submission Failed: The request timed out."
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.RequestException as e:
+        status_message = f"Submission Failed: Network error - {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except Exception as e:
+        status_message = f"An unexpected error occurred during submission: {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+
+
+# --- Build Gradio Interface using Blocks ---
+with gr.Blocks() as demo:
+    gr.Markdown("# Basic Agent Evaluation Runner")
+    gr.Markdown(
+        """
+        **Instructions:**
+
+        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
+        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
+        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
+
+        ---
+        **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.LoginButton()
+
+    run_button = gr.Button("Run Evaluation & Submit All Answers")
+
+    status_output = gr.Textbox(
+        label="Run Status / Submission Result", lines=5, interactive=False
+    )
+    # Removed max_rows=10 from DataFrame constructor
+    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
+
+    run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])
+
+if __name__ == "__main__":
+    print("\n" + "-" * 30 + " App Starting " + "-" * 30)
+    # Check for SPACE_HOST and SPACE_ID at startup for information
+    space_host_startup = os.getenv("SPACE_HOST")
+    space_id_startup = os.getenv("SPACE_ID")  # Get SPACE_ID at startup
+
+    if space_host_startup:
+        print(f"✅ SPACE_HOST found: {space_host_startup}")
+        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
+    else:
+        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
+
+    if space_id_startup:  # Print repo URLs if SPACE_ID is found
+        print(f"✅ SPACE_ID found: {space_id_startup}")
+        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
+        print(
+            f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main"
+        )
+    else:
+        print(
+            "ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined."
+        )
+
+    print("-" * (60 + len(" App Starting ")) + "\n")
+
+    print("Launching Gradio Interface for Basic Agent Evaluation...")
+    demo.launch(debug=True, share=False)

requirements.txt

@@ -0,0 +1,18 @@
+gradio
+langchain>=0.1.0
+langchain-core>=0.1.0
+langchain-community>=0.0.10
+langchain-google-genai>=0.0.6
+google-generativeai>=0.3.0
+python-dotenv>=1.0.0
+google-api-python-client>=2.108.0
+duckduckgo-search>=4.4
+tiktoken>=0.5.2
+google-cloud-speech>=2.24.0
+requests>=2.31.0
+pydub>=0.25.1
+yt-dlp>=2023.12.30
+smolagents>=0.1.3
+wikipedia>=1.4.0
+Pillow>=10.2.0
+wikipedia-api>=0.6.0

system_prompt.txt

@@ -0,0 +1,5 @@
+You are a helpful assistant tasked with answering questions using a set of tools.
+Now, I will ask you a question. Report your thoughts, and finish your answer with the following template:
+FINAL ANSWER: [YOUR FINAL ANSWER].
+YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
+Your answer should only start with "FINAL ANSWER: ", then follows with the answer.