app.py

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
    )