Agents_Course_Final_Assignment

Sleeping

Added GoogleSearchTool

98f88a8 verified over 1 year ago

13 kB

import os
import gradio as gr
import pandas as pd
# Transformers and torch might not be directly needed by BasicAgent if smolgent handles its own LLM
# from transformers import AutoModelForCausalLM, AutoTokenizer
# import torch
import asyncio
import aiohttp

from smolagents import CodeAgent, LiteLLMModel, GoogleSearchTool

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

# --- Basic Agent Definition ---
MODEL_ID = "gemini/gemini-2.5-pro-preview-06-05" 
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")

SYSTEM_PROMPT = """
You are a highly capable and precise code generation and information retrieval agent. Your primary goal is to answer questions from various datasets.
Instructions:
Analyze the problem: Understand the user's query, specifically focusing on the factual questions.
Utilize Web Search: Before attempting to answer, use web search (Google Search or Browsing tool) to find the most accurate and up-to-date information relevant to the question. Formulate precise search queries. Prioritize authoritative sources.
Generate Code (if necessary): If the question requires computation, data manipulation, or logical reasoning that can be solved by code, generate the necessary code. The code should be correct, efficient, and directly solve the problem posed by the question. Use appropriate programming languages for the task. Ensure the code produces the final answer.
If you call any API and it returns rate limit then attempt again but add 30 seconds delay.
Precision is paramount: Your final answer must be as precise and accurate as possible.
No Explanations or Commentary: Crucially, provide ONLY the final answer to the question. Do NOT include any introductory phrases, concluding remarks, explanations of your thought process, code comments (unless the code itself is the answer), or any other conversational text. Your output should be the direct, concise answer. If the answer is code, provide only the code. If the answer is a numerical value or a string, provide only that value or string.
Your output format for questions will always be:
[THE PRECISE ANSWER (e.g., a number, a string, or the complete code if the code itself is the answer)]
"""

class BasicAgent:
    def __init__(self):
        print("BasicAgent initialized to use smolagents.")

        # --- smolagents Initialization ---
        try:    
            search_tool = GoogleSearchTool()
            
            llm_model = LiteLLMModel(
                model_id=MODEL_ID,
                api_key=GOOGLE_API_KEY,
            )
            self.agent_executor = CodeAgent(
                model=llm_model,
                tools=[search_tool],
                max_steps=12,
                planning_interval=4,
            )
            print("smolagents AgentExecutor initialized successfully with GoogleSearchTool.")

        except ImportError:
            print("ERROR: smolagents library not found. Please ensure 'smolagents' is installed and import paths are correct.")
            raise
        except Exception as e:
            print(f"Error initializing smolagents: {e}")
            raise


    def __call__(self, question: str) -> str:
        print(f"\nAgent (smolagents) received question (first 100 chars): {question[:100]}...")
        try:
            response = self.agent_executor.run(question)

            print(f"Agent (smolagents) returning answer (first 50 chars): {str(response)[:50]}...")
            return str(response)
        except Exception as e:
            print(f"Error during smolagents agent's __call__: {e}")
            return f"Error: Could not process the question with smolagents. {e}"

async def run_and_submit_all(profile: gr.OAuthProfile | None):
    """
    Fetches all questions, runs the BasicAgent on them, submits all answers,
    and displays the results.
    """
    space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code

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

    if profile:
        username= f"{profile.username}"
        print(f"User logged in: {username}")

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

    questions_data = []
    async with aiohttp.ClientSession() as session:
        # 2. Fetch Questions
        print(f"Fetching questions from: {questions_url}")
        try:
            async with session.get(questions_url, timeout=aiohttp.ClientTimeout(total=20)) as response:
                response.raise_for_status()
                questions_data = await 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 aiohttp.ClientResponseError as e:
            err_msg = f"Error fetching questions (HTTP {e.status}): {e.message}"
            print(err_msg)
            return err_msg, None
        except asyncio.TimeoutError:
            err_msg = "Timeout error fetching questions."
            print(err_msg)
            return err_msg, None
        except aiohttp.ClientError as e: # Includes connection errors, etc.
            err_msg = f"Error fetching questions (ClientError): {e}"
            print(err_msg)
            return err_msg, None
        except Exception as e: # Catch-all for others like JSONDecodeError
            err_msg = f"An unexpected error occurred fetching questions: {e}"
            print(err_msg)
            if 'response' in locals() and hasattr(response, 'text'):
                try:
                    response_text_snippet = (await response.text())[:500]
                    print(f"Response text snippet: {response_text_snippet}")
                except: pass
            return err_msg, None

    # 3. Run your Agent Concurrently
    results_log = []
    answers_payload = []
    print(f"Running agent on {len(questions_data)} questions...")
    
    # Note: If BasicAgent initialization is very lightweight, this is fine.
    # If it's heavy (e.g., downloads large models every time),
    # then ensuring BasicAgent and its components are stateless/reentrant (Option 2)
    # would be more efficient, but also more complex to verify.
    async def process_question(item):
        # Instantiate a new agent for each question to ensure isolation
        agent_instance = BasicAgent() 
        task_id = item.get("task_id")
        question_text = item.get("question")
        if not task_id or question_text is None:
            print(f"Skipping item with missing task_id or question: {item}")
            return None, None
        try:
            loop = asyncio.get_event_loop()
            # Run the synchronous agent call in a thread pool executor
            submitted_answer = await loop.run_in_executor(None, agent_instance, question_text)
            return {"task_id": task_id, "submitted_answer": submitted_answer}, \
                   {"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}")
            return None, {"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"}

    tasks = [process_question(item) for item in questions_data]
    processed_results = await asyncio.gather(*tasks)

    for payload, log_entry in processed_results:
        if payload: answers_payload.append(payload)
        if log_entry: results_log.append(log_entry)

    if not answers_payload:
        print("Agent did not produce any answers to submit.")
        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
    
    # 4. Prepare Submission
    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
    print(status_update)

    async with aiohttp.ClientSession() as session:
        # 5. Submit
        print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
        try:
            async with session.post(submit_url, json=submission_data, timeout=aiohttp.ClientTimeout(total=60)) as response:
                response.raise_for_status()
                result_data = await response.json()
                final_status = (
                    f"Submission Successful!\n"
                    f"User: {result_data.get('username')}\n"
                    f"Overall Score: {result_data.get('score', 'N/A')}% "
                    f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
                    f"Message: {result_data.get('message', 'No message received.')}"
                )
                print("Submission successful.", final_status)
                return final_status, pd.DataFrame(results_log)
        except aiohttp.ClientResponseError as e:
            error_detail = f"Server responded with status {e.status}."
            try:
                # Try to get JSON error detail, then text
                error_body = await e.json()
                error_detail += f" Detail: {error_body.get('detail', str(error_body))}"
            except Exception: # If not JSON or other error
                try:
                    error_text = (await response.text())[:500]
                    error_detail += f" Response: {error_text}"
                except:
                     error_detail += " Could not retrieve error response body."
            status_message = f"Submission Failed: {error_detail}"
        except asyncio.TimeoutError:
            status_message = "Submission Failed: The request timed out."
        except aiohttp.ClientError as e:
            status_message = f"Submission Failed: Network error - {e}"
        except Exception as e:
            status_message = f"An unexpected error occurred during submission: {e}"

        print(status_message)
        return status_message, pd.DataFrame(results_log)

if __name__ == "__main__":
    print("\n" + "-"*30 + " App Starting " + "-"*30)
    # Check for SPACE_HOST and SPACE_ID at startup for information
    space_host_startup = os.getenv("SPACE_HOST")
    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup

    if space_host_startup:
        print(f"✅ SPACE_HOST found: {space_host_startup}")
        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
    else:
        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
        loop = asyncio.get_event_loop()
        loop.run_until_complete(run_and_submit_all(None))
        exit(0)

    if space_id_startup: # Print repo URLs if SPACE_ID is found
        print(f"✅ SPACE_ID found: {space_id_startup}")
        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
    else:
        print("ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")
        exit(0)

    print("-"*(60 + len(" App Starting ")) + "\n")
    print("Launching Gradio Interface for Basic Agent Evaluation...")
    # --- Build Gradio Interface using Blocks ---
    with gr.Blocks() as demo:
        gr.Markdown("# Basic Agent Evaluation Runner")
        gr.Markdown(
            """
            **Instructions:**
            1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
            2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
            3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
            ---
            **Disclaimers:**
            Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
            This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
            """
        )

        gr.LoginButton()

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

        status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
        # Removed max_rows=10 from DataFrame constructor
        results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

        run_button.click(
            fn=run_and_submit_all,
            outputs=[status_output, results_table]
        )
        demo.launch(debug=True, share=False)

	import os
	import gradio as gr
	import pandas as pd
	# Transformers and torch might not be directly needed by BasicAgent if smolgent handles its own LLM
	# from transformers import AutoModelForCausalLM, AutoTokenizer
	# import torch
	import asyncio
	import aiohttp

	from smolagents import CodeAgent, LiteLLMModel, GoogleSearchTool

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

	# --- Basic Agent Definition ---
	MODEL_ID = "gemini/gemini-2.5-pro-preview-06-05"
	GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")

	SYSTEM_PROMPT = """
	You are a highly capable and precise code generation and information retrieval agent. Your primary goal is to answer questions from various datasets.
	Instructions:
	Analyze the problem: Understand the user's query, specifically focusing on the factual questions.
	Utilize Web Search: Before attempting to answer, use web search (Google Search or Browsing tool) to find the most accurate and up-to-date information relevant to the question. Formulate precise search queries. Prioritize authoritative sources.
	Generate Code (if necessary): If the question requires computation, data manipulation, or logical reasoning that can be solved by code, generate the necessary code. The code should be correct, efficient, and directly solve the problem posed by the question. Use appropriate programming languages for the task. Ensure the code produces the final answer.
	If you call any API and it returns rate limit then attempt again but add 30 seconds delay.
	Precision is paramount: Your final answer must be as precise and accurate as possible.
	No Explanations or Commentary: Crucially, provide ONLY the final answer to the question. Do NOT include any introductory phrases, concluding remarks, explanations of your thought process, code comments (unless the code itself is the answer), or any other conversational text. Your output should be the direct, concise answer. If the answer is code, provide only the code. If the answer is a numerical value or a string, provide only that value or string.
	Your output format for questions will always be:
	[THE PRECISE ANSWER (e.g., a number, a string, or the complete code if the code itself is the answer)]
	"""

	class BasicAgent:
	def __init__(self):
	print("BasicAgent initialized to use smolagents.")

	# --- smolagents Initialization ---
	try:
	search_tool = GoogleSearchTool()

	llm_model = LiteLLMModel(
	model_id=MODEL_ID,
	api_key=GOOGLE_API_KEY,
	)
	self.agent_executor = CodeAgent(
	model=llm_model,
	tools=[search_tool],
	max_steps=12,
	planning_interval=4,
	)
	print("smolagents AgentExecutor initialized successfully with GoogleSearchTool.")

	except ImportError:
	print("ERROR: smolagents library not found. Please ensure 'smolagents' is installed and import paths are correct.")
	raise
	except Exception as e:
	print(f"Error initializing smolagents: {e}")
	raise


	def __call__(self, question: str) -> str:
	print(f"\nAgent (smolagents) received question (first 100 chars): {question[:100]}...")
	try:
	response = self.agent_executor.run(question)

	print(f"Agent (smolagents) returning answer (first 50 chars): {str(response)[:50]}...")
	return str(response)
	except Exception as e:
	print(f"Error during smolagents agent's __call__: {e}")
	return f"Error: Could not process the question with smolagents. {e}"

	async def run_and_submit_all(profile: gr.OAuthProfile \| None):
	"""
	Fetches all questions, runs the BasicAgent on them, submits all answers,
	and displays the results.
	"""
	space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code

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

	if profile:
	username= f"{profile.username}"
	print(f"User logged in: {username}")

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

	questions_data = []
	async with aiohttp.ClientSession() as session:
	# 2. Fetch Questions
	print(f"Fetching questions from: {questions_url}")
	try:
	async with session.get(questions_url, timeout=aiohttp.ClientTimeout(total=20)) as response:
	response.raise_for_status()
	questions_data = await 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 aiohttp.ClientResponseError as e:
	err_msg = f"Error fetching questions (HTTP {e.status}): {e.message}"
	print(err_msg)
	return err_msg, None
	except asyncio.TimeoutError:
	err_msg = "Timeout error fetching questions."
	print(err_msg)
	return err_msg, None
	except aiohttp.ClientError as e: # Includes connection errors, etc.
	err_msg = f"Error fetching questions (ClientError): {e}"
	print(err_msg)
	return err_msg, None
	except Exception as e: # Catch-all for others like JSONDecodeError
	err_msg = f"An unexpected error occurred fetching questions: {e}"
	print(err_msg)
	if 'response' in locals() and hasattr(response, 'text'):
	try:
	response_text_snippet = (await response.text())[:500]
	print(f"Response text snippet: {response_text_snippet}")
	except: pass
	return err_msg, None

	# 3. Run your Agent Concurrently
	results_log = []
	answers_payload = []
	print(f"Running agent on {len(questions_data)} questions...")

	# Note: If BasicAgent initialization is very lightweight, this is fine.
	# If it's heavy (e.g., downloads large models every time),
	# then ensuring BasicAgent and its components are stateless/reentrant (Option 2)
	# would be more efficient, but also more complex to verify.
	async def process_question(item):
	# Instantiate a new agent for each question to ensure isolation
	agent_instance = BasicAgent()
	task_id = item.get("task_id")
	question_text = item.get("question")
	if not task_id or question_text is None:
	print(f"Skipping item with missing task_id or question: {item}")
	return None, None
	try:
	loop = asyncio.get_event_loop()
	# Run the synchronous agent call in a thread pool executor
	submitted_answer = await loop.run_in_executor(None, agent_instance, question_text)
	return {"task_id": task_id, "submitted_answer": submitted_answer}, \
	{"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}")
	return None, {"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"}

	tasks = [process_question(item) for item in questions_data]
	processed_results = await asyncio.gather(*tasks)

	for payload, log_entry in processed_results:
	if payload: answers_payload.append(payload)
	if log_entry: results_log.append(log_entry)

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

	# 4. Prepare Submission
	submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
	status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
	print(status_update)

	async with aiohttp.ClientSession() as session:
	# 5. Submit
	print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
	try:
	async with session.post(submit_url, json=submission_data, timeout=aiohttp.ClientTimeout(total=60)) as response:
	response.raise_for_status()
	result_data = await response.json()
	final_status = (
	f"Submission Successful!\n"
	f"User: {result_data.get('username')}\n"
	f"Overall Score: {result_data.get('score', 'N/A')}% "
	f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
	f"Message: {result_data.get('message', 'No message received.')}"
	)
	print("Submission successful.", final_status)
	return final_status, pd.DataFrame(results_log)
	except aiohttp.ClientResponseError as e:
	error_detail = f"Server responded with status {e.status}."
	try:
	# Try to get JSON error detail, then text
	error_body = await e.json()
	error_detail += f" Detail: {error_body.get('detail', str(error_body))}"
	except Exception: # If not JSON or other error
	try:
	error_text = (await response.text())[:500]
	error_detail += f" Response: {error_text}"
	except:
	error_detail += " Could not retrieve error response body."
	status_message = f"Submission Failed: {error_detail}"
	except asyncio.TimeoutError:
	status_message = "Submission Failed: The request timed out."
	except aiohttp.ClientError as e:
	status_message = f"Submission Failed: Network error - {e}"
	except Exception as e:
	status_message = f"An unexpected error occurred during submission: {e}"

	print(status_message)
	return status_message, pd.DataFrame(results_log)

	if __name__ == "__main__":
	print("\n" + "-"30 + " App Starting " + "-"30)
	# Check for SPACE_HOST and SPACE_ID at startup for information
	space_host_startup = os.getenv("SPACE_HOST")
	space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup

	if space_host_startup:
	print(f"✅ SPACE_HOST found: {space_host_startup}")
	print(f" Runtime URL should be: https://{space_host_startup}.hf.space")
	else:
	print("ℹ️ SPACE_HOST environment variable not found (running locally?).")
	loop = asyncio.get_event_loop()
	loop.run_until_complete(run_and_submit_all(None))
	exit(0)

	if space_id_startup: # Print repo URLs if SPACE_ID is found
	print(f"✅ SPACE_ID found: {space_id_startup}")
	print(f" Repo URL: https://huggingface.co/spaces/{space_id_startup}")
	print(f" Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
	else:
	print("ℹ️ SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")
	exit(0)

	print("-"*(60 + len(" App Starting ")) + "\n")
	print("Launching Gradio Interface for Basic Agent Evaluation...")
	# --- Build Gradio Interface using Blocks ---
	with gr.Blocks() as demo:
	gr.Markdown("# Basic Agent Evaluation Runner")
	gr.Markdown(
	"""
	Instructions:

	1. Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
	2. Log in to your Hugging Face account using the button below. This uses your HF username for submission.
	3. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.

	---
	Disclaimers:
	Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
	This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
	"""
	)

	gr.LoginButton()

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

	status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
	# Removed max_rows=10 from DataFrame constructor
	results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

	run_button.click(
	fn=run_and_submit_all,
	outputs=[status_output, results_table]
	)
	demo.launch(debug=True, share=False)