Final_Assignment_Template

Paused

Update app.py

7da69d5 verified about 1 month ago

12.5 kB

import os
import re
import traceback

import gradio as gr
import pandas as pd
import requests

from smolagents import (
    CodeAgent,
    DuckDuckGoSearchTool,
    InferenceClientModel,
    VisitWebpageTool,
)


# =========================================================
# CONFIG
# =========================================================

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

# Explicit model instead of relying on a changing default.
MODEL_ID = "Qwen/Qwen3.8-27B"


# =========================================================
# AGENT
# =========================================================

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

        hf_token = os.getenv("HF_TOKEN")

        if not hf_token:
            raise RuntimeError(
                "HF_TOKEN is missing. "
                "Add HF_TOKEN in Space -> Settings -> Variables and secrets."
            )

        print(f"Using model: {MODEL_ID}")

        self.model = InferenceClientModel(
            model_id=MODEL_ID,
            provider="auto",
            token=hf_token,
            timeout=180,
            max_tokens=4096,
        )

        self.agent = CodeAgent(
            model=self.model,
            tools=[
                DuckDuckGoSearchTool(
                    max_results=10,
                    rate_limit=1.0,
                ),
                VisitWebpageTool(
                    max_output_length=30000,
                ),
            ],
            additional_authorized_imports=[
                "requests",
                "re",
                "json",
                "math",
                "datetime",
                "statistics",
                "collections",
                "itertools",
                "urllib.parse",
            ],
            max_steps=15,
        )

        print("GAIA agent initialized successfully.")
        print("=" * 80)

    def clean_answer(self, value) -> str:
        if value is None:
            return ""

        answer = str(value).strip()

        # Remove common prefixes if model ignored instructions.
        answer = re.sub(
            r"^\s*(FINAL\s+ANSWER|Final\s+Answer|Final\s+answer)\s*:\s*",
            "",
            answer,
            flags=re.IGNORECASE,
        )

        # Remove surrounding whitespace/newlines.
        answer = answer.strip()

        return answer

    def __call__(self, question: str) -> str:
        print("\n" + "=" * 80)
        print("QUESTION")
        print("=" * 80)
        print(question)
        print("=" * 80)

        prompt = f"""
You are solving one GAIA benchmark question.
Your goal is to obtain the exact answer, not merely explain how to
find it.
You have web-search and webpage-reading tools. You can also execute
Python code.
Work autonomously.
Rules:
1. Carefully determine exactly what the question asks.
2. If the question requires factual information, search the web.
3. Do not trust only a search-result snippet when the underlying page
   can be inspected. Open relevant pages and verify the information.
4. If Wikipedia is explicitly mentioned, search for the relevant
   Wikipedia page and inspect the required information.
5. For counting, arithmetic, dates, string transformations, reversing
   text, decoding, sorting, comparison, or other deterministic work,
   use Python when useful.
6. Cross-check ambiguous information when possible.
7. Pay extremely close attention to the requested output format.
8. Your final answer must be minimal. Examples:
   7
   Paris
   Mercedes Sosa
   12, 15, 18
   Qh4
9. Do NOT put explanations, citations, reasoning, Markdown, or
   "FINAL ANSWER:" in the final answer.
10. If the question contains a URL, investigate that URL or search for
    information related to it when useful.
QUESTION:
{question}
Solve the problem and return only the exact final answer.
"""

        result = self.agent.run(prompt)

        answer = self.clean_answer(result)

        if not answer:
            raise RuntimeError(
                "The agent completed without returning a final answer."
            )

        print("\n" + "-" * 80)
        print("FINAL ANSWER:")
        print(answer)
        print("-" * 80)

        return answer


# =========================================================
# FETCH QUESTIONS
# =========================================================

def fetch_questions():
    questions_url = f"{DEFAULT_API_URL}/questions"

    print(f"Fetching questions from: {questions_url}")

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

    response.raise_for_status()

    data = response.json()

    if not isinstance(data, list) or not data:
        raise RuntimeError(
            "The scoring API returned an empty or invalid questions list."
        )

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

    return data


# =========================================================
# EVALUATION
# =========================================================

def run_and_submit_all(profile: gr.OAuthProfile | None):
    if not profile:
        return (
            "Please log in to Hugging Face first.",
            pd.DataFrame(),
        )

    username = str(profile.username).strip()

    print("\n" + "#" * 80)
    print(f"USER: {username}")
    print("#" * 80)

    space_id = os.getenv("SPACE_ID")

    if not space_id:
        return (
            "SPACE_ID environment variable is missing.",
            pd.DataFrame(),
        )

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

    # -----------------------------------------------------
    # Initialize agent
    # -----------------------------------------------------

    try:
        agent = BasicAgent()

    except Exception as e:
        traceback.print_exc()

        return (
            f"Error initializing agent:\n{type(e).__name__}: {e}",
            pd.DataFrame(),
        )

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

    try:
        questions_data = fetch_questions()

    except Exception as e:
        traceback.print_exc()

        return (
            f"Error fetching questions:\n{type(e).__name__}: {e}",
            pd.DataFrame(),
        )

    # -----------------------------------------------------
    # Solve
    # -----------------------------------------------------

    results_log = []
    answers_payload = []

    success_count = 0
    error_count = 0

    total = len(questions_data)

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

        if not task_id or question is None:
            print(f"Skipping malformed task: {item}")
            continue

        print("\n" + "#" * 80)
        print(f"TASK {index}/{total}")
        print(f"TASK ID: {task_id}")
        print("#" * 80)

        try:
            answer = agent(question)

            success_count += 1

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

            results_log.append(
                {
                    "Task ID": task_id,
                    "Question": question,
                    "Submitted Answer": answer,
                    "Status": "OK",
                }
            )

        except Exception as e:
            error_count += 1

            traceback.print_exc()

            error_message = (
                f"{type(e).__name__}: {str(e)}"
            )

            print(
                f"\nERROR ON TASK {task_id}: "
                f"{error_message}"
            )

            # IMPORTANT:
            # Do NOT send an empty answer to the scoring API.
            results_log.append(
                {
                    "Task ID": task_id,
                    "Question": question,
                    "Submitted Answer": "",
                    "Status": error_message,
                }
            )

    results_df = pd.DataFrame(results_log)

    # -----------------------------------------------------
    # Prevent another useless 0/20 submission
    # -----------------------------------------------------

    if success_count == 0:
        return (
            "Evaluation stopped.\n\n"
            f"Successful answers: 0/{total}\n"
            f"Errors: {error_count}/{total}\n\n"
            "Nothing was submitted because the model failed to "
            "produce any answers.\n\n"
            "Open Logs and inspect the Status column below.",
            results_df,
        )

    print("\n" + "=" * 80)
    print("AGENT RUN COMPLETE")
    print(f"Successful answers: {success_count}")
    print(f"Errors: {error_count}")
    print("=" * 80)

    # -----------------------------------------------------
    # Submit only successfully generated answers
    # -----------------------------------------------------

    submission_data = {
        "username": username,
        "agent_code": agent_code,
        "answers": answers_payload,
    }

    submit_url = f"{DEFAULT_API_URL}/submit"

    print(
        f"Submitting {len(answers_payload)} answers "
        f"to {submit_url}"
    )

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

        response.raise_for_status()

        result = response.json()

        score = result.get("score", "N/A")
        correct = result.get("correct_count", "?")
        attempted = result.get("total_attempted", "?")
        message = result.get(
            "message",
            "No message received.",
        )

        status = (
            "Submission Successful!\n\n"
            f"User: {result.get('username', username)}\n"
            f"Generated answers: {success_count}/{total}\n"
            f"Agent errors: {error_count}/{total}\n"
            f"Score: {score}%\n"
            f"Correct: {correct}/{attempted}\n\n"
            f"{message}"
        )

        print(status)

        return status, results_df

    except requests.exceptions.HTTPError as e:
        traceback.print_exc()

        try:
            detail = e.response.json()
        except Exception:
            detail = e.response.text[:1000]

        return (
            "Submission failed.\n\n"
            f"HTTP {e.response.status_code}\n"
            f"{detail}",
            results_df,
        )

    except Exception as e:
        traceback.print_exc()

        return (
            "Submission failed.\n\n"
            f"{type(e).__name__}: {e}",
            results_df,
        )


# =========================================================
# UI
# =========================================================

with gr.Blocks() as demo:
    gr.Markdown(
        """
# GAIA Agent Evaluation Runner
### Instructions
1. Log in with your Hugging Face account.
2. Click **Run Evaluation & Submit All Answers**.
3. The agent will fetch the course questions.
4. It will research and solve them.
5. Successful answers will be submitted to the course scoring API.
6. Errors will be displayed in the table instead of silently
   submitting blank answers.
Evaluation may take several minutes.
"""
    )

    gr.LoginButton()

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

    status_output = gr.Textbox(
        label="Run Status / Submission Result",
        lines=10,
        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" + "=" * 80)
    print("STARTING GAIA AGENT SPACE")
    print(f"SPACE_ID: {os.getenv('SPACE_ID', 'NOT FOUND')}")
    print(f"SPACE_HOST: {os.getenv('SPACE_HOST', 'NOT FOUND')}")
    print(f"HF_TOKEN configured: {bool(os.getenv('HF_TOKEN'))}")
    print(f"MODEL: {MODEL_ID}")
    print("=" * 80 + "\n")

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

	import os
	import re
	import traceback

	import gradio as gr
	import pandas as pd
	import requests

	from smolagents import (
	CodeAgent,
	DuckDuckGoSearchTool,
	InferenceClientModel,
	VisitWebpageTool,
	)


	# =========================================================
	# CONFIG
	# =========================================================

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

	# Explicit model instead of relying on a changing default.
	MODEL_ID = "Qwen/Qwen3.8-27B"


	# =========================================================
	# AGENT
	# =========================================================

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

	hf_token = os.getenv("HF_TOKEN")

	if not hf_token:
	raise RuntimeError(
	"HF_TOKEN is missing. "
	"Add HF_TOKEN in Space -> Settings -> Variables and secrets."
	)

	print(f"Using model: {MODEL_ID}")

	self.model = InferenceClientModel(
	model_id=MODEL_ID,
	provider="auto",
	token=hf_token,
	timeout=180,
	max_tokens=4096,
	)

	self.agent = CodeAgent(
	model=self.model,
	tools=[
	DuckDuckGoSearchTool(
	max_results=10,
	rate_limit=1.0,
	),
	VisitWebpageTool(
	max_output_length=30000,
	),
	],
	additional_authorized_imports=[
	"requests",
	"re",
	"json",
	"math",
	"datetime",
	"statistics",
	"collections",
	"itertools",
	"urllib.parse",
	],
	max_steps=15,
	)

	print("GAIA agent initialized successfully.")
	print("=" * 80)

	def clean_answer(self, value) -> str:
	if value is None:
	return ""

	answer = str(value).strip()

	# Remove common prefixes if model ignored instructions.
	answer = re.sub(
	r"^\s(FINAL\s+ANSWER\|Final\s+Answer\|Final\s+answer)\s:\s*",
	"",
	answer,
	flags=re.IGNORECASE,
	)

	# Remove surrounding whitespace/newlines.
	answer = answer.strip()

	return answer

	def __call__(self, question: str) -> str:
	print("\n" + "=" * 80)
	print("QUESTION")
	print("=" * 80)
	print(question)
	print("=" * 80)

	prompt = f"""
	You are solving one GAIA benchmark question.

	Your goal is to obtain the exact answer, not merely explain how to
	find it.

	You have web-search and webpage-reading tools. You can also execute
	Python code.

	Work autonomously.

	Rules:

	1. Carefully determine exactly what the question asks.

	2. If the question requires factual information, search the web.

	3. Do not trust only a search-result snippet when the underlying page
	can be inspected. Open relevant pages and verify the information.

	4. If Wikipedia is explicitly mentioned, search for the relevant
	Wikipedia page and inspect the required information.

	5. For counting, arithmetic, dates, string transformations, reversing
	text, decoding, sorting, comparison, or other deterministic work,
	use Python when useful.

	6. Cross-check ambiguous information when possible.

	7. Pay extremely close attention to the requested output format.

	8. Your final answer must be minimal. Examples:
	7
	Paris
	Mercedes Sosa
	12, 15, 18
	Qh4

	9. Do NOT put explanations, citations, reasoning, Markdown, or
	"FINAL ANSWER:" in the final answer.

	10. If the question contains a URL, investigate that URL or search for
	information related to it when useful.

	QUESTION:

	{question}

	Solve the problem and return only the exact final answer.
	"""

	result = self.agent.run(prompt)

	answer = self.clean_answer(result)

	if not answer:
	raise RuntimeError(
	"The agent completed without returning a final answer."
	)

	print("\n" + "-" * 80)
	print("FINAL ANSWER:")
	print(answer)
	print("-" * 80)

	return answer


	# =========================================================
	# FETCH QUESTIONS
	# =========================================================

	def fetch_questions():
	questions_url = f"{DEFAULT_API_URL}/questions"

	print(f"Fetching questions from: {questions_url}")

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

	response.raise_for_status()

	data = response.json()

	if not isinstance(data, list) or not data:
	raise RuntimeError(
	"The scoring API returned an empty or invalid questions list."
	)

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

	return data


	# =========================================================
	# EVALUATION
	# =========================================================

	def run_and_submit_all(profile: gr.OAuthProfile \| None):
	if not profile:
	return (
	"Please log in to Hugging Face first.",
	pd.DataFrame(),
	)

	username = str(profile.username).strip()

	print("\n" + "#" * 80)
	print(f"USER: {username}")
	print("#" * 80)

	space_id = os.getenv("SPACE_ID")

	if not space_id:
	return (
	"SPACE_ID environment variable is missing.",
	pd.DataFrame(),
	)

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

	# -----------------------------------------------------
	# Initialize agent
	# -----------------------------------------------------

	try:
	agent = BasicAgent()

	except Exception as e:
	traceback.print_exc()

	return (
	f"Error initializing agent:\n{type(e).__name__}: {e}",
	pd.DataFrame(),
	)

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

	try:
	questions_data = fetch_questions()

	except Exception as e:
	traceback.print_exc()

	return (
	f"Error fetching questions:\n{type(e).__name__}: {e}",
	pd.DataFrame(),
	)

	# -----------------------------------------------------
	# Solve
	# -----------------------------------------------------

	results_log = []
	answers_payload = []

	success_count = 0
	error_count = 0

	total = len(questions_data)

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

	if not task_id or question is None:
	print(f"Skipping malformed task: {item}")
	continue

	print("\n" + "#" * 80)
	print(f"TASK {index}/{total}")
	print(f"TASK ID: {task_id}")
	print("#" * 80)

	try:
	answer = agent(question)

	success_count += 1

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

	results_log.append(
	{
	"Task ID": task_id,
	"Question": question,
	"Submitted Answer": answer,
	"Status": "OK",
	}
	)

	except Exception as e:
	error_count += 1

	traceback.print_exc()

	error_message = (
	f"{type(e).__name__}: {str(e)}"
	)

	print(
	f"\nERROR ON TASK {task_id}: "
	f"{error_message}"
	)

	# IMPORTANT:
	# Do NOT send an empty answer to the scoring API.
	results_log.append(
	{
	"Task ID": task_id,
	"Question": question,
	"Submitted Answer": "",
	"Status": error_message,
	}
	)

	results_df = pd.DataFrame(results_log)

	# -----------------------------------------------------
	# Prevent another useless 0/20 submission
	# -----------------------------------------------------

	if success_count == 0:
	return (
	"Evaluation stopped.\n\n"
	f"Successful answers: 0/{total}\n"
	f"Errors: {error_count}/{total}\n\n"
	"Nothing was submitted because the model failed to "
	"produce any answers.\n\n"
	"Open Logs and inspect the Status column below.",
	results_df,
	)

	print("\n" + "=" * 80)
	print("AGENT RUN COMPLETE")
	print(f"Successful answers: {success_count}")
	print(f"Errors: {error_count}")
	print("=" * 80)

	# -----------------------------------------------------
	# Submit only successfully generated answers
	# -----------------------------------------------------

	submission_data = {
	"username": username,
	"agent_code": agent_code,
	"answers": answers_payload,
	}

	submit_url = f"{DEFAULT_API_URL}/submit"

	print(
	f"Submitting {len(answers_payload)} answers "
	f"to {submit_url}"
	)

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

	response.raise_for_status()

	result = response.json()

	score = result.get("score", "N/A")
	correct = result.get("correct_count", "?")
	attempted = result.get("total_attempted", "?")
	message = result.get(
	"message",
	"No message received.",
	)

	status = (
	"Submission Successful!\n\n"
	f"User: {result.get('username', username)}\n"
	f"Generated answers: {success_count}/{total}\n"
	f"Agent errors: {error_count}/{total}\n"
	f"Score: {score}%\n"
	f"Correct: {correct}/{attempted}\n\n"
	f"{message}"
	)

	print(status)

	return status, results_df

	except requests.exceptions.HTTPError as e:
	traceback.print_exc()

	try:
	detail = e.response.json()
	except Exception:
	detail = e.response.text[:1000]

	return (
	"Submission failed.\n\n"
	f"HTTP {e.response.status_code}\n"
	f"{detail}",
	results_df,
	)

	except Exception as e:
	traceback.print_exc()

	return (
	"Submission failed.\n\n"
	f"{type(e).__name__}: {e}",
	results_df,
	)


	# =========================================================
	# UI
	# =========================================================

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

	### Instructions

	1. Log in with your Hugging Face account.
	2. Click Run Evaluation & Submit All Answers.
	3. The agent will fetch the course questions.
	4. It will research and solve them.
	5. Successful answers will be submitted to the course scoring API.
	6. Errors will be displayed in the table instead of silently
	submitting blank answers.

	Evaluation may take several minutes.
	"""
	)

	gr.LoginButton()

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

	status_output = gr.Textbox(
	label="Run Status / Submission Result",
	lines=10,
	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" + "=" * 80)
	print("STARTING GAIA AGENT SPACE")
	print(f"SPACE_ID: {os.getenv('SPACE_ID', 'NOT FOUND')}")
	print(f"SPACE_HOST: {os.getenv('SPACE_HOST', 'NOT FOUND')}")
	print(f"HF_TOKEN configured: {bool(os.getenv('HF_TOKEN'))}")
	print(f"MODEL: {MODEL_ID}")
	print("=" * 80 + "\n")

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