app.py

import os
import re
import gradio as gr
import requests
import pandas as pd

from smolagents import CodeAgent, InferenceClientModel, WebSearchTool


# ---------------------------------------------------------
# Configuration
# ---------------------------------------------------------

DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

MODEL_ID = "Qwen/Qwen2.5-Coder-32B-Instruct"


# ---------------------------------------------------------
# GAIA Agent
# ---------------------------------------------------------

class BasicAgent:
    def __init__(self):
        print("Initializing GAIA agent...")

        hf_token = os.getenv("HF_TOKEN")

        if not hf_token:
            raise ValueError(
                "HF_TOKEN is missing. Add it in "
                "Settings → Variables and secrets."
            )

        self.model = InferenceClientModel(
            model_id=MODEL_ID,
            token=hf_token,
        )

        self.agent = CodeAgent(
            tools=[
                WebSearchTool(),
            ],
            model=self.model,
            max_steps=12,
            additional_authorized_imports=[
                "math",
                "statistics",
                "datetime",
                "re",
                "json",
            ],
            instructions="""
You are an AI agent solving Level 1 GAIA benchmark questions.

Carefully solve each question using web search and Python when needed.

Important rules:

1. Search the web for factual or obscure information.
2. Verify important facts before answering.
3. Use Python for calculations when useful.
4. Follow the answer format requested in the question exactly.
5. Return only the final requested answer.
6. Do not include explanations, reasoning, citations, or introductions.
7. Do not write "FINAL ANSWER".
8. Do not write "The answer is".
9. Preserve requested capitalization, ordering, punctuation, units,
   separators, singular/plural forms, and date formats.
""",
        )

        print("GAIA agent initialized successfully.")

    @staticmethod
    def clean_answer(answer) -> str:
        """
        Remove common prefixes that can cause exact-match failure.
        """

        text = str(answer).strip()

        unwanted_prefixes = [
            r"^final answer\s*:\s*",
            r"^answer\s*:\s*",
            r"^the answer is\s*",
        ]

        for pattern in unwanted_prefixes:
            text = re.sub(
                pattern,
                "",
                text,
                flags=re.IGNORECASE,
            ).strip()

        # Remove accidental surrounding quotation marks.
        if (
            len(text) >= 2
            and text[0] == text[-1]
            and text[0] in {"'", '"'}
        ):
            text = text[1:-1].strip()

        return text

    def __call__(self, question: str) -> str:
        print(f"Question received: {question[:100]}...")

        prompt = f"""
Solve this GAIA benchmark question carefully.

Question:
{question}

Use web search and Python tools when necessary.

Return only the exact answer requested by the question.
Do not include an explanation.
Do not include citations.
Do not write FINAL ANSWER.
Do not write "The answer is".
"""

        result = self.agent.run(prompt)

        cleaned_answer = self.clean_answer(result)

        print(f"Agent answer: {cleaned_answer}")

        return cleaned_answer


# ---------------------------------------------------------
# Evaluation and submission
# ---------------------------------------------------------

def run_and_submit_all(profile: gr.OAuthProfile | None):
    """
    Fetch all GAIA questions, run the agent, submit the answers,
    and display the score.
    """

    space_id = os.getenv("SPACE_ID")

    if profile:
        username = profile.username
        print(f"Logged-in user: {username}")
    else:
        return (
            "Please log in to Hugging Face using the login button.",
            None,
        )

    if not space_id:
        return (
            "SPACE_ID was not found. Make sure this app is running "
            "inside a Hugging Face Space.",
            None,
        )

    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    submit_url = f"{api_url}/submit"

    # Initialize agent.
    try:
        agent = BasicAgent()
    except Exception as error:
        print(f"Agent initialization error: {error}")

        return (
            f"Error initializing agent: {error}",
            None,
        )

    agent_code = (
        f"https://huggingface.co/spaces/"
        f"{space_id}/tree/main"
    )

    print(f"Agent code URL: {agent_code}")

    # Fetch questions.
    try:
        response = requests.get(
            questions_url,
            timeout=30,
        )

        response.raise_for_status()
        questions_data = response.json()

        if not questions_data:
            return (
                "The questions list is empty.",
                None,
            )

        print(f"Fetched {len(questions_data)} questions.")

    except requests.exceptions.RequestException as error:
        return (
            f"Error fetching questions: {error}",
            None,
        )

    except ValueError as error:
        return (
            f"Invalid response from questions API: {error}",
            None,
        )

    # Run the agent.
    results_log = []
    answers_payload = []

    for question_number, item in enumerate(
        questions_data,
        start=1,
    ):
        task_id = item.get("task_id")
        question_text = item.get("question")

        if not task_id or not question_text:
            print(f"Skipping invalid question item: {item}")
            continue

        print(
            f"Processing question "
            f"{question_number}/{len(questions_data)}"
        )

        try:
            submitted_answer = agent(question_text)

        except Exception as error:
            print(
                f"Error on task {task_id}: {error}"
            )

            submitted_answer = ""

        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,
            }
        )

    results_df = pd.DataFrame(results_log)

    if not answers_payload:
        return (
            "The agent did not produce any answers.",
            results_df,
        )

    # Prepare submission.
    submission_data = {
        "username": username.strip(),
        "agent_code": agent_code,
        "answers": answers_payload,
    }

    print(
        f"Submitting {len(answers_payload)} answers "
        f"for {username}."
    )

    # Submit answers.
    try:
        response = requests.post(
            submit_url,
            json=submission_data,
            timeout=120,
        )

        response.raise_for_status()
        result_data = response.json()

        final_status = (
            "Submission Successful!\n\n"
            f"User: {result_data.get('username', username)}\n"
            f"Overall Score: "
            f"{result_data.get('score', 'N/A')}%\n"
            f"Correct Answers: "
            f"{result_data.get('correct_count', '?')}/"
            f"{result_data.get('total_attempted', '?')}\n"
            f"Message: "
            f"{result_data.get('message', 'No message received.')}"
        )

        return final_status, results_df

    except requests.exceptions.HTTPError as error:
        error_detail = (
            f"Server returned status "
            f"{error.response.status_code}."
        )

        try:
            error_json = error.response.json()
            error_detail += (
                f"\nDetails: "
                f"{error_json.get('detail', error.response.text)}"
            )

        except ValueError:
            error_detail += (
                f"\nResponse: "
                f"{error.response.text[:500]}"
            )

        return (
            f"Submission failed.\n{error_detail}",
            results_df,
        )

    except requests.exceptions.Timeout:
        return (
            "Submission failed because the request timed out.",
            results_df,
        )

    except requests.exceptions.RequestException as error:
        return (
            f"Submission failed because of a network error: {error}",
            results_df,
        )

    except Exception as error:
        return (
            f"Unexpected submission error: {error}",
            results_df,
        )


# ---------------------------------------------------------
# Gradio interface
# ---------------------------------------------------------

with gr.Blocks() as demo:
    gr.Markdown("# GAIA Agent Evaluation Runner")

    gr.Markdown(
        """
        ### Instructions

        1. Log in using your Hugging Face account.
        2. Click **Run Evaluation & Submit All Answers**.
        3. The agent will solve all 20 GAIA questions.
        4. Your answers will be submitted automatically.

        The target score for the course certificate is **30% or higher**.
        """
    )

    gr.LoginButton()

    run_button = gr.Button(
        "Run Evaluation & Submit All Answers",
        variant="primary",
    )

    status_output = gr.Textbox(
        label="Run Status / Submission Result",
        lines=8,
        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,
        ],
    )


# ---------------------------------------------------------
# Start application
# ---------------------------------------------------------

if __name__ == "__main__":
    print("Starting GAIA Agent Evaluation Runner...")

    demo.launch(
        debug=True,
        share=False,
    )