| import os |
| import gradio as gr |
| import requests |
| import pandas as pd |
|
|
| from smolagents import ( |
| CodeAgent, |
| DuckDuckGoSearchTool, |
| InferenceClientModel, |
| tool, |
| LiteLLMModel, |
| ) |
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
|
|
| @tool |
| def download_file_from_task(task_id: str) -> str: |
| """ |
| Downloads a file associated with a GAIA task and returns its content as text. |
| Use this when a question references an attached file. |
| Args: |
| task_id: The task ID string of the GAIA question. |
| """ |
| url = f"{DEFAULT_API_URL}/files/{task_id}" |
| try: |
| response = requests.get(url, timeout=30) |
| response.raise_for_status() |
| try: |
| text = response.content.decode("utf-8") |
| if len(text) > 8000: |
| text = text[:8000] + "\n[... truncated ...]" |
| return text |
| except UnicodeDecodeError: |
| return f"Binary file. Size: {len(response.content)} bytes." |
| except Exception as e: |
| return f"Error downloading file: {e}" |
|
|
| @tool |
| def python_calculator(code: str) -> str: |
| """ |
| Executes a Python expression and returns the result. |
| Use for arithmetic, unit conversions, date calculations. |
| Args: |
| code: A Python expression to evaluate (e.g. '2 ** 10') |
| """ |
| import math, datetime |
| allowed = {"__builtins__": {}, "math": math, "datetime": datetime, |
| "abs": abs, "round": round, "int": int, "float": float, |
| "str": str, "len": len, "sum": sum, "min": min, "max": max, |
| "sorted": sorted, "range": range, "list": list, "dict": dict, |
| "set": set, "zip": zip, "enumerate": enumerate} |
| try: |
| return str(eval(code, allowed)) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
| SYSTEM_PROMPT = """You are an expert research assistant answering GAIA benchmark questions. |
| CRITICAL RULES: |
| 1. Your final answer must be EXACT and CONCISE. No explanations, no sentences. |
| 2. If the answer is a number, return ONLY the number. |
| 3. If the answer is a name, return ONLY the name. |
| 4. If the answer is a list, return items separated by commas. |
| 5. If a question references a file, use download_file_from_task with the task_id. |
| 6. Always search the web for factual questions. |
| 7. Never include FINAL ANSWER in your response. |
| 8. Match the exact format requested in the question. |
| 9. If you cannot find the answer, return your best single-word or single-number guess. Never return a long explanation. |
| """ |
|
|
| def build_agent(): |
| model = LiteLLMModel( |
| model_id="groq/llama-3.1-8b-instant", |
| api_key=os.getenv("HF"), |
| max_tokens=1024, |
| temperature=0.1, |
| ) |
| agent = CodeAgent( |
| model=model, |
| tools=[DuckDuckGoSearchTool(), download_file_from_task, python_calculator], |
| max_steps=8, |
| verbosity_level=1, |
| additional_authorized_imports=["math", "datetime", "re", "json", "csv", "io"], |
| ) |
| agent.prompt_templates["system_prompt"] = SYSTEM_PROMPT + "\n\n" + agent.prompt_templates["system_prompt"] |
| return agent |
|
|
| def run_and_submit_all(profile: gr.OAuthProfile | None): |
| space_id = os.getenv("SPACE_ID") |
| if not profile: |
| return "Please login to Hugging Face first.", None |
| username = profile.username |
| print(f"User logged in: {username}") |
| try: |
| agent = build_agent() |
| except Exception as e: |
| return f"Error initializing agent: {e}", None |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" |
| try: |
| response = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15) |
| response.raise_for_status() |
| questions_data = response.json() |
| except Exception as e: |
| return f"Error fetching questions: {e}", None |
|
|
| results_log = [] |
| answers_payload = [] |
|
|
| 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: |
| continue |
| try: |
| raw = str(agent.run(f"[task_id: {task_id}]\n\n{question_text}")).strip() |
| lines = [l.strip() for l in raw.split('\n') if l.strip()] |
| submitted_answer = lines[0] if lines else raw |
| for prefix in ["The answer is", "Answer:", "Result:", "Final answer:", "FINAL ANSWER:"]: |
| if submitted_answer.lower().startswith(prefix.lower()): |
| submitted_answer = submitted_answer[len(prefix):].strip().strip(":") |
| except Exception as e: |
| submitted_answer = f"AGENT ERROR: {e}" |
| print(f"Error on task {task_id}: {e}") |
|
|
| answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer}) |
| results_log.append({"Task ID": task_id, "Question": question_text[:120], "Submitted Answer": submitted_answer}) |
| print(f"Task {task_id}: {submitted_answer[:80]}") |
|
|
| if not answers_payload: |
| return "Agent produced no answers.", pd.DataFrame(results_log) |
|
|
| try: |
| response = requests.post(f"{DEFAULT_API_URL}/submit", |
| json={"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}, |
| timeout=120) |
| response.raise_for_status() |
| r = response.json() |
| return (f"Submission Successful!\nUser: {r.get('username')}\n" |
| f"Score: {r.get('score')}% ({r.get('correct_count')}/{r.get('total_attempted')} correct)\n" |
| f"Message: {r.get('message')}"), pd.DataFrame(results_log) |
| except Exception as e: |
| return f"Submission Failed: {e}", pd.DataFrame(results_log) |
|
|
| with gr.Blocks() as demo: |
| gr.Markdown("# GAIA Agent - Unit 4 Final Assignment") |
| gr.Markdown(""" |
| 1. Log in with your Hugging Face account. |
| 2. Click Run Evaluation to start the agent on all 20 GAIA questions. |
| 3. Target: >= 30% to earn the certificate. |
| """) |
| gr.LoginButton() |
| run_button = gr.Button("Run Evaluation & Submit All Answers", variant="primary") |
| status_output = gr.Textbox(label="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__": |
| 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: {space_host_startup}") |
| if space_id_startup: |
| print(f"SPACE_ID: {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") |
| print("-"*74 + "\n") |
| print("Launching Gradio Interface...") |
| demo.launch(debug=True, share=False) |
