Spaces:
Running
Running
File size: 4,044 Bytes
969f7c5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Unit 4: fetch the 20 GAIA questions, answer them with the agent, submit for the certificate.
Local: py -3.12 app.py -> answer all questions (cached in answers.json)
py -3.12 app.py --only 3 9 -> (re)answer only those question numbers
py -3.12 app.py --submit -> submit the cached answers and print the score
Space: same code with a Gradio login + button (runs when SPACE_ID is set).
"""
import json
import os
import sys
from pathlib import Path
import requests
API = "https://agents-course-unit4-scoring.hf.space"
USERNAME = "brunopaschoal"
AGENT_CODE = "https://huggingface.co/spaces/brunopaschoal/Final_Assignment/tree/main"
HERE = Path(__file__).parent
CACHE = HERE / "answers.json"
FILES = HERE / "files"
TEXT_FILES = (".py", ".txt", ".csv", ".md", ".json")
def get_questions() -> list[dict]:
r = requests.get(f"{API}/questions", timeout=30)
r.raise_for_status()
return r.json()
def build_prompt(q: dict) -> str:
prompt = q["question"]
if not q.get("file_name"):
return prompt
FILES.mkdir(exist_ok=True)
path = FILES / q["file_name"]
if not path.exists():
r = requests.get(f"{API}/files/{q['task_id']}", timeout=60)
if r.ok:
path.write_bytes(r.content)
else: # course server often 404s; same file is in the (gated) GAIA validation set
from huggingface_hub import hf_hub_download
hf_hub_download("gaia-benchmark/GAIA", f"2023/validation/{q['file_name']}", repo_type="dataset", local_dir=HERE / "gaia")
path.write_bytes((HERE / "gaia/2023/validation" / q["file_name"]).read_bytes())
prompt += f"\n\nAttached file: {path.resolve().as_posix()}"
if path.suffix in TEXT_FILES: # the agent's sandbox can't open() files, so inline text files
prompt += f"\nFile contents:\n```\n{path.read_text(encoding='utf-8', errors='replace')}\n```"
return prompt
def answer_all(only: set[int] | None = None) -> dict:
from agent import build_agent, clean_answer
cache = json.loads(CACHE.read_text(encoding="utf-8")) if CACHE.exists() else {}
for i, q in enumerate(get_questions(), 1):
if (only and i not in only) or (not only and q["task_id"] in cache):
continue
print(f"\n===== Question {i}/20 =====\n{q['question'][:200]}")
try:
cache[q["task_id"]] = clean_answer(build_agent().run(build_prompt(q)))
except Exception as e: # one failure (e.g. quota) must not lose the other answers
print(f"!! Question {i} failed: {e}")
continue
CACHE.write_text(json.dumps(cache, indent=2, ensure_ascii=False), encoding="utf-8")
print(f">> Answer {i}: {cache[q['task_id']]}")
return cache
def submit(cache: dict, username: str = USERNAME) -> str:
answers = [{"task_id": t, "submitted_answer": a} for t, a in cache.items()]
r = requests.post(f"{API}/submit", json={"username": username, "agent_code": AGENT_CODE, "answers": answers}, timeout=120)
r.raise_for_status()
d = r.json()
return f"Score: {d.get('score')}% ({d.get('correct_count')}/{d.get('total_attempted')} correct). {d.get('message', '')}"
def launch_space():
import gradio as gr
def run(profile: gr.OAuthProfile | None):
if not profile:
return "Please log in with Hugging Face first."
return submit(answer_all(), profile.username)
with gr.Blocks() as demo:
gr.Markdown("# GAIA agent (HF Agents course, Unit 4)\nsmolagents CodeAgent + Gemini. Log in, then run.")
gr.LoginButton()
out = gr.Textbox(label="Result")
gr.Button("Run evaluation & submit").click(run, outputs=out)
demo.launch()
if __name__ == "__main__":
if os.environ.get("SPACE_ID"):
launch_space()
elif "--submit" in sys.argv:
print(submit(json.loads(CACHE.read_text(encoding="utf-8"))))
else:
only = {int(a) for a in sys.argv[sys.argv.index("--only") + 1:]} if "--only" in sys.argv else None
answer_all(only)
|