Update app.py

app.py

@@ -1,5 +1,6 @@
 import os
 import re
+import base64
 from io import BytesIO
 from pathlib import Path
 from zipfile import ZipFile
@@ -7,7 +8,6 @@ from zipfile import ZipFile
 import gradio as gr
 import pandas as pd
 import requests
-from litellm import completion
 from smolagents import (
     CodeAgent,
     DuckDuckGoSearchTool,
@@ -22,7 +22,9 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 RESULT_COLUMNS = ["Task ID", "Question", "Submitted Answer"]
 HTTP_TIMEOUT = 45
 MAX_EXTRACTED_CHARS = 35_000
-MODEL_ID = "huggingface/Qwen/Qwen2.5-Coder-32B-Instruct"
+MODEL_ID = os.getenv(
+    "GAIA_MODEL_ID", "huggingface/openai/gpt-oss-120b"
+)
 
 
 def clean_filename(value: str) -> str:
@@ -101,12 +103,35 @@ def extract_attachment_text(data: bytes, filename: str) -> str:
             BytesIO(data), sep=separator, encoding_errors="replace"
         )
         text = dataframe.to_csv(index=False)
-    elif suffix in {".txt", ".md", ".json", ".html", ".htm", ".xml"}:
+    elif suffix in {
+        ".txt",
+        ".md",
+        ".json",
+        ".html",
+        ".htm",
+        ".xml",
+        ".py",
+    }:
         text = data.decode("utf-8", errors="replace")
         if suffix in {".html", ".htm"}:
             from bs4 import BeautifulSoup
 
             text = BeautifulSoup(text, "html.parser").get_text("\n")
+    elif suffix in {".mp3", ".wav", ".flac", ".m4a", ".ogg"}:
+        from huggingface_hub import InferenceClient
+
+        token = os.getenv("HF_TOKEN")
+        if not token:
+            text = "Audio transcription failed: HF_TOKEN is not configured."
+        else:
+            client = InferenceClient(api_key=token, provider="auto")
+            asr_model = os.getenv(
+                "GAIA_ASR_MODEL", "openai/whisper-large-v3"
+            )
+            transcript = client.automatic_speech_recognition(
+                data, model=asr_model
+            )
+            text = f"Audio transcript:\n{transcript.text}"
     elif suffix == ".zip":
         with ZipFile(BytesIO(data)) as archive:
             text = "Files inside ZIP:\n" + "\n".join(archive.namelist())
@@ -167,6 +192,122 @@ class InspectGaiaAttachmentTool(Tool):
             return f"Could not inspect attachment for task {task_id}: {exc}"
 
 
+class YouTubeTranscriptTool(Tool):
+    name = "youtube_transcript"
+    description = (
+        "Retrieves the spoken transcript or subtitles of a YouTube video. "
+        "Use it for questions asking what a person says in a linked video. "
+        "It cannot determine purely visual events."
+    )
+    inputs = {
+        "url": {
+            "type": "string",
+            "description": "Full YouTube URL or the 11-character video ID.",
+        }
+    }
+    output_type = "string"
+
+    def forward(self, url: str) -> str:
+        from youtube_transcript_api import YouTubeTranscriptApi
+
+        value = str(url or "").strip()
+        match = re.search(
+            r"(?:v=|youtu\.be/|shorts/)([A-Za-z0-9_-]{11})", value
+        )
+        video_id = match.group(1) if match else value
+        if not re.fullmatch(r"[A-Za-z0-9_-]{11}", video_id):
+            return "Could not identify a valid YouTube video ID."
+
+        try:
+            api = YouTubeTranscriptApi()
+            transcript = api.fetch(video_id)
+            lines = []
+            for snippet in transcript:
+                text = getattr(snippet, "text", None)
+                if text is None and isinstance(snippet, dict):
+                    text = snippet.get("text")
+                if text:
+                    lines.append(str(text))
+            result = " ".join(lines).strip()
+            return result or "The video has no available transcript."
+        except Exception as exc:
+            return f"Could not retrieve YouTube transcript: {exc}"
+
+
+class AnalyzeGaiaImageTool(Tool):
+    name = "analyze_gaia_image"
+    description = (
+        "Downloads the official image for a GAIA task and analyzes it with a "
+        "vision model. Use this for questions whose answer depends on image "
+        "pixels, diagrams, chess positions, or visual details."
+    )
+    inputs = {
+        "task_id": {
+            "type": "string",
+            "description": "The exact GAIA task_id associated with the image.",
+        },
+        "question": {
+            "type": "string",
+            "description": "The complete question the image must answer.",
+        },
+    }
+    output_type = "string"
+
+    def forward(self, task_id: str, question: str) -> str:
+        from openai import OpenAI
+
+        token = os.getenv("HF_TOKEN")
+        if not token:
+            return "Image analysis failed: HF_TOKEN is not configured."
+
+        try:
+            response = requests.get(
+                f"{DEFAULT_API_URL}/files/{str(task_id).strip()}",
+                timeout=HTTP_TIMEOUT,
+            )
+            response.raise_for_status()
+            mime = response.headers.get("content-type", "image/png").split(";")[0]
+            encoded = base64.b64encode(response.content).decode("ascii")
+
+            client = OpenAI(
+                base_url="https://router.huggingface.co/v1",
+                api_key=token,
+            )
+            vision_model = os.getenv(
+                "GAIA_VISION_MODEL",
+                "Qwen/Qwen3-VL-235B-A22B-Instruct:cheapest",
+            )
+            result = client.chat.completions.create(
+                model=vision_model,
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "text",
+                                "text": (
+                                    "Analyze the supplied image carefully and "
+                                    "answer this task. Explain visual evidence "
+                                    f"briefly so another agent can verify it:\n{question}"
+                                ),
+                            },
+                            {
+                                "type": "image_url",
+                                "image_url": {
+                                    "url": f"data:{mime};base64,{encoded}"
+                                },
+                            },
+                        ],
+                    }
+                ],
+                temperature=0,
+                max_tokens=600,
+            )
+            return str(result.choices[0].message.content).strip()
+        except Exception as exc:
+            return f"Could not analyze the GAIA image: {exc}"
+
+
 class BasicAgent:
     def __init__(self):
         print("Inicializando o agente GAIA...")
@@ -194,6 +335,8 @@ class BasicAgent:
                     language="en",
                 ),
                 InspectGaiaAttachmentTool(),
+                YouTubeTranscriptTool(),
+                AnalyzeGaiaImageTool(),
             ],
             model=self.model,
             max_steps=10,
@@ -220,6 +363,12 @@ Research carefully before answering and cross-check uncertain facts.
 Use web_search to find sources and visit_webpage to read a result in detail.
 If the task mentions an attached file, call inspect_gaia_attachment with the
 task_id given in the task.
+If the task asks what someone says in a YouTube video, call youtube_transcript
+with the exact video URL before searching the web.
+If the task depends on an attached image, call analyze_gaia_image with the
+task_id and complete question. Do not try to infer image contents from metadata.
+Prefer primary or official sources. When search snippets conflict, open the
+source and verify the relevant passage instead of guessing.
 Only call tools that are explicitly available. Never invent a function such as
 visit_webpage if it is not listed, and never use subprocess or shell commands.
 Do not repeat nearly identical searches. If one approach fails, change source
@@ -295,43 +444,34 @@ If a number is requested, return only that number.
         return text.replace("FINAL ANSWER", "").strip()
 
     def format_exact_answer(self, question: str, raw_answer: str) -> str:
-        """Usa uma chamada curta para converter o resultado em exact match."""
-        raw_answer = self.deterministic_answer_cleanup(raw_answer)
-        formatter_prompt = f"""
-You are a strict answer formatter for an exact-match benchmark.
-
-Original question:
-{question}
-
-Candidate answer:
-{raw_answer}
-
-Extract only the final value requested by the original question.
-Do not solve the question again and do not explain anything.
-Do not include labels such as "answer" or "final answer".
-Do not use Markdown, citations, full sentences, or quotation marks unless the
-question explicitly requires them.
-Preserve the requested ordering, spelling, capitalization, units, plural form,
-and separators. If a comma-separated list is requested, output only that list.
-Your entire response must be the exact answer and nothing else.
-""".strip()
-
-        try:
-            response = completion(
-                model=MODEL_ID,
-                api_key=self.hf_token,
-                messages=[{"role": "user", "content": formatter_prompt}],
-                temperature=0,
-                max_tokens=180,
+        """Limpa o resultado mecanicamente, sem pedir a outro modelo para alterá-lo."""
+        del question
+        cleaned = self.deterministic_answer_cleanup(raw_answer)
+        lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
+
+        # Quando o agente ainda inclui uma explicação e deixa uma resposta curta
+        # isolada na última linha, conserva somente essa última linha.
+        if len(lines) > 1:
+            last_line = lines[-1]
+            reasoning_cues = (
+                "because",
+                "therefore",
+                "research",
+                "source",
+                "conclude",
+                "analysis",
+                "porque",
+                "portanto",
+                "pesquisa",
+                "conclu",
             )
-            formatted = response.choices[0].message.content
-            cleaned = self.deterministic_answer_cleanup(formatted)
-            if cleaned:
-                return cleaned
-        except Exception as exc:
-            print(f"Exact-answer formatter failed; using cleaned result: {exc}")
+            preceding = " ".join(lines[:-1]).lower()
+            if len(last_line) <= 250 and any(
+                cue in preceding for cue in reasoning_cues
+            ):
+                cleaned = last_line
 
-        return raw_answer
+        return self.deterministic_answer_cleanup(cleaned)
 
 
 def empty_results() -> pd.DataFrame: