Update app.py

app.py

@@ -4,10 +4,12 @@ import requests
 import pandas as pd
 from io import BytesIO
 import re
-import ffmpeg # Using the ffmpeg-python wrapper
+import ffmpeg
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 # --- Tool-specific Imports ---
 from pytube import YouTube
+from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound
 
 # --- LangChain & Dependency Imports ---
 from groq import Groq
@@ -22,6 +24,7 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 TEMP_DIR = "/tmp"
 
 # --- Tool Definition: Audio File Transcription ---
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
 def transcribe_audio_file(task_id: str) -> str:
     """
     Downloads an audio file (.mp3) for a given task_id, transcribes it, and returns the text.
@@ -40,31 +43,31 @@ def transcribe_audio_file(task_id: str) -> str:
     except Exception as e:
         return f"Error during audio file transcription: {e}"
 
-# --- Tool Definition: Video Transcription (using ffmpeg-python) ---
+# --- Tool Definition: Video Transcription (prioritizing transcripts) ---
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
 def transcribe_youtube_video(video_url: str) -> str:
     """
-    Downloads a YouTube video from a URL, extracts its audio using FFmpeg, and transcribes it.
+    Fetches a transcript for a YouTube video from a URL, falling back to download and transcription if needed.
     Use this tool ONLY when a question provides a youtube.com URL.
     """
-    print(f"Tool 'transcribe_youtube_video' (ffmpeg-python) called with URL: {video_url}")
+    print(f"Tool 'transcribe_youtube_video' called with URL: {video_url}")
     video_path, audio_path = None, None
     try:
+        # Extract video ID and try to fetch official transcript
+        video_id = re.search(r"(?:v=|\/)([0-9A-Za-z_-]{11}).*", video_url).group(1)
+        transcript = YouTubeTranscriptApi.get_transcript(video_id)
+        return ' '.join([entry['text'] for entry in transcript])
+    except NoTranscriptFound:
+        print("No transcript found; falling back to download and transcribe.")
+        # Fallback to original download logic
         os.makedirs(TEMP_DIR, exist_ok=True)
         yt = YouTube(video_url)
         stream = yt.streams.filter(only_audio=True).first()
         video_path = stream.download(output_path=TEMP_DIR)
         audio_path = os.path.join(TEMP_DIR, "output.mp3")
-        
-        # Use ffmpeg-python to convert the downloaded file to mp3
-        (
-            ffmpeg
-            .input(video_path)
-            .output(audio_path, **{'q:a': 0, 'map': 'a'}) # Set audio quality and select audio stream
-            .overwrite_output() # Corresponds to the -y flag
-            .run(quiet=True) # Use quiet=True to avoid printing ffmpeg logs
-        )
-        
-        print(f"Audio extracted to: {audio_path}")
+        stream = ffmpeg.input(video_path)
+        stream = ffmpeg.output(stream, audio_path, q=0, map='a', y='y')
+        ffmpeg.run(stream)
         client = Groq(api_key=os.getenv("GROQ_API_KEY"))
         with open(audio_path, "rb") as audio_file:
             transcription = client.audio.transcriptions.create(file=audio_file, model="whisper-large-v3", response_format="text")
@@ -72,10 +75,8 @@ def transcribe_youtube_video(video_url: str) -> str:
     except Exception as e:
         return f"Error during YouTube transcription: {e}"
     finally:
-        # Clean up temporary files
         if video_path and os.path.exists(video_path): os.remove(video_path)
         if audio_path and os.path.exists(audio_path): os.remove(audio_path)
-        print("Cleaned up temporary files.")
 
 # --- Agent Definition ---
 class LangChainAgent:
@@ -95,9 +96,10 @@ class LangChainAgent:
                 "2.  **Select ONE tool based on the question:**\n"
                 "    - For general knowledge, facts, or current events: use `web_search`.\n"
                 "    - For an audio file, .mp3, or voice memo: use `audio_file_transcriber` with the `task_id`.\n"
-                "    - For a youtube.com URL: use `youtube_video_transcriber` with the URL.\n"
+                "    - For a youtube.com URL: use `youtube_video_transcriber` with the URL. If transcription fails, fall back to web_search for video transcripts or summaries.\n"
                 "    - For anything else (like images, which you cannot see, or math), you must answer directly without using a tool.\n"
-                "3.  **Execute and Answer:** After using a tool, analyze the result and provide ONLY THE FINAL ANSWER."
+                "3.  **Handle Errors:** If a tool fails (e.g., download error), retry once or use web_search to find alternatives.\n"
+                "4.  **Execute and Answer:** After using a tool, analyze the result and provide ONLY THE FINAL ANSWER without extra text like 'FINAL ANSWER'."
             )),
             ("human", "Question: {input}\nTask ID: {task_id}"),
             ("placeholder", "{agent_scratchpad}"),
@@ -169,7 +171,7 @@ with gr.Blocks() as demo:
 
 if __name__ == "__main__":
     print("\n" + "-"*30 + " App Starting " + "-"*30)
-    for key in ["GROQ_API_KEY", "TAVILY_API_KEY"]:
+    for key in ["GROQ_API_KEY", "TAVILY_API_KEY", "SPACE_ID"]:
         print(f"✅ {key} secret is set." if os.getenv(key) else f"⚠️ WARNING: {key} secret is not set.")
     print("-"*(60 + len(" App Starting ")) + "\n")
-    demo.launch(debug=True, share=False)
+    demo.launch(debug=True, share=False)