|
|
| import os |
| import gradio as gr |
| import requests |
| import pandas as pd |
|
|
| from smolagents import CodeAgent, InferenceClientModel |
| from tools import DuckDuckGoSearchTool, WeatherInfoTool, HubStatsTool |
| from retriever import load_guest_dataset |
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
|
|
| |
| |
| |
|
|
| class BasicAgent: |
|
|
| def __init__(self): |
| token = os.getenv("HF_TOKEN") |
|
|
| if not token: |
| raise ValueError( |
| "HF_TOKEN is missing. Add it in Space Secrets." |
| ) |
|
|
| model = InferenceClientModel( |
| model_id="Qwen/Qwen2.5-Coder-32B-Instruct", |
| token=token |
| ) |
|
|
| self.search_tool = DuckDuckGoSearchTool() |
| self.weather_tool = WeatherInfoTool() |
| self.hub_tool = HubStatsTool() |
| self.guest_tool = load_guest_dataset() |
|
|
| self.agent = CodeAgent( |
| tools=[ |
| self.guest_tool, |
| self.weather_tool, |
| self.hub_tool, |
| self.search_tool |
| ], |
| model=model, |
| max_steps=8, |
| planning_interval=3 |
| ) |
|
|
| def __call__(self, question: str) -> str: |
| try: |
| answer = self.agent.run( |
| question, |
| reset=True |
| ) |
|
|
| if answer is None: |
| return "No answer generated." |
|
|
| return str(answer).strip() |
|
|
| except Exception as e: |
| print(f"Agent error: {e}") |
| return f"Agent execution error: {e}" |
|
|
|
|
| |
| |
| |
|
|
| def run_and_submit_all(profile: gr.OAuthProfile | None): |
|
|
| if not profile: |
| return "Please login to Hugging Face first.", None |
|
|
| username = profile.username |
| space_id = os.getenv("SPACE_ID") |
|
|
| if not space_id: |
| return "SPACE_ID not found. Run this inside your Hugging Face Space.", None |
|
|
| agent_code = ( |
| f"https://huggingface.co/spaces/{space_id}/tree/main" |
| ) |
|
|
| |
| try: |
| agent = BasicAgent() |
| except Exception as e: |
| return f"Agent initialization failed: {e}", None |
|
|
| |
| try: |
| response = requests.get( |
| f"{DEFAULT_API_URL}/questions", |
| timeout=30 |
| ) |
| response.raise_for_status() |
| questions = response.json() |
|
|
| except Exception as e: |
| return f"Could not fetch questions: {e}", None |
|
|
| if not questions: |
| return "No questions received.", None |
|
|
| answers = [] |
| results = [] |
|
|
| |
| for index, item in enumerate(questions, start=1): |
|
|
| task_id = item.get("task_id") |
| question = item.get("question") |
|
|
| if not task_id or not question: |
| continue |
|
|
| print(f"Solving question {index}/{len(questions)}") |
|
|
| try: |
| answer = agent(question) |
|
|
| answers.append({ |
| "task_id": task_id, |
| "submitted_answer": answer |
| }) |
|
|
| results.append({ |
| "Task ID": task_id, |
| "Question": question, |
| "Submitted Answer": answer |
| }) |
|
|
| except Exception as e: |
| print(f"Question error: {e}") |
|
|
| if not answers: |
| return "Agent could not generate any answers.", pd.DataFrame(results) |
|
|
| |
| submission_data = { |
| "username": username, |
| "agent_code": agent_code, |
| "answers": answers |
| } |
|
|
| try: |
| response = requests.post( |
| f"{DEFAULT_API_URL}/submit", |
| json=submission_data, |
| timeout=300 |
| ) |
|
|
| response.raise_for_status() |
| result = response.json() |
|
|
| status = ( |
| "Submission Successful!\n" |
| f"User: {result.get('username')}\n" |
| f"Score: {result.get('score')}%\n" |
| f"Correct: {result.get('correct_count')}/" |
| f"{result.get('total_attempted')}\n" |
| f"Message: {result.get('message')}" |
| ) |
|
|
| return status, pd.DataFrame(results) |
|
|
| except Exception as e: |
| return f"Submission failed: {e}", pd.DataFrame(results) |
|
|
|
|
| |
| |
| |
|
|
| with gr.Blocks() as demo: |
|
|
| gr.Markdown("# Alfred - GAIA AI Agent") |
|
|
| gr.Markdown( |
| "Login to Hugging Face and run your AI agent evaluation." |
| ) |
|
|
| gr.LoginButton() |
|
|
| run_button = gr.Button( |
| "Run Evaluation & Submit All Answers" |
| ) |
|
|
| status_output = gr.Textbox( |
| label="Submission Result", |
| lines=6 |
| ) |
|
|
| results_table = gr.DataFrame( |
| label="Agent Answers", |
| wrap=True |
| ) |
|
|
| run_button.click( |
| fn=run_and_submit_all, |
| outputs=[status_output, results_table] |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| demo.launch() |
|
|
