Update app.py

app.py

@@ -1,321 +1,210 @@
 import os
-from typing import Annotated, Optional, TypedDict
 import gradio as gr
-from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
-from langchain_openai import ChatOpenAI
-from langgraph.graph.message import add_messages
-from langgraph.graph import StateGraph, START
-from langgraph.prebuilt import tools_condition, ToolNode
 import requests
 import pandas as pd
-from langchain.tools import Tool
-from dotenv import load_dotenv
+from smolagents import CodeAgent, OpenAIServerModel, DuckDuckGoSearchTool, VisitWebpageTool, tool, \
+    FinalAnswerTool, PythonInterpreterTool, SpeechToTextTool, ToolCallingAgent
+import yaml
+import importlib
+from io import BytesIO
+import tempfile
+import base64
+from youtube_transcript_api import YouTubeTranscriptApi
+from youtube_transcript_api._errors import TranscriptsDisabled, NoTranscriptFound, VideoUnavailable
+from urllib.parse import urlparse, parse_qs
+import json
+import whisper
+import re
 
-import arxiv
-from chess_algebraic_notation_retriever import ChessAlgebraicNotationMoveRetriever
-from excel_file_reader import ExcelFileReader
-from image_question_answer_tool import ImageQuestionAnswerTool
-from python_code_question_answer_tool import PythonCodeQuestionAnswerTool
-from tavily_searcher import TavilySearcher
-from transcriber import Transcriber
-from wikipedia_searcher import WikipediaSearcher
-from youtube_video_question_answer_tool import YoutubeVideoQuestionAnswerTool
 
-load_dotenv()
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-ASSOCIATED_FILE_ENDPOINT = f"{DEFAULT_API_URL}/files/"
 
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-#search_tool = DuckDuckGoSearchRun()
-
-#search_tool = DuckDuckGoSearcherTool()
 
-def retrieve_task_file(task_id: str) -> Optional[bytes]:
+@tool
+def transcribe_audio_file(file_path: str) -> str:
     """
-    Retrieve the task file for a given task ID.
+    Transcribes a local MP3 audio file using Whisper.
+    Args:
+        file_path: Full path to the .mp3 audio file.
+    Returns:
+        A JSON-formatted string containing either the transcript or an error message.
+        {
+            "success": true,
+            "transcript": [
+                {"start": 0.0, "end": 5.2, "text": "Hello and welcome"},
+                ...
+            ]
+        }
+        OR
+        {
+            "success": false,
+            "error": "Reason why transcription failed"
+        }
     """
     try:
-        response = requests.get(ASSOCIATED_FILE_ENDPOINT + task_id, timeout=15)
-        response.raise_for_status()
-        if response.status_code != 200:
-            print(f"Error fetching file: {response.status_code}")
-            return None
-        #print(f"Fetched file: {response.content}")
-        return response.content
-    except requests.exceptions.RequestException as e:
-        print(f"Error fetching file: {e}")
-        return None
+        if not os.path.exists(file_path):
+            return json.dumps({"success": False, "error": "File does not exist."})
+
+        if not file_path.lower().endswith(".mp3"):
+            return json.dumps({"success": False, "error": "Invalid file type. Only MP3 files are supported."})
+
+        model = whisper.load_model("base")  # You can use 'tiny', 'base', 'small', 'medium', or 'large'
+        result = model.transcribe(file_path, verbose=False, word_timestamps=False)
+
+        transcript_data = [
+            {
+                "start": segment["start"],
+                "end": segment["end"],
+                "text": segment["text"].strip()
+            }
+            for segment in result["segments"]
+        ]
+
+        return json.dumps({"success": True, "transcript": transcript_data})
+
     except Exception as e:
-        print(f"An unexpected error occurred fetching file: {e}")
-        return None
+        return json.dumps({"success": False, "error": str(e)})
 
-def retrieve_next_chess_move_in_algebraic_notation(task_file_path: str, is_black_turn: bool) -> str:
-    """
-    Retrieve the next chess move in algebraic notation from an image path.
-    """
-    if task_file_path is None:
-        return "Error: Task file not found."
-    # Retrieve the next chess move in algebraic notation
-    next_chess_move = ChessAlgebraicNotationMoveRetriever().retrieve(task_file_path, is_black_turn)
-    return next_chess_move
-
-# Initialize the tool
-retrieve_next_chess_move_in_algebraic_notation_tool = Tool(
-    name="retrieve_next_chess_move_in_algebraic_notation",
-    func=retrieve_next_chess_move_in_algebraic_notation,
-    description="Retrieve the next chess move in algebraic notation from an image path."
-)
-
-def transcribe_audio(file_path: str) -> str:
-    if file_path is None:
-        return "Error: Audio path not found."
-    # Transcribe the audio
-    return Transcriber().transcribe(file_path)
-
-# Initialize the tool
-transcribe_audio_tool = Tool(
-    name="transcribe_audio",
-    func=transcribe_audio,
-    description="Transcribe the audio from an audio path."
-)
-
-# Initialize the tool
-answer_python_code_tool = PythonCodeQuestionAnswerTool()
-
-# Initialize the tool
-answer_image_question_tool = ImageQuestionAnswerTool()
-
-# Initialize the tool
-answer_youtube_video_question_tool = YoutubeVideoQuestionAnswerTool()
-
-'''def answer_youtube_video_question(youtube_video_url: str, question: str) -> str:
-    """
-    Answer the question based on the youtube video.
+
+@tool
+def get_youtube_transcript(video_url: str) -> str:
     """
-    if youtube_video_url is None:
-        return "Error: Video not found."
-    # Download the video
-    video_path = YoutubeVideoDownloader().download_video(youtube_video_url)
-    # Answer the question
-    return VideoQuestionAnswer().answer(video_path, question)
-# Initialize the tool
-answer_youtube_video_question_tool = Tool(
-    name="answer_youtube_video_question",
-    func=answer_youtube_video_question,
-    description="Answer the question based on the youtube video."
-)'''
-
-def read_excel_file(file_path: str) -> str:
-    if file_path is None:
-        return "Error: File not found."
-    return ExcelFileReader().read_file(file_path)
-
-# Initialize the tool
-read_excel_file_tool = Tool(
-    name="read_excel_file",
-    func=read_excel_file,
-    description="Read the excel file."
-)
-
-# Initialize the tool
-wikipedia_search_tool = Tool(
-    name="wikipedia_search",
-    func=WikipediaSearcher().search,
-    description="Search Wikipedia for a given query."
-)
-
-# Initialize the tool
-arxiv_search_tool = Tool(
-    name="arxiv_search",
-    func=ArxivSearcher().search,
-    description="Search Arxiv for a given query."
-)
-
-tavily_search_tool = Tool(
-    name="tavily_search",
-    func=TavilySearcher().search,
-    description="Search the web for a given query."
-)
-
-def format_gaia_answer(answer: str) -> str:
-    llm = ChatOpenAI(model="o3-mini", openai_api_key=os.getenv("OPENAI_API_KEY"))
-    prompt = f"""
-    You are formatting answers for the GAIA benchmark, which requires responses to be concise and unambiguous.
-    Given the answer: {answer}
-    Return the answer in the correct GAIA format:
-    - If the answer is a single word or number, return it without any additional text or formatting.
-    - If the answer is a list, return a comma-separated list without any additional text or formatting.
-    - If the answer is a string, return it without any additional text or formatting.
-    Do not include any prefixes, dots, enumerations, explanations, or quotation marks.
-    Do not include any additional text or formatting.
+    Retrieves the transcript from a YouTube video URL, including timestamps.
+    This tool fetches the English transcript for a given YouTube video. Automatically generated subtitles
+    are also supported. The result includes each snippet's start time, duration, and text.
+    Args:
+        video_url: The full URL of the YouTube video (e.g., https://www.youtube.com/watch?v=12345)
+    Returns:
+        A JSON-formatted string containing either the transcript with timestamps or an error message.
+        {
+            "success": true,
+            "transcript": [
+                {"start": 0.0, "duration": 1.54, "text": "Hey there"},
+                {"start": 1.54, "duration": 4.16, "text": "how are you"},
+                ...
+            ]
+        }
+        OR
+        {
+            "success": false,
+            "error": "Reason why the transcript could not be retrieved"
+        }
     """
-    response = llm.invoke(prompt)
-    # Delete double quotes
-    return response.content.strip().replace('"', '')
-
-class AgentState(TypedDict):
-    # The document provided
-    messages: Annotated[list[AnyMessage], add_messages]
-    file_path: Optional[str]
-    
+    try:
+        # Extract video ID from URL
+        parsed_url = urlparse(video_url)
+        query_params = parse_qs(parsed_url.query)
+        video_id = query_params.get("v", [None])[0]
+
+        if not video_id:
+            return json.dumps({"success": False, "error": "Invalid YouTube URL. Could not extract video ID."})
+
+        fetched_transcript = YouTubeTranscriptApi().fetch(video_id)
+        transcript_data = [
+            {
+                "start": snippet.start,
+                "duration": snippet.duration,
+                "text": snippet.text
+            }
+            for snippet in fetched_transcript
+        ]
+
+        return json.dumps({"success": True, "transcript": transcript_data})
+
+    except VideoUnavailable:
+        return json.dumps({"success": False, "error": "The video is unavailable."})
+    except TranscriptsDisabled:
+        return json.dumps({"success": False, "error": "Transcripts are disabled for this video."})
+    except NoTranscriptFound:
+        return json.dumps({"success": False, "error": "No transcript found for this video."})
+    except Exception as e:
+        return json.dumps({"success": False, "error": str(e)})
+
+# --- Basic Agent Definition ---
+# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
+
     def __init__(self):
-        tools = [
-            tavily_search_tool, 
-            arxiv_search_tool,
-            wikipedia_search_tool, 
-            transcribe_audio_tool, 
-            answer_python_code_tool, 
-            answer_image_question_tool, 
-            answer_youtube_video_question_tool,
-            read_excel_file_tool
-        ]
-        '''llm = ChatGoogleGenerativeAI(
-            model="gemini-2.0-flash", 
-            temperature=0.2,
-            api_key=os.getenv("GEMINI_API_KEY")
-        )'''
-        llm = ChatOpenAI(model="o3-mini", openai_api_key=os.getenv("OPENAI_API_KEY"))
-        self.llm_with_tools = llm.bind_tools(tools)
-        builder = StateGraph(AgentState)
-
-        # Define nodes: these do the work
-        builder.add_node("assistant", self.assistant)
-        builder.add_node("tools", ToolNode(tools))
-
-        # Define edges: these determine how the control flow moves
-        builder.add_edge(START, "assistant")
-        builder.add_conditional_edges(
-            "assistant",
-            # If the latest message requires a tool, route to tools
-            # Otherwise, provide a direct response
-            tools_condition,
-        )
-        builder.add_edge("tools", "assistant")
-        self.agent = builder.compile()
+        model = OpenAIServerModel(api_key=os.environ.get("OPENAI_API_KEY"), model_id="gpt-4o")
+
+        self.code_agent = CodeAgent(
+            tools=[PythonInterpreterTool(), DuckDuckGoSearchTool(), VisitWebpageTool(), transcribe_audio_file,
+                   get_youtube_transcript,
+                   FinalAnswerTool()],
+            model=model,
+            max_steps=20,
+            name="hf_agent_course_final_assignment_solver",
+            prompt_templates=yaml.safe_load(
+                importlib.resources.files("prompts").joinpath("code_agent.yaml").read_text()
+            )
 
+        )
         print("BasicAgent initialized.")
 
-    def assistant(self, state: AgentState):
-        # System message
-        textual_description_of_tools="""
-            tavily_search(query: str) -> str:
-                Search the web for a given query.
-                Args:
-                    query: Query to search the web for (string).
-                Returns:
-                    A single string containing the information found on the web.
-            arxiv_search(query: str) -> str:
-                Search Arxiv, that contains scientific papers, for a given query.
-                Args:
-                    query: Query to search Arxiv for (string).
-                Returns:
-                    A single string containing the answer to the question.
-            wikipedia_search(query: str) -> str:
-                Search Wikipedia for a given query.
-                Args:
-                    query: Query to search Wikipedia for (string).
-                Returns:
-                    A single string containing the answer to the question.
-            transcribe_audio(file_path: str) -> str:
-                Transcribe the audio from an audio path.
-                Args:
-                    file_path: File path of the audio file (string).
-                Returns:
-                    A single string containing the transcribed text from the audio.
-            
-            answer_python_code(file_path: str, question: str) -> str:
-                Answer the question based on the python code.
-                Args:
-                    file_path: File path of the python file (string).
-                    question: Question to answer (string).
-                Returns:
-                    A single string containing the answer to the question.
-            
-            answer_image_question(file_path: str, question: str) -> str:
-                Answer the question based on the image.
-                Args:
-                    file_path: File path of the image (string).
-                    question: Question to answer (string).
-                Returns:
-                    A single string containing the answer to the question.
-            
-            download_youtube_video(youtube_video_url: str) -> str:
-                Download the Youtube video into a local file based on the URL
-                Args:
-                    youtube_video_url: A youtube video url (string).
-                Returns:
-                    A single string containing the file path of the downloaded youtube video.
-            answer_youtube_video_question(file_path: str, question: str) -> str:
-                Answer the question based on file path of the downloaded youtube video
-                Args:
-                    file_path: File path of the downloaded youtube video (string).
-                    question: Question to answer (string).
-                Returns:
-                    A single string containing the answer to the question.
-            
-            read_excel_file(file_path: str) -> str:
-                Read the excel file.
-                Args:
-                    file_path: File path of the excel file (string).
-                Returns:
-                    A markdown formatted string containing the contents of the excel file.
-        """
-        file_path=state["file_path"]
-        prompt = f"""
-            You are a helpful assistant that can analyse images, videos, excel files and Python scripts and run computations with provided tools:
-            {textual_description_of_tools} 
-            You have access to the file path of the attached file in case it's informed. Currently the file path is: {file_path}
-            Be direct and specific. GAIA benchmark requires exact matching answers.
-            For example, if asked "What is the capital of France?", respond simply with "Paris".
-            Do not include any prefixes, dots, enumerations, explanations, or quotation marks.
-            Do not include any additional text or formatting.
-            If you are required a number, return a number, not the items.
-        """
-        sys_msg = SystemMessage(content=prompt)
+    def __call__(self, task_id: str, question: str, file_name: str) -> str:
+        if file_name:
+            question = self.enrich_question_with_associated_file_details(task_id, question, file_name)
+
+        final_result = self.code_agent.run(question)
+
+        # Extract text after "FINAL ANSWER:" (case-insensitive, and trims whitespace)
+        match = re.search(r'final answer:\s*(.*)', str(final_result), re.IGNORECASE | re.DOTALL)
+        if match:
+            return match.group(1).strip()
+
+        # Fallback in case the pattern is not found
+        return str(final_result).strip()
+
+    def enrich_question_with_associated_file_details(self, task_id:str, question: str, file_name: str) -> str:
+        api_url = DEFAULT_API_URL
+        get_associated_files_url = f"{api_url}/files/{task_id}"
+        response = requests.get(get_associated_files_url, timeout=15)
+        response.raise_for_status()
+
+        if file_name.endswith(".mp3"):
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
+                tmp_file.write(response.content)
+                file_path = tmp_file.name
+                return question + "\n\nMentioned .mp3 file local path is: " + file_path
+        elif file_name.endswith(".py"):
+            file_content = response.text
+            return question + "\n\nBelow is mentioned Python file:\n\n```python\n" + file_content + "\n```\n"
+        elif file_name.endswith(".xlsx"):
+            xlsx_io = BytesIO(response.content)
+            df = pd.read_excel(xlsx_io)
+            file_content = df.to_csv(index=False)
+            return question + "\n\nBelow is mentioned excel file in CSV format:\n\n```csv\n" + file_content + "\n```\n"
+        elif file_name.endswith(".png"):
+            base64_str = base64.b64encode(response.content).decode('utf-8')
+            return question + "\n\nBelow is the .png image in base64 format:\n\n```base64\n" + base64_str + "\n```\n"
+
+
+    def enrich_question_with_associated_file_details(self, task_id:str, question: str, file_name: str) -> str:
+        api_url = DEFAULT_API_URL
+        get_associated_files_url = f"{api_url}/files/{task_id}"
+        response = requests.get(get_associated_files_url, timeout=15)
+        response.raise_for_status()
+
+        if file_name.endswith(".mp3"):
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
+                tmp_file.write(response.content)
+                file_path = tmp_file.name
+                return question + "\n\nMentioned .mp3 file local path is: " + file_path
+        elif file_name.endswith(".py"):
+            file_content = response.text
+            return question + "\n\nBelow is mentioned Python file:\n\n```python\n" + file_content + "\n```\n"
+        elif file_name.endswith(".xlsx"):
+            xlsx_io = BytesIO(response.content)
+            df = pd.read_excel(xlsx_io)
+            file_content = df.to_csv(index=False)
+            return question + "\n\nBelow is mentioned excel file in CSV format:\n\n```csv\n" + file_content + "\n```\n"
+        elif file_name.endswith(".png"):
+            base64_str = base64.b64encode(response.content).decode('utf-8')
+            return question + "\n\nBelow is the .png image in base64 format:\n\n```base64\n" + base64_str + "\n```\n"
 
-        return {
-            "messages": [self.llm_with_tools.invoke([sys_msg] + state["messages"], config={"configurable": {"file_path": state["file_path"]}})],
-            "file_path": state["file_path"]
-        }
-        '''return {
-            "messages": [self.llm_with_tools.invoke(
-                state["messages"],
-                config={"configurable": {"file_path": state["file_path"]}}  # Aquí pasas el task_id
-            )],
-            "file_path": state["file_path"]
-        }'''
-    
-    def __call__(self, question: str, task_id: str, file_name: str) -> str:
-        print(f"######################### Agent received question (first 50 chars): {question[:50]}... with file_name: {file_name}")
-
-        # Get the file path
-        tmp_file_path = None
-        if file_name is not None and file_name != "":
-            file_content = retrieve_task_file(task_id)
-            if file_content is not None:
-                print(f"Saving file {file_name} to tmp folder")
-                tmp_file_path = f"tmp/{file_name}"
-                with open(tmp_file_path, "wb") as f:
-                    f.write(file_content)
-        # Show the file path
-        print(f"File path: {tmp_file_path}")
-
-        messages = self.agent.invoke({"messages": [HumanMessage(question)], "file_path": tmp_file_path})
-        # Show the messages
-        for m in messages['messages']:
-            m.pretty_print()
-        answer = messages["messages"][-1].content
-        answer = format_gaia_answer(answer)
-        print(f"######################### Agent returning answer: {answer}\n")
-        # Delete the file
-        if tmp_file_path is not None:
-            os.remove(tmp_file_path)
-        return answer
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """