Implement final assignment agent

app.py

@@ -1,23 +1,137 @@
 import os
 import gradio as gr
 import requests
-import inspect
 import pandas as pd
+from dotenv import load_dotenv
+from smolagents import CodeAgent, OpenAIServerModel, DuckDuckGoSearchTool, VisitWebpageTool, tool, \
+    FinalAnswerTool, PythonInterpreterTool, SpeechToTextTool
+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
+
+load_dotenv()
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
+
+@tool
+def get_youtube_transcript(video_url: str) -> str:
+    """
+    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"
+        }
+    """
+    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):
+        model = OpenAIServerModel(api_key=os.environ.get("OPENAI_API_KEY"), model_id="gpt-4o")
+
+        self.code_agent = CodeAgent(
+            tools=[PythonInterpreterTool(), DuckDuckGoSearchTool(), VisitWebpageTool(), SpeechToTextTool(),
+                   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 __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 __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)
+        return str(final_result)
+
+    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 run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -76,11 +190,12 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     for item in questions_data:
         task_id = item.get("task_id")
         question_text = item.get("question")
+        file_name = item.get("file_name")
         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(task_id, question_text, file_name)
             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:
@@ -91,7 +206,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
         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 
+    # 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)