app.py

import logging
import sys
import tempfile
from pathlib import Path

import gradio as gr
import pandas as pd

sys.path.insert(0, str(Path(__file__).parent / "src"))

try:
    import spaces
except ImportError:

    class _LocalSpaces:
        @staticmethod
        def GPU(*_args, **_kwargs):
            return lambda function: function

    spaces = _LocalSpaces()

from gaia_agent.agent import GaiaAgent
from gaia_agent.cli import AGENT_CODE_URL
from gaia_agent.client import ScoringClient
from gaia_agent.models import Answer, Question

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
LOGGER = logging.getLogger(__name__)


@spaces.GPU(duration=1)
def zero_gpu_healthcheck() -> bool:
    """Satisfy the ZeroGPU runtime contract; inference itself is hosted remotely."""
    return True


def solve_question(question_text: str) -> str:
    question_text = question_text.strip()
    if not question_text:
        raise gr.Error("Enter a question first.")
    question = Question(task_id="interactive", question=question_text)
    return GaiaAgent().solve(question).submitted_answer


def run_and_submit_all(
    profile: gr.OAuthProfile | None,
) -> tuple[str, pd.DataFrame]:
    if profile is None:
        return "Sign in with Hugging Face before submitting.", pd.DataFrame()

    rows: list[dict[str, str]] = []
    answers: list[Answer] = []
    agent = GaiaAgent()

    try:
        with tempfile.TemporaryDirectory(prefix="gaia-evaluation-") as directory:
            download_directory = Path(directory)
            with ScoringClient() as client:
                questions = client.questions()
                for index, question in enumerate(questions, start=1):
                    LOGGER.info("Solving question %s/%s", index, len(questions))
                    attachment = client.download_attachment(question, download_directory)
                    record = agent.solve(question, attachment)
                    answer = Answer(
                        task_id=question.task_id,
                        submitted_answer=record.submitted_answer,
                    )
                    answers.append(answer)
                    rows.append(
                        {
                            "Task ID": question.task_id,
                            "Question": question.question,
                            "Answer": answer.submitted_answer,
                        }
                    )
                score = client.submit(profile.username, AGENT_CODE_URL, answers)
    except Exception:
        LOGGER.exception("Evaluation failed")
        return "Evaluation failed. Check the Space logs and retry.", pd.DataFrame(rows)

    status = (
        f"Submission complete: {score.score:.1f}% "
        f"({score.correct_count}/{score.total_attempted}) for {score.username}."
    )
    return status, pd.DataFrame(rows)


with gr.Blocks(title="GAIA Final Agent") as demo:
    gr.Markdown("# GAIA Final Agent")
    gr.Markdown("Tool-using research agent for the Hugging Face Agents Course evaluation.")

    with gr.Tab("Try the agent"):
        question_input = gr.Textbox(label="Question", lines=4)
        solve_button = gr.Button("Solve", variant="primary")
        answer_output = gr.Textbox(label="Exact answer", interactive=False)
        solve_button.click(solve_question, question_input, answer_output)

    with gr.Tab("Course evaluation"):
        gr.LoginButton()
        run_button = gr.Button("Run all 20 questions and submit", variant="primary")
        status_output = gr.Textbox(label="Status", interactive=False)
        results_table = gr.DataFrame(label="Evaluation results", wrap=True)
        run_button.click(
            run_and_submit_all,
            outputs=[status_output, results_table],
            concurrency_limit=1,
        )


if __name__ == "__main__":
    demo.launch()