app.py

import os
import gradio as gr
import requests
import pandas as pd
import tempfile
import json
import subprocess
import sys
from pathlib import Path

from smolagents import (
    CodeAgent,
    HfApiModel,
    DuckDuckGoSearchTool,
    tool,
)

# ============================================================
# CONSTANTS
# ============================================================

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


# ============================================================
# CUSTOM TOOLS
# ============================================================

@tool
def calculate(expression: str) -> str:
    """
    Calculate a mathematical expression.

    Args:
        expression: A Python mathematical expression such as
                    "17 * 23" or "(100 / 4) + 7".

    Returns:
        The calculated result as a string.
    """
    try:
        allowed = {
            "abs": abs,
            "round": round,
            "min": min,
            "max": max,
            "sum": sum,
        }

        result = eval(
            expression,
            {"__builtins__": {}},
            allowed
        )

        return str(result)

    except Exception as e:
        return f"Calculation error: {e}"


@tool
def analyze_csv_or_excel(file_path: str, operation: str) -> str:
    """
    Analyze a CSV or Excel file.

    Args:
        file_path: Path to the CSV or Excel file.
        operation: Describe what should be calculated or inspected.

    Returns:
        A textual summary of the file and requested operation.
    """

    try:
        path = Path(file_path)

        if not path.exists():
            return "File does not exist."

        if path.suffix.lower() in [".xlsx", ".xls"]:
            df = pd.read_excel(path)
        elif path.suffix.lower() == ".csv":
            df = pd.read_csv(path)
        else:
            return "Unsupported file type."

        result = {
            "columns": list(df.columns),
            "rows": len(df),
            "preview": df.head(10).to_dict(orient="records"),
            "numeric_summary": df.describe(
                include="all"
            ).to_dict()
        }

        return json.dumps(result, default=str)

    except Exception as e:
        return f"Spreadsheet analysis error: {e}"


@tool
def read_text_file(file_path: str) -> str:
    """
    Read a text-based file.

    Args:
        file_path: Path to the text file.

    Returns:
        The contents of the file.
    """

    try:
        path = Path(file_path)

        if not path.exists():
            return "File does not exist."

        return path.read_text(
            encoding="utf-8",
            errors="ignore"
        )

    except Exception as e:
        return f"Could not read file: {e}"


@tool
def execute_python_file(file_path: str) -> str:
    """
    Execute a Python file and return its output.

    Args:
        file_path: Path to the Python file.

    Returns:
        Standard output produced by the Python program.
    """

    try:
        path = Path(file_path)

        if not path.exists():
            return "Python file does not exist."

        result = subprocess.run(
            [sys.executable, str(path)],
            capture_output=True,
            text=True,
            timeout=30
        )

        output = result.stdout.strip()

        if result.stderr:
            output += "\nSTDERR:\n" + result.stderr.strip()

        return output

    except subprocess.TimeoutExpired:
        return "Python execution timed out."

    except Exception as e:
        return f"Python execution error: {e}"


# ============================================================
# GAIA AGENT
# ============================================================

class BasicAgent:

    def __init__(self):

        print("=" * 60)
        print("Initializing GAIA Agent...")
        print("=" * 60)

        # ----------------------------------------------------
        # Hugging Face model
        # ----------------------------------------------------

        self.model = HfApiModel(
            model_id="Qwen/Qwen2.5-Coder-32B-Instruct"
        )

        # ----------------------------------------------------
        # Web search
        # ----------------------------------------------------

        self.search_tool = DuckDuckGoSearchTool()

        # ----------------------------------------------------
        # Agent
        # ----------------------------------------------------

        self.agent = CodeAgent(

            model=self.model,

            tools=[
                self.search_tool,
                calculate,
                analyze_csv_or_excel,
                read_text_file,
                execute_python_file,
            ],

            additional_authorized_imports=[
                "requests",
                "bs4",
                "datetime",
                "pandas",
                "math",
                "statistics",
                "json",
                "re",
            ],

            max_steps=15,

            verbosity_level=1,
        )

        print("GAIA Agent initialized successfully.")


    # --------------------------------------------------------
    # Clean final answer
    # --------------------------------------------------------

    def clean_answer(self, answer):

        if answer is None:
            return "unknown"

        answer = str(answer).strip()

        # Remove common unwanted prefixes
        prefixes = [
            "FINAL ANSWER:",
            "Final Answer:",
            "FINAL ANSWER",
            "Final answer:",
            "Answer:",
            "answer:",
        ]

        for prefix in prefixes:

            if answer.startswith(prefix):
                answer = answer[len(prefix):].strip()

        # Remove markdown code fences
        answer = answer.replace("```", "").strip()

        # Remove accidental surrounding quotes
        if (
            len(answer) >= 2
            and answer[0] == '"'
            and answer[-1] == '"'
        ):
            answer = answer[1:-1].strip()

        return answer


    # --------------------------------------------------------
    # Run Agent
    # --------------------------------------------------------

    def __call__(self, question: str) -> str:

        print("\n" + "=" * 70)
        print("QUESTION")
        print(question)
        print("=" * 70)

        prompt = f"""
You are a highly capable general-purpose AI agent solving a
GAIA Level-1 benchmark question.

Your goal is to obtain the CORRECT final answer.

You have access to:

1. Web search
2. Python calculations
3. Spreadsheet analysis
4. Python file execution
5. Text file reading

GENERAL RULES:

- Carefully understand the question.
- Do not guess if you can verify the information.
- Use web search for factual/current/research questions.
- Use Python for calculations, sorting, counting and data processing.
- Use spreadsheet tools for Excel/CSV questions.
- Use the Python execution tool when a Python file is provided.
- Perform multi-step reasoning when necessary.
- Verify important results before answering.

IMPORTANT OUTPUT RULE:

The benchmark uses exact matching.

Therefore your FINAL RESPONSE must contain ONLY the answer.

Do NOT write:

"FINAL ANSWER:"

Do NOT write:

"The answer is..."

Do NOT explain your reasoning.

Do NOT add unnecessary punctuation.

Examples:

If the answer is a number:

391

If the answer is a name:

Smith

If the question asks for a comma-separated list:

apples, bananas, oranges

Question:

{question}
"""

        try:

            result = self.agent.run(prompt)

            answer = self.clean_answer(result)

            print("\nAGENT ANSWER:")
            print(answer)

            return answer

        except Exception as e:

            print("\nAGENT ERROR:")
            print(e)

            return "unknown"


# ============================================================
# DOWNLOAD ATTACHED GAIA FILE
# ============================================================

def download_task_file(task_id):

    file_url = f"{DEFAULT_API_URL}/files/{task_id}"

    try:

        response = requests.get(
            file_url,
            timeout=30
        )

        if response.status_code != 200:
            return None

        # Try to determine extension from headers
        content_type = response.headers.get(
            "content-type",
            ""
        ).lower()

        extension = ".bin"

        if "png" in content_type:
            extension = ".png"

        elif "jpeg" in content_type:
            extension = ".jpg"

        elif "audio" in content_type:
            extension = ".mp3"

        elif "excel" in content_type:
            extension = ".xlsx"

        elif "csv" in content_type:
            extension = ".csv"

        elif "python" in content_type:
            extension = ".py"

        # Temporary file
        temp = tempfile.NamedTemporaryFile(
            delete=False,
            suffix=extension
        )

        temp.write(response.content)
        temp.close()

        print(
            f"Downloaded task file: {temp.name}"
        )

        return temp.name

    except Exception as e:

        print(
            f"Could not download task file: {e}"
        )

        return None


# ============================================================
# MAIN EVALUATION FUNCTION
# ============================================================

def run_and_submit_all(
    profile: gr.OAuthProfile | None
):

    """
    Fetch all GAIA questions,
    run the Agent,
    and submit answers.
    """

    # --------------------------------------------------------
    # Get username
    # --------------------------------------------------------

    space_id = os.getenv("SPACE_ID")

    if profile:

        username = f"{profile.username}"

        print(
            f"User logged in: {username}"
        )

    else:

        print("User not logged in.")

        return (
            "Please Login to Hugging Face with the button.",
            None
        )


    # --------------------------------------------------------
    # API endpoints
    # --------------------------------------------------------

    api_url = DEFAULT_API_URL

    questions_url = f"{api_url}/questions"

    submit_url = f"{api_url}/submit"


    # --------------------------------------------------------
    # Create Agent
    # --------------------------------------------------------

    try:

        agent = BasicAgent()

    except Exception as e:

        print(
            f"Error initializing agent: {e}"
        )

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


    # --------------------------------------------------------
    # Code link
    # --------------------------------------------------------

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

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


    # --------------------------------------------------------
    # Fetch questions
    # --------------------------------------------------------

    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 (
                "Question list is empty.",
                None
            )

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

    except Exception as e:

        print(
            f"Error fetching questions: {e}"
        )

        return (
            f"Error fetching questions: {e}",
            None
        )


    # --------------------------------------------------------
    # Run Agent
    # --------------------------------------------------------

    results_log = []

    answers_payload = []

    print(
        f"\nRunning Agent on "
        f"{len(questions_data)} questions..."
    )


    for index, item in enumerate(
        questions_data,
        start=1
    ):

        task_id = item.get("task_id")

        question_text = item.get(
            "question"
        )

        file_name = item.get(
            "file_name",
            ""
        )


        if not task_id or question_text is None:

            print(
                "Skipping invalid question."
            )

            continue


        print("\n")
        print("#" * 80)
        print(
            f"QUESTION {index}/{len(questions_data)}"
        )
        print(
            f"TASK ID: {task_id}"
        )

        if file_name:

            print(
                f"ATTACHED FILE: {file_name}"
            )


        # ----------------------------------------------------
        # Handle attached file
        # ----------------------------------------------------

        enhanced_question = question_text

        if file_name:

            file_path = download_task_file(
                task_id
            )

            if file_path:

                enhanced_question += f"""

IMPORTANT:

This GAIA question has an attached file.

Original file name:
{file_name}

The file has been downloaded locally to:

{file_path}

You MUST use the appropriate tool to inspect this file
when the question requires information from it.

For an Excel/CSV file:
use analyze_csv_or_excel.

For a Python file:
use execute_python_file.

For a text file:
use read_text_file.

Do not ignore the attached file.
"""

            else:

                enhanced_question += f"""

IMPORTANT:

The question references this attached file:

{file_name}

The file could not be downloaded automatically.
Try to solve the question using the information available
in the question or other available tools.
"""


        # ----------------------------------------------------
        # Run Agent
        # ----------------------------------------------------

        try:

            submitted_answer = agent(
                enhanced_question
            )

            submitted_answer = str(
                submitted_answer
            ).strip()

            answers_payload.append(
                {
                    "task_id": task_id,
                    "submitted_answer": submitted_answer
                }
            )

            results_log.append(
                {
                    "Task ID": task_id,
                    "Question": question_text,
                    "File": file_name,
                    "Submitted Answer": submitted_answer
                }
            )

            print(
                f"FINAL ANSWER: {submitted_answer}"
            )


        except Exception as e:

            print(
                f"Agent failed on task "
                f"{task_id}: {e}"
            )

            results_log.append(
                {
                    "Task ID": task_id,
                    "Question": question_text,
                    "File": file_name,
                    "Submitted Answer":
                        f"AGENT ERROR: {e}"
                }
            )


    # --------------------------------------------------------
    # Check answers
    # --------------------------------------------------------

    if not answers_payload:

        return (
            "Agent did not produce any answers.",
            pd.DataFrame(results_log)
        )


    # --------------------------------------------------------
    # Prepare submission
    # --------------------------------------------------------

    submission_data = {

        "username":
            username.strip(),

        "agent_code":
            agent_code,

        "answers":
            answers_payload
    }


    status_update = (
        f"Agent finished.\n"
        f"Submitting {len(answers_payload)} "
        f"answers for user '{username}'..."
    )

    print(status_update)


    # --------------------------------------------------------
    # Submit
    # --------------------------------------------------------

    try:

        response = requests.post(

            submit_url,

            json=submission_data,

            timeout=180
        )

        response.raise_for_status()

        result_data = response.json()


        final_status = (

            "Submission Successful!\n\n"

            f"User: "
            f"{result_data.get('username')}\n"

            f"Overall Score: "
            f"{result_data.get('score', 'N/A')}%\n"

            f"Correct: "
            f"{result_data.get('correct_count', '?')}/"
            f"{result_data.get('total_attempted', '?')}\n\n"

            f"Message: "
            f"{result_data.get('message', '')}"
        )


        print(final_status)


        results_df = pd.DataFrame(
            results_log
        )


        return (
            final_status,
            results_df
        )


    except requests.exceptions.HTTPError as e:

        error_detail = (
            f"Server responded with "
            f"status {e.response.status_code}."
        )

        try:

            error_json = e.response.json()

            error_detail += (
                f" Detail: "
                f"{error_json.get('detail', '')}"
            )

        except Exception:

            error_detail += (
                f" Response: "
                f"{e.response.text[:500]}"
            )


        status_message = (
            f"Submission Failed: "
            f"{error_detail}"
        )


        return (
            status_message,
            pd.DataFrame(results_log)
        )


    except requests.exceptions.Timeout:

        status_message = (
            "Submission Failed: "
            "The request timed out."
        )

        return (
            status_message,
            pd.DataFrame(results_log)
        )


    except requests.exceptions.RequestException as e:

        status_message = (
            f"Submission Failed: "
            f"Network error - {e}"
        )

        return (
            status_message,
            pd.DataFrame(results_log)
        )


    except Exception as e:

        status_message = (
            f"Unexpected submission error: {e}"
        )

        return (
            status_message,
            pd.DataFrame(results_log)
        )


# ============================================================
# GRADIO INTERFACE
# ============================================================

with gr.Blocks() as demo:

    gr.Markdown(
        "# 🤖 GAIA Agent Evaluation Runner"
    )

    gr.Markdown(
        """
        ## Instructions

        1. Log in to your Hugging Face account.
        2. Your Agent will receive the GAIA evaluation questions.
        3. Your Agent will use tools to solve them.
        4. The answers will automatically be submitted.
        5. Your score will be displayed below.

        **Important:** The benchmark uses exact matching,
        so the Agent should return only the final answer.
        """
    )

    gr.LoginButton()

    run_button = gr.Button(
        "🚀 Run Evaluation & Submit All Answers"
    )

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

if __name__ == "__main__":

    print(
        "\n" +
        "=" * 60 +
        "\n GAIA AGENT STARTING \n" +
        "=" * 60
    )

    space_id = os.getenv(
        "SPACE_ID"
    )

    if space_id:

        print(
            f"Space ID: {space_id}"
        )

        print(
            "Code URL: "
            f"https://huggingface.co/spaces/"
            f"{space_id}/tree/main"
        )

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