app.py

import os
import json
import gradio as gr
import requests
import inspect
import torch
import pandas as pd
import yaml

# Tool disponibili di default:
# ApiWebSearchTool
# PythonInterpreterTool
# FinalAnswerTool
# UserInputTool
# WebSearchTool
# DuckDuckGoSearchTool
# GoogleSearchTool
# VisitWebpageTool
# WikipediaSearchTool
# SpeechToTextTool
from smolagents import CodeAgent, ToolCallingAgent, TransformersModel, VisitWebpageTool, PythonInterpreterTool, WebSearchTool, WikipediaSearchTool, FinalAnswerTool, Tool, tool # InferenceClientModel, GoogleSearchTool (usa SERPAPI_API_KEY), DuckDuckGoSearchTool
from smolagents.agents import PromptTemplates, EMPTY_PROMPT_TEMPLATES

# (Keep Constants as is)
# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

@tool
def invert_sentence_tool(sentence: str) -> str:
	"""
	Inverts the order of all characters in a sentence.
	Args:
		sentence (str): The sentence to invert.
	Returns:
		str: The sentence with characters in reverse order.
	"""
	return sentence[::-1]

# --- First Agent Definition ---
# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
class FirstAgent:
	### First Agent is the first attempt to develop an agent for the course. ###
	def __init__(self):
		""" Initializes the FirstAgent with a TransformersModel and a CodeAgent. """
		# microsoft/Phi-4-reasoning-plus, microsoft/Phi-4-reasoning, microsoft/Phi-4-mini-reasoning, microsoft/Phi-4-mini-instruct, microsoft/Phi-4-multimodal-instruct, "Qwen/Qwen2.5-Coder-32B-Instruct", "HuggingFaceTB/SmolLM2-1.7B-Instruct"
		model_id = "HuggingFaceTB/SmolLM2-1.7B-Instruct"
		model = TransformersModel(model_id=model_id, trust_remote_code=True, max_new_tokens=4096)

		# Inizializza l'agente
		with open("prompts.yaml", "r", encoding="utf-8") as file:
			prompts = yaml.safe_load(file)

		prompt_templates = PromptTemplates(**prompts)

		# prompt_templates = EMPTY_PROMPT_TEMPLATES
		prompt_templates["system_prompt"] = """You are an intelligent agent that answers questions and uses tools to help users.
			Think step by step and use tools when needed.
			If you need information in Wikipedia, try to use the WikipediaSearchTool. If you cannot find a page with this tool, try to use wikipediaapi directly to get the page and extract content.
			If you need to search the web, use the WebSearchTool.
			If you need to visit a webpage, use the VisitWebpageTool.
			If you need to run Python code, use the PythonInterpreterTool.
			If you need to generate code to extract information, generate the code and use the PythonInterpreterTool to execute the code and extract the informations.
			Verify your final answer and the format of the final answer before returning it.
			At each step, in the 'Thought:' sequence, you should first explain your reasoning towards solving the task and the tools that you want to use.
			Then in the '<code>' sequence, you should write the code in simple Python. The code sequence must end with '</code>' sequence.
			During each intermediate step, you can use 'print()' to save whatever important information you will then need.
			These print outputs will then appear in the 'Observation:' field, which will be available as input for the next step.
			In the end you have to return a final answer using the `final_answer` tool, like this: <code>final_answer(answer)</code>.
			Here are the rules you should always follow to solve your task:
			1. Always provide a 'Thought:' sequence, and a '<code>' sequence ending with '</code>', else you will fail.
			2. Use only variables that you have defined!
			3. Always use the right arguments for the tools. DO NOT pass the arguments as a dict as in 'answer = wikipedia_search({'query': "What is the place where James Bond lives?"})', but use the arguments directly as in 'answer = wikipedia_search(query="What is the place where James Bond lives?")'.
			4. Take care to not chain too many sequential tool calls in the same code block, especially when the output format is unpredictable. For instance, a call to wikipedia_search has an unpredictable return format, so do not have another tool call that depends on its output in the same block: rather output results with print() to use them in the next block.
			5. Call a tool only when needed, and never re-do a tool call that you previously did with the exact same parameters.
			6. Don't name any new variable with the same name as a tool: for instance don't name a variable 'final_answer'.
			7. Never create any notional variables in our code, as having these in your logs will derail you from the true variables.
			8. You can use imports in your code, but only from the following list of modules: {{authorized_imports}}
			9. The state persists between code executions: so if in one step you've created variables or imported modules, these will all persist.
			10. Don't give up! You're in charge of solving the task, not providing directions to solve it.
			If you don't know the answer, say 'I don't know'."""

		self.agent = CodeAgent(
			model=model,
			# prompt_templates=prompt_templates,
			stream_outputs=True,  # Enable streaming outputs
			additional_authorized_imports=['bs4', 'collections', 'datetime', 'itertools', 'math', 'queue', 'random', 're', 'requests', 'stat', 'statistics', 'time', 'unicodedata', 'wikipediaapi'],
			# add_base_tools=True,  # Add base tools like UserInputTool
			# use_structured_outputs_internally=True,  # Use structured outputs internally
			tools=[
				WebSearchTool(),
				PythonInterpreterTool(),
				# WikipediaSearchTool(
				# 		user_agent="FinalAssignmentAgent ([email protected])",
				# 		language="en",
				# 		content_type="text",  # "summary" or "text"
				# 		extract_format="WIKI", # "WIKI" or "HTML"
				# 	),
				VisitWebpageTool(),
				invert_sentence_tool, # Custom tool to invert sentences
				FinalAnswerTool()  # Final answer tool to extract the final answer
			]
		)
		print("FirstAgent inizializzato.")
	
	def __call__(self, question: str) -> str:
		""" Runs the agent with the provided question and returns the answer. """
		print(f"Agent ricevuto domanda (primi 50 char): {question[:50]}...")
		try:
			answer = self.agent.run(question)
			print(f"Agent restituisce risposta: {str(answer)[:100]}...")
			return answer
		except Exception as e:
			print(f"Errore nell'agente: {e}")
			return f"Errore nell'agente: {str(e)}"

# --- Basic Agent Definition ---
# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
class BasicAgent:
	### Basic Agent is a placeholder for a simple agent that always returns a fixed answer. ###
	### It is used to demonstrate the structure of an agent. ###
	def __init__(self):
		print("BasicAgent initialized.")
	def __call__(self, question: str) -> str:
		print(f"Agent received question (first 50 chars): {question[:50]}...")
		fixed_answer = "This is a default answer."
		print(f"Agent returning fixed answer: {fixed_answer}")
		return fixed_answer

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

	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"

	# 1. Instantiate Agent ( modify this part to create your agent)
	try:
		agent = FirstAgent()
	except Exception as e:
		print(f"Error instantiating agent: {e}")
		return f"Error initializing agent: {e}", None
	# In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
	agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
	print(agent_code)

	# 2. Fetch Questions
	print(f"Fetching questions from: {questions_url}")
	try:
		response = requests.get(questions_url, timeout=15)
		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 from API: {e}")
		print("Attempting to load questions from local file 'questions.json'...")
		try:
			with open("questions.json", "r", encoding="utf-8") as f:
				questions_data = json.load(f)
			if not questions_data:
				return "Both API and local questions file are empty.", None
			print(f"Successfully loaded {len(questions_data)} questions from local file.")
		except FileNotFoundError:
			return "Error: Could not fetch questions from API and 'questions.json' file not found.", None
		except json.JSONDecodeError as json_e:
			return f"Error: Could not fetch questions from API and local file has invalid JSON: {json_e}", None
		except Exception as file_e:
			return f"Error: Could not fetch questions from API and failed to read local file: {file_e}", None
	except requests.exceptions.JSONDecodeError as e:
		print(f"Error decoding JSON response from questions endpoint: {e}")
		print(f"Response text: {response.text[:500]}")
		print("Attempting to load questions from local file 'questions.json'...")
		try:
			with open("questions.json", "r", encoding="utf-8") as f:
				questions_data = json.load(f)
			if not questions_data:
				return "Both API response is invalid and local questions file is empty.", None
			print(f"Successfully loaded {len(questions_data)} questions from local file.")
		except FileNotFoundError:
			return "Error: Could not decode API response and 'questions.json' file not found.", None
		except json.JSONDecodeError as json_e:
			return f"Error: Could not decode API response and local file has invalid JSON: {json_e}", None
		except Exception as file_e:
			return f"Error: Could not decode API response and failed to read local file: {file_e}", None
	except Exception as e:
		print(f"An unexpected error occurred fetching questions: {e}")
		print("Attempting to load questions from local file 'questions.json'...")
		try:
			with open("questions.json", "r", encoding="utf-8") as f:
				questions_data = json.load(f)
			if not questions_data:
				return "Unexpected API error occurred and local questions file is empty.", None
			print(f"Successfully loaded {len(questions_data)} questions from local file.")
		except FileNotFoundError:
			return "Error: Unexpected API error occurred and 'questions.json' file not found.", None
		except json.JSONDecodeError as json_e:
			return f"Error: Unexpected API error occurred and local file has invalid JSON: {json_e}", None
		except Exception as file_e:
			return f"Error: Unexpected API error occurred and failed to read local file: {file_e}", None

	# 3. Run your Agent
	results_log = []
	answers_payload = []
	print(f"Running agent on {len(questions_data)} questions...")
	for item in questions_data:
		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}")
			continue
		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}"})

	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)

	# 5. Submit
	print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
	try:
		response = requests.post(submit_url, json=submission_data, timeout=60)
		response.raise_for_status()
		result_data = 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.")
		results_df = pd.DataFrame(results_log)
		return final_status, results_df
	except requests.exceptions.HTTPError as e:
		error_detail = f"Server responded with status {e.response.status_code}."
		try:
			error_json = e.response.json()
			error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
		except requests.exceptions.JSONDecodeError:
			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 during submission: {e}"
		print(status_message)
		results_df = pd.DataFrame(results_log)
		return status_message, results_df


# --- Build Gradio Interface using Blocks ---
with gr.Blocks() as demo:
	gr.Markdown("# Raffaele Agent Evaluation Runner")
	gr.Markdown(
		"""
		1.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
		2.  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]
	)

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?).")

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

	print("-"*(60 + len(" App Starting ")) + "\n")

	print("Launching Gradio Interface for Basic Agent Evaluation...")
	demo.launch(debug=True, share=False)