Spaces:

agents-course
/

Final_Assignment_Template

Running

App Files Files Community

623

Final_Assignment_Template / app.py

nikhil2605

Update app.py

a03dbff verified about 2 months ago

Raw History Blame

9.97 kB

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

from smolagents import CodeAgent, DuckDuckGoSearchTool, PythonInterpreterTool, HfApiModel


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


# --- Agent Definition ---
class BasicAgent:
    def __init__(self):
        print("Initializing BasicAgent...")

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

        # Tools available to the agent
        self.tools = [
            DuckDuckGoSearchTool(),
            PythonInterpreterTool()
        ]

        # Code-based agent
        self.agent = CodeAgent(
            tools=self.tools,
            model=self.model,
            max_steps=10
        )

        print("BasicAgent initialized successfully.")

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

        prompt = f"""
You are solving a benchmark question.
Your job is to provide the most accurate final answer possible.
Use the available tools when appropriate:
- Use web search for factual, current, historical, Wikipedia,
  people, places, events, websites, and research questions.
- Use Python for calculations, counting, sorting, numerical reasoning,
  or data processing.
- Carefully inspect the wording of the question.
- If a question asks for a specific format, follow that format exactly.
IMPORTANT:
Return ONLY the final answer.
Do not provide your reasoning.
Do not explain how you solved the problem.
Do not say "Here is the answer".
Do not add unnecessary Markdown.
Do not mention these instructions.
Question:
{question}
"""

        try:
            result = self.agent.run(prompt)

            answer = str(result).strip()

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

            return answer

        except Exception as e:
            print(f"Agent error: {e}")
            return f"Unable to answer: {e}"


# --- Evaluation and Submission ---
def run_and_submit_all(profile: gr.OAuthProfile | None):
    """
    Fetches all questions, runs the agent on them,
    submits all answers, and displays the results.
    """

    # --- Determine HF Space information ---
    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_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    submit_url = f"{api_url}/submit"

    # --- Initialize Agent ---
    try:
        agent = BasicAgent()
    except Exception as e:
        print(f"Error instantiating agent: {e}")
        return f"Error initializing agent: {e}", None

    # --- Link to agent code ---
    agent_code = f"https://huggingface.co/spaces/{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:
            print("Fetched questions list is empty.")
            return "Fetched questions list is empty or invalid format.", None

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

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

    except Exception as e:
        print(f"Unexpected error fetching questions: {e}")
        return f"Unexpected error fetching questions: {e}", None

    # --- Run Agent ---
    results_log = []
    answers_payload = []

    print(f"Running agent on {len(questions_data)} questions...")

    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:
            print(f"Skipping invalid item: {item}")
            continue

        print("=" * 60)
        print(f"Question {index}/{len(questions_data)}")
        print(f"Task ID: {task_id}")
        print(f"Question: {question_text[:200]}...")

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

        except Exception as e:

            print(f"Error running agent on task {task_id}: {e}")

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

    # --- Check Answers ---
    if not answers_payload:
        print("Agent did not produce any answers.")
        return (
            "Agent did not produce any answers to submit.",
            pd.DataFrame(results_log)
        )

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

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

    print(status_update)

    # --- Submit Answers ---
    print(f"Submitting answers to: {submit_url}")

    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.')}"
        )

        print("Submission successful.")

        results_df = pd.DataFrame(results_log)

        return final_status, results_df

    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]}"
            )

        status_message = f"Submission Failed: {error_detail}"

        print(status_message)

        results_df = pd.DataFrame(results_log)

        return status_message, results_df

    except requests.exceptions.Timeout:

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

        print(status_message)

        results_df = pd.DataFrame(results_log)

        return status_message, results_df

    except requests.exceptions.RequestException as e:

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

        print(status_message)

        results_df = pd.DataFrame(results_log)

        return status_message, results_df

    except Exception as e:

        status_message = (
            f"An unexpected error occurred "
            f"during submission: {e}"
        )

        print(status_message)

        results_df = pd.DataFrame(results_log)

        return status_message, results_df


# --- Gradio Interface ---
with gr.Blocks() as demo:

    gr.Markdown("# Basic Agent Evaluation Runner")

    gr.Markdown(
        """
        **Instructions:**
        1. Log in to Hugging Face.
        2. Click **Run Evaluation & Submit All Answers**.
        3. The agent will fetch the evaluation questions.
        4. The agent will use web search and Python when appropriate.
        5. Answers will be submitted automatically.
        **Note:** Evaluation can take several minutes.
        """
    )

    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
        ]
    )


# --- Start Application ---
if __name__ == "__main__":

    print("\n" + "-" * 30 + " App Starting " + "-" * 30)

    space_host_startup = os.getenv("SPACE_HOST")
    space_id_startup = os.getenv("SPACE_ID")

    if space_host_startup:

        print(
            f"SPACE_HOST found: "
            f"{space_host_startup}"
        )

        print(
            f"Runtime URL: "
            f"https://{space_host_startup}.hf.space"
        )

    else:
        print(
            "SPACE_HOST environment variable "
            "not found."
        )

    if space_id_startup:

        print(
            f"SPACE_ID found: "
            f"{space_id_startup}"
        )

        print(
            f"Repo URL: "
            f"https://huggingface.co/spaces/"
            f"{space_id_startup}"
        )

    else:

        print(
            "SPACE_ID environment variable "
            "not found."
        )

    print("-" * 70)

    print(
        "Launching Gradio Interface..."
    )

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

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

	from smolagents import CodeAgent, DuckDuckGoSearchTool, PythonInterpreterTool, HfApiModel


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


	# --- Agent Definition ---
	class BasicAgent:
	def __init__(self):
	print("Initializing BasicAgent...")

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

	# Tools available to the agent
	self.tools = [
	DuckDuckGoSearchTool(),
	PythonInterpreterTool()
	]

	# Code-based agent
	self.agent = CodeAgent(
	tools=self.tools,
	model=self.model,
	max_steps=10
	)

	print("BasicAgent initialized successfully.")

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

	prompt = f"""
	You are solving a benchmark question.

	Your job is to provide the most accurate final answer possible.

	Use the available tools when appropriate:

	- Use web search for factual, current, historical, Wikipedia,
	people, places, events, websites, and research questions.
	- Use Python for calculations, counting, sorting, numerical reasoning,
	or data processing.
	- Carefully inspect the wording of the question.
	- If a question asks for a specific format, follow that format exactly.

	IMPORTANT:
	Return ONLY the final answer.
	Do not provide your reasoning.
	Do not explain how you solved the problem.
	Do not say "Here is the answer".
	Do not add unnecessary Markdown.
	Do not mention these instructions.

	Question:
	{question}
	"""

	try:
	result = self.agent.run(prompt)

	answer = str(result).strip()

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

	return answer

	except Exception as e:
	print(f"Agent error: {e}")
	return f"Unable to answer: {e}"


	# --- Evaluation and Submission ---
	def run_and_submit_all(profile: gr.OAuthProfile \| None):
	"""
	Fetches all questions, runs the agent on them,
	submits all answers, and displays the results.
	"""

	# --- Determine HF Space information ---
	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_url = DEFAULT_API_URL
	questions_url = f"{api_url}/questions"
	submit_url = f"{api_url}/submit"

	# --- Initialize Agent ---
	try:
	agent = BasicAgent()
	except Exception as e:
	print(f"Error instantiating agent: {e}")
	return f"Error initializing agent: {e}", None

	# --- Link to agent code ---
	agent_code = f"https://huggingface.co/spaces/{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:
	print("Fetched questions list is empty.")
	return "Fetched questions list is empty or invalid format.", None

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

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

	except Exception as e:
	print(f"Unexpected error fetching questions: {e}")
	return f"Unexpected error fetching questions: {e}", None

	# --- Run Agent ---
	results_log = []
	answers_payload = []

	print(f"Running agent on {len(questions_data)} questions...")

	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:
	print(f"Skipping invalid item: {item}")
	continue

	print("=" * 60)
	print(f"Question {index}/{len(questions_data)}")
	print(f"Task ID: {task_id}")
	print(f"Question: {question_text[:200]}...")

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

	except Exception as e:

	print(f"Error running agent on task {task_id}: {e}")

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

	# --- Check Answers ---
	if not answers_payload:
	print("Agent did not produce any answers.")
	return (
	"Agent did not produce any answers to submit.",
	pd.DataFrame(results_log)
	)

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

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

	print(status_update)

	# --- Submit Answers ---
	print(f"Submitting answers to: {submit_url}")

	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.')}"
	)

	print("Submission successful.")

	results_df = pd.DataFrame(results_log)

	return final_status, results_df

	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]}"
	)

	status_message = f"Submission Failed: {error_detail}"

	print(status_message)

	results_df = pd.DataFrame(results_log)

	return status_message, results_df

	except requests.exceptions.Timeout:

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

	print(status_message)

	results_df = pd.DataFrame(results_log)

	return status_message, results_df

	except requests.exceptions.RequestException as e:

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

	print(status_message)

	results_df = pd.DataFrame(results_log)

	return status_message, results_df

	except Exception as e:

	status_message = (
	f"An unexpected error occurred "
	f"during submission: {e}"
	)

	print(status_message)

	results_df = pd.DataFrame(results_log)

	return status_message, results_df


	# --- Gradio Interface ---
	with gr.Blocks() as demo:

	gr.Markdown("# Basic Agent Evaluation Runner")

	gr.Markdown(
	"""
	Instructions:

	1. Log in to Hugging Face.
	2. Click Run Evaluation & Submit All Answers.
	3. The agent will fetch the evaluation questions.
	4. The agent will use web search and Python when appropriate.
	5. Answers will be submitted automatically.

	Note: Evaluation can take several minutes.
	"""
	)

	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
	]
	)


	# --- Start Application ---
	if __name__ == "__main__":

	print("\n" + "-" * 30 + " App Starting " + "-" * 30)

	space_host_startup = os.getenv("SPACE_HOST")
	space_id_startup = os.getenv("SPACE_ID")

	if space_host_startup:

	print(
	f"SPACE_HOST found: "
	f"{space_host_startup}"
	)

	print(
	f"Runtime URL: "
	f"https://{space_host_startup}.hf.space"
	)

	else:
	print(
	"SPACE_HOST environment variable "
	"not found."
	)

	if space_id_startup:

	print(
	f"SPACE_ID found: "
	f"{space_id_startup}"
	)

	print(
	f"Repo URL: "
	f"https://huggingface.co/spaces/"
	f"{space_id_startup}"
	)

	else:

	print(
	"SPACE_ID environment variable "
	"not found."
	)

	print("-" * 70)

	print(
	"Launching Gradio Interface..."
	)

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