Update app.py

app.py

@@ -1,451 +1,3 @@
-import os
-import re
-import io
-import json
-import time
-import base64
-import mimetypes
-from pathlib import Path
-from urllib.parse import urlparse
-
-import gradio as gr
-import pandas as pd
-import requests
-import spaces
-from bs4 import BeautifulSoup
-from openai import OpenAI
-
-# ---------------------------------------------------------------------
-# Configuration
-# ---------------------------------------------------------------------
-DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-
-# Configure these in Hugging Face Space → Settings → Variables and secrets.
-OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
-OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o")
-OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
-AUDIO_MODEL = os.getenv("AUDIO_MODEL", "whisper-1")
-
-MAX_ATTACHMENT_CHARS = 24000
-MAX_WEB_CHARS = 14000
-HTTP_TIMEOUT = 45
-
-# ZeroGPU startup compatibility. This function is deliberately never called.
-# The agent uses OpenAI's remote API, not the Space GPU.
-@spaces.GPU(duration=1)
-def _zerogpu_startup_probe():
-    return None
-
-
-# ---------------------------------------------------------------------
-# Utility functions
-# ---------------------------------------------------------------------
-def clean_answer(text: str) -> str:
-    """Remove common answer wrappers while preserving the answer itself."""
-    answer = (text or "").strip()
-    answer = re.sub(
-        r"^\s*(?:FINAL ANSWER|FINAL|ANSWER)\s*[:\-]\s*",
-        "",
-        answer,
-        flags=re.IGNORECASE,
-    )
-    answer = answer.strip()
-    if len(answer) >= 2 and answer[0] == answer[-1] and answer[0] in {"'", '"'}:
-        answer = answer[1:-1].strip()
-    return answer
-
-
-def extract_urls(text: str) -> list[str]:
-    urls = re.findall(r"https?://[^\s<>\])}]+", text or "")
-    return [u.rstrip(".,;:!?") for u in urls]
-
-
-def image_to_data_url(filename: str, content: bytes) -> str:
-    mime, _ = mimetypes.guess_type(filename)
-    if not mime:
-        mime = "image/jpeg"
-    encoded = base64.b64encode(content).decode("ascii")
-    return f"data:{mime};base64,{encoded}"
-
-
-# ---------------------------------------------------------------------
-# GAIA Agent
-# ---------------------------------------------------------------------
-class BasicAgent:
-    def __init__(self):
-        if not OPENAI_API_KEY:
-            raise ValueError(
-                "Missing OPENAI_API_KEY. Add it under Space Settings → "
-                "Variables and secrets."
-            )
-
-        self.client = OpenAI(
-            api_key=OPENAI_API_KEY,
-            base_url=OPENAI_BASE_URL,
-            timeout=120,
-            max_retries=2,
-        )
-        self.api_url = DEFAULT_API_URL
-        print(f"BasicAgent initialized with model: {OPENAI_MODEL}")
-
-    def download_attachment(self, task_id: str, file_name: str) -> bytes | None:
-        if not file_name:
-            return None
-
-        url = f"{self.api_url}/files/{task_id}"
-        print(f"Downloading attachment for task {task_id}: {file_name}")
-        response = requests.get(url, timeout=HTTP_TIMEOUT)
-        response.raise_for_status()
-        return response.content
-
-    def extract_attachment(
-        self, file_name: str, content: bytes
-    ) -> tuple[str, list[dict]]:
-        """
-        Return extracted textual context and optional GPT-4o image content.
-        Handles common GAIA attachment types.
-        """
-        suffix = Path(file_name).suffix.lower()
-        text_parts: list[str] = []
-        image_parts: list[dict] = []
-
-        # Plain text and structured text.
-        if suffix in {
-            ".txt", ".md", ".json", ".csv", ".tsv", ".py", ".html",
-            ".xml", ".log", ".yaml", ".yml",
-        }:
-            text_parts.append(
-                content.decode("utf-8", errors="replace")[:MAX_ATTACHMENT_CHARS]
-            )
-
-        # PDF documents.
-        elif suffix == ".pdf":
-            try:
-                from pypdf import PdfReader
-                reader = PdfReader(io.BytesIO(content))
-                pages = []
-                for page in reader.pages[:35]:
-                    pages.append(page.extract_text() or "")
-                text_parts.append("\n".join(pages)[:MAX_ATTACHMENT_CHARS])
-            except Exception as exc:
-                text_parts.append(f"[PDF extraction error: {exc}]")
-
-        # Excel workbooks. Include rows, not just a small preview, where possible.
-        elif suffix in {".xlsx", ".xls"}:
-            try:
-                engine = "openpyxl" if suffix == ".xlsx" else "xlrd"
-                sheets = pd.read_excel(
-                    io.BytesIO(content),
-                    sheet_name=None,
-                    engine=engine,
-                )
-                for sheet_name, dataframe in list(sheets.items())[:8]:
-                    text_parts.append(
-                        f"SHEET: {sheet_name}\n"
-                        f"ROWS: {len(dataframe)}\n"
-                        f"DATA:\n{dataframe.to_csv(index=False)}"
-                    )
-                    if sum(len(part) for part in text_parts) > MAX_ATTACHMENT_CHARS:
-                        break
-                text_parts = ["\n\n".join(text_parts)[:MAX_ATTACHMENT_CHARS]]
-            except Exception as exc:
-                text_parts.append(f"[Spreadsheet extraction error: {exc}]")
-
-        # Word documents.
-        elif suffix == ".docx":
-            try:
-                from docx import Document
-                doc = Document(io.BytesIO(content))
-                paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
-                for table in doc.tables:
-                    for row in table.rows:
-                        paragraphs.append(" | ".join(cell.text for cell in row.cells))
-                text_parts.append("\n".join(paragraphs)[:MAX_ATTACHMENT_CHARS])
-            except Exception as exc:
-                text_parts.append(f"[DOCX extraction error: {exc}]")
-
-        # PowerPoint presentations.
-        elif suffix == ".pptx":
-            try:
-                from pptx import Presentation
-                presentation = Presentation(io.BytesIO(content))
-                slides_text = []
-                for slide_number, slide in enumerate(presentation.slides, start=1):
-                    slide_lines = []
-                    for shape in slide.shapes:
-                        if getattr(shape, "has_text_frame", False):
-                            slide_lines.append(shape.text)
-                        if getattr(shape, "has_table", False):
-                            for row in shape.table.rows:
-                                slide_lines.append(
-                                    " | ".join(cell.text for cell in row.cells)
-                                )
-                    slides_text.append(
-                        f"SLIDE {slide_number}\n" + "\n".join(slide_lines)
-                    )
-                text_parts.append("\n\n".join(slides_text)[:MAX_ATTACHMENT_CHARS])
-            except Exception as exc:
-                text_parts.append(f"[PPTX extraction error: {exc}]")
-
-        # Images: send directly to GPT-4o vision.
-        elif suffix in {".png", ".jpg", ".jpeg", ".webp", ".gif"}:
-            image_parts.append({
-                "type": "image_url",
-                "image_url": {"url": image_to_data_url(file_name, content)},
-            })
-
-        # Audio: transcribe through OpenAI's audio transcription endpoint.
-        elif suffix in {".mp3", ".wav", ".m4a", ".mpeg", ".mpga", ".webm"}:
-            try:
-                mime, _ = mimetypes.guess_type(file_name)
-                with io.BytesIO(content) as audio_file:
-                    audio_file.name = Path(file_name).name
-                    transcription = self.client.audio.transcriptions.create(
-                        model=AUDIO_MODEL,
-                        file=audio_file,
-                    )
-                text_parts.append(
-                    "AUDIO TRANSCRIPTION:\n" + (transcription.text or "")
-                )
-            except Exception as exc:
-                text_parts.append(f"[Audio transcription error: {exc}]")
-
-        else:
-            text_parts.append(
-                f"[Attachment '{file_name}' was downloaded, but its format "
-                "is not supported by this implementation.]"
-            )
-
-        return "\n\n".join(text_parts), image_parts
-
-    def search_web(self, query: str) -> str:
-        """Best-effort DuckDuckGo search plus extraction of a few result pages."""
-        try:
-            from ddgs import DDGS
-            results = DDGS().text(query, max_results=5)
-        except Exception as exc:
-            print(f"Web search unavailable: {exc}")
-            return ""
-
-        snippets = []
-        urls_to_read = []
-
-        for result in results or []:
-            title = result.get("title", "")
-            body = result.get("body", "")
-            url = result.get("href") or result.get("url") or ""
-            snippets.append(f"TITLE: {title}\nURL: {url}\nSNIPPET: {body}")
-            if url.startswith("http"):
-                urls_to_read.append(url)
-
-        page_texts = []
-        for url in urls_to_read[:2]:
-            try:
-                response = requests.get(
-                    url,
-                    timeout=20,
-                    headers={"User-Agent": "Mozilla/5.0 (compatible; GAIAAgent/1.0)"},
-                )
-                response.raise_for_status()
-                content_type = response.headers.get("content-type", "").lower()
-                if "text/html" not in content_type:
-                    continue
-                soup = BeautifulSoup(response.text, "html.parser")
-                for tag in soup(["script", "style", "nav", "footer", "header"]):
-                    tag.decompose()
-                page_text = soup.get_text(" ", strip=True)
-                if page_text:
-                    page_texts.append(
-                        f"SOURCE URL: {url}\nPAGE TEXT:\n{page_text[:5000]}"
-                    )
-            except Exception as exc:
-                print(f"Could not read search result {url}: {exc}")
-
-        combined = "\n\n".join(snippets + page_texts)
-        return combined[:MAX_WEB_CHARS]
-
-    def fetch_given_urls(self, question: str) -> str:
-        """Read URLs explicitly supplied in the question."""
-        chunks = []
-        for url in extract_urls(question)[:3]:
-            try:
-                response = requests.get(
-                    url,
-                    timeout=25,
-                    headers={"User-Agent": "Mozilla/5.0 (compatible; GAIAAgent/1.0)"},
-                )
-                response.raise_for_status()
-                content_type = response.headers.get("content-type", "").lower()
-                if "text/html" in content_type or "text/plain" in content_type:
-                    soup = BeautifulSoup(response.text, "html.parser")
-                    for tag in soup(["script", "style", "nav", "footer", "header"]):
-                        tag.decompose()
-                    content = soup.get_text(" ", strip=True)
-                    chunks.append(f"URL: {url}\nCONTENT:\n{content[:5000]}")
-            except Exception as exc:
-                chunks.append(f"URL: {url}\nCould not read page: {exc}")
-        return "\n\n".join(chunks)[:MAX_WEB_CHARS]
-
-    def ask_model(
-        self,
-        question: str,
-        attachment_context: str = "",
-        web_context: str = "",
-        image_parts: list[dict] | None = None,
-    ) -> str:
-        image_parts = image_parts or []
-
-        system_prompt = """
-You are a highly capable research and problem-solving assistant answering
-GAIA benchmark questions.
-
-Your objective is to return the exact answer requested by the question.
-
-Rules:
-1. Analyze the question carefully and identify every constraint.
-2. Use the supplied attachment content when relevant.
-3. Use web evidence when the question asks for obscure, historical, current,
-   or externally verifiable facts. Prefer reliable primary sources.
-4. For calculations, calculate precisely. For spreadsheet questions, inspect
-   all relevant rows and distinguish the requested categories.
-5. Do not guess when the supplied evidence contradicts your assumptions.
-6. Follow the requested output format exactly: alphabetize when requested,
-   use comma-separated values when requested, and use the requested name form.
-7. For numeric answers, do not add units, currency symbols, commas, or
-   explanatory text unless the question explicitly requests them.
-8. Return ONLY the final answer. No explanation, citations, markdown, or
-   'FINAL ANSWER:' prefix.
-"""
-
-        user_content = [
-            {
-                "type": "text",
-                "text": (
-                    f"QUESTION:\n{question}\n\n"
-                    f"ATTACHMENT CONTENT:\n"
-                    f"{attachment_context or 'No attachment content.'}\n\n"
-                    f"WEB RESEARCH:\n"
-                    f"{web_context or 'No web research retrieved.'}"
-                ),
-            }
-        ]
-        user_content.extend(image_parts[:3])
-
-        last_error = None
-        for attempt in range(3):
-            try:
-                response = self.client.chat.completions.create(
-                    model=OPENAI_MODEL,
-                    temperature=0,
-                    max_tokens=700,
-                    messages=[
-                        {"role": "system", "content": system_prompt},
-                        {"role": "user", "content": user_content},
-                    ],
-                )
-                answer = response.choices[0].message.content or ""
-                return clean_answer(answer)
-            except Exception as exc:
-                last_error = exc
-                print(f"OpenAI request attempt {attempt + 1} failed: {exc}")
-                if attempt < 2:
-                    time.sleep(2 ** attempt)
-
-        raise RuntimeError(f"OpenAI model request failed: {last_error}")
-
-    def __call__(self, task: dict) -> str:
-        task_id = str(task.get("task_id", ""))
-        question = str(task.get("question", ""))
-        file_name = str(task.get("file_name") or "")
-
-        print(f"Processing task {task_id}; attachment={file_name or 'none'}")
-
-        attachment_context = ""
-        image_parts = []
-
-        if file_name:
-            try:
-                file_bytes = self.download_attachment(task_id, file_name)
-                if file_bytes:
-                    attachment_context, image_parts = self.extract_attachment(
-                        file_name, file_bytes
-                    )
-            except Exception as exc:
-                attachment_context = f"[Attachment download failed: {exc}]"
-
-        # Search for externally verifiable questions. Avoid unnecessary search
-        # when the question is clearly about a provided file or simple sorting.
-        search_context = self.fetch_given_urls(question)
-        lower_question = question.lower()
-        needs_research = any(
-            marker in lower_question
-            for marker in (
-                "as of", "who is", "who was", "which", "what year",
-                "what date", "first name", "last name", "nationality",
-                "recipient", "according to", "number before", "number after",
-                "historical", "born", "award", "competition",
-            )
-        )
-
-        if needs_research and not attachment_context:
-            query = question[:450]
-            found = self.search_web(query)
-            if found:
-                search_context = (search_context + "\n\n" + found).strip()
-
-        answer = self.ask_model(
-            question=question,
-            attachment_context=attachment_context,
-            web_context=search_context,
-            image_parts=image_parts,
-        )
-
-        print(f"Answer for {task_id}: {answer}")
-        return answer
-
-
-# ---------------------------------------------------------------------
-# Course evaluation / submission workflow
-# ---------------------------------------------------------------------
-def run_and_submit_all(profile: gr.OAuthProfile | None, progress=gr.Progress()):
-    space_id = os.getenv("SPACE_ID", "")
-
-    if not profile:
-        return "Please sign in to Hugging Face using the login button.", pd.DataFrame()
-
-    username = str(profile.username).strip()
-    questions_url = f"{DEFAULT_API_URL}/questions"
-    submit_url = f"{DEFAULT_API_URL}/submit"
-
-    try:
-        agent = BasicAgent()
-    except Exception as exc:
-        return f"Agent initialization failed: {exc}", pd.DataFrame()
-
-    agent_code = (
-        f"https://huggingface.co/spaces/{space_id}/tree/main"
-        if space_id else ""
-    )
-
-    try:
-        response = requests.get(questions_url, timeout=HTTP_TIMEOUT)
-        response.raise_for_status()
-        questions_data = response.json()
-        if not isinstance(questions_data, list) or not questions_data:
-            return "The scoring API returned no questions.", pd.DataFrame()
-    except Exception as exc:
-        return f"Could not fetch benchmark questions: {exc}", pd.DataFrame()
-
-    results_log = []
-    answers_payload = []
-    total = len(questions_data)
-
-    for index, item in enumerate(questions_data, start=1):
-        task_id = item.get("task_id")
-        question_text = item.get("question")
-
-        if not task_id or question_text is None:
-            continue
 
         progress((index - 1) / max(total, 1), desc=f"Solving question {index}/{total}")
 
@@ -570,4 +122,4 @@ if __name__ == "__main__":
     print("Starting GAIA GPT-4o evaluation app...")
     print(f"Model: {OPENAI_MODEL}")
     print(f"Space ID: {os.getenv('SPACE_ID', 'local')}")
-    demo.launch(debug=True, share=False)
+    demo.launch(debug=True, share=False)