"""GAIA benchmark agent for the HF Agents Course final assignment. Brain: Gemini (free-tier API) with model fallback + retry. Tools: web_search, visit_webpage, run_python. Multimodal: images (.png) and audio (.mp3) passed natively to Gemini; .xlsx parsed to text; .py provided as text + runnable. """ import json import os import pathlib import re import subprocess import sys import tempfile import time import urllib.request import requests from bs4 import BeautifulSoup from duckduckgo_search import DDGS from google import genai from google.genai import types API_BASE = "https://agents-course-unit4-scoring.hf.space" MODELS = ["gemini-3.8-flash", "gemini-3.7-flash", "gemini-3.1-flash-lite", "gemini-3.1-flash-lite-preview", "gemini-flash-lite-latest", "gemini-3.6-flash", "gemini-3.5-flash"] MAX_STEPS = 10 WORK_DIR = pathlib.Path(tempfile.mkdtemp(prefix="gaia_")) SYSTEM_PROMPT = """You are a general AI agent solving GAIA benchmark questions. Strategy rules: 1. NEVER rely on memory alone for specifics (counts, dates, names, numbers, statistics). Verify with tools. 2. Do at most 2-3 web_search calls per question. If search results contain promising URLs, IMMEDIATELY use visit_webpage to read the actual page (e.g. Wikipedia articles) and find the exact fact. 3. For Wikipedia facts, visit the exact article page, not just search snippets. 4. For any arithmetic, counting, or data analysis, use run_python to compute - never compute in your head. 5. Think step by step between tool calls. Be efficient: you have limited steps. 6. When you are confident, output the final line EXACTLY as: FINAL: Answer format requirements (CRITICAL - scoring is exact match): - The answer must be ONLY the requested value, no explanations, no units, no currency symbols, no "FINAL ANSWER" text. - Numbers: plain digits/decimals, no thousands separators (e.g. 3717.29). Keep two decimals if the question asks for it. - If the question asks for a name, give only the name in the form requested (e.g. "first name only"). - If a question asks for multiple items, give them in the exact format the question requests (e.g. "A, B"). - If the question asks for a word, give just that word. - If the question asks for an IOC country code, give just the 3-letter code. """ TOOLS = [ {"name": "web_search", "description": "Search the web with DuckDuckGo. Returns titles, URLs and snippets.", "parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "The search query"}}, "required": ["query"]}}, {"name": "visit_webpage", "description": "Fetch a webpage and return its readable text content (truncated).", "parameters": {"type": "object", "properties": {"url": {"type": "string", "description": "The URL to fetch"}}, "required": ["url"]}}, {"name": "run_python", "description": "Execute Python code in a subprocess (has pandas, openpyxl, requests installed). Return the printed output. Use for calculations, file parsing, data analysis.", "parameters": {"type": "object", "properties": {"code": {"type": "string", "description": "Python source code to execute"}}, "required": ["code"]}}, ] # ---------------- tool implementations ---------------- def web_search(query: str) -> str: try: with DDGS() as ddgs: results = list(ddgs.text(query, max_results=8)) if not results: return "No results found." out = [] for r in results: out.append(f"TITLE: {r.get('title','')}\nURL: {r.get('href','')}\nSNIPPET: {r.get('body','')}") return "\n---\n".join(out) except Exception as e: return f"Search error: {e}" def visit_webpage(url: str) -> str: try: headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36"} resp = requests.get(url, headers=headers, timeout=25) resp.raise_for_status() soup = BeautifulSoup(resp.text, "html.parser") for tag in soup(["script", "style", "noscript", "header", "footer", "nav", "iframe", "aside"]): tag.decompose() text = re.sub(r"\s+", " ", soup.get_text(" ", strip=True)) return text[:9000] except Exception as e: return f"Fetch error: {e}" def run_python(code: str) -> str: try: proc = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=40, cwd=str(WORK_DIR)) out = (proc.stdout or "") + (("\nSTDERR:\n" + proc.stderr) if proc.stderr else "") return out[:6000] if out.strip() else "(no output)" except subprocess.TimeoutExpired: return "Timeout: code took too long." except Exception as e: return f"Execution error: {e}" TOOL_FUNCS = {"web_search": web_search, "visit_webpage": visit_webpage, "run_python": run_python} # ---------------- gemini wrapper ---------------- class GeminiBrain: def __init__(self, api_key: str, daily_cap: int = 18): self.client = genai.Client(api_key=api_key, http_options=types.HttpOptions(timeout=180000)) self.daily_cap = daily_cap self.usage = {} # model -> successful request count self.blocked = set() # models blacklisted for this session (daily quota exhausted) def _pick_model(self): for m in MODELS: if m in self.blocked or self.usage.get(m, 0) >= self.daily_cap: continue return m return None def generate(self, contents, tools=None, retries=14): last_err = None for attempt in range(retries): model = self._pick_model() if model is None: raise RuntimeError(f"All models exhausted/blocked. Usage: {self.usage}, blocked: {self.blocked}") try: config = types.GenerateContentConfig(system_instruction=SYSTEM_PROMPT, temperature=0.1) if tools: config.tools = [types.Tool(function_declarations=tools)] resp = self.client.models.generate_content(model=model, contents=contents, config=config) self.usage[model] = self.usage.get(model, 0) + 1 time.sleep(2) return resp, model except Exception as e: last_err = e msg = str(e) if "404" in msg or "no longer available" in msg: self.blocked.add(model) continue if "429" in msg or "RESOURCE_EXHAUSTED" in msg: if "PerDay" in msg: print(f" [429 {model}] daily quota hit, blacklisting", flush=True) self.blocked.add(model) continue m = re.search(r"Please retry in ([\d.]+)s", msg) wait = float(m.group(1)) + 3 if m else 62 print(f" [429 {model}] waiting {wait:.0f}s", flush=True) time.sleep(wait) continue if "503" in msg or "UNAVAILABLE" in msg: print(f" [503 {model}] attempt {attempt}", flush=True) time.sleep(min(2 ** attempt + 1, 30)) continue if "timeout" in msg.lower() or "timed out" in msg.lower(): print(f" [timeout {model}] retrying", flush=True) time.sleep(5) continue print(f" [error {model}] {msg[:120]}", flush=True) time.sleep(min(2 ** attempt + 1, 30)) raise RuntimeError(f"All models failed: {last_err}") # ---------------- file handling ---------------- GAIA_FILE_URL = "https://huggingface.co/datasets/gaia-benchmark/GAIA/resolve/main/2023/validation/{name}" def download_task_file(task_id: str, file_name: str): if not file_name: return None dest = WORK_DIR / file_name token = os.environ.get("HF_TOKEN", "").strip() if not token: token = (pathlib.Path(__file__).parent / ".hf_token").read_text().strip() url = GAIA_FILE_URL.format(name=file_name) try: req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) with urllib.request.urlopen(req, timeout=90) as r, open(dest, "wb") as f: f.write(r.read()) return dest except Exception as e: print(f" [file download error] {e}") return None def build_file_part(path: pathlib.Path): """Return (inline_part, text_context) for a downloaded task file.""" ext = path.suffix.lower() if ext in (".png", ".jpg", ".jpeg"): data = path.read_bytes() return types.Part.from_bytes(data=data, mime_type="image/png" if ext == ".png" else "image/jpeg"), None if ext in (".mp3", ".wav", ".m4a"): mime = {"mp3": "audio/mpeg", "wav": "audio/wav", "m4a": "audio/mp4"}[ext[1:]] return types.Part.from_bytes(data=path.read_bytes(), mime_type=mime), None if ext in (".xlsx", ".xls"): df_text = run_python(f""" import pandas as pd xl = pd.ExcelFile({str(path)!r}) for sheet in xl.sheet_names: df = xl.parse(sheet) print(f"=== SHEET: {{sheet}} ({{len(df)}} rows) ===") print(df.to_string(index=False)) """) return None, f"The task file {path.name} is an Excel file. Its full contents printed via pandas:\n{df_text}" # default: text file (.py, .txt, .csv, .json ...) content = path.read_text(errors="replace")[:30000] return None, f"The task file {path.name} contains:\n```\n{content}\n```" # ---------------- agent ---------------- def answer_question(brain: GeminiBrain, question: dict, log=print) -> str: task_id = question["task_id"] qtext = question["question"] contents = [] file_part, file_text = None, None if question.get("file_name"): p = download_task_file(task_id, question["file_name"]) if p: file_part, file_text = build_file_part(p) user_text = f"Question: {qtext}" if file_text: user_text += f"\n\n{file_text}\nIf useful, you can process the data further with run_python (it runs in the same working directory where this file context came from)." yt = re.search(r'https://(?:www\.)?youtube\.com/watch\?v=[\w-]+', qtext) yt_part = types.Part.from_uri(file_uri=yt.group(0), mime_type="video/mp4") if yt else None if yt_part is not None: user_text += "\n\n[The video is attached - you can watch and listen to it directly.]" text_part = types.Part.from_text(text=user_text) parts_to_send = [p for p in (text_part, file_part, yt_part) if p is not None] contents.append(types.Content(role="user", parts=parts_to_send)) for step in range(MAX_STEPS): resp, model = brain.generate(contents, tools=TOOLS) parts = resp.candidates[0].content.parts or [] func_calls = [p.function_call for p in parts if p.function_call] text_out = "".join(p.text or "" for p in parts if p.text) if func_calls: contents.append(resp.candidates[0].content) for fc in func_calls: args = dict(fc.args or {}) log(f" step {step+1}: {fc.name}({str(args)[:140]})") result = TOOL_FUNCS[fc.name](**args) contents.append(types.Content(role="user", parts=[types.Part.from_function_response(name=fc.name, response={"result": result})])) continue m = re.search(r"FINAL:\s*(.+)", text_out, re.S) if m: return m.group(1).strip() contents.append(types.Content(role="user", parts=[types.Part.from_text(text="Continue. When sure, output exactly 'FINAL: ' with only the concise answer value.")])) # All steps exhausted - force a final answer for demand in [ "You have no steps left. Based on everything gathered, output your best answer NOW as 'FINAL: ' with ONLY the concise answer value.", "You MUST give a concrete answer now. Output 'FINAL: ' where value is a number, name, or short phrase - never empty. Make your best educated guess based on the information gathered.", ]: try: contents.append(types.Content(role="user", parts=[types.Part.from_text(text=demand)])) resp, _ = brain.generate(contents, tools=None) final_text = "".join(p.text or "" for p in (resp.candidates[0].content.parts or []) if p.text) m2 = re.search(r"FINAL:\s*(.+)", final_text, re.S) ans = (m2.group(1) if m2 else final_text).strip() if ans and ans.lower() not in ("none", "n/a", "unknown", "i don't know"): return ans[:200] except Exception as e: return f"No answer produced. ({e})" return "No answer produced."