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 ... blocks if emitted by reasoning models
text = re.sub(r".*?", "", 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)