app.py

import os
import tempfile
from base64 import b64encode
from contextlib import suppress
from io import BytesIO
from pprint import pprint
from time import sleep
from typing import TypedDict, List, Dict, Any, Optional, Tuple
from typing_extensions import Annotated

import openai
import gradio as gr
import requests
import inspect
import pandas as pd

from langgraph.graph import MessagesState, StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AnyMessage
from langchain_core.runnables.config import RunnableConfig
from langchain_core.tools import tool
from langchain_tavily import TavilySearch


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

model = ChatOpenAI(model="gpt-4o", temperature=0)


class State(MessagesState):
    question: str


class BasicAgent:
    def __init__(self):
        self.tools = [
            BasicAgent.search_tool,
            BasicAgent.find_local_files_tool,
            BasicAgent.read_text_file_tool,
            BasicAgent.vision_tool,
            BasicAgent.audio_qa_tool,
            BasicAgent.excel_tool
        ]

        # Chat model with tool support
        self.model_with_tools = model.bind_tools(self.tools, parallel_tool_calls=False)

        # LangGraph
        self.graph = StateGraph(State)
        self.graph.add_node("assistant", self.assistant)
        self.graph.add_node("tools", ToolNode(self.tools))

        self.graph.add_edge(START, "assistant")
        self.graph.add_conditional_edges("assistant", tools_condition)  # decide if tools should be called
        self.graph.add_edge("tools", "assistant")  # loop back

        self.compiled_graph = self.graph.compile()
        print("BasicAgent initialized.")

    def __call__(self, question: str) -> Tuple[str, List[Dict[str, Any]]]:
        print(f"\nAgent received question: {question}")
        sys_msg = SystemMessage(
            content="""
        You are a ReAct (Reasoning and Acting) agent with self-reflection. For each question:

        1. **Thought:** Briefly outline your reasoning step.  
        2. **Reflect:** Check “Did I use all observations? Did my tool call succeed?”  
        3. **Action:** Either call a tool (with arguments) or prepare your final answer.  
        4. **Final Answer:** Provide only the bare result (no labels, no extra text, no actions, no thoughts, no reflection, no "Final Answer" string in the result). For question that contain phrases like `what is the number` or 
        `what is the highest number` return just the number, e.g., 2.
        
        **Answer Format Rules**  
        - If the answer is a number, output digits only (no commas, no units, no strings like “one”, “twenty three”).  
        - If it’s a word or phrase, don't use articles, neither abbreviations (e.g. for cities - Saint Louis, not St. Louis).  
        - If it’s a comma separated list, output a comma-separated list following the above rules for each element.  
        - **Always** output exactly one line as an answer and nothing else.
        
        **Example 1**  
        Q: What is 7 × 6?  
        Thought: Multiply 7 by 6.  
        Reflect: Simple arithmetic, no tool needed.  
        Final Answer: 42
        
        **Example 2**  
        Q: How many prime numbers are there under 20?  
        Thought: Primes under 20 are 2, 3, 5, 7, 11, 13, 17, 19 (8 total).  
        Reflect: Count is correct.  
        Final Answer: 8
        
        **Example 3**  
        Q: Sort “banana”, “apple”, “cherry” alphabetically descending.  
        Thought: Alphabetical descending: cherry, banana, apple.  
        Reflect: Order and formatting confirmed.  
        Final Answer: cherry, banana, apple

        **Example 4**
        Q: The attached csv file contains the amount of impressions for an ad campaign. What were the total amount of clicks crevenue that occurred after 2024-01-01? Express your answer in EUR with two decimal places.
        Thought: Calculate the total amount of revenue for clicks across all dates after 2024-01-01.
        Reflect: I have all the necessary data from the csv file.
        Action: Multiple clicks amount by revenue per click for each row after 2024-01-01 and then sum these values.
        Final Answer: 283934.00

        **Example 5**
        Q: What is the number of the most performant desktop processor model from Ryzen 1000 series?
        Thought: The number of the most performant desktop processor model from Ryzen 1000 series is 1800X.
        Reflect: I know the answer, displaying only the model number without anything else.
        Final Answer: 1800X
        ---
        
        Now answer the next question following this chain-of-thought + reflection pattern, and output **only** the `Final Answer` in the required format.

        """
        )
            
        state = State(
            question=question, 
            messages=[sys_msg, HumanMessage(content=question)]
        )
        config = RunnableConfig(recursion_limit=15)
        result = self.compiled_graph.invoke(state, config)
        final_answer = result["messages"][-1].content
        print(f"\nFinal Answer: {final_answer}")
        return final_answer, result["messages"]

    def assistant(self, state: State):
        print("\nAssistant invoked. State:\n")
        pprint(state)
        response = self.model_with_tools.invoke(state["messages"])
        print("\nAssistant response:", response)
        return {
            "messages": [response]
        }

    @staticmethod
    @tool(
      description="Search the web using TavilySearch and return the final snippet.",
    )
    def search_tool(question: str, max_length: int = 100000) -> str:
        print(f"\nCalling search tool with: {question}, max_lentgh: {max_length}")
        search_ = TavilySearch(
            max_results=4,
            topic="general",
        )
        info = search_.invoke({"query": question})
        result = "\n".join(m["content"] for m in info["results"])
        print("f\nSearch result: {result}")
        return result[:max_length]

    @staticmethod
    @tool(
      description="List task files.",
    )
    def find_local_files_tool() -> list[str]:
        print(f"\nCalling find local files tool")
        files = [f for f in os.listdir() if os.path.isfile(f) and f.startswith('task_file_')]
        print(f"\nReturning", files)
        return files

    @staticmethod
    @tool(
      description="Read the text file and return it's content.",
    )
    def read_text_file_tool(file_name: str) -> str:
        print(f"\nCalling read text file tool for", file_name)
        print("File metadata:", os.stat(file_name))
        with open(file_name, 'r') as f:
            return f.read()

    @staticmethod
    @tool(
        description="Analyze an image file and answer a follow-up question about its content."
    )
    def vision_tool(path: str, question: str) -> str:
        """
        Args:
          path: Path to a local image file.
          question: What you want to know (e.g. 'How many people are in this photo?').
        Returns:
          The LLM’s answer based on the image content.
        """
        if not os.path.exists(path):
            return f"Error: file not found at {path}"

        print("File metadata:", os.stat(path))

        with open(path, "rb") as f:
            b64 = b64encode(f.read()).decode()

        ext = os.path.splitext(path)[1].lower().lstrip(".")
        mime = f"image/{'jpeg' if ext in ('jpg','jpeg') else 'png'}"
        
        # 2) Build the multimodal message
        msg = HumanMessage(content=[
            {"type": "text", "text": question},
            {
              "type": "image_url",
              "image_url": {"url": f"data:{mime};base64,{b64}"}
            }
        ])
        
        response = model.invoke([SystemMessage(content="Analyze the image and answer the question."), msg])
        result = response.content
        print("Result:", result)
        return result

    @staticmethod
    @tool(
        description="Transcribe an audio file with Whisper and answer a question about its content."
    )
    def audio_qa_tool(path: str, question: str, max_chars: int = 10000) -> str:
        """
        Args:
          path: Local filesystem path to an audio file (mp3, wav, etc.).
          question: What to ask about the audio content.
          max_chars: Maximum length of the returned answer.
        Returns:
          The LLM’s answer, based on the transcript (truncated if necessary).
        """
        if not os.path.exists(path):
            return f"Error: file not found at {path}"

        print("File metadata:", os.stat(path))
        with open(path, "rb") as audio_file:
            client = openai.OpenAI()
            transcription = client.audio.transcriptions.create(
                file=audio_file,
                model="whisper-1"
            )
        transcript = transcription.text
        prompt = f"""
            Here is a transcript of an audio file:
            '''{transcript}'''

            Question: '''{question}'''
                    
            Please answer briefly based on this transcript, and give only the answer.
        """
        response = model.invoke([{"role": "user", "content": prompt}])
    
        answer = response.content.strip()
        return answer[:max_chars]


    @staticmethod
    @tool(
        description="Load an Excel file and returns it's text representation."
    )
    def excel_tool(path: str) -> str:
        """
        Args:
          path: Path to the .xlsx file.
        Returns:
          The string form of the content.
        """
        df = pd.read_excel(path, engine='openpyxl')
        return str(df.to_csv(index=False))
    

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 = BasicAgent()
    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: {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...")
    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:
            file_url = f"{api_url}/files/{task_id}"
            file_name = f"task_file_{task_id}"
            with open(file_name, "wb") as file:
                response = requests.get(file_url, timeout=15)
                file.write(response.content)
        except Exception as e:
            print(f"Expection occurred while trying to download {file_name} from {file_url}:", e)
            print("Didn't manage to download a file, probably it's not expected for this task")
        
        try:
            submitted_answer, logs = 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})
            print(f"\n\n\n==============Finishing task id: {task_id}, question_text: {question_text}==============\n\n\n")
            sleep(2)
        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}"})
        finally:
            with suppress(Exception):
                os.remove(file_name)

    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


def check_agent(question: str):
    agent = BasicAgent()
    final_answer, msgs = agent(question)
    return final_answer, "\n\n".join([str(msg) for msg in msgs])


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

    with gr.Row():
        question_input = gr.Textbox(label="Enter your question", placeholder="e.g., What is the capital of France?", lines=10)
        check_button = gr.Button("Check Answer")

    final_output = gr.Textbox(label="✅ Final Answer", lines=10, interactive=False)
    logs_output = gr.Textbox(label="📝 Agent Logs", lines=20, interactive=False)

    check_button.click(
        fn=check_agent,
        inputs=question_input,
        outputs=[final_output, logs_output]
    )

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)