Update app.py

app.py

@@ -1,6 +1,14 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Optional, Union, List
+import re
+import tempfile
+import requests
+import urllib.parse as _urlparse
 import os
 import gradio as gr
-import requests
 import inspect
 import pandas as pd
 from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel
@@ -9,27 +17,274 @@ from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel
 # --- 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.")
 
-        hf_token = os.getenv("HF_TOKEN")
-        if not hf_token:
-            raise ValueError("HF_TOKEN environment variable is not set.")
 
+# ‑‑‑ smol‑agents base imports (provided by the framework) ‑‑‑
+from smol_agents import (
+    Tool,
+    PipelineTool,
+    CodeAgent,
+    DuckDuckGoSearchTool,
+    WikipediaSearchTool,
+    OpenAIServerModel,
+)
+
+# ---------------------------------------------------------------------------
+# Speech‑to‑Text (OpenAI Whisper)
+# ---------------------------------------------------------------------------
+class SpeechToTextTool(PipelineTool):
+    """Transcribe *local* audio files via OpenAI Whisper (cached)."""
+
+    name = "transcriber"
+    description = (
+        "Send a local audio file to OpenAI Whisper (model **whisper‑1**) and "
+        "return the plain‑text transcript."
+    )
+    inputs = {
+        "audio": {
+            "type": "string",
+            "description": "Absolute or relative path to a local audio file.",
+        }
+    }
+    output_type = "string"
+
+    def __call__(self, audio: str) -> str:  # noqa: D401
+        return self._transcribe(audio)
+
+    @staticmethod
+    @lru_cache(maxsize=64)
+    def _transcribe(audio_path: str) -> str:
+        path = Path(audio_path).expanduser().resolve()
+        if not path.is_file():
+            raise FileNotFoundError(f"No such audio file: {path}")
+
+        from openai import audio as _audio  # late import
+
+        with path.open("rb") as fp:
+            resp = _audio.transcriptions.create(
+                file=fp,
+                model="whisper-1",
+                response_format="text",
+            )
+        return resp
+
+
+# ---------------------------------------------------------------------------
+# Excel → Markdown helper
+# ---------------------------------------------------------------------------
+class ExcelToTextTool(Tool):
+    """Render an Excel worksheet as a Markdown table (GitHub flavour)."""
+
+    name = "excel_to_text"
+    description = (
+        "Convert an Excel sheet to Markdown. Accepts sheet name *or* index "
+        "(as string). Returns a GitHub‑style table without index column."
+    )
+
+    inputs = {
+        "excel_path": {
+            "type": "string",
+            "description": "Path to the Excel file (.xlsx / .xls).",
+        },
+        "sheet_name": {
+            "type": "string",
+            "nullable": True,
+            "description": (
+                "Worksheet name or 0‑based index *as string* (optional; "
+                "default=first sheet)."
+            ),
+        },
+    }
+    output_type = "string"
+
+    @lru_cache(maxsize=32)
+    def forward(self, excel_path: str, sheet_name: Optional[str] = None) -> str:  # type: ignore[override]
+        path = Path(excel_path).expanduser().resolve()
+        if not path.is_file():
+            return f"Error: Excel file not found at {path}"
+
+        import importlib.util as _imp
+        if not _imp.find_spec("pandas"):
+            return "Error: pandas library not available in this environment."
+        import pandas as pd
+
+        try:
+            sheet: Union[int, str] = 0
+            if sheet_name and sheet_name.strip():
+                sheet = int(sheet_name) if sheet_name.isdigit() else sheet_name
+            df = pd.read_excel(path, sheet_name=sheet)
+            if hasattr(pd.DataFrame, "to_markdown"):
+                return df.to_markdown(index=False)
+            from tabulate import tabulate  # pragma: no cover
+            return tabulate(df, headers="keys", tablefmt="github", showindex=False)
+        except Exception as exc:  # pragma: no cover – user‑visible error
+            return f"Error reading Excel file: {exc}"
+
+
+# ---------------------------------------------------------------------------
+# NEW: YouTube Question‑Answer Tool
+# ---------------------------------------------------------------------------
+class YouTubeQATool(PipelineTool):
+    """Answer questions about the spoken content of a YouTube video.
+
+    • Downloads the auto‑generated or creator‑provided transcript using
+      **youtube‑transcript‑api** (no API key needed for most public videos).
+    • Feeds a compressed transcript + user question to GPT‑4o for an answer.
+    • Caches transcripts locally to avoid repeated network calls.
+    """
+
+    name = "youtube_qa"
+    description = (
+        "Given a YouTube URL and a natural‑language *question*, return an answer "
+        "based solely on the video transcript (no hallucinations)."
+    )
+
+    inputs = {
+        "url": {
+            "type": "string",
+            "description": "Full YouTube video URL or just the watch ID.",
+        },
+        "question": {
+            "type": "string",
+            "description": "Question about the video content (English / French).",
+        },
+    }
+    output_type = "string"
+
+    # ––––– internal helpers ––––– ------------------------------------------------
+    _TRANSCRIPT_CACHE: dict[str, str] = {}  # simple in‑proc cache
+
+    @staticmethod
+    def _extract_video_id(url: str) -> str:
+        """Return the 11‑char YouTube ID from a watch/shorts URL or raw ID."""
+        if len(url) == 11 and "/" not in url:
+            return url
+        parsed = _urlparse.urlparse(url)
+        if parsed.hostname in ("youtu.be",):
+            return parsed.path.lstrip("/")
+        if parsed.hostname and "youtube" in parsed.hostname:
+            qs = _urlparse.parse_qs(parsed.query)
+            if "v" in qs:
+                return qs["v"][0]
+            # shorts/embedded
+            return parsed.path.split("/")[-1]
+        raise ValueError("Could not parse YouTube video ID from URL")
+
+    @classmethod
+    def _get_transcript(cls, video_id: str) -> str:
+        if video_id in cls._TRANSCRIPT_CACHE:
+            return cls._TRANSCRIPT_CACHE[video_id]
+        try:
+            from youtube_transcript_api import YouTubeTranscriptApi  # type: ignore
+        except ModuleNotFoundError:
+            return "Error: youtube‑transcript‑api library not installed."
+        try:
+            segments: List[dict] = YouTubeTranscriptApi.get_transcript(video_id)
+        except Exception as exc:  # private video, disabled captions, …
+            return f"Error fetching transcript: {exc}"
+        text = " ".join(seg["text"] for seg in segments)
+        cls._TRANSCRIPT_CACHE[video_id] = text
+        return text
+
+    # ––––– main entry point ––––– -------------------------------------------
+    def forward(self, url: str, question: str) -> str:  # type: ignore[override]
+        try:
+            vid = self._extract_video_id(url)
+        except ValueError as e:
+            return str(e)
+
+        transcript = self._get_transcript(vid)
+        if transcript.startswith("Error"):
+            return transcript
+
+        # Keep prompt under ~15k chars – truncate transcript if necessary
+        max_chars = 15000
+        if len(transcript) > max_chars:
+            transcript = transcript[:max_chars] + " …(truncated)…"
+
+        from openai import chat  # lazy import OpenAI client only here
+
+        system = (
+            "You are a meticulous assistant. Answer the user's question about "
+            "the provided YouTube transcript. If the transcript lacks the "
+            "information, reply 'I don't know based on the transcript.'"
+        )
+        messages = [
+            {"role": "system", "content": system},
+            {"role": "user", "content": f"Transcript:\n{transcript}"},
+            {"role": "user", "content": f"Question: {question}"},
+        ]
+        try:
+            resp = chat.completions.create(
+                model="gpt-4o",  # uses the same hosted model as the agent
+                messages=messages,
+                temperature=0.2,
+                max_tokens=256,
+            )
+            return resp.choices[0].message.content.strip()
+        except Exception as exc:  # pragma: no cover
+            return f"Error generating answer: {exc}"
+
+
+# ---------------------------------------------------------------------------
+# Helper: download attachment (if any)
+# ---------------------------------------------------------------------------
+
+def download_file_if_any(base_api_url: str, task_id: str) -> str | None:
+    url = f"{base_api_url}/files/{task_id}"
+    try:
+        resp = requests.get(url, timeout=30)
+        if resp.status_code == 404:
+            return None
+        resp.raise_for_status()
+    except requests.HTTPError:
+        raise
+
+    filename = task_id
+    if cd := resp.headers.get("content-disposition"):
+        if match := re.search(r'filename="([^"]+)"', cd):
+            filename = match.group(1)
+
+    tmp_dir = Path(tempfile.gettempdir(), "gaia_files")
+    tmp_dir.mkdir(exist_ok=True)
+    file_path = tmp_dir / filename
+    file_path.write_bytes(resp.content)
+    return str(file_path)
+
+
+# ---------------------------------------------------------------------------
+# Minimal agent wired with our custom tools
+# ---------------------------------------------------------------------------
+class BasicAgent:
+    _model = OpenAIServerModel(model_id="gpt-4o")
+    _tools = [
+        DuckDuckGoSearchTool(),
+        WikipediaSearchTool(),
+        SpeechToTextTool(),
+        ExcelToTextTool(),
+        YouTubeQATool(),  # <-- NEW
+    ]
+
+    def __init__(self) -> None:
         self.agent = CodeAgent(
-            tools=[DuckDuckGoSearchTool()],
-            model=HfApiModel(token=hf_token, model="microsoft/phi-2")
+            model=self._model,
+            tools=self._tools,
+            add_base_tools=True,
+            additional_authorized_imports=[
+                "numpy",
+                "pandas",
+                "csv",
+                "subprocess",
+            ],
         )
+        print("BasicAgent initialized with YouTubeQATool.")
+
+    def __call__(self, question: str) -> str:  # noqa: D401
+        print(f"Agent received question (first 80 chars): {question[:80]}…")
+        answer = self.agent.run(question)
+        print(f"Agent returning answer: {answer}")
+        return answer
 
-    def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = self.agent.run("You are a helpful assistant answering short factual questions. Answer the question: " + question, max_steps=1)
-        print(fixed_answer)
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """