| import os |
| import re |
| import subprocess |
| import time |
|
|
| import gradio as gr |
| import pandas as pd |
| import requests |
| from smolagents import ( |
| CodeAgent, |
| DuckDuckGoSearchTool, |
| LiteLLMModel, |
| VisitWebpageTool, |
| tool, |
| ) |
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
| MODEL_ID = os.getenv("MODEL_ID", "gemini/gemini-2.0-flash") |
| PAUSE_BETWEEN_QUESTIONS = 8 |
| FILES_DIR = "/tmp/task_files" |
| os.makedirs(FILES_DIR, exist_ok=True) |
|
|
|
|
| |
| |
| |
| @tool |
| def read_excel_file(path: str) -> str: |
| """Reads an Excel file (.xlsx) and returns all its sheets as text. |
| |
| Args: |
| path: Local path of the Excel file. |
| """ |
| sheets = pd.read_excel(path, sheet_name=None) |
| out = [] |
| for name, df in sheets.items(): |
| out.append(f"--- Sheet: {name} ---\n{df.to_string()}") |
| return "\n\n".join(out) |
|
|
|
|
| @tool |
| def read_text_file(path: str) -> str: |
| """Reads a text or code file (.py, .txt, .csv...) and returns its content. |
| |
| Args: |
| path: Local path of the file. |
| """ |
| with open(path, "r", encoding="utf-8", errors="ignore") as f: |
| return f.read() |
|
|
|
|
| @tool |
| def run_python_file(path: str) -> str: |
| """Runs a Python file and returns what it prints (stdout and stderr). |
| |
| Args: |
| path: Local path of the Python file to run. |
| """ |
| try: |
| result = subprocess.run( |
| ["python", path], capture_output=True, text=True, timeout=60 |
| ) |
| return (result.stdout + result.stderr).strip() |
| except Exception as e: |
| return f"Execution error: {e}" |
|
|
|
|
| @tool |
| def transcribe_audio(path: str) -> str: |
| """Transcribes an audio file (mp3, wav...) to text using Whisper on Groq. |
| |
| Args: |
| path: Local path of the audio file. |
| """ |
| api_key = os.getenv("GROQ_API_KEY") |
| if not api_key: |
| return "Error: GROQ_API_KEY is not set in the Space secrets." |
| with open(path, "rb") as f: |
| response = requests.post( |
| "https://api.groq.com/openai/v1/audio/transcriptions", |
| headers={"Authorization": f"Bearer {api_key}"}, |
| files={"file": (os.path.basename(path), f)}, |
| data={"model": "whisper-large-v3"}, |
| timeout=120, |
| ) |
| if response.status_code != 200: |
| return f"Transcription error {response.status_code}: {response.text[:300]}" |
| return response.json().get("text", "") |
|
|
|
|
| |
| |
| |
| SYSTEM_RULES = """You are answering questions from the GAIA benchmark. Be precise. |
| Rules for your final answer: |
| - Reply with ONLY the final answer. No explanation, no sentence, no 'FINAL ANSWER'. |
| - Numbers: write digits only, without units or thousands separators, unless the question asks otherwise. |
| - Lists: comma-separated, in the order requested (alphabetical or ascending if asked). |
| - Strings: no articles, no abbreviations unless the question asks for them. |
| - Use web search and visit_webpage to verify facts instead of guessing. |
| - If a file path is given, use the provided file tools to read it. |
| """ |
|
|
|
|
| class GaiaAgent: |
| def __init__(self): |
| model = LiteLLMModel( |
| model_id=MODEL_ID, |
| api_key=os.getenv("GEMINI_API_KEY"), |
| ) |
| self.agent = CodeAgent( |
| tools=[ |
| DuckDuckGoSearchTool(), |
| VisitWebpageTool(), |
| read_excel_file, |
| read_text_file, |
| run_python_file, |
| transcribe_audio, |
| ], |
| model=model, |
| max_steps=12, |
| additional_authorized_imports=[ |
| "pandas", "numpy", "re", "json", "math", "datetime", |
| "itertools", "collections", "statistics", "unicodedata", |
| "requests", "bs4", |
| ], |
| verbosity_level=0, |
| ) |
|
|
| def __call__(self, question: str, file_path: str | None = None) -> str: |
| prompt = f"{SYSTEM_RULES}\nQuestion: {question}" |
| if file_path: |
| prompt += f"\n\nAn attached file is available locally at: {file_path}" |
| answer = str(self.agent.run(prompt)).strip() |
| answer = re.sub(r"^\s*final answer\s*:\s*", "", answer, flags=re.IGNORECASE) |
| return answer.strip().strip('"').strip() |
|
|
|
|
| def download_task_file(api_url: str, task_id: str, file_name: str) -> str | None: |
| """Telecharge le fichier associe a une question (si elle en a un).""" |
| if not file_name: |
| return None |
| try: |
| r = requests.get(f"{api_url}/files/{task_id}", timeout=60) |
| r.raise_for_status() |
| local_path = os.path.join(FILES_DIR, file_name) |
| with open(local_path, "wb") as f: |
| f.write(r.content) |
| return local_path |
| except Exception as e: |
| print(f"Could not download file for {task_id}: {e}") |
| return None |
|
|
|
|
| |
| |
| |
| def run_and_submit_all(profile: gr.OAuthProfile | None): |
| space_id = os.getenv("SPACE_ID") |
|
|
| if not profile: |
| return "Please log in to Hugging Face with the button.", None |
| username = profile.username |
|
|
| api_url = DEFAULT_API_URL |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" |
|
|
| try: |
| agent = GaiaAgent() |
| except Exception as e: |
| return f"Error initializing agent: {e}", None |
|
|
| try: |
| response = requests.get(f"{api_url}/questions", timeout=30) |
| response.raise_for_status() |
| questions = response.json() |
| except Exception as e: |
| return f"Error fetching questions: {e}", None |
|
|
| results_log = [] |
| answers_payload = [] |
| for i, item in enumerate(questions): |
| task_id = item.get("task_id") |
| question_text = item.get("question") |
| file_name = item.get("file_name", "") |
| if not task_id or question_text is None: |
| continue |
| file_path = download_task_file(api_url, task_id, file_name) |
| try: |
| answer = agent(question_text, file_path) |
| except Exception as e: |
| print(f"Error on task {task_id}: {e}") |
| answer = "AGENT ERROR" |
| answers_payload.append({"task_id": task_id, "submitted_answer": answer}) |
| results_log.append( |
| { |
| "Task ID": task_id, |
| "Question": question_text[:150], |
| "Submitted Answer": answer, |
| } |
| ) |
| print(f"[{i + 1}/{len(questions)}] {answer}") |
| time.sleep(PAUSE_BETWEEN_QUESTIONS) |
|
|
| if not answers_payload: |
| return "Agent did not produce any answers.", pd.DataFrame(results_log) |
|
|
| submission = { |
| "username": username.strip(), |
| "agent_code": agent_code, |
| "answers": answers_payload, |
| } |
| try: |
| response = requests.post(f"{api_url}/submit", json=submission, timeout=120) |
| response.raise_for_status() |
| data = response.json() |
| status = ( |
| "Submission successful!\n" |
| f"User: {data.get('username')}\n" |
| f"Score: {data.get('score')}% " |
| f"({data.get('correct_count')}/{data.get('total_attempted')} correct)\n" |
| f"Message: {data.get('message')}" |
| ) |
| return status, pd.DataFrame(results_log) |
| except Exception as e: |
| return f"Submission failed: {e}", pd.DataFrame(results_log) |
|
|
|
|
| |
| |
| |
| with gr.Blocks() as demo: |
| gr.Markdown("# Agent Evaluation Runner") |
| gr.Markdown( |
| "1. Log in with your Hugging Face account.\n" |
| "2. Click the button: the agent answers the 20 questions and submits them.\n" |
| "3. This can take 10 to 20 minutes. Keep the page open." |
| ) |
| gr.LoginButton() |
| run_button = gr.Button("Run Evaluation & Submit All Answers") |
| status_output = gr.Textbox(label="Run Status / Submission Result", lines=6, 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) |
