Update app.py

app.py

@@ -3,23 +3,390 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
+from smolagents import (
+    CodeAgent, 
+    LiteLLMModel,
+    DuckDuckGoSearchTool, 
+    LogLevel, 
+    load_tool,
+    PythonInterpreterTool
+)
+from dotenv import load_dotenv
+from smolagents import Tool
+import base64
+import anthropic
+from PIL import Image
+import io
+
+class SimpleExcelTool(Tool):
+    name = "SimpleExcelTool"
+    description = "Load a downloaded Excel file associated with a task ID and perform basic operations like reading data"
+    inputs = {
+        "task_id": {
+            "type": "string",
+            "description": "Task ID for which the Excel file has been downloaded"
+        },
+        "operation": {
+            "type": "string",
+            "description": "Operation to perform on the Excel file (currently only 'read' is supported)",
+            "nullable": True
+        }
+    }
+    output_type = "string"
+
+    def forward(self, task_id: str, operation: str = "read") -> str:
+        try:
+            filename = f"{task_id}_downloaded_file"
+            df = pd.read_excel(filename, engine="openpyxl")
+            
+            if operation == "read":
+                return df.head().to_string()
+            else:
+                return f"Unsupported operation: {operation}"
+        except Exception as e:
+            return f"Error reading Excel file: {str(e)}"
+
+class ImageAnalysisTool(Tool):
+    name = "ImageAnalysisTool"
+    description = "Analyze a downloaded image file associated with a task ID using Claude Vision. Provide a detailed description of what's in the image."
+    inputs = {
+        "task_id": {
+            "type": "string",
+            "description": "Task ID for which the image file has been downloaded"
+        },
+        "prompt": {
+            "type": "string",
+            "description": "Optional specific question or aspect to analyze about the image",
+            "nullable": True
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        super().__init__()
+        self.client = anthropic.Client(api_key="")
+
+    def forward(self, task_id: str, prompt: str = "Describe what you see in this image in detail.") -> str:
+        try:
+            filename = f"{task_id}_downloaded_file"
+
+            with open(filename, 'rb') as img_file:
+                img_bytes = img_file.read()
+
+            img = Image.open(io.BytesIO(img_bytes))
+            if img.mode != 'RGB':
+                img = img.convert('RGB')
+
+            img_byte_arr = io.BytesIO()
+            img.save(img_byte_arr, format='JPEG')
+            img_byte_arr = img_byte_arr.getvalue()
+
+            base64_image = base64.b64encode(img_byte_arr).decode('utf-8')
+
+            message = self.client.messages.create(
+                model="claude-3-7-sonnet-20250219",
+                max_tokens=1000,
+                messages=[{
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "image",
+                            "source": {
+                                "type": "base64",
+                                "media_type": "image/jpeg",
+                                "data": base64_image
+                            }
+                        },
+                        {
+                            "type": "text",
+                            "text": prompt
+                        }
+                    ]
+                }]
+            )
+
+            return message.content[0].text
+
+        except Exception as e:
+            return f"Error analyzing image: {str(e)}"
+
+
+# New: TaskFileDownloaderTool
+class TaskFileDownloaderTool(Tool):
+    name = "TaskFileDownloaderTool"
+    description = "Download a specific file associated with a given task ID and save it locally"
+    inputs = {
+        "task_id": {
+            "type": "string",
+            "description": "Task ID for which to download the associated file"
+        }
+    }
+    output_type = "string"
+
+    def forward(self, task_id: str) -> str:
+        try:
+            download_url = f"{DEFAULT_API_URL}/files/{task_id}"
+            response = requests.get(download_url)
+            response.raise_for_status()
+
+            filename = f"{task_id}_downloaded_file"
+            with open(filename, "wb") as f:
+                f.write(response.content)
+            
+            return f"File downloaded successfully and saved as: {filename}"
+        except Exception as e:
+            return f"Error downloading file: {str(e)}"
+
+# New: FileOpenerTool
+class FileOpenerTool(Tool):
+    name = "FileOpenerTool"
+    description = "Open a downloaded file associated with a task ID and read its contents as plain text."
+    inputs = {
+        "task_id": {
+            "type": "string",
+            "description": "Task ID for which the file has been downloaded"
+        },
+        "num_lines": {
+            "type": "integer",
+            "description": "Number of lines to read from the file",
+            "nullable": True
+        }
+    }
+    output_type = "string"
+
+    def forward(self, task_id: str, num_lines: int = 10) -> str:
+        try:
+            filename = f"{task_id}_downloaded_file"
+            if not os.path.exists(filename):
+                return f"Error: File {filename} does not exist."
+
+            with open(filename, "r", encoding="utf-8", errors="ignore") as file:
+                lines = []
+                for _ in range(num_lines):
+                    line = file.readline()
+                    if not line:
+                        break
+                    lines.append(line.strip())
+
+            return "\n".join(lines)
+        except Exception as e:
+            return f"Error reading file: {str(e)}"
+
+# New: SpeechToTextTool
+import mlx_whisper
+
+class SpeechToTextTool(Tool):
+    name = "SpeechToTextTool"
+    description = "Transcribe a downloaded MP3 audio file associated with a task ID into text."
+    inputs = {
+        "task_id": {
+            "type": "string",
+            "description": "Task ID for which the MP3 audio file has been downloaded"
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        super().__init__()
+
+    def forward(self, task_id: str) -> str:
+        try:
+            filename = f"{task_id}_downloaded_file"
+            if not os.path.exists(filename):
+                return f"Error: Audio file {filename} does not exist."
+
+            result = mlx_whisper.transcribe(filename)
+            return result["text"]
+        except Exception as e:
+            return f"Error transcribing audio file: {str(e)}"
+
+
+
+import wikipedia
+
+class WikipediaSearchTool(Tool):
+    name = "WikipediaSearchTool"
+    description = "Search Wikipedia for a query and return a brief summary."
+    inputs = {
+        "query": {
+            "type": "string",
+            "description": "Query to search on Wikipedia"
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        super().__init__()
+        wikipedia.set_lang("en")  # Ensure English Wikipedia
+
+    def forward(self, query: str) -> str:
+        try:
+            summary = wikipedia.summary(query, sentences=3000)
+            return summary
+        except wikipedia.exceptions.DisambiguationError as e:
+            return f"Disambiguation error. Possible options: {e.options[:5]}"
+        except wikipedia.exceptions.PageError:
+            return f"Page not found for query: {query}"
+        except Exception as e:
+            return f"Error searching Wikipedia: {str(e)}"
+        
+
+import os
+from youtube_transcript_api import YouTubeTranscriptApi, TextFormatter
+import yt_dlp
+import mlx_whisper
+
+class YouTubeTranscriptTool(Tool):
+    name = "YouTubeTranscriptTool"
+    description = "Fetches or transcribes the text from a YouTube video ID."
+    inputs = {
+        "video_id": {
+            "type": "string",
+            "description": "YouTube Video ID (the part after 'watch?v=')"
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        super().__init__()
+
+    def forward(self, video_id: str) -> str:
+        try:
+            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
+
+            try:
+                # First try manually created transcript
+                transcript = transcript_list.find_manually_created_transcript(['en'])
+            except Exception:
+                # If not found, try auto-generated transcript
+                transcript = transcript_list.find_generated_transcript(['en'])
+            
+            transcript_data = transcript.fetch()
+
+            # Format nicely
+            formatter = TextFormatter()
+            text = formatter.format_transcript(transcript_data)
+
+            return text
+
+        except Exception as e:
+            print(f"No direct transcript found: {e}")
+            print("Trying to download and transcribe audio with Whisper...")
+
+            # Step 1: Download audio using yt_dlp
+            audio_filename = f"{video_id}.mp3"
+            try:
+                ydl_opts = {
+                    'format': 'bestaudio/best',
+                    'outtmpl': audio_filename,
+                    'postprocessors': [{
+                        'key': 'FFmpegExtractAudio',
+                        'preferredcodec': 'mp3',
+                        'preferredquality': '192',
+                    }],
+                    'quiet': True,
+                }
+                with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                    ydl.download([f"https://www.youtube.com/watch?v={video_id}"])
+
+                # Step 2: Transcribe audio using mlx_whisper
+                result = mlx_whisper.transcribe(audio_filename)
+                return result["text"]
+
+            except Exception as download_error:
+                return f"Error downloading or transcribing YouTube audio: {str(download_error)}"
+            finally:
+                if os.path.exists(audio_filename):
+                    os.remove(audio_filename)  # Clean up downloaded file
+
+# Load environment variables
+load_dotenv()
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
     def __init__(self):
-        print("BasicAgent initialized.")
-    def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = "This is a default answer."
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
-
-def run_and_submit_all( profile: gr.OAuthProfile | None):
+        print("Initializing Agent with tools...")
+
+        # Initialize the model using Claude via LiteLLM
+        self.model = LiteLLMModel(
+            model="claude-3-7-sonnet-20250219",
+            api_key="",
+            temperature=0.7,
+            max_tokens=4096
+        )
+
+        # Initialize tools
+        youtube_transcript_tool = YouTubeTranscriptTool()
+        excel_tool = SimpleExcelTool()
+        image_analysis_tool = ImageAnalysisTool()
+        file_opener_tool = FileOpenerTool()
+        speech_to_text_tool = SpeechToTextTool()
+        task_file_downloader_tool = TaskFileDownloaderTool()
+        wikipedia_search_tool = WikipediaSearchTool()
+
+        self.tools = [
+            DuckDuckGoSearchTool(),
+            wikipedia_search_tool,
+            youtube_transcript_tool,
+            PythonInterpreterTool(),
+            excel_tool,
+            image_analysis_tool,
+            file_opener_tool,
+            speech_to_text_tool,
+            task_file_downloader_tool
+        ]
+
+        # Initialize the agent
+        self.agent = CodeAgent(
+            tools=self.tools,
+            model=self.model,
+            verbosity_level=LogLevel.INFO
+        )
+
+        print("Agent initialized successfully")
+
+    def __call__(self, question: str, task_id: str) -> str:
+        print(f"Agent received question: {question[:100]}...")
+        try:
+            # Step 1: Download the file associated with the task first
+            download_result = self.tools[-1](task_id=task_id)  # TaskFileDownloaderTool is the last in self.tools
+            print(download_result)
+
+            # Step 2: Create a comprehensive prompt for the agent
+            prompt = f"""Please answer the following question. Use the available tools (web search) 
+            to gather relevant information before providing a comprehensive answer.
+            
+            Question: {question}
+            Task_id: {task_id}
+            
+            Instructions:
+            1. Search for relevant information using web search.
+            2. Look for relevant YouTube content if applicable.
+            3. If the task requires working with an Excel or image file:
+            - First, download the file associated with the task ID using the file download tool.
+            - Then, perform analysis on the downloaded file.
+            4. Extract and analyze data from Excel files after downloading.
+            5. Convert images to text after downloading the image file.
+            6. Convert attached mp3 to text as seepch to text
+            7. Make Wikipedia search on facts and for a query and return a brief summary
+            78. Synthesize all gathered and analyzed information into a clear, well-structured final answer.
+            Answer:"""
+
+            # Step 3: Get response from the agent
+            response = self.agent.run(prompt)
+            print(f"Agent generated response: {response[:100]}...")
+            return response
+
+        except Exception as e:
+            error_msg = f"Error generating answer: {str(e)}"
+            print(error_msg)
+            return error_msg
+
+def run_and_submit_all(profile: gr.OAuthProfile | None):
     """
     Fetches all questions, runs the BasicAgent on them, submits all answers,
     and displays the results.
@@ -28,7 +395,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
 
     if profile:
-        username= f"{profile.username}"
+        username = f"{profile.username}"
         print(f"User logged in: {username}")
     else:
         print("User not logged in.")
@@ -38,13 +405,13 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     questions_url = f"{api_url}/questions"
     submit_url = f"{api_url}/submit"
 
-    # 1. Instantiate Agent ( modify this part to create your agent)
+    # 1. Instantiate 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)
 
@@ -54,6 +421,9 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
         response = requests.get(questions_url, timeout=15)
         response.raise_for_status()
         questions_data = response.json()
+
+        print(questions_data)
+        
         if not questions_data:
              print("Fetched questions list is empty.")
              return "Fetched questions list is empty or invalid format.", None
@@ -69,30 +439,44 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
         print(f"An unexpected error occurred fetching questions: {e}")
         return f"An unexpected error occurred fetching questions: {e}", None
 
-    # 3. Run your Agent
+    # 3. Run 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)
+            submitted_answer = agent(question_text, task_id)
             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})
+            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}"})
+            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}
+    # 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)
 
@@ -139,31 +523,28 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
         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("# Advanced 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.
+        1. Make sure you have set up your environment variables:
+           - HF_TOKEN: Your Hugging Face API token
+           - YOUTUBE_API_KEY: Your YouTube API key (optional)
+        2. Log in to your Hugging Face account using the button below
+        3. Click 'Run Evaluation & Submit All Answers' to process all questions
+        
+        The agent will use:
+        - Web search (DuckDuckGo)
+        - YouTube search (if API key provided)
+        - Mistral-7B-Instruct LLM
         """
     )
 
     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(
@@ -173,24 +554,33 @@ with gr.Blocks() as demo:
 
 if __name__ == "__main__":
     print("\n" + "-"*30 + " App Starting " + "-"*30)
-    # Check for SPACE_HOST and SPACE_ID at startup for information
+    
+    # Check for required environment variables
+    hf_token = os.getenv("HF_TOKEN")
+    if not hf_token:
+        print("⚠️ Warning: HF_TOKEN not found in environment variables")
+    
+    youtube_api_key = os.getenv("YOUTUBE_API_KEY")
+    if not youtube_api_key:
+        print("ℹ️ Note: YOUTUBE_API_KEY not found. YouTube search will be disabled")
+    
+    # Check for SPACE_HOST and SPACE_ID
     space_host_startup = os.getenv("SPACE_HOST")
-    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup
+    space_id_startup = os.getenv("SPACE_ID")
 
     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?).")
+        print("ℹ️ SPACE_HOST environment variable not found (running locally?).")
 
-    if space_id_startup: # Print repo URLs if SPACE_ID is found
+    if space_id_startup:
         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("ℹ️ SPACE_ID environment variable not found (running locally?)")
 
     print("-"*(60 + len(" App Starting ")) + "\n")
-
-    print("Launching Gradio Interface for Basic Agent Evaluation...")
+    print("Launching Gradio Interface for Advanced Agent Evaluation...")
     demo.launch(debug=True, share=False)