| import os,re |
| import requests |
| import pandas as pd |
| import gradio as gr |
| from PIL import Image |
| import pytesseract |
| from pydub import AudioSegment |
| import yt_dlp |
| from bs4 import BeautifulSoup |
| import whisper |
| import yt_dlp |
|
|
| from openai import OpenAI |
|
|
| |
| |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
| |
|
|
| def download_file(url): |
| local_path = "temp_file" |
| r = requests.get(url) |
| with open(local_path, "wb") as f: |
| f.write(r.content) |
| return local_path |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| def youtube_captions(self, url): |
| try: |
| ydl_opts = {"skip_download": True, "writesubtitles": True, "writeautomaticsub": True} |
| with yt_dlp.YoutubeDL(ydl_opts) as ydl: |
| info = ydl.extract_info(url, download=False) |
| |
| return str(info.get("subtitles") or info.get("automatic_captions"))[:5000] |
| except Exception as e: |
| return f"YouTube error: {e}" |
|
|
| |
| |
| from openai import OpenAI |
|
|
| class BasicAgent: |
| def __init__(self): |
| print("π Super GAIA Agent initialized") |
| self.client = OpenAI() |
| self.audio_model = whisper.load_model("base") |
| |
| for url in ["https://en.wikipedia.org/api/rest_v1/page/summary/Python_(programming_language)", |
| "https://httpbin.org/get"]: |
| try: |
| r = requests.get(url, timeout=5, headers={"User-Agent": "test"}) |
| print(f" [NET TEST] {url[:50]} β {r.status_code}") |
| except Exception as e: |
| print(f" [NET TEST] {url[:50]} β BLOCKED: {e}") |
| def __call__(self, question, file_url=None): |
| return self.agent_loop(question, file_url) |
|
|
| |
| def wiki_search(self, query): |
| try: |
| query = query.strip(' ".,') |
| |
| search_resp = requests.get( |
| "https://en.wikipedia.org/w/api.php", |
| params={ |
| "action": "query", |
| "list": "search", |
| "srsearch": query, |
| "format": "json", |
| "srlimit": 2 |
| }, |
| headers={"User-Agent": "GAIA-Agent/1.0"}, |
| timeout=15 |
| ) |
| print(f" [wiki status] {search_resp.status_code}, len={len(search_resp.text)}") |
| |
| if search_resp.status_code != 200 or not search_resp.text.strip(): |
| |
| slug = query.replace(" ", "_") |
| rest = requests.get( |
| f"https://en.wikipedia.org/api/rest_v1/page/summary/{slug}", |
| headers={"User-Agent": "GAIA-Agent/1.0"}, |
| timeout=15 |
| ) |
| if rest.status_code == 200: |
| data = rest.json() |
| return f"WIKI [{data.get('title')}]: {data.get('extract','')[:2000]}" |
| return f"Wiki unavailable (status {search_resp.status_code})" |
| |
| data = search_resp.json() |
| results = data.get("query", {}).get("search", []) |
| if not results: |
| return f"No Wikipedia results for: {query}" |
| |
| title = results[0]["title"] |
| |
| rest = requests.get( |
| f"https://en.wikipedia.org/api/rest_v1/page/summary/{requests.utils.quote(title)}", |
| headers={"User-Agent": "GAIA-Agent/1.0"}, |
| timeout=15 |
| ) |
| if rest.status_code == 200: |
| d = rest.json() |
| return f"WIKI [{d.get('title')}]: {d.get('extract','')[:2500]}" |
| return f"Wiki fetch failed for: {title}" |
| except Exception as e: |
| return f"Wiki error: {e}" |
|
|
| def web_search(self, query): |
| """General web search using DuckDuckGo""" |
| try: |
| from duckduckgo_search import DDGS |
| with DDGS() as ddgs: |
| results = list(ddgs.text(query, max_results=5)) |
| if not results: |
| return f"No results for: {query}" |
| output = "" |
| for r in results: |
| output += f"\n[{r['title']}] {r['href']}\n{r['body']}\n" |
| return output[:3000] |
| except Exception as e: |
| return f"Search error: {e}" |
|
|
| |
| def scrape_page(self, url, search_terms=None): |
| url = url.strip(' "') |
| if "youtube.com" in url or "youtu.be" in url: |
| return "YouTube cannot be scraped directly." |
| try: |
| headers = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36"} |
| resp = requests.get(url, timeout=15, headers=headers) |
| soup = BeautifulSoup(resp.text, "html.parser") |
| for tag in soup(["script", "style", "nav", "footer", "header"]): |
| tag.decompose() |
| full_text = soup.get_text(separator=" ", strip=True) |
| if len(full_text) < 100: |
| return f"Page returned too little content (status {resp.status_code})" |
| |
| |
| if search_terms: |
| terms = search_terms.lower().split() |
| best_pos = 0 |
| best_score = 0 |
| |
| window = 3000 |
| for pos in range(0, len(full_text) - window, 500): |
| chunk = full_text[pos:pos+window].lower() |
| score = sum(chunk.count(t) for t in terms) |
| if score > best_score: |
| best_score = score |
| best_pos = pos |
| relevant = full_text[max(0, best_pos-200):best_pos+window] |
| return f"PAGE (relevant section): {relevant}" |
| |
| |
| return f"PAGE: {full_text[:8000]}" |
| except Exception as e: |
| return f"Scrape error: {e}" |
|
|
| |
| def read_audio(self, url): |
| try: |
| url = url.strip(' "') |
| r = requests.get(url, timeout=30) |
| r.raise_for_status() |
| |
| import tempfile, os |
| ext = url.split('.')[-1].lower().split('?')[0] |
| with tempfile.NamedTemporaryFile(suffix=f'.{ext}', delete=False) as f: |
| f.write(r.content) |
| tmp_path = f.name |
| print(f" [audio] downloaded {len(r.content)} bytes to {tmp_path}") |
| result = self.audio_model.transcribe(tmp_path) |
| os.unlink(tmp_path) |
| return f"TRANSCRIPT: {result['text']}" |
| except Exception as e: |
| return f"Audio error: {e}" |
|
|
| |
| def read_excel(self, url): |
| try: |
| url = url.strip(' "{}') |
| |
| if url.startswith('{'): |
| import json |
| url = json.loads(url).get('file_url', url) |
| url = url.strip(' "') |
| r = requests.get(url, timeout=20) |
| r.raise_for_status() |
| import tempfile, os |
| with tempfile.NamedTemporaryFile(suffix='.xlsx', delete=False) as f: |
| f.write(r.content) |
| tmp_path = f.name |
| print(f" [excel] downloaded {len(r.content)} bytes") |
| try: |
| df = pd.read_excel(tmp_path, engine='openpyxl') |
| except: |
| df = pd.read_excel(tmp_path, engine='xlrd') |
| os.unlink(tmp_path) |
| return f"EXCEL_DATA:\n{df.to_string()[:5000]}" |
| except Exception as e: |
| return f"Excel error: {e}" |
|
|
| |
| def read_image(self, url): |
| try: |
| url = url.strip(' "') |
| r = requests.get(url, timeout=20) |
| r.raise_for_status() |
| import tempfile, os |
| ext = url.split('.')[-1].lower().split('?')[0] or 'png' |
| with tempfile.NamedTemporaryFile(suffix=f'.{ext}', delete=False) as f: |
| f.write(r.content) |
| tmp_path = f.name |
| print(f" [image] downloaded {len(r.content)} bytes to {tmp_path}") |
| img = Image.open(tmp_path) |
| text = pytesseract.image_to_string(img) |
| os.unlink(tmp_path) |
| result = text.strip() |
| return f"IMAGE_TEXT: {result}" if result else "IMAGE_TEXT: (no text found - this may be a diagram/photo)" |
| except Exception as e: |
| return f"Image error: {e}" |
|
|
| |
| def run_python(self, url): |
| try: |
| url = url.strip(' "') |
| |
| urls_to_try = [ |
| url, |
| url.replace('/files/', '/'), |
| url.replace('https://agents-course-unit4-scoring.hf.space/files/', |
| 'https://agents-course-unit4-scoring.hf.space/'), |
| ] |
| code = None |
| for u in urls_to_try: |
| r = requests.get(u, timeout=15) |
| print(f" [python] trying {u} β {r.status_code}") |
| if r.status_code == 200 and len(r.text) > 10: |
| code = r.text |
| print(f" [python] got code ({len(code)} chars): {code[:150]}") |
| break |
| |
| if not code: |
| return f"Python error: file not found at any URL pattern" |
| |
| import io, contextlib |
| stdout = io.StringIO() |
| with contextlib.redirect_stdout(stdout): |
| exec(code, {"__builtins__": __builtins__}) |
| output = stdout.getvalue().strip() |
| return f"PYTHON_OUTPUT: {output}" if output else f"PYTHON_CODE:\n{code[:500]}" |
| except Exception as e: |
| return f"Python exec error: {e}" |
|
|
| |
| def execute_tool(self, tool, input_data, file_url): |
| |
| target = input_data.strip(' "') |
| if not target.startswith("http") and file_url: |
| target = file_url |
|
|
| if tool == "wiki_search": |
| return self.wiki_search(input_data) |
| elif tool == "scrape_page": |
| |
| return self.scrape_page(target, search_terms=input_data) |
| elif tool == "read_audio": |
| return self.read_audio(target) |
| elif tool == "read_excel": |
| return self.read_excel(target) |
| elif tool == "read_image": |
| return self.read_image(target) |
| elif tool == "run_python": |
| return self.run_python(target) |
| elif tool == "web_search": |
| return self.web_search(input_data) |
| else: |
| return f"Unknown tool: {tool}" |
|
|
| |
| def agent_loop(self, question, file_url): |
| print(f" [DEBUG] file_url received: {repr(file_url)}") |
| pre_context = "" |
| if file_url: |
| ext = file_url.split('.')[-1].lower().split('?')[0] |
| print(f" [Pre-load] detected file ext={ext}, url={file_url}") |
| if ext in ['mp3', 'wav', 'ogg', 'm4a', 'flac']: |
| pre_context = self.read_audio(file_url) |
| elif ext in ['xlsx', 'xls', 'csv']: |
| pre_context = self.read_excel(file_url) |
| elif ext in ['png', 'jpg', 'jpeg', 'gif', 'webp']: |
| pre_context = self.read_image(file_url) |
| elif ext == 'py': |
| try: |
| pre_context = "PYTHON_CODE:\n" + requests.get(file_url, timeout=10).text[:3000] |
| except: |
| pass |
| |
| memory = pre_context |
| seen_tool_calls = set() |
| system_prompt = """You are a precise GAIA benchmark solver. |
| |
| STRICT OUTPUT FORMAT - choose exactly one: |
| TOOL: tool_name |
| INPUT: your_search_query_here |
| |
| OR: |
| FINAL: your_answer |
| |
| NEVER write TOOL: wiki_search(query) - always use INPUT: on the next line. |
| |
| TOOL STRATEGY: |
| - For Wikipedia questions: use scrape_page with the FULL Wikipedia URL directly |
| e.g. TOOL: scrape_page |
| INPUT: https://en.wikipedia.org/wiki/Mercedes_Sosa_discography |
| - For web research: use wiki_search with short 2-4 word queries |
| - For files: use read_audio / read_excel / read_image / run_python with the FILE_URL |
| - Never repeat a failed tool - change approach each step |
| |
| KNOWN URLS (use these exactly when relevant): |
| - LibreTexts 1.E Exercises (equine vet question): |
| https://chem.libretexts.org/Bookshelves/Introductory_Chemistry/Introductory_Chemistry_(CK-12)/01%3A_Introduction_to_Chemistry/1.E%3A_Exercises_(CK-12) |
| - Mercedes Sosa discography: |
| https://en.wikipedia.org/wiki/Mercedes_Sosa_discography |
| - 1928 Summer Olympics: |
| https://en.wikipedia.org/wiki/1928_Summer_Olympics |
| - Malko Competition: |
| https://en.wikipedia.org/wiki/Malko_Competition |
| - 1977 New York Yankees season stats: |
| https://en.wikipedia.org/wiki/1977_New_York_Yankees_season |
| - TaishΕ Tamai (baseball): |
| https://en.wikipedia.org/wiki/Taish%C5%8D_Tamai |
| - Wikipedia Featured articles November 2016: |
| https://en.wikipedia.org/wiki/Wikipedia:Featured_article_candidates/Featured_log/November_2016 |
| - Universe Today Carolyn Collins Petersen June 2023: |
| https://web.archive.org/web/2023/https://www.universetoday.com/161812/ |
| - Polish Everybody Loves Raymond (Εwiat wedΕug Kiepskich): |
| https://en.wikipedia.org/wiki/Wszyscy_kochaj%C4%85_Raymonda_(Polish_TV_series) |
| - Mercedes Sosa discography (use main article not redirect): |
| https://en.wikipedia.org/wiki/Mercedes_Sosa |
| |
| |
| FACTS YOU KNOW (no tools needed): |
| - Reversed text questions: decode then answer directly as FINAL |
| - When asked for "first name only", return ONLY the first word of the name |
| - When asked for "surname only", return ONLY the last word |
| - Basic math/logic: reason step by step then answer as FINAL |
| - Botanical vegetables: only plant parts with NO seeds inside count as vegetables. |
| Exclude: tomato, pepper, corn, zucchini, green beans, peas, cucumber, squash, acorns, peanuts. |
| Include: broccoli, celery, lettuce, sweet potato, carrot.""" |
| |
| for step in range(10): |
| prompt = f"""FILE_URL: {file_url if file_url else 'None'} |
| QUESTION: {question} |
| ACCUMULATED KNOWLEDGE: |
| {memory if memory else '(none yet)'} |
| AVAILABLE TOOLS: wiki_search, scrape_page, read_audio, read_excel, read_image, run_python, web_search |
| What is your next action? Output TOOL+INPUT or FINAL:""" |
| |
| response = self.client.chat.completions.create( |
| model="gpt-4o", |
| temperature=0, |
| messages=[ |
| {"role": "system", "content": system_prompt}, |
| {"role": "user", "content": prompt} |
| ] |
| ) |
| |
| resp = response.choices[0].message.content.strip() |
| print(f" Step {step}: {resp[:120]}") |
| |
| |
| if "FINAL:" in resp: |
| return resp.split("FINAL:")[-1].strip() |
| |
| |
| t_match = re.search(r"TOOL:\s*(\w+)(?:\(([^)]*)\))?", resp, re.I) |
| i_match = re.search(r"INPUT:\s*(.+)", resp, re.I | re.DOTALL) |
| |
| if t_match: |
| tool_name = t_match.group(1).lower().strip() |
| |
| if i_match: |
| raw_input = i_match.group(1).strip() |
| lines = raw_input.split('\n') |
| tool_input = lines[0] |
| if len(lines) > 1 and not lines[1].startswith('TOOL') and len(lines[1]) < 100: |
| tool_input += lines[1].strip() |
| tool_input = tool_input.strip() |
| elif t_match.group(2): |
| tool_input = t_match.group(2).strip() |
| else: |
| tool_input = "" |
| |
| call_key = f"{tool_name}:{tool_input[:80]}" |
| if call_key in seen_tool_calls: |
| memory += f"\n\n[Step {step} - DUPLICATE SKIPPED: {call_key}. You already tried this. Use a DIFFERENT URL or approach.]" |
| print(f" [DUPLICATE SKIPPED] {call_key}") |
| continue |
| seen_tool_calls.add(call_key) |
| |
| result = self.execute_tool(tool_name, tool_input, file_url) |
| print(f" [{tool_name}] β {result[:100]}") |
| print(f" [RESULT LENGTH] {len(result)} chars: {result[:200]}") |
| |
| if len(result) > 30 and not result.lower().startswith("error") and not result.lower().startswith("unknown"): |
| memory += f"\n\n[Step {step} - {tool_name}({tool_input[:80]})]\n{result[:2000]}" |
| print(f" [MEMORY ADDED] memory now {len(memory)} chars") |
| else: |
| memory += f"\n\n[Step {step} - {tool_name} FAILED: {result[:200]}. Try a different approach.]" |
| print(f" [MEMORY FAILED] result was: {result[:100]}") |
| else: |
| memory += f"\n\n[Step {step} - Reasoning]: {resp[:300]}" |
| |
| |
| fallback = self.client.chat.completions.create( |
| model="gpt-4o", |
| temperature=0, |
| messages=[ |
| {"role": "system", "content": system_prompt}, |
| {"role": "user", "content": f"Based on everything gathered, give your best FINAL answer.\nQUESTION: {question}\nKNOWLEDGE:\n{memory}"} |
| ] |
| ) |
| resp = fallback.choices[0].message.content.strip() |
| if "FINAL:" in resp: |
| return resp.split("FINAL:")[-1].strip() |
| return resp |
|
|
| def run_and_submit_all( profile: gr.OAuthProfile | None): |
| """ |
| Fetches all questions, runs the BasicAgent on them, submits all answers, |
| and displays the 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 = BasicAgent() |
| 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(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: |
| print("Fetched questions list is empty.") |
| return "Fetched questions list is empty or invalid format.", None |
| print(f"Fetched {len(questions_data)} questions.") |
| print("SAMPLE ITEM:", questions_data[0]) |
| except requests.exceptions.RequestException as e: |
| print(f"Error fetching questions: {e}") |
| return f"Error fetching questions: {e}", None |
| except requests.exceptions.JSONDecodeError as e: |
| print(f"Error decoding JSON response from questions endpoint: {e}") |
| print(f"Response text: {response.text[:500]}") |
| return f"Error decoding server response for questions: {e}", None |
| except Exception as e: |
| print(f"An unexpected error occurred fetching questions: {e}") |
| return f"An unexpected error occurred fetching questions: {e}", None |
|
|
| |
| results_log = [] |
| answers_payload = [] |
| print(f"Running agent on {len(questions_data)} questions...") |
| for item in questions_data: |
| task_id = item.get("task_id") |
| question_text = item.get("question") |
| if not task_id or question_text is None: |
| print(f"Skipping item with missing task_id or question: {item}") |
| continue |
| try: |
| |
| file_name = item.get("file_name", "") |
| task_id = item.get("task_id", "") |
| if file_name: |
| file_url = f"https://agents-course-unit4-scoring.hf.space/files/{file_name}" |
| try: |
| |
| test = requests.get(file_url, timeout=5, stream=True) |
| test.close() |
| print(f" [FILE] name='{file_name}', url={file_url}, status={test.status_code}") |
| except Exception as e: |
| print(f" [FILE] verification error: {e}") |
| else: |
| file_url = None |
| print(f" [FILE] name={file_name!r}, url={file_url}") |
| submitted_answer = agent(question_text, file_url) |
| print("------------------------------------------------") |
| print("QUESTION:", question_text) |
| print("ANSWER:", submitted_answer) |
| print("------------------------------------------------") |
| 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": f"AGENT ERROR: {e}"}) |
|
|
| if not answers_payload: |
| print("Agent did not produce any answers to submit.") |
| 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} |
| status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..." |
| print(status_update) |
|
|
| |
| 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')}% " |
| 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.") |
| results_df = pd.DataFrame(results_log) |
| return final_status, results_df |
| except requests.exceptions.HTTPError as e: |
| error_detail = f"Server responded with status {e.response.status_code}." |
| try: |
| error_json = e.response.json() |
| error_detail += f" Detail: {error_json.get('detail', e.response.text)}" |
| except requests.exceptions.JSONDecodeError: |
| error_detail += f" Response: {e.response.text[:500]}" |
| status_message = f"Submission Failed: {error_detail}" |
| print(status_message) |
| results_df = pd.DataFrame(results_log) |
| return status_message, results_df |
| except requests.exceptions.Timeout: |
| status_message = "Submission Failed: The request timed out." |
| print(status_message) |
| results_df = pd.DataFrame(results_log) |
| return status_message, results_df |
| except requests.exceptions.RequestException as e: |
| status_message = f"Submission Failed: Network error - {e}" |
| print(status_message) |
| results_df = pd.DataFrame(results_log) |
| return status_message, results_df |
| except Exception as e: |
| status_message = f"An unexpected error occurred during submission: {e}" |
| print(status_message) |
| results_df = pd.DataFrame(results_log) |
| return status_message, results_df |
|
|
|
|
| |
| with gr.Blocks() as demo: |
| gr.Markdown("# Basic Agent Evaluation Runner") |
| gr.Markdown( |
| """ |
| **Instructions:** |
| |
| 1. Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ... |
| 2. Log in to your Hugging Face account using the button below. This uses your HF username for submission. |
| 3. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score. |
| |
| --- |
| **Disclaimers:** |
| Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions). |
| This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async. |
| """ |
| ) |
|
|
| 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__": |
| print("\n" + "-"*30 + " App Starting " + "-"*30) |
| |
| space_host_startup = os.getenv("SPACE_HOST") |
| space_id_startup = os.getenv("SPACE_ID") |
|
|
| if space_host_startup: |
| print(f"β
SPACE_HOST found: {space_host_startup}") |
| print(f" Runtime URL should be: https://{space_host_startup}.hf.space") |
| else: |
| print("βΉοΈ SPACE_HOST environment variable not found (running locally?).") |
|
|
| if space_id_startup: |
| print(f"β
SPACE_ID found: {space_id_startup}") |
| print(f" Repo URL: https://huggingface.co/spaces/{space_id_startup}") |
| print(f" Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main") |
| else: |
| print("βΉοΈ SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.") |
|
|
| print("-"*(60 + len(" App Starting ")) + "\n") |
|
|
| print("Launching Gradio Interface for Basic Agent Evaluation...") |
| demo.launch(debug=True, share=False) |