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") # modele du LLM (changeable via les variables du Space) PAUSE_BETWEEN_QUESTIONS = 8 # secondes, pour eviter les limites de requetes gratuites FILES_DIR = "/tmp/task_files" os.makedirs(FILES_DIR, exist_ok=True) # ---------------------------------------------------------------------------- # Outils personnalises # ---------------------------------------------------------------------------- @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", "") # ---------------------------------------------------------------------------- # Agent # ---------------------------------------------------------------------------- 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 # ---------------------------------------------------------------------------- # Evaluation + soumission # ---------------------------------------------------------------------------- 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) # ---------------------------------------------------------------------------- # Interface Gradio # ---------------------------------------------------------------------------- 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)