app.py

# 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: '<title> — <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)