Final_Assignment_Template

Sleeping

Update app.py

2b9de41 verified 16 days ago

25.6 kB

import os
import io
import re
import time
import json
import base64
import zipfile
import gradio as gr
import requests
import pandas as pd
import spaces
from types import SimpleNamespace
from openai import OpenAI
from smolagents import CodeAgent, DuckDuckGoSearchTool, OpenAIServerModel, tool
from huggingface_hub import InferenceClient

# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
GROQ_BASE_URL = "https://api.groq.com/openai/v1"

PREFERRED_GROQ_MODELS = [
    "qwen/qwen3.8-27b",
    "openai/gpt-oss-120b",
    "openai/gpt-oss-20b",
    "allam-2-7b"
]

BROWSER_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9"
}

# Satisfies ZeroGPU scanner requirement
@spaces.GPU
def check_gpu_startup():
    return True

# --- Dedicated Tools (Hardened against 403, PDF binary dumps, and Truncation) ---

@tool
def search_wikipedia(query: str) -> str:
    """Search English Wikipedia for a topic and return the full relevant article content (including discography/tables).
    Args:
        query: The topic, person, or article title to search on Wikipedia.
    """
    import wikipedia
    wikipedia.set_lang("en")
    wikipedia.set_user_agent("GAIAResearchAgent/1.0 (https://huggingface.co; contact@huggingface.co)")
    try:
        search_results = wikipedia.search(query, results=5)
        if not search_results:
            return f"No Wikipedia pages found for '{query}'."

        page = None
        for title in search_results:
            try:
                page = wikipedia.page(title, auto_suggest=False)
                break
            except wikipedia.exceptions.DisambiguationError as e:
                if e.options:
                    try:
                        page = wikipedia.page(e.options[0], auto_suggest=False)
                        break
                    except Exception:
                        continue
            except Exception:
                continue

        if page is None:
            return f"Could not load Wikipedia page for '{query}'."

        content = page.content
        # If article is short enough, return all of it (up to 14,000 chars)
        if len(content) <= 14000:
            return f"Title: {page.title}\nURL: {page.url}\n\n{content}"

        # If article is very long, keep intro + matching sections (e.g., Discography, Albums, Career)
        intro = content[:3500]
        sections = re.split(r"(?=\n==+\s*[^=]+\s*==+)", content)
        query_words = [w.lower() for w in re.findall(r"\w+", query) if len(w) > 3]
        priority_keywords = query_words + ["discography", "album", "studio", "list", "work", "award", "career", "filmography", "result"]

        matched_sections = []
        for sec in sections:
            sec_lower = sec.lower()
            if any(k in sec_lower for k in priority_keywords):
                matched_sections.append(sec)

        combined = f"Title: {page.title}\nURL: {page.url}\n\n[INTRO]\n{intro}\n\n[RELEVANT SECTIONS]\n" + "\n".join(matched_sections)
        if len(combined) > 14000:
            # Also include the tail of the article where discographies/lists usually live
            combined = combined[:9000] + "\n...\n[END OF ARTICLE]\n" + content[-5000:]
        return combined
    except Exception as e:
        return f"Wikipedia search error: {e}"

@tool
def visit_webpage(url: str) -> str:
    """Fetch and extract clean readable text from a webpage URL or online PDF (safe against binary dumps and 403 blocks).
    Args:
        url: The full HTTP/HTTPS URL to read.
    """
    from markdownify import markdownify
    from pypdf import PdfReader
    try:
        resp = requests.get(url, headers=BROWSER_HEADERS, timeout=15)
        resp.raise_for_status()
        content_type = resp.headers.get("Content-Type", "").lower()

        # Safely parse online PDFs instead of dumping raw binary bytes!
        if "application/pdf" in content_type or url.lower().endswith(".pdf") or resp.content.startswith(b"%PDF"):
            reader = PdfReader(io.BytesIO(resp.content))
            pdf_text = []
            for i, page in enumerate(reader.pages[:10]):
                txt = page.extract_text() or ""
                if txt.strip():
                    pdf_text.append(f"--- Page {i+1} ---\n{txt.strip()}")
            joined_pdf = "\n".join(pdf_text)
            return joined_pdf[:6000] if joined_pdf else "PDF parsed but no extractable text found."

        # Parse HTML cleanly
        md_text = markdownify(resp.text)
        lines = [line.strip() for line in md_text.splitlines() if line.strip()]
        cleaned = "\n".join(lines)
        return cleaned[:6000]
    except Exception as e:
        return f"Could not fetch webpage ({url}): {e}. Try using web_search or search_wikipedia instead."

@tool
def transcribe_audio(file_path: str) -> str:
    """Transcribe a local audio file (.mp3, .wav) into text using Whisper speech-to-text.
    Args:
        file_path: The local path to the audio file.
    """
    groq_key = os.getenv("GROQ_API_KEY")
    if groq_key:
        try:
            client = OpenAI(base_url=GROQ_BASE_URL, api_key=groq_key)
            for whisper_model in ["whisper-large-v3-turbo", "whisper-large-v3"]:
                try:
                    with open(file_path, "rb") as audio_file:
                        transcription = client.audio.transcriptions.create(
                            model=whisper_model,
                            file=audio_file
                        )
                    return transcription.text
                except Exception as inner_e:
                    print(f"Groq Whisper ({whisper_model}) error: {inner_e}")
        except Exception as e:
            print(f"Groq audio transcription error: {e}")

    token = os.getenv("HF_TOKEN")
    client = InferenceClient(token=token)
    try:
        with open(file_path, "rb") as f:
            audio_bytes = f.read()
        res = client.automatic_speech_recognition(
            audio_bytes,
            model="openai/whisper-large-v3-turbo"
        )
        if hasattr(res, "text"):
            return res.text
        if isinstance(res, dict) and "text" in res:
            return res["text"]
        return str(res)
    except Exception as e:
        return f"Speech-to-text transcription failed: {e}"

@tool
def analyze_image(file_path: str, question: str) -> str:
    """Analyze a local image file (.png, .jpg) to answer visual questions (e.g., chess boards, charts, text in images).
    Args:
        file_path: Local path to the image file.
        question: Specific question or instruction about the image content.
    """
    groq_key = os.getenv("GROQ_API_KEY")
    try:
        with open(file_path, "rb") as f:
            b64_img = base64.b64encode(f.read()).decode("utf-8")
        ext = os.path.splitext(file_path)[-1].lower().lstrip(".")
        mime = "image/jpeg" if ext in ["jpg", "jpeg"] else "image/png"

        if groq_key:
            client = OpenAI(base_url=GROQ_BASE_URL, api_key=groq_key)
            models_resp = client.models.list()
            avail = [m.id for m in models_resp.data]
            vision_candidates = [
                m for m in avail
                if any(k in m.lower() for k in ["vision", "scout", "maverick", "llava"])
            ]
            for v_model in vision_candidates:
                try:
                    res = client.chat.completions.create(
                        model=v_model,
                        messages=[
                            {
                                "role": "user",
                                "content": [
                                    {"type": "text", "text": question},
                                    {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64_img}"}}
                                ]
                            }
                        ],
                        max_tokens=500
                    )
                    return res.choices[0].message.content
                except Exception as ve:
                    print(f"Vision model {v_model} failed: {ve}")
        return "No vision model available on account to inspect image pixels directly."
    except Exception as e:
        return f"Image analysis error: {e}"

# --- Dynamic Model Discovery & Context-Safe Resilient Wrapper ---

def discover_groq_models(api_key: str) -> list[str]:
    """Identify active Groq chat models and prioritize Qwen."""
    client = OpenAI(base_url=GROQ_BASE_URL, api_key=api_key)
    try:
        models_resp = client.models.list()
        available_ids = [m.id for m in models_resp.data]
    except Exception as e:
        print(f"Could not list Groq models: {e}")
        available_ids = PREFERRED_GROQ_MODELS

    verified = []
    for pref in PREFERRED_GROQ_MODELS:
        if pref in available_ids:
            try:
                client.chat.completions.create(
                    model=pref,
                    messages=[{"role": "user", "content": "1"}],
                    max_tokens=5
                )
                print(f"Verified working Groq model: {pref}")
                verified.append(pref)
            except Exception as e:
                print(f"Model {pref} ping failed: {e}")

    return verified if verified else ["qwen/qwen3.8-27b"]

def sanitize_messages_length(messages: list) -> list:
    """Truncate overly long tool observations in message history to prevent Error 400 Context Overflow."""
    if not isinstance(messages, list):
        return messages
    sanitized = []
    for idx, msg in enumerate(messages):
        if isinstance(msg, dict):
            msg_copy = dict(msg)
            content = msg_copy.get("content", "")
            # Keep system prompt (idx 0) intact, cap any oversized observation message
            if idx > 0 and isinstance(content, str) and len(content) > 4500:
                msg_copy["content"] = content[:4500] + "\n...[TRUNCATED TO FIT CONTEXT WINDOW]..."
            sanitized.append(msg_copy)
        else:
            sanitized.append(msg)
    return sanitized

def create_resilient_groq_model(api_key: str, verified_models: list[str]) -> OpenAIServerModel:
    """Create OpenAIServerModel with automatic context trimming, tool rescue, and failover."""
    primary_model = verified_models[0]

    model = OpenAIServerModel(
        model_id=primary_model,
        api_base=GROQ_BASE_URL,
        api_key=api_key
    )

    original_create = model.client.chat.completions.create

    def resilient_create(*args, **kwargs):
        time.sleep(1.2)
        if "messages" in kwargs:
            kwargs["messages"] = sanitize_messages_length(kwargs["messages"])

        last_err = None
        for attempt in range(4):
            active_model = verified_models[attempt % len(verified_models)]
            kwargs["model"] = active_model

            try:
                return original_create(*args, **kwargs)
            except Exception as e:
                last_err = e
                err_str = str(e)

                # 1. Rescue if context length still exceeded (Error 400 reduce length)
                if "reduce the length" in err_str.lower() or "context_length" in err_str.lower():
                    print("Context overflow detected! Aggressively trimming intermediate messages...")
                    msgs = kwargs.get("messages", [])
                    if len(msgs) > 3:
                        # Keep system prompt, first user task, and last 2 messages
                        kwargs["messages"] = [msgs[0], msgs[1]] + msgs[-2:]
                    continue

                # 2. Rescue Groq's 'Tool choice is none, but model called a tool' bug
                if "tool choice is none" in err_str.lower() or "tool_use_failed" in err_str.lower():
                    print("Intercepted Groq tool-choice error. Rescuing code from failed_generation...")
                    try:
                        extracted = None
                        if hasattr(e, "body") and isinstance(e.body, dict):
                            fg = e.body.get("error", {}).get("failed_generation", "")
                        else:
                            m = re.search(r"'failed_generation':\s*'({.*?})'", err_str, re.DOTALL)
                            fg = m.group(1) if m else ""

                        if '"arguments":' in fg:
                            code_candidate = fg.split('"arguments":', 1)[-1].strip()
                            if code_candidate.endswith("}"):
                                code_candidate = code_candidate[:-1].strip()
                            extracted = code_candidate.strip("'\"")

                        if extracted:
                            return SimpleNamespace(
                                choices=[
                                    SimpleNamespace(
                                        message=SimpleNamespace(
                                            role="assistant",
                                            content=f"```python\n{extracted}\n```"
                                        )
                                    )
                                ]
                            )
                    except Exception as rescue_err:
                        print(f"Rescue parse failed: {rescue_err}")

                # 3. Rate-limit or capacity retry
                if any(k in err_str.lower() for k in ["429", "rate limit", "503", "413"]):
                    wait_time = 8 * (attempt + 1)
                    print(f"Rate limit on {active_model}. Sleeping {wait_time}s and switching...")
                    time.sleep(wait_time)
                else:
                    raise e

        raise last_err

    model.client.chat.completions.create = resilient_create
    return model

# --- Answer Post-Processing (Fixes Monologue Leaks) ---

def clean_gaia_answer(raw_result: str) -> str:
    """Extract pure exact-match value even if the model leaked a chain-of-thought paragraph."""
    if raw_result is None:
        return ""

    text = str(raw_result).strip()

    # Remove <think>...</think> blocks if emitted by reasoning models
    text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

    # If the model leaked a monologue ending with 'raw final value. X' or 'FINAL ANSWER: X'
    patterns = [
        r"(?:final answer|raw final value|the answer)\s*(?:is|:|-|\.)\s*([^\n\.]+)\.?$",
        r"Wait, the rules say output ONLY the raw final value\.?\s*(.+)$",
    ]
    for pat in patterns:
        m = re.search(pat, text, flags=re.IGNORECASE | re.MULTILINE)
        if m:
            text = m.group(1).strip()
            break

    # If still multi-line, take the last non-empty line
    lines = [line.strip() for line in text.splitlines() if line.strip()]
    if len(lines) > 1:
        text = lines[-1]

    # If it's a long reasoning paragraph (>120 chars) with multiple sentences, extract the final sentence/fragment
    if len(text) > 120 and ". " in text:
        parts = [p.strip() for p in text.split(". ") if p.strip()]
        if parts:
            text = parts[-1]

    # Strip prefixes
    for prefix in ["final answer:", "answer:", "the answer is", "raw final value:"]:
        if text.lower().startswith(prefix):
            text = text[len(prefix):].strip(" :\n*`")

    text = text.strip(" '\"`*")
    # Strip trailing period unless it's an abbreviation
    if text.endswith(".") and not text.endswith(".."):
        text = text[:-1].strip()

    # Normalize whole-number floats ('42.0' -> '42')
    try:
        as_float = float(text)
        if as_float == int(as_float):
            text = str(int(as_float))
    except ValueError:
        pass

    return text

class GAIAgentWrapper:
    """Wrapper that initializes a fresh CodeAgent per question to prevent memory bleed."""
    def __init__(self):
        groq_key = os.getenv("GROQ_API_KEY")
        if not groq_key:
            raise RuntimeError("GROQ_API_KEY secret is missing in Space settings.")
        self.groq_key = groq_key
        self.verified_models = discover_groq_models(groq_key)
        self.model = create_resilient_groq_model(groq_key, self.verified_models)
        self.search_tool = DuckDuckGoSearchTool()

    def __call__(self, prompt: str) -> str:
        # Fresh CodeAgent instance per question for clean state
        agent = CodeAgent(
            tools=[self.search_tool, visit_webpage, search_wikipedia, transcribe_audio, analyze_image],
            model=self.model,
            additional_authorized_imports=[
                "pandas", "numpy", "openpyxl", "pypdf", "PIL", "math", "re",
                "datetime", "json", "requests", "bs4", "csv", "statistics",
                "os", "io", "urllib", "collections", "itertools", "docx",
                "pptx", "wikipedia", "time", "zipfile", "base64"
            ],
            max_steps=8
        )

        strict_instruction = (
            f"{prompt}\n\n"
            "CRITICAL EXECUTION & FORMAT INSTRUCTIONS:\n"
            "1. ALWAYS call `final_answer(your_answer)` inside a Python code block as soon as you know the answer.\n"
            "2. For Wikipedia questions, use `search_wikipedia(query=...)` first.\n"
            "3. NEVER print raw binary bytes or full base64 strings of files.\n"
            "4. Output ONLY the raw final value inside `final_answer(...)`: NO explanations, NO units, NO comma separators in numbers (write 16000, not 16,000), and for lists use comma + single space ('a, b, c')."
        )

        try:
            raw_result = agent.run(strict_instruction)
            cleaned = clean_gaia_answer(raw_result)
            print(f"==> Final Cleaned Answer: {cleaned}")
            return cleaned
        except Exception as e:
            print(f"Agent execution error: {e}")
            return "unknown"

# --- File Handling ---

def detect_file_extension(content: bytes) -> str:
    """Detect file extension accurately by inspecting magic bytes and ZIP structure."""
    if content.startswith(b"%PDF"):
        return ".pdf"
    if content.startswith(b"\x89PNG\r\n\x1a\n"):
        return ".png"
    if content.startswith(b"\xff\xd8\xff"):
        return ".jpg"
    if content.startswith(b"ID3") or content.startswith(b"\xff\xfb"):
        return ".mp3"
    if content.startswith(b"RIFF") and b"WAVE" in content[:12]:
        return ".wav"
    if content.startswith(b"PK\x03\x04"):
        try:
            with zipfile.ZipFile(io.BytesIO(content)) as z:
                names = z.namelist()
                if any(n.startswith("word/") for n in names):
                    return ".docx"
                if any(n.startswith("ppt/") for n in names):
                    return ".pptx"
                if any(n.startswith("xl/") for n in names):
                    return ".xlsx"
        except Exception:
            pass
        return ".xlsx"
    return ".txt"

def download_task_file(api_url: str, task_id: str, suggested_name: str | None = None, save_dir: str = "./downloads") -> str | None:
    """Download task attachment from API with accurate extension."""
    os.makedirs(save_dir, exist_ok=True)
    file_url = f"{api_url}/files/{task_id}"
    try:
        response = requests.get(file_url, stream=True, timeout=30)
        if response.status_code == 200:
            content = response.content
            if suggested_name and "." in suggested_name:
                fname = suggested_name
            else:
                cd = response.headers.get("content-disposition", "")
                if "filename=" in cd:
                    fname = cd.split("filename=")[-1].strip('"\'; ')
                else:
                    ext = detect_file_extension(content)
                    fname = f"{task_id}{ext}"

            file_path = os.path.join(save_dir, fname)
            with open(file_path, "wb") as f:
                f.write(content)
            print(f"Downloaded file for task {task_id}: {file_path}")
            return file_path
    except Exception as e:
        print(f"File download error for task {task_id}: {e}")
    return None

# --- Main Evaluation Logic ---

def run_and_submit_all(profile: gr.OAuthProfile | None):
    """Fetch questions, run CodeAgent, submit answers, and display results."""
    space_id = os.getenv("SPACE_ID")

    if profile:
        username = f"{profile.username}"
        print(f"User logged in: {username}")
    else:
        print("User not logged in.")
        return "Please Login to Hugging Face with the button.", None

    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    submit_url = f"{api_url}/submit"

    try:
        agent = GAIAgentWrapper()
    except Exception as e:
        print(f"Error instantiating agent: {e}")
        return f"Error initializing agent: {e}", None

    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
    print(f"Agent code URL: {agent_code}")

    print(f"Fetching questions from: {questions_url}")
    try:
        response = requests.get(questions_url, timeout=15)
        response.raise_for_status()
        questions_data = response.json()
        if not questions_data:
            return "Fetched questions list is empty or invalid format.", None
        print(f"Fetched {len(questions_data)} questions.")
    except Exception as e:
        print(f"Error fetching questions: {e}")
        return f"Error fetching questions: {e}", None

    results_log = []
    answers_payload = []
    print(f"Running agent on {len(questions_data)} questions...")
    for idx, item in enumerate(questions_data, 1):
        task_id = item.get("task_id")
        question_text = item.get("question")
        suggested_filename = item.get("file_name")
        if not task_id or question_text is None:
            continue

        print(f"\n==========================================")
        print(f"Solving Question {idx}/{len(questions_data)} (Task {task_id})")
        print(f"==========================================")
        file_path = download_task_file(api_url, task_id, suggested_name=suggested_filename)
        if file_path:
            file_ext = os.path.splitext(file_path)[-1].lower()
            if file_ext in [".mp3", ".wav"]:
                full_prompt = (
                    f"{question_text}\n\n"
                    f"An audio file is attached at: '{file_path}'. "
                    f"Call `transcribe_audio(file_path='{file_path}')` to get its transcript."
                )
            elif file_ext in [".png", ".jpg", ".jpeg"]:
                full_prompt = (
                    f"{question_text}\n\n"
                    f"An image file is attached at: '{file_path}'. "
                    f"Call `analyze_image(file_path='{file_path}', question=...)` to inspect it. Do NOT print raw image bytes."
                )
            else:
                full_prompt = (
                    f"{question_text}\n\n"
                    f"A task file is provided at local path: '{file_path}' (Detected type: {file_ext}). "
                    f"Inspect and parse this file using Python (pandas, openpyxl, pypdf, docx, pptx, etc.)."
                )
        else:
            full_prompt = question_text

        try:
            submitted_answer = agent(full_prompt)
            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
        except Exception as e:
            print(f"Error running agent on task {task_id}: {e}")
            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": "unknown"})

        time.sleep(1.5)

    if not answers_payload:
        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)

    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
    try:
        response = requests.post(submit_url, json=submission_data, timeout=60)
        response.raise_for_status()
        result_data = response.json()
        final_status = (
            f"Submission Successful!\n"
            f"User: {result_data.get('username')}\n"
            f"Overall Score: {result_data.get('score', 'N/A')}%\n"
            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
            f"Message: {result_data.get('message', 'No message received.')}"
        )
        print("Submission successful.")
        return final_status, pd.DataFrame(results_log)
    except Exception as e:
        status_message = f"Submission error: {e}"
        print(status_message)
        return status_message, pd.DataFrame(results_log)

# --- UI Setup ---
with gr.Blocks() as demo:
    gr.Markdown("# GAIA CodeAgent Evaluation Runner")
    gr.LoginButton()
    run_button = gr.Button("Run Evaluation & Submit All Answers")
    status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

    run_button.click(
        fn=run_and_submit_all,
        outputs=[status_output, results_table]
    )

if __name__ == "__main__":
    demo.launch(debug=True, share=False)

	import os
	import io
	import re
	import time
	import json
	import base64
	import zipfile
	import gradio as gr
	import requests
	import pandas as pd
	import spaces
	from types import SimpleNamespace
	from openai import OpenAI
	from smolagents import CodeAgent, DuckDuckGoSearchTool, OpenAIServerModel, tool
	from huggingface_hub import InferenceClient

	# --- Constants ---
	DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
	GROQ_BASE_URL = "https://api.groq.com/openai/v1"

	PREFERRED_GROQ_MODELS = [
	"qwen/qwen3.8-27b",
	"openai/gpt-oss-120b",
	"openai/gpt-oss-20b",
	"allam-2-7b"
	]

	BROWSER_HEADERS = {
	"User-Agent": (
	"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
	"AppleWebKit/537.36 (KHTML, like Gecko) "
	"Chrome/124.0.0.0 Safari/537.36"
	),
	"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,/;q=0.8",
	"Accept-Language": "en-US,en;q=0.9"
	}

	# Satisfies ZeroGPU scanner requirement
	@spaces.GPU
	def check_gpu_startup():
	return True

	# --- Dedicated Tools (Hardened against 403, PDF binary dumps, and Truncation) ---

	@tool
	def search_wikipedia(query: str) -> str:
	"""Search English Wikipedia for a topic and return the full relevant article content (including discography/tables).
	Args:
	query: The topic, person, or article title to search on Wikipedia.
	"""
	import wikipedia
	wikipedia.set_lang("en")
	wikipedia.set_user_agent("GAIAResearchAgent/1.0 (https://huggingface.co; contact@huggingface.co)")
	try:
	search_results = wikipedia.search(query, results=5)
	if not search_results:
	return f"No Wikipedia pages found for '{query}'."

	page = None
	for title in search_results:
	try:
	page = wikipedia.page(title, auto_suggest=False)
	break
	except wikipedia.exceptions.DisambiguationError as e:
	if e.options:
	try:
	page = wikipedia.page(e.options[0], auto_suggest=False)
	break
	except Exception:
	continue
	except Exception:
	continue

	if page is None:
	return f"Could not load Wikipedia page for '{query}'."

	content = page.content
	# If article is short enough, return all of it (up to 14,000 chars)
	if len(content) <= 14000:
	return f"Title: {page.title}\nURL: {page.url}\n\n{content}"

	# If article is very long, keep intro + matching sections (e.g., Discography, Albums, Career)
	intro = content[:3500]
	sections = re.split(r"(?=\n==+\s[^=]+\s==+)", content)
	query_words = [w.lower() for w in re.findall(r"\w+", query) if len(w) > 3]
	priority_keywords = query_words + ["discography", "album", "studio", "list", "work", "award", "career", "filmography", "result"]

	matched_sections = []
	for sec in sections:
	sec_lower = sec.lower()
	if any(k in sec_lower for k in priority_keywords):
	matched_sections.append(sec)

	combined = f"Title: {page.title}\nURL: {page.url}\n\n[INTRO]\n{intro}\n\n[RELEVANT SECTIONS]\n" + "\n".join(matched_sections)
	if len(combined) > 14000:
	# Also include the tail of the article where discographies/lists usually live
	combined = combined[:9000] + "\n...\n[END OF ARTICLE]\n" + content[-5000:]
	return combined
	except Exception as e:
	return f"Wikipedia search error: {e}"

	@tool
	def visit_webpage(url: str) -> str:
	"""Fetch and extract clean readable text from a webpage URL or online PDF (safe against binary dumps and 403 blocks).
	Args:
	url: The full HTTP/HTTPS URL to read.
	"""
	from markdownify import markdownify
	from pypdf import PdfReader
	try:
	resp = requests.get(url, headers=BROWSER_HEADERS, timeout=15)
	resp.raise_for_status()
	content_type = resp.headers.get("Content-Type", "").lower()

	# Safely parse online PDFs instead of dumping raw binary bytes!
	if "application/pdf" in content_type or url.lower().endswith(".pdf") or resp.content.startswith(b"%PDF"):
	reader = PdfReader(io.BytesIO(resp.content))
	pdf_text = []
	for i, page in enumerate(reader.pages[:10]):
	txt = page.extract_text() or ""
	if txt.strip():
	pdf_text.append(f"--- Page {i+1} ---\n{txt.strip()}")
	joined_pdf = "\n".join(pdf_text)
	return joined_pdf[:6000] if joined_pdf else "PDF parsed but no extractable text found."

	# Parse HTML cleanly
	md_text = markdownify(resp.text)
	lines = [line.strip() for line in md_text.splitlines() if line.strip()]
	cleaned = "\n".join(lines)
	return cleaned[:6000]
	except Exception as e:
	return f"Could not fetch webpage ({url}): {e}. Try using web_search or search_wikipedia instead."

	@tool
	def transcribe_audio(file_path: str) -> str:
	"""Transcribe a local audio file (.mp3, .wav) into text using Whisper speech-to-text.
	Args:
	file_path: The local path to the audio file.
	"""
	groq_key = os.getenv("GROQ_API_KEY")
	if groq_key:
	try:
	client = OpenAI(base_url=GROQ_BASE_URL, api_key=groq_key)
	for whisper_model in ["whisper-large-v3-turbo", "whisper-large-v3"]:
	try:
	with open(file_path, "rb") as audio_file:
	transcription = client.audio.transcriptions.create(
	model=whisper_model,
	file=audio_file
	)
	return transcription.text
	except Exception as inner_e:
	print(f"Groq Whisper ({whisper_model}) error: {inner_e}")
	except Exception as e:
	print(f"Groq audio transcription error: {e}")

	token = os.getenv("HF_TOKEN")
	client = InferenceClient(token=token)
	try:
	with open(file_path, "rb") as f:
	audio_bytes = f.read()
	res = client.automatic_speech_recognition(
	audio_bytes,
	model="openai/whisper-large-v3-turbo"
	)
	if hasattr(res, "text"):
	return res.text
	if isinstance(res, dict) and "text" in res:
	return res["text"]
	return str(res)
	except Exception as e:
	return f"Speech-to-text transcription failed: {e}"

	@tool
	def analyze_image(file_path: str, question: str) -> str:
	"""Analyze a local image file (.png, .jpg) to answer visual questions (e.g., chess boards, charts, text in images).
	Args:
	file_path: Local path to the image file.
	question: Specific question or instruction about the image content.
	"""
	groq_key = os.getenv("GROQ_API_KEY")
	try:
	with open(file_path, "rb") as f:
	b64_img = base64.b64encode(f.read()).decode("utf-8")
	ext = os.path.splitext(file_path)[-1].lower().lstrip(".")
	mime = "image/jpeg" if ext in ["jpg", "jpeg"] else "image/png"

	if groq_key:
	client = OpenAI(base_url=GROQ_BASE_URL, api_key=groq_key)
	models_resp = client.models.list()
	avail = [m.id for m in models_resp.data]
	vision_candidates = [
	m for m in avail
	if any(k in m.lower() for k in ["vision", "scout", "maverick", "llava"])
	]
	for v_model in vision_candidates:
	try:
	res = client.chat.completions.create(
	model=v_model,
	messages=[
	{
	"role": "user",
	"content": [
	{"type": "text", "text": question},
	{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64_img}"}}
	]
	}
	],
	max_tokens=500
	)
	return res.choices[0].message.content
	except Exception as ve:
	print(f"Vision model {v_model} failed: {ve}")
	return "No vision model available on account to inspect image pixels directly."
	except Exception as e:
	return f"Image analysis error: {e}"

	# --- Dynamic Model Discovery & Context-Safe Resilient Wrapper ---

	def discover_groq_models(api_key: str) -> list[str]:
	"""Identify active Groq chat models and prioritize Qwen."""
	client = OpenAI(base_url=GROQ_BASE_URL, api_key=api_key)
	try:
	models_resp = client.models.list()
	available_ids = [m.id for m in models_resp.data]
	except Exception as e:
	print(f"Could not list Groq models: {e}")
	available_ids = PREFERRED_GROQ_MODELS

	verified = []
	for pref in PREFERRED_GROQ_MODELS:
	if pref in available_ids:
	try:
	client.chat.completions.create(
	model=pref,
	messages=[{"role": "user", "content": "1"}],
	max_tokens=5
	)
	print(f"Verified working Groq model: {pref}")
	verified.append(pref)
	except Exception as e:
	print(f"Model {pref} ping failed: {e}")

	return verified if verified else ["qwen/qwen3.8-27b"]

	def sanitize_messages_length(messages: list) -> list:
	"""Truncate overly long tool observations in message history to prevent Error 400 Context Overflow."""
	if not isinstance(messages, list):
	return messages
	sanitized = []
	for idx, msg in enumerate(messages):
	if isinstance(msg, dict):
	msg_copy = dict(msg)
	content = msg_copy.get("content", "")
	# Keep system prompt (idx 0) intact, cap any oversized observation message
	if idx > 0 and isinstance(content, str) and len(content) > 4500:
	msg_copy["content"] = content[:4500] + "\n...[TRUNCATED TO FIT CONTEXT WINDOW]..."
	sanitized.append(msg_copy)
	else:
	sanitized.append(msg)
	return sanitized

	def create_resilient_groq_model(api_key: str, verified_models: list[str]) -> OpenAIServerModel:
	"""Create OpenAIServerModel with automatic context trimming, tool rescue, and failover."""
	primary_model = verified_models[0]

	model = OpenAIServerModel(
	model_id=primary_model,
	api_base=GROQ_BASE_URL,
	api_key=api_key
	)

	original_create = model.client.chat.completions.create

	def resilient_create(args, *kwargs):
	time.sleep(1.2)
	if "messages" in kwargs:
	kwargs["messages"] = sanitize_messages_length(kwargs["messages"])

	last_err = None
	for attempt in range(4):
	active_model = verified_models[attempt % len(verified_models)]
	kwargs["model"] = active_model

	try:
	return original_create(args, *kwargs)
	except Exception as e:
	last_err = e
	err_str = str(e)

	# 1. Rescue if context length still exceeded (Error 400 reduce length)
	if "reduce the length" in err_str.lower() or "context_length" in err_str.lower():
	print("Context overflow detected! Aggressively trimming intermediate messages...")
	msgs = kwargs.get("messages", [])
	if len(msgs) > 3:
	# Keep system prompt, first user task, and last 2 messages
	kwargs["messages"] = [msgs[0], msgs[1]] + msgs[-2:]
	continue

	# 2. Rescue Groq's 'Tool choice is none, but model called a tool' bug
	if "tool choice is none" in err_str.lower() or "tool_use_failed" in err_str.lower():
	print("Intercepted Groq tool-choice error. Rescuing code from failed_generation...")
	try:
	extracted = None
	if hasattr(e, "body") and isinstance(e.body, dict):
	fg = e.body.get("error", {}).get("failed_generation", "")
	else:
	m = re.search(r"'failed_generation':\s'({.?})'", err_str, re.DOTALL)
	fg = m.group(1) if m else ""

	if '"arguments":' in fg:
	code_candidate = fg.split('"arguments":', 1)[-1].strip()
	if code_candidate.endswith("}"):
	code_candidate = code_candidate[:-1].strip()
	extracted = code_candidate.strip("'\"")

	if extracted:
	return SimpleNamespace(
	choices=[
	SimpleNamespace(
	message=SimpleNamespace(
	role="assistant",
	content=f"```python\n{extracted}\n```"
	)
	)
	]
	)
	except Exception as rescue_err:
	print(f"Rescue parse failed: {rescue_err}")

	# 3. Rate-limit or capacity retry
	if any(k in err_str.lower() for k in ["429", "rate limit", "503", "413"]):
	wait_time = 8 * (attempt + 1)
	print(f"Rate limit on {active_model}. Sleeping {wait_time}s and switching...")
	time.sleep(wait_time)
	else:
	raise e

	raise last_err

	model.client.chat.completions.create = resilient_create
	return model

	# --- Answer Post-Processing (Fixes Monologue Leaks) ---

	def clean_gaia_answer(raw_result: str) -> str:
	"""Extract pure exact-match value even if the model leaked a chain-of-thought paragraph."""
	if raw_result is None:
	return ""

	text = str(raw_result).strip()

	# Remove <think>...</think> blocks if emitted by reasoning models
	text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

	# If the model leaked a monologue ending with 'raw final value. X' or 'FINAL ANSWER: X'
	patterns = [
	r"(?:final answer\|raw final value\|the answer)\s(?:is\|:\|-\|\.)\s([^\n\.]+)\.?$",
	r"Wait, the rules say output ONLY the raw final value\.?\s*(.+)$",
	]
	for pat in patterns:
	m = re.search(pat, text, flags=re.IGNORECASE \| re.MULTILINE)
	if m:
	text = m.group(1).strip()
	break

	# If still multi-line, take the last non-empty line
	lines = [line.strip() for line in text.splitlines() if line.strip()]
	if len(lines) > 1:
	text = lines[-1]

	# If it's a long reasoning paragraph (>120 chars) with multiple sentences, extract the final sentence/fragment
	if len(text) > 120 and ". " in text:
	parts = [p.strip() for p in text.split(". ") if p.strip()]
	if parts:
	text = parts[-1]

	# Strip prefixes
	for prefix in ["final answer:", "answer:", "the answer is", "raw final value:"]:
	if text.lower().startswith(prefix):
	text = text[len(prefix):].strip(" :\n*`")

	text = text.strip(" '\"`*")
	# Strip trailing period unless it's an abbreviation
	if text.endswith(".") and not text.endswith(".."):
	text = text[:-1].strip()

	# Normalize whole-number floats ('42.0' -> '42')
	try:
	as_float = float(text)
	if as_float == int(as_float):
	text = str(int(as_float))
	except ValueError:
	pass

	return text

	class GAIAgentWrapper:
	"""Wrapper that initializes a fresh CodeAgent per question to prevent memory bleed."""
	def __init__(self):
	groq_key = os.getenv("GROQ_API_KEY")
	if not groq_key:
	raise RuntimeError("GROQ_API_KEY secret is missing in Space settings.")
	self.groq_key = groq_key
	self.verified_models = discover_groq_models(groq_key)
	self.model = create_resilient_groq_model(groq_key, self.verified_models)
	self.search_tool = DuckDuckGoSearchTool()

	def __call__(self, prompt: str) -> str:
	# Fresh CodeAgent instance per question for clean state
	agent = CodeAgent(
	tools=[self.search_tool, visit_webpage, search_wikipedia, transcribe_audio, analyze_image],
	model=self.model,
	additional_authorized_imports=[
	"pandas", "numpy", "openpyxl", "pypdf", "PIL", "math", "re",
	"datetime", "json", "requests", "bs4", "csv", "statistics",
	"os", "io", "urllib", "collections", "itertools", "docx",
	"pptx", "wikipedia", "time", "zipfile", "base64"
	],
	max_steps=8
	)

	strict_instruction = (
	f"{prompt}\n\n"
	"CRITICAL EXECUTION & FORMAT INSTRUCTIONS:\n"
	"1. ALWAYS call `final_answer(your_answer)` inside a Python code block as soon as you know the answer.\n"
	"2. For Wikipedia questions, use `search_wikipedia(query=...)` first.\n"
	"3. NEVER print raw binary bytes or full base64 strings of files.\n"
	"4. Output ONLY the raw final value inside `final_answer(...)`: NO explanations, NO units, NO comma separators in numbers (write 16000, not 16,000), and for lists use comma + single space ('a, b, c')."
	)

	try:
	raw_result = agent.run(strict_instruction)
	cleaned = clean_gaia_answer(raw_result)
	print(f"==> Final Cleaned Answer: {cleaned}")
	return cleaned
	except Exception as e:
	print(f"Agent execution error: {e}")
	return "unknown"

	# --- File Handling ---

	def detect_file_extension(content: bytes) -> str:
	"""Detect file extension accurately by inspecting magic bytes and ZIP structure."""
	if content.startswith(b"%PDF"):
	return ".pdf"
	if content.startswith(b"\x89PNG\r\n\x1a\n"):
	return ".png"
	if content.startswith(b"\xff\xd8\xff"):
	return ".jpg"
	if content.startswith(b"ID3") or content.startswith(b"\xff\xfb"):
	return ".mp3"
	if content.startswith(b"RIFF") and b"WAVE" in content[:12]:
	return ".wav"
	if content.startswith(b"PK\x03\x04"):
	try:
	with zipfile.ZipFile(io.BytesIO(content)) as z:
	names = z.namelist()
	if any(n.startswith("word/") for n in names):
	return ".docx"
	if any(n.startswith("ppt/") for n in names):
	return ".pptx"
	if any(n.startswith("xl/") for n in names):
	return ".xlsx"
	except Exception:
	pass
	return ".xlsx"
	return ".txt"

	def download_task_file(api_url: str, task_id: str, suggested_name: str \| None = None, save_dir: str = "./downloads") -> str \| None:
	"""Download task attachment from API with accurate extension."""
	os.makedirs(save_dir, exist_ok=True)
	file_url = f"{api_url}/files/{task_id}"
	try:
	response = requests.get(file_url, stream=True, timeout=30)
	if response.status_code == 200:
	content = response.content
	if suggested_name and "." in suggested_name:
	fname = suggested_name
	else:
	cd = response.headers.get("content-disposition", "")
	if "filename=" in cd:
	fname = cd.split("filename=")[-1].strip('"\'; ')
	else:
	ext = detect_file_extension(content)
	fname = f"{task_id}{ext}"

	file_path = os.path.join(save_dir, fname)
	with open(file_path, "wb") as f:
	f.write(content)
	print(f"Downloaded file for task {task_id}: {file_path}")
	return file_path
	except Exception as e:
	print(f"File download error for task {task_id}: {e}")
	return None

	# --- Main Evaluation Logic ---

	def run_and_submit_all(profile: gr.OAuthProfile \| None):
	"""Fetch questions, run CodeAgent, submit answers, and display results."""
	space_id = os.getenv("SPACE_ID")

	if profile:
	username = f"{profile.username}"
	print(f"User logged in: {username}")
	else:
	print("User not logged in.")
	return "Please Login to Hugging Face with the button.", None

	api_url = DEFAULT_API_URL
	questions_url = f"{api_url}/questions"
	submit_url = f"{api_url}/submit"

	try:
	agent = GAIAgentWrapper()
	except Exception as e:
	print(f"Error instantiating agent: {e}")
	return f"Error initializing agent: {e}", None

	agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
	print(f"Agent code URL: {agent_code}")

	print(f"Fetching questions from: {questions_url}")
	try:
	response = requests.get(questions_url, timeout=15)
	response.raise_for_status()
	questions_data = response.json()
	if not questions_data:
	return "Fetched questions list is empty or invalid format.", None
	print(f"Fetched {len(questions_data)} questions.")
	except Exception as e:
	print(f"Error fetching questions: {e}")
	return f"Error fetching questions: {e}", None

	results_log = []
	answers_payload = []
	print(f"Running agent on {len(questions_data)} questions...")
	for idx, item in enumerate(questions_data, 1):
	task_id = item.get("task_id")
	question_text = item.get("question")
	suggested_filename = item.get("file_name")
	if not task_id or question_text is None:
	continue

	print(f"\n==========================================")
	print(f"Solving Question {idx}/{len(questions_data)} (Task {task_id})")
	print(f"==========================================")
	file_path = download_task_file(api_url, task_id, suggested_name=suggested_filename)
	if file_path:
	file_ext = os.path.splitext(file_path)[-1].lower()
	if file_ext in [".mp3", ".wav"]:
	full_prompt = (
	f"{question_text}\n\n"
	f"An audio file is attached at: '{file_path}'. "
	f"Call `transcribe_audio(file_path='{file_path}')` to get its transcript."
	)
	elif file_ext in [".png", ".jpg", ".jpeg"]:
	full_prompt = (
	f"{question_text}\n\n"
	f"An image file is attached at: '{file_path}'. "
	f"Call `analyze_image(file_path='{file_path}', question=...)` to inspect it. Do NOT print raw image bytes."
	)
	else:
	full_prompt = (
	f"{question_text}\n\n"
	f"A task file is provided at local path: '{file_path}' (Detected type: {file_ext}). "
	f"Inspect and parse this file using Python (pandas, openpyxl, pypdf, docx, pptx, etc.)."
	)
	else:
	full_prompt = question_text

	try:
	submitted_answer = agent(full_prompt)
	answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
	results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
	except Exception as e:
	print(f"Error running agent on task {task_id}: {e}")
	results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": "unknown"})

	time.sleep(1.5)

	if not answers_payload:
	return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)

	submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
	print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
	try:
	response = requests.post(submit_url, json=submission_data, timeout=60)
	response.raise_for_status()
	result_data = response.json()
	final_status = (
	f"Submission Successful!\n"
	f"User: {result_data.get('username')}\n"
	f"Overall Score: {result_data.get('score', 'N/A')}%\n"
	f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
	f"Message: {result_data.get('message', 'No message received.')}"
	)
	print("Submission successful.")
	return final_status, pd.DataFrame(results_log)
	except Exception as e:
	status_message = f"Submission error: {e}"
	print(status_message)
	return status_message, pd.DataFrame(results_log)

	# --- UI Setup ---
	with gr.Blocks() as demo:
	gr.Markdown("# GAIA CodeAgent Evaluation Runner")
	gr.LoginButton()
	run_button = gr.Button("Run Evaluation & Submit All Answers")
	status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
	results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

	run_button.click(
	fn=run_and_submit_all,
	outputs=[status_output, results_table]
	)

	if __name__ == "__main__":
	demo.launch(debug=True, share=False)