# Standard library import os import io import base64 import shutil import subprocess from pathlib import Path from typing import Any, Dict, List, TypedDict # Third-party libraries import gradio as gr import pandas as pd import requests from dotenv import load_dotenv import PIL.Image as Image from openai import AzureOpenAI # LangChain and LangGraph from langchain_openai import AzureChatOpenAI from langchain_core.tools import tool from langchain.tools import Tool from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents import AgentExecutor from langchain.agents.openai_functions_agent.base import create_openai_functions_agent from langchain.tools import tool from langgraph.graph import START, StateGraph, END from langchain_community.tools.tavily_search import TavilySearchResults # LangChain Community Tools from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.utilities.wikipedia import WikipediaAPIWrapper from langchain_community.tools.wikipedia.tool import WikipediaQueryRun # Custom Tools from tools import add, subtract, divide, multiply, modulus, string_reverse # (Keep Constants as is) # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- Basic Agent Definition --- # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------ load_dotenv() # Azure OpenAI model llm = AzureChatOpenAI( deployment_name=os.environ["AZURE_OPENAI_GPT41MINI_ID"], api_key=os.environ["AZURE_OPENAI_API_KEY"], api_version=os.environ["AZURE_OPENAI_GPT41MINI_VERSION"], azure_endpoint=os.environ["AZURE_OPENAI_GPT41MINI_ENDPOINT"], temperature=0 ) o4_mini = AzureChatOpenAI( deployment_name = os.environ.get("AZURE_OPENAI_O4MINI_ID"), api_key=os.environ.get("AZURE_OPENAI_API_KEY"), api_version=os.environ.get("AZURE_OPENAI_O4MINI_VERSION"), azure_endpoint=os.environ.get("AZURE_OPENAI_O4MINI_ENDPOINT") ) class AgentState(TypedDict, total=False): file_path: str | None # Contains file path question: str # Contains tabular file path (CSV) answer: str | None agent_type: str | None messages: list[AIMessage | HumanMessage | SystemMessage] ## Tools duckduckgo_search = Tool( name="duckduckgo_search", func=DuckDuckGoSearchRun().run, description="""A wrapper around DuckDuckGo Search. Useful for when you need to answer questions about current events. Input should be a search query.""" ) @tool def web_search(query: str): """ Description: A web search tool. Scrapes the top results and returns each on its own line. Arguments: • query (str) : question you want to web search. Return: str – A newline-separated text summary: ' — <url> : <snippet>' or 'No results found' """ search = TavilySearchResults() results = search.run(query) return "\n".join([f"- {r['content']} ({r['url']})" for r in results]) @tool def wikipedia_query(query: str): """ Description: Query the English-language Wikipedia via the MediaWiki API and return a short plain-text extract. Arguments: • query (str) : Page title or free-text search string. Return: str – Extracted summary paragraph. """ wiki = WikipediaAPIWrapper() return wiki.run(query) @tool def python_handler(filepath: str) -> str: """ Description: Execute a stand-alone Python script in a sandboxed subprocess and capture anything the script prints to stdout. Stderr is returned instead if the script exits with a non-zero status. Arguments: • filepath (str): Path to the .py file to run. Return: str – The final output of the .py file. """ try: result = subprocess.run( ["python", filepath], capture_output=True, text=True, timeout=30 # Safety ) return result.stdout.strip() if result.returncode == 0 else result.stderr except Exception as e: return f"Execution failed: {str(e)}" @tool def addition_tool(list: List[float]) -> float: """ Description: A simple addition tool that takes a list of numbers and returns their sum. Arguments: • list (List[float]): List of numbers to add. Return: float – The sum of the numbers in the list. """ return sum(list) @tool def xlsx_handler(filepath: str) -> List[Dict[str, Any]]: """ Description: Load the first sheet of an Excel workbook and convert it into a JSON-serialisable list of row dictionaries (records). Arguments: • filepath (str): Absolute or relative path to the .xlsx file. Return: str – A list of dictionaries representing the column names and their values. """ # Load the Excel file df = pd.read_excel(filepath) columns = df.columns.tolist() result = [] for col in columns: result.append({"column": col, "values": df[col].tolist()}) return result ## Functions def img_to_data(img: Image.Image) -> str: buf = io.BytesIO(); img.save(buf, format="PNG", optimize=True) b64 = base64.b64encode(buf.getvalue()).decode() return f"data:image/png;base64,{b64}" def task_examiner(state: AgentState): file_path = state["file_path"] if file_path != None: p = Path(file_path) suffix = p.suffix if suffix == ".png": state["agent_type"] = "image" elif suffix == ".mp3": state["agent_type"] = "audio" elif suffix == ".py" or suffix == ".xlsx": state["agent_type"] = "code" else: state["agent_type"] = "general" return state def task_router(state: AgentState) -> str: return state["agent_type"] ## Agents def general_agent(state: AgentState): question = state["question"] tools = [web_search, wikipedia_query, string_reverse] system_prompt = ChatPromptTemplate.from_messages([ ("system", """ SYSTEM GUIDELINES: You are a general-purpose AI assistant tasked with accurately answering the user's questions. BEHAVIOR RULES: - You have access to a set of specialized tools to help with certain tasks. - You MUST reason step-by-step internally before calling any tool. - Only call a tool when you are confident it is required to answer the question. - Tool calls should be minimal and purposeful. TOOL REUSE RULE: - Maintain an internal list of tools already used in the current answer. - You MUST NOT call the same tool more than once per answer. (However, you may still use a different tool.) AVAILABLE TOOLS: - `web_search`: Searches the web for up-to-date information not present in your training data. - `wikipedia_query`: Searches Wikipedia for factual information not present in your training data. - `string_reverse`: Reverses a sentence. Use this if the input appears garbled, backward, or unreadable. INPUT FORMAT: - A single user question (free-form text). OUTPUT FORMAT: - Output ONLY the final answer to the user's question. - NEVER include explanations, reasoning steps, or tool usage metadata. - Your output must strictly follow the required format described below. SPECIAL CASE FORMATTING RULES: - If the question includes a YouTube link (e.g. `https://www.youtube.com/watch?...`), respond ONLY with: `Don't know` - For questions beginning with: - **"How many..."** → Respond with a **single numeral** (e.g., `5`). Do **not** include punctuation, words, or units. - **"What number..."** → Respond with a **single numeral** (e.g., `42`). No extra text. - **"Who did..."** → Respond with the **full name of the person only**, without any punctuation or additional commentary. - If asked for a **comma-separated list**, respond in the format: `[item1,item2,item3]` NEVER use `a,b,c,d` formatting outside brackets. - If asked to **output a list**, respond with: `[item1,item2,item3]` - If the question says: **"What does person A say when..."** → Respond with **only the quoted phrase**, with no extra punctuation, commentary, or formatting. FAILURE TO FOLLOW THESE FORMATTING RULES WILL RESULT IN AN INVALID RESPONSE. """), ("user", "{input}"), MessagesPlaceholder("agent_scratchpad"), ]) agent = create_openai_functions_agent( llm=llm, tools=tools, prompt=system_prompt ) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) response = agent_executor.invoke({"input": question}) state["answer"] = response["output"] return state def audio_agent(state: AgentState): with open(state["file_path"], "rb") as f: client = AzureOpenAI( api_key=os.environ["AZURE_OPENAI_API_KEY"], api_version='2024-06-01', azure_endpoint=os.environ["AZURE_OPENAI_WHISPER_ENDPOINT"], ) transcript = client.audio.transcriptions.create(model='whisper', file=f, response_format="text") question = state["question"] system_msg = SystemMessage( content=("You are an AI assistant that answers the user's question based solely on the provided transcript." "When the user asks for a “comma-delimited / comma-separated list”, you must:" " - Filter the items exactly as requested." " - Output one single line that contains the items separated by commas and a space enclosed in square brackets." " - Output nothing else- no extra words or explanations" "OUTPUT FORMAT EXAMPLES:" "If asked to output a list -> Output: [item1,item2,item3]" "If asked something else -> Output: text answering exactly that question and nothing more" ) ) messages = [ system_msg, HumanMessage( content=[ { "type": "text", "text": f"Transcript:\n{transcript}\n\nQuestion:\n{question}" } ] ) ] response = llm.invoke(messages) state["answer"] = response.content.strip() return state def image_agent(state: AgentState): file_path = state["file_path"] question = state["question"] with open(file_path, "rb") as image_file: image_bytes = image_file.read() image_base64 = base64.b64encode(image_bytes).decode("utf-8") system_msg = SystemMessage( content=(""" You are a Image AI assistant that can process images and answer correctly the user's questions" **OUTPUT** only the final answer and absolutely nothing else (no punctuation, no sentence, no units). """) ) messages = [ system_msg, HumanMessage( content=[ { "type": "text", "text": (f"{question}") }, { "type": "image_url", "image_url": { "url": f"data:image/png;base64,{image_base64}" }, } ] ) ] response = llm.invoke(messages) state["answer"] = response.content.strip() return state def code_agent(state: AgentState): file_path = state["file_path"] question = state["question"] tools = [xlsx_handler, python_handler, addition_tool] system_prompt = ChatPromptTemplate.from_messages([ ("system", """ SYSTEM GUIDELINES: - You are a data AI assistant and your job is to answer questions that depend on .xlsx or .py files. - You have in your disposal 2 tools that are mandatory for solving the tasks. - You **MUST** use the tools as instructed below and you **MUST** output only the final numeric result of the task. INPUT FORMAT: - A question (text) based on a file which will be either .py or .xlsx. - The path of the file related to the question. TOOLS: - Tool name: xlsx_handler, Purpose: This is the tool you should use if the file contained in the file_path is an .xlsx file and it's purpose is to return the contents of the file in a list of dictionaries for you to process, reason **INTERNALLY** and output only the final numeric result. - Tool name: python_handler, Purpose: This is the tool you should use if the file contained in the file_path is a .py file and it's purpose is to execute the python file and return the final numeric result of it. - Tool name: addition_tool, Purpose: This is the tool you should use if the question asks you to sum a list of numbers and return the final numeric result. EXAMPLE OUTPUTS: - Input: "What is the result of the code in the file?" Output: "5" - Input: "What is the total sales mentioned in the file. Your answer must have 2 decimal places?" Output: "305.00" - YOU MUST OUTPUT ONLY THE FINAL NUMBER. The file relevant to the task is at: {file_path}."""), ("user", "{input}"), MessagesPlaceholder("agent_scratchpad"), ]) agent = create_openai_functions_agent( llm=llm, tools=tools, prompt=system_prompt # Optional – remove if you want default prompt behavior ) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) response = agent_executor.invoke({"input": question, "file_path": file_path}) state["answer"] = response["output"] return state # def math_agent(state: AgentState): # file_path = state["file_path"] # question = state["question"] # tools = [add, subtract, divide, multiply, modulus] # system_prompt = ChatPromptTemplate.from_messages([ # ("system", # """ SYSTEM GUIDELINES: # - You are a data AI assistant and your job is to answer questions that are math-related. # - You have in your disposal 5 tools that are mandatory for solving the tasks. # - You **MUST** use the tools as instructed below and you **MUST** output only the final numeric result of the task. # INPUT FORMAT: # - A question (text) based that needs a math function to be solved. # TOOLS: # - Tool name: add, Purpose: A simple addition tool that takes two numbers and returns their sum. # - Tool name: subtract, Purpose: A simple subtraction tool that takes two numbers and returns the result of the first number minus the second number. # - Tool name: divide, Purpose: A simple division tool that takes two numbers and returns the result of the first number divided by the second number. # - Tool name: multiply, Purpose: A simple multiplication tool that takes two numbers and returns the result of the first number multiplied by the second number. # - Tool name: modulus, Purpose: A simple modulus tool that takes two numbers and returns the result of the first number modulo the second number. # EXAMPLE OUTPUTS: # - Input: "What is 10 divided by 2" Output: "5" # - YOU MUST OUTPUT ONLY THE FINAL NUMBER. # """), # ("user", "{input}"), # MessagesPlaceholder("agent_scratchpad"), # ]) # agent = OpenAIFunctionsAgent( # llm=llm, # tools=tools, # prompt=system_prompt # ) # agent_executor = AgentExecutor.from_agent_and_tools( # agent=agent, # tools=tools, # verbose=True, # ) # response = agent_executor.invoke({"input": question, "file_path": file_path}) # state["answer"] = response["output"] # return state ## Agent Workflow class Agent_Workflow: def __init__(self): print("Agent Workflow initialized.") def __call__(self, question: str, filepath: str) -> str: builder = StateGraph(AgentState) # Agent Nodes builder.add_node("task_examiner", task_examiner) builder.add_node("general_agent", general_agent) builder.add_node("audio_agent", audio_agent) builder.add_node("image_agent", image_agent) builder.add_node("code_agent", code_agent) # Edges that connect agent nodes builder.add_edge(START, "task_examiner") builder.add_conditional_edges("task_examiner", task_router, { "general": "general_agent", "audio": "audio_agent", "image": "image_agent", "code": "code_agent", } ) builder.add_edge("general_agent", END) builder.add_edge("audio_agent", END) builder.add_edge("image_agent", END) builder.add_edge("code_agent", END) workflow_graph = builder.compile() state = workflow_graph.invoke({"file_path": filepath, "question": question, "answer": "",}) return state["answer"] def fetch_task_file_static(task_id: str, file_name: str | None = None, session: requests.Session | None = None) -> Path: """ Download the attachment for `task_id` to temp_files/<task_id>.<suffix> """ if file_name == None: return None # Decide the suffix suffix = Path(file_name).suffix if file_name else "" dest = "temp/"+task_id+suffix url = f"{DEFAULT_API_URL}/files/{task_id}" s = session or requests with s.get(url, stream=True, timeout=30) as r: r.raise_for_status() with open(dest, "wb") as f: shutil.copyfileobj(r.raw, f) return dest 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 = Agent_Workflow() 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() questions_data = questions_data 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 requests.exceptions.JSONDecodeError as e: print(f"Error decoding JSON response from questions endpoint: {e}") print(f"Response text: {response.text[:500]}") return f"Error decoding server response for questions: {e}", None except Exception as e: print(f"An unexpected error occurred fetching questions: {e}") return f"An unexpected error occurred fetching questions: {e}", None # 3. Run your Agent results_log = [] answers_payload = [] print(f"Running agent on {len(questions_data)} questions...") session = requests.Session() # Reuse session for fetching files for item in questions_data: task_id = item.get("task_id") question_text = item.get("question") file_name = item.get("file_name") file_path = None if file_name: try: file_path = fetch_task_file_static(task_id, file_name, session=session) except requests.HTTPError as e: print(f"⚠️ Couldn’t fetch file for {task_id}: {e}") 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, filepath=file_path) 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("# 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] ) 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)