| import os |
| import gradio as gr |
| import requests |
| import pandas as pd |
|
|
| from smolagents import CodeAgent, DuckDuckGoSearchTool, PythonInterpreterTool, HfApiModel |
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
|
|
|
|
| class BasicAgent: |
| def __init__(self): |
| print("Initializing agent...") |
|
|
| self.model = HfApiModel( |
| model_id="Qwen/Qwen2.5-Coder-32B-Instruct" |
| ) |
|
|
| self.agent = CodeAgent( |
| tools=[ |
| DuckDuckGoSearchTool(), |
| PythonInterpreterTool() |
| ], |
| model=self.model, |
| max_steps=10 |
| ) |
|
|
| print("Agent initialized successfully.") |
|
|
| def __call__(self, question: str) -> str: |
| prompt = f""" |
| Answer the following benchmark question accurately. |
| |
| Use web search whenever the question requires factual research, |
| Wikipedia, current information, or information from the internet. |
| |
| Use Python when calculations, counting, sorting, or data processing |
| are useful. |
| |
| Follow the requested answer format exactly. |
| |
| Return ONLY the final answer. Do not include your reasoning. |
| |
| Question: |
| {question} |
| """ |
|
|
| try: |
| answer = self.agent.run(prompt) |
| return str(answer).strip() |
|
|
| except Exception as e: |
| print(f"Agent error: {e}") |
| return f"Unable to answer: {e}" |
|
|
|
|
| def run_and_submit_all(profile: gr.OAuthProfile | None): |
|
|
| space_id = os.getenv("SPACE_ID") |
|
|
| if profile: |
| username = f"{profile.username}" |
| print(f"User logged in: {username}") |
| else: |
| 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: |
| return f"Error initializing agent: {e}", None |
|
|
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" |
|
|
| print(f"Fetching questions from: {questions_url}") |
|
|
| try: |
| response = requests.get( |
| questions_url, |
| timeout=30 |
| ) |
| response.raise_for_status() |
|
|
| questions_data = response.json() |
|
|
| if not questions_data: |
| return "No questions were returned.", None |
|
|
| print(f"Fetched {len(questions_data)} questions.") |
|
|
| except Exception as e: |
| return f"Error fetching questions: {e}", None |
|
|
| results_log = [] |
| answers_payload = [] |
|
|
| for index, item in enumerate(questions_data, start=1): |
|
|
| task_id = item.get("task_id") |
| question_text = item.get("question") |
|
|
| if not task_id or question_text is None: |
| continue |
|
|
| print("=" * 60) |
| print(f"Running question {index}/{len(questions_data)}") |
|
|
| try: |
| submitted_answer = agent(question_text) |
|
|
| 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 |
| }) |
|
|
| print(f"Answer: {submitted_answer}") |
|
|
| except Exception as e: |
|
|
| print(f"Agent error: {e}") |
|
|
| results_log.append({ |
| "Task ID": task_id, |
| "Question": question_text, |
| "Submitted Answer": f"AGENT ERROR: {e}" |
| }) |
|
|
| if not answers_payload: |
| return ( |
| "Agent did not produce any answers.", |
| pd.DataFrame(results_log) |
| ) |
|
|
| submission_data = { |
| "username": username.strip(), |
| "agent_code": agent_code, |
| "answers": answers_payload |
| } |
|
|
| print( |
| f"Submitting {len(answers_payload)} answers " |
| f"for {username}..." |
| ) |
|
|
| try: |
|
|
| response = requests.post( |
| submit_url, |
| json=submission_data, |
| timeout=120 |
| ) |
|
|
| response.raise_for_status() |
|
|
| result_data = response.json() |
|
|
| final_status = ( |
| f"Submission Successful!\n" |
| f"User: {result_data.get('username')}\n" |
| f"Overall Score: " |
| f"{result_data.get('score', 'N/A')}% " |
| f"({result_data.get('correct_count', '?')}/" |
| f"{result_data.get('total_attempted', '?')} correct)\n" |
| f"Message: " |
| f"{result_data.get('message', 'No message received.')}" |
| ) |
|
|
| return final_status, pd.DataFrame(results_log) |
|
|
| except requests.exceptions.HTTPError as e: |
|
|
| error_detail = ( |
| f"Server responded with status " |
| f"{e.response.status_code}." |
| ) |
|
|
| try: |
| error_json = e.response.json() |
| error_detail += ( |
| f" Detail: " |
| f"{error_json.get('detail', e.response.text)}" |
| ) |
| except Exception: |
| error_detail += ( |
| f" Response: {e.response.text[:500]}" |
| ) |
|
|
| return error_detail, pd.DataFrame(results_log) |
|
|
| except requests.exceptions.Timeout: |
|
|
| return ( |
| "Submission failed: request timed out.", |
| pd.DataFrame(results_log) |
| ) |
|
|
| except requests.exceptions.RequestException as e: |
|
|
| return ( |
| f"Submission failed: network error - {e}", |
| pd.DataFrame(results_log) |
| ) |
|
|
| except Exception as e: |
|
|
| return ( |
| f"Unexpected submission error: {e}", |
| pd.DataFrame(results_log) |
| ) |
|
|
|
|
| with gr.Blocks() as demo: |
|
|
| gr.Markdown("# Basic Agent Evaluation Runner") |
|
|
| gr.Markdown( |
| """ |
| Log in to Hugging Face and run the evaluation. |
| |
| The agent uses a CodeAgent with web search and Python tools. |
| """ |
| ) |
|
|
| 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("Starting agent evaluation Space...") |
|
|
| demo.launch( |
| debug=True, |
| share=False |
| ) |
|
|
