Update app.py

app.py

@@ -1,155 +1,145 @@
+"""
+GAIA Agent — Final Assignment
+Direct Groq API calls, NO smolagents CodeAgent (too slow).
+Manual tool routing: detect URLs, files, etc. and fetch context before asking Groq.
+Target: 6/20 (30%) to pass.
+"""
 import os
 import re
 import io
-import json
 import time
 import traceback
 import gradio as gr
 import requests
 import pandas as pd
 from bs4 import BeautifulSoup
-from smolagents import (
-    CodeAgent,
-    DuckDuckGoSearchTool,
-    LiteLLMModel,
-    tool,
-)
 
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-GROQ_MODEL = "groq/llama-3.3-70b-versatile"
 
 
-@tool
-def visit_webpage(url: str) -> str:
-    """Visits a webpage and extracts its main clean text content.
-    Use this to read Wikipedia pages, news articles, or any online resource.
-    Args:
-        url: The full URL of the webpage to visit.
-    """
+# ==========================================
+# 🔧 TOOLS (plain functions, no smolagents)
+# ==========================================
+
+def fetch_webpage(url: str) -> str:
     try:
         headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
-        response = requests.get(url, headers=headers, timeout=15)
-        response.raise_for_status()
-        soup = BeautifulSoup(response.text, "html.parser")
+        resp = requests.get(url, headers=headers, timeout=15)
+        resp.raise_for_status()
+        soup = BeautifulSoup(resp.text, "html.parser")
         for el in soup(["script", "style", "nav", "footer", "header", "aside", "noscript"]):
             el.extract()
-        lines = [l.strip() for l in soup.get_text(separator="\n", strip=True).splitlines() if l.strip()]
-        return "\n".join(lines)[:10000]
+        lines = [l.strip() for l in soup.get_text("\n", strip=True).splitlines() if l.strip()]
+        return "\n".join(lines)[:8000]
     except Exception as e:
-        return f"Error: {str(e)}"
+        return f"Error: {e}"
 
 
-@tool
-def get_youtube_transcript(video_url: str) -> str:
-    """Fetches the transcript/captions of a YouTube video.
-    Use this whenever the question refers to a YouTube video URL.
-    Args:
-        video_url: The full YouTube video URL (or just the video ID).
-    """
+def fetch_youtube_transcript(url: str) -> str:
     try:
         from youtube_transcript_api import YouTubeTranscriptApi
-        match = re.search(r"(?:v=|youtu\.be/|embed/)([^&\n?#]+)", video_url)
-        video_id = match.group(1) if match else video_url.strip()
+        match = re.search(r"(?:v=|youtu\.be/|embed/)([^&\n?#]+)", url)
+        vid = match.group(1) if match else url.strip()
         try:
-            entries = YouTubeTranscriptApi.get_transcript(video_id, languages=["en"])
+            entries = YouTubeTranscriptApi.get_transcript(vid, languages=["en"])
         except Exception:
-            try:
-                entries = YouTubeTranscriptApi.get_transcript(video_id)
-            except Exception:
-                tl = YouTubeTranscriptApi.list_transcripts(video_id)
-                entries = tl.find_generated_transcript(["en", "it", "fr", "de", "es"]).fetch()
-        return " ".join([e["text"] for e in entries])[:8000]
+            entries = YouTubeTranscriptApi.get_transcript(vid)
+        return " ".join([e["text"] for e in entries])[:6000]
     except Exception as e:
-        return f"Transcript error: {str(e)}"
+        return f"Transcript error: {e}"
 
 
-@tool
-def download_task_file(task_id: str) -> str:
-    """Downloads and reads the file attached to a GAIA task.
-    Handles text, CSV, JSON, PDF, Excel (.xlsx/.xls), and Python files.
-    Always try this tool first if the question might reference an attached file.
-    Args:
-        task_id: The task_id string from the GAIA question.
-    """
+def fetch_task_file(task_id: str) -> str:
     try:
-        file_url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
-        response = requests.get(file_url, timeout=20)
-        if response.status_code == 404:
-            return "No file attached to this task."
-        response.raise_for_status()
-        ct = response.headers.get("Content-Type", "")
-        cd = response.headers.get("Content-Disposition", "")
-        filename = ""
-        if "filename=" in cd:
-            filename = cd.split("filename=")[-1].strip('" ')
+        resp = requests.get(f"{DEFAULT_API_URL}/files/{task_id}", timeout=20)
+        if resp.status_code != 200:
+            return ""
+        ct = resp.headers.get("Content-Type", "")
+        cd = resp.headers.get("Content-Disposition", "")
+        filename = cd.split("filename=")[-1].strip('" ') if "filename=" in cd else ""
         ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
 
+        # Text/CSV/JSON
         if any(t in ct for t in ["text", "json", "csv"]) or ext in ["txt", "csv", "json", "py", "md"]:
-            text = response.text
             if ext == "csv" or "csv" in ct:
                 try:
-                    df = pd.read_csv(io.StringIO(text))
-                    return f"CSV with {len(df)} rows, columns: {list(df.columns)}\n\n{df.to_string()}"[:8000]
+                    df = pd.read_csv(io.StringIO(resp.text))
+                    return f"CSV: {len(df)} rows, columns={list(df.columns)}\n{df.to_string()}"[:6000]
                 except Exception:
                     pass
-            return text[:8000]
+            return resp.text[:6000]
 
+        # Excel
         if "spreadsheet" in ct or "excel" in ct or ext in ["xlsx", "xls"]:
             try:
-                df = pd.read_excel(io.BytesIO(response.content), engine="openpyxl")
-                return f"Excel with {len(df)} rows, columns: {list(df.columns)}\n\n{df.to_string()}"[:8000]
-            except Exception as e:
-                return f"Excel read error: {e}"
+                df = pd.read_excel(io.BytesIO(resp.content), engine="openpyxl")
+                return f"Excel: {len(df)} rows, columns={list(df.columns)}\n{df.to_string()}"[:6000]
+            except Exception:
+                return "Excel file (could not parse)"
 
+        # PDF
         if "pdf" in ct or ext == "pdf":
             try:
                 import PyPDF2
-                reader = PyPDF2.PdfReader(io.BytesIO(response.content))
-                return "\n".join([p.extract_text() or "" for p in reader.pages])[:8000]
-            except Exception as e:
-                return f"PDF read error: {e}"
-
-        if "audio" in ct or ext in ["mp3", "wav", "m4a", "ogg"]:
-            return f"Audio file ({ct}, {len(response.content)} bytes)."
+                reader = PyPDF2.PdfReader(io.BytesIO(resp.content))
+                return "\n".join([p.extract_text() or "" for p in reader.pages])[:6000]
+            except Exception:
+                return "PDF file (could not parse)"
 
-        if "image" in ct or ext in ["png", "jpg", "jpeg", "gif", "webp"]:
-            return f"Image file ({ct}, {len(response.content)} bytes)."
+        # Audio/Image
+        if "audio" in ct or ext in ["mp3", "wav"]:
+            return f"Audio file ({ext}, {len(resp.content)} bytes)"
+        if "image" in ct or ext in ["png", "jpg", "jpeg"]:
+            return f"Image file ({ext}, {len(resp.content)} bytes)"
 
         try:
-            return response.content.decode("utf-8")[:8000]
+            return resp.content.decode("utf-8")[:6000]
         except Exception:
-            return f"Binary file ({ct}, {len(response.content)} bytes)."
+            return f"Binary file ({ct})"
+    except Exception:
+        return ""
+
+
+def web_search(query: str) -> str:
+    """Quick DuckDuckGo search via ddgs."""
+    try:
+        from ddgs import DDGS
+        with DDGS() as ddgs:
+            results = list(ddgs.text(query, max_results=3))
+        if not results:
+            return "No results found."
+        return "\n\n".join([f"{r.get('title','')}: {r.get('body','')}" for r in results])[:4000]
     except Exception as e:
-        return f"File error: {str(e)}"
+        return f"Search error: {e}"
+
 
+# ==========================================
+# 🧹 UTILITIES
+# ==========================================
 
 def preprocess_question(question: str) -> str:
     stripped = question.strip()
-    reversed_q = stripped[::-1]
-    keywords = ["answer", "what", "who", "how", "find", "list", "which", "where", "when", "the"]
-    if sum(1 for w in keywords if w in reversed_q.lower()) > sum(1 for w in keywords if w in stripped.lower()) and len(stripped) > 20:
-        return reversed_q
+    rev = stripped[::-1]
+    kw = ["answer", "what", "who", "how", "find", "list", "which", "where", "when", "the"]
+    if sum(1 for w in kw if w in rev.lower()) > sum(1 for w in kw if w in stripped.lower()) and len(stripped) > 20:
+        return rev
     return question
 
 
 def clean_answer(raw: str) -> str:
     answer = str(raw).strip()
-    lines = [l.strip() for l in answer.split("\n") if l.strip()]
-    if lines:
-        answer = lines[0]
-    prefixes = [
-        "the answer is:", "the answer is", "final answer:", "final answer is:",
-        "final answer is", "answer:", "answer is:", "answer is",
-        "the result is:", "the result is", "the correct answer is:",
-        "the correct answer is", "based on", "according to",
-        "here is the answer:", "sure,", "sure!",
-    ]
-    lower = answer.lower()
-    for prefix in prefixes:
-        if lower.startswith(prefix):
+    # First non-empty line
+    for line in answer.split("\n"):
+        line = line.strip()
+        if line:
+            answer = line
+            break
+    # Remove prefixes
+    for prefix in ["the answer is:", "the answer is", "final answer:", "final answer is",
+                    "answer:", "answer is", "the result is", "result:", "the correct answer is",
+                    "based on", "according to", "sure,"]:
+        if answer.lower().startswith(prefix):
             answer = answer[len(prefix):].strip()
-            if answer and answer[0] in '"\'':
-                answer = answer[1:]
             break
     if answer.endswith(".") and not re.search(r"\d\.$", answer):
         answer = answer[:-1].strip()
@@ -157,143 +147,120 @@ def clean_answer(raw: str) -> str:
     return answer
 
 
-def prefetch_file(task_id: str) -> str:
-    if not task_id:
-        return ""
-    try:
-        resp = requests.get(f"https://agents-course-unit4-scoring.hf.space/files/{task_id}", timeout=10)
-        if resp.status_code != 200:
-            return ""
-        ct = resp.headers.get("Content-Type", "")
-        cd = resp.headers.get("Content-Disposition", "")
-        filename = cd.split("filename=")[-1].strip('" ') if "filename=" in cd else ""
-        ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
-
-        if any(t in ct for t in ["text", "json", "csv"]) or ext in ["txt", "csv", "json", "py"]:
-            if ext == "csv" or "csv" in ct:
-                try:
-                    df = pd.read_csv(io.StringIO(resp.text))
-                    return f"CSV: {len(df)} rows, cols={list(df.columns)}\n{df.to_string()}"[:5000]
-                except Exception:
-                    pass
-            return resp.text[:5000]
-        if "spreadsheet" in ct or "excel" in ct or ext in ["xlsx", "xls"]:
-            try:
-                df = pd.read_excel(io.BytesIO(resp.content), engine="openpyxl")
-                return f"Excel: {len(df)} rows, cols={list(df.columns)}\n{df.to_string()}"[:5000]
-            except Exception:
-                pass
-        if "pdf" in ct or ext == "pdf":
-            try:
-                import PyPDF2
-                reader = PyPDF2.PdfReader(io.BytesIO(resp.content))
-                return "\n".join([p.extract_text() or "" for p in reader.pages])[:5000]
-            except Exception:
-                pass
-        return ""
-    except Exception:
-        return ""
-
-
 def is_valid(answer: str) -> bool:
     if not answer:
         return False
-    invalid = {"i don't know", "unknown", "n/a", "none", "error", "i cannot", "i can't", "not available"}
-    return answer.lower().strip() not in invalid
+    return answer.lower().strip() not in {"i don't know", "unknown", "n/a", "none", "error", "i cannot", "i can't"}
 
 
-def direct_groq(question: str, file_context: str = "") -> str:
-    groq_key = os.getenv("GROQ_API_KEY", "")
-    if not groq_key:
-        return "I don't know"
-    extra = f"\n\nAttached file:\n{file_context[:2500]}\n" if file_context else ""
-    prompt = f"""Answer with ONLY the final answer. No explanation. No preamble.
-- Numbers: just digits. Names: just the name. Lists: comma-separated.
-{extra}
-Question: {question}
-Answer:"""
-
-    for model in ["llama-3.3-70b-versatile", "llama-3.1-8b-instant"]:
+# ==========================================
+# 🤖 GROQ DIRECT CALL
+# ==========================================
+
+def ask_groq(system: str, user: str, groq_key: str) -> str:
+    """Single Groq API call. Fast and simple."""
+    for attempt in range(3):
         try:
             resp = requests.post(
                 "https://api.groq.com/openai/v1/chat/completions",
                 headers={"Authorization": f"Bearer {groq_key}", "Content-Type": "application/json"},
-                json={"model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "max_tokens": 150},
-                timeout=20,
+                json={
+                    "model": "llama-3.3-70b-versatile",
+                    "messages": [
+                        {"role": "system", "content": system},
+                        {"role": "user", "content": user},
+                    ],
+                    "temperature": 0.1,
+                    "max_tokens": 300,
+                },
+                timeout=25,
             )
             if resp.status_code == 200:
-                raw = resp.json()["choices"][0]["message"]["content"].strip()
-                answer = clean_answer(raw)
-                if is_valid(answer):
-                    return answer
+                return resp.json()["choices"][0]["message"]["content"].strip()
             elif resp.status_code == 429:
-                time.sleep(10)
-        except Exception:
-            pass
-    return "I don't know"
-
-
-class GaiaAgent:
-    def __init__(self):
-        print("🚀 Init GaiaAgent...")
-        groq_key = os.getenv("GROQ_API_KEY", "")
-        if not groq_key:
-            raise ValueError("❌ GROQ_API_KEY mancante! Vai su console.groq.com")
-
-        self.model = LiteLLMModel(model_id=GROQ_MODEL, api_key=groq_key, temperature=0.1, max_tokens=1024)
-        self.agent = CodeAgent(
-            tools=[DuckDuckGoSearchTool(), visit_webpage, get_youtube_transcript, download_task_file],
-            model=self.model,
-            max_steps=5,  # RIDOTTO da 8 a 5 per velocità
-            additional_authorized_imports=[
-                "requests", "bs4", "json", "time", "math", "datetime",
-                "pandas", "numpy", "re", "csv", "urllib", "collections",
-                "itertools", "string", "unicodedata", "statistics",
-            ],
-        )
-        print("✅ Agent pronto!")
-
-    def __call__(self, question: str, task_id: str = "") -> str:
-        print(f"\n[Q]: {question[:120]}")
-        processed = preprocess_question(question)
-        file_context = prefetch_file(task_id)
-
-        file_hint = f'\nTask has task_id="{task_id}". Call download_task_file("{task_id}") for attached files.' if task_id else ""
-        extra = f"\n\n--- FILE ---\n{file_context[:3000]}\n---\n" if file_context else ""
-
-        prompt = f"""You are solving GAIA benchmark questions. Find the EXACT answer.
-
-STRATEGY:
-1. YouTube URL → get_youtube_transcript(url)
-2. Any URL → visit_webpage(url)
-3. Attached file → download_task_file(task_id)
-4. Factual → DuckDuckGoSearchTool + visit_webpage
-5. Math → Python code
-6. Reversed text → text[::-1]
-
-OUTPUT ONLY the bare answer. No "The answer is". No explanation.
-Numbers: just digits. Names: just the name. Lists: comma-separated.
-{file_hint}{extra}
-Question: {processed}"""
-
-        try:
-            time.sleep(1)  # RIDOTTO da 3 a 1
-            raw = self.agent.run(prompt)
-            answer = clean_answer(str(raw))
-            if is_valid(answer):
-                print(f"  ✅ {answer}")
-                return answer
+                wait = 5 * (attempt + 1)
+                print(f"    Rate limited, waiting {wait}s...")
+                time.sleep(wait)
+            else:
+                print(f"    Groq HTTP {resp.status_code}: {resp.text[:100]}")
+                return ""
         except Exception as e:
-            err = str(e)
-            print(f"  ⚠️ Agent error: {err[:150]}")
-            if "429" in err:
-                time.sleep(10)
+            print(f"    Groq error: {e}")
+            time.sleep(3)
+    return ""
+
+
+# ==========================================
+# 🧠 MAIN LOGIC: gather context, then ask
+# ==========================================
+
+SYSTEM_PROMPT = """You are an expert AI solving GAIA benchmark questions.
+You will be given a question and possibly some context (web search results, file content, webpage text, video transcript).
+Use the context to find the EXACT answer.
+
+RULES:
+- Output ONLY the final answer. Nothing else.
+- No "The answer is", no explanation, no preamble.
+- Numbers: just digits (e.g. 42)
+- Names: just the name (e.g. Einstein)
+- Lists: comma-separated (e.g. cat, dog, bird)
+- No period at the end."""
+
+
+def solve_question(question: str, task_id: str, groq_key: str) -> str:
+    """Gather context, then ask Groq once."""
+    print(f"\n[Q]: {question[:120]}")
+
+    processed = preprocess_question(question)
+    context_parts = []
+
+    # 1. Always try to fetch task file
+    file_content = fetch_task_file(task_id)
+    if file_content:
+        context_parts.append(f"ATTACHED FILE:\n{file_content}")
+        print(f"  📁 File: {len(file_content)} chars")
+
+    # 2. If YouTube URL in question
+    yt_match = re.search(r'(https?://(?:www\.)?(?:youtube\.com/watch\?v=|youtu\.be/)[^\s]+)', processed)
+    if yt_match:
+        transcript = fetch_youtube_transcript(yt_match.group(1))
+        context_parts.append(f"YOUTUBE TRANSCRIPT:\n{transcript}")
+        print(f"  🎬 YouTube transcript: {len(transcript)} chars")
+
+    # 3. If any other URL in question
+    url_match = re.search(r'(https?://[^\s]+)', processed)
+    if url_match and not yt_match:
+        page = fetch_webpage(url_match.group(1))
+        context_parts.append(f"WEBPAGE CONTENT:\n{page}")
+        print(f"  🌐 Webpage: {len(page)} chars")
+
+    # 4. For questions without URLs/files, or to supplement: web search
+    if not context_parts or (not yt_match and not url_match):
+        # Extract search query from question
+        search_q = processed[:200]  # Use question as search query
+        search_results = web_search(search_q)
+        if search_results and "error" not in search_results.lower():
+            context_parts.append(f"WEB SEARCH RESULTS:\n{search_results}")
+            print(f"  🔍 Search: {len(search_results)} chars")
+
+    # Build user message
+    context = "\n\n".join(context_parts) if context_parts else "No additional context available."
+    user_msg = f"CONTEXT:\n{context}\n\nQUESTION: {processed}"
+
+    # Ask Groq
+    raw = ask_groq(SYSTEM_PROMPT, user_msg, groq_key)
+    if not raw:
+        # Retry with simpler prompt (no context, just question)
+        raw = ask_groq(SYSTEM_PROMPT, f"QUESTION: {processed}", groq_key)
+
+    answer = clean_answer(raw) if raw else "I don't know"
+    print(f"  → {answer}")
+    return answer
 
-        # Fallback
-        answer = direct_groq(processed, file_context)
-        print(f"  🔄 Fallback: {answer}")
-        return answer
 
+# ==========================================
+# ⚙️ RUNNER
+# ==========================================
 
 def run_and_submit_all(profile: gr.OAuthProfile | None):
     space_id = os.getenv("SPACE_ID")
@@ -301,47 +268,54 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
         return "Fai il Login con Hugging Face.", None
 
     username = profile.username
-    print(f"👤 {username}")
+    groq_key = os.getenv("GROQ_API_KEY", "")
+    if not groq_key:
+        return "❌ GROQ_API_KEY mancante! Mettila nei Secrets dello Space.", None
 
-    try:
-        agent = GaiaAgent()
-    except Exception as e:
-        return f"Errore init: {e}", None
+    print(f"\n{'='*50}\n👤 {username}\n{'='*50}")
 
     agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
 
     try:
         resp = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15)
         resp.raise_for_status()
-        questions_data = resp.json()
+        questions = resp.json()
     except Exception as e:
         return f"Errore domande: {e}", None
 
-    results_log = []
-    answers_payload = []
+    print(f"📋 {len(questions)} domande\n")
+
+    results = []
+    answers = []
 
-    for i, item in enumerate(questions_data):
+    for i, item in enumerate(questions):
         task_id = item.get("task_id", "")
-        question_text = item.get("question")
-        if not task_id or question_text is None:
+        q = item.get("question")
+        if not task_id or q is None:
             continue
-        print(f"\n[{i+1}/{len(questions_data)}] ──────────")
+
+        print(f"[{i+1}/{len(questions)}] ──────────")
         try:
-            answer = agent(question_text, task_id=task_id)
+            ans = solve_question(q, task_id, groq_key)
         except Exception as e:
-            answer = "I don't know"
+            ans = "I don't know"
             print(f"  ERROR: {e}")
-        answers_payload.append({"task_id": task_id, "submitted_answer": answer})
-        results_log.append({"Task ID": task_id, "Question": question_text[:100], "Answer": answer})
 
-    if not answers_payload:
-        return "Nessuna risposta.", pd.DataFrame(results_log)
+        answers.append({"task_id": task_id, "submitted_answer": ans})
+        results.append({"Task ID": task_id, "Question": q[:100], "Answer": ans})
+
+        # Small delay between questions to avoid rate limits
+        time.sleep(1)
 
+    if not answers:
+        return "Nessuna risposta.", pd.DataFrame(results)
+
+    print(f"\n📤 Invio {len(answers)} risposte...")
     try:
         resp = requests.post(
             f"{DEFAULT_API_URL}/submit",
-            json={"username": username, "agent_code": agent_code, "answers": answers_payload},
-            timeout=120,
+            json={"username": username, "agent_code": agent_code, "answers": answers},
+            timeout=60,
         )
         resp.raise_for_status()
         r = resp.json()
@@ -350,13 +324,18 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
             f"🏆 {r.get('score', 'N/A')}% ({r.get('correct_count', '?')}/{r.get('total_attempted', '?')})\n"
             f"📝 {r.get('message', '')}"
         )
-        return status, pd.DataFrame(results_log)
+        print(f"\n{status}")
+        return status, pd.DataFrame(results)
     except Exception as e:
-        return f"❌ Invio fallito: {e}", pd.DataFrame(results_log)
+        return f"❌ Invio fallito: {e}", pd.DataFrame(results)
+
 
+# ==========================================
+# 🖥️ GRADIO
+# ==========================================
 
 with gr.Blocks() as demo:
-    gr.Markdown("# 🚀 GAIA Agent — Final Assignment\nPowered by **Groq** (Llama 3.3 70B)")
+    gr.Markdown("# 🚀 GAIA Agent — Final Assignment\nPowered by **Groq** (Llama 3.3 70B) — direct & fast")
     gr.LoginButton()
     run_button = gr.Button("🔥 Avvia Valutazione", variant="primary")
     status_output = gr.Textbox(label="Risultato", lines=5, interactive=False)