app.py

import os
import time
import gradio as gr
import requests
import inspect
import pandas as pd
from langchain_core.documents import Document
import base64
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_google_community import GoogleSearchAPIWrapper
from langchain_community.document_loaders import YoutubeLoader,PyPDFLoader,Docx2txtLoader,TextLoader,ArxivLoader
from langchain_community.document_loaders import WikipediaLoader
from langchain_tavily import TavilySearch
import wikipedia
import speech_recognition as sr
import tempfile
import ast
import pytesseract
from PIL import Image
from langchain_core.tools import tool
# Or using AudioTranscriptTool.


# (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 ------

############### tool
@tool
def arvix_search(query: str) -> str:
    """Search Arxiv papers for a query and return maximum 5 result.
    
    Args:
        query: The search query."""
    
    search_docs = ArxivLoader(query=query, load_max_docs=5).load()

    print(f"Search query arvix: {query}")
    print(f"Search results count arvix: {len(search_docs)}, type: {type(search_docs)}")
    for doc in search_docs:
        print(doc.metadata.keys())

    formatted_search_docs = "\n\n---\n\n".join(
        [
            f'<Document source="{doc.metadata.get("Title", "")}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content[:1000]}\n</Document>'
            for doc in search_docs
        ])
    return {"arvix_results": formatted_search_docs}
    
@tool
def divide(a: int|float, b: int|float) -> float:
    """Divide a and b."""
    if b == 0:
        return "Cannot divide by zero."
    return a / b
    
@tool    
def multiply(a: int|float, b: int|float) -> float:
    """Multiply a and b."""
    return a * b
def add(a: int|float, b: int|float) -> float:
    """Add a and b."""
    return a + b
def subtract(a: int|float, b: int|float) -> float:
    """Subtract a with b."""
    return a - b
    
@tool
def pdf_loader_tool(file_url: str) -> str:
    """Load and extract text from a PDF file downloaded from given file_url."""
    try:
        # Download file into temporary file
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"
            temp_file.write(response.content)
            temp_file.flush()  # Make sure data is written
            # Read from temp file
            loader = PyPDFLoader(temp_file.name)
            docs = loader.load()
            return "\n".join([doc.page_content for doc in docs])
            
    except Exception as e:
        return f"Reading failed: {str(e)}"

@tool
def docx_loader_tool(file_url: str) -> str:
    """Load and extract text from a docx file downloaded from given file_url."""
    try:
        # Download file into temporary file
        with tempfile.NamedTemporaryFile(suffix=".docx", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"
            temp_file.write(response.content)
            temp_file.flush()  # Make sure data is written
            # Read from temp file
            loader = Docx2txtLoader(temp_file.name)
            docs = loader.load()
            return "\n".join([doc.page_content for doc in docs])
            
    except Exception as e:
        return f"Reading failed: {str(e)}"

@tool
def excel_loader_tool(file_url: str) -> str:
    """Load and extract text from an Excel file downloaded from given file_url."""
    try:
        with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"
            temp_file.write(response.content)
            temp_file.flush()
            dfs = pd.read_excel(temp_file.name, sheet_name=None)
            output = ""
            for sheet, df in dfs.items():
                output += f"Sheet: {sheet}\n{df.to_string()}\n\n"
            return output
    except Exception as e:
        return f"Reading failed: {str(e)}"

@tool
def txt_loader_tool(file_url: str) -> str:
    """Load and extract text from a txt file downloaded from given file_url."""
    try:
        # Download file into temporary file
        with tempfile.NamedTemporaryFile(suffix=".txt", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"
            temp_file.write(response.content)
            temp_file.flush()  # Make sure data is written
            # Read from temp file
            loader = TextLoader(temp_file.name)
            docs = loader.load()
            return "\n".join([doc.page_content for doc in docs])
            
    except Exception as e:
        return f"Reading failed: {str(e)}"
        
@tool    
def read_image_text(file_URL: str) -> str:
    """Extract text from image downloaded from given file_URL using OCR."""
    try:
        # Download file into temporary file
        with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"

            temp_file.write(response.content)
            temp_file.flush()  # Make sure data is written
            # Read from temp file
            image = Image.open(temp_file.name)
            return pytesseract.image_to_string(image)

    except Exception as e:
        return f"Reading failed: {str(e)}"
        
@tool        
def youtube_transcript_tool(url: str) -> str:
    """Load transcript from a YouTube video URL.Example of correct URL: https://www.youtube.com/watch?v=tPY0LjBKZUE """
    loader = YoutubeLoader.from_youtube_url(url, add_video_info=True)
    docs = loader.load()
    return "\n".join([f"Title: {doc.metadata.get('title', 'Unknown')}\nTranscript: {doc.page_content}" for doc in docs])

@tool    
def analyze_python_code(file_url: str) -> str:
    """Downloads an python code from a URL, then analyze it and summarize its structure."""
    try:
        # Download file into temporary file
        with tempfile.NamedTemporaryFile(suffix=".py", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"
                
            temp_file.write(response.content)
            temp_file.flush()  # Make sure data is written
            # Transcribe from temp file
            tree = ast.parse(temp_file.name)
            functions = [node.name for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)]
            classes = [node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef)]
            return f"Functions: {functions}\nClasses: {classes}"

    except Exception as e:
        return f"Reading failed: {str(e)}"

@tool
def transcribe_audio(file_url: str) -> str:
    """Downloads an audio file from a URL into a temporary file and transcribes it using SpeechRecognition."""
    try:
        # Download file into temporary file
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=True) as temp_file:
            response = requests.get(file_url, timeout=15)
            if response.status_code != 200:
                return f"Failed to download file: {response.status_code}"

            temp_file.write(response.content)
            temp_file.flush()  # Make sure data is written
            # Transcribe from temp file
            r = sr.Recognizer()
            with sr.AudioFile(temp_file.name) as source:
                audio = r.record(source)
                return r.recognize_google(audio)

    except Exception as e:
        return f"Transcription failed: {str(e)}"
        
@tool
def modulus(a: int, b: int) -> int:
    """Get the modulus of two numbers.
    
    Args:
        a: first int
        b: second int
    """
    return a % b
    
@tool
def wiki_search(query: str) -> str:
    """Search Wikipedia for a query and return maximum 5 results.
    Args:
        query: The search query."""
    search_docs = WikipediaLoader(query=query, load_max_docs=5).load()
    formatted_search_docs = "\n\n---\n\n".join(
        [
            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
            for doc in search_docs
        ]
    )
    return {"wiki_results": formatted_search_docs}

@tool
def web_search(query: str) -> str:
    """Search Tavily on the web for a query and return maximum 5 results.
    
    Args:
        query: The search query."""
    search_docs = TavilySearch(max_results=5).invoke({'query': query})

    print(f"Search query: {query}")
    print(f"Search results count: {len(search_docs)}")
    
    if isinstance(search_docs, dict):
        search_docs = search_docs.get('results', [])
    
    
    formatted_search_docs = "\n\n---\n\n".join(
        [
            f"""<Document source="{doc.get('url', '')}" page="{doc.get('title', '')}"/>\n{doc.get('content', '')}\n</Document>"""
            for doc in search_docs
        ])
    return {"web_results": formatted_search_docs}
    
#### add duckduckGoSearch into tool
tools = [
    divide,
    multiply,
    add,
    subtract,
    modulus,
    # read_image_text,
    # pdf_loader_tool,
    # docx_loader_tool,
    # excel_loader_tool,
    # txt_loader_tool,
    # youtube_transcript_tool,
    # transcribe_audio,
    # analyze_python_code,
    web_search,
    arvix_search,
    wiki_search,
    GoogleSearchAPIWrapper(k=5).run,
    # DuckDuckGoSearchResults(),
]
###### state and behavior
from typing import TypedDict, Annotated, Optional
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    # The input file download URL
    input_file: Optional[str]  
    
from langchain_core.messages import HumanMessage, SystemMessage

# from langchain_core.tools import tool # non necessary with toolNode

# @tool
# def multiply(a: int, b: int) -> int:
#     """Multiply a and b."""
#     return a * b

def assistant(state: AgentState, llm_with_tools):
    # System message

    file_info = ""
    # if state["input_file"]:
        # file_info = f"""\nIf the question refers to a file, it is available for download at this URL: {state['input_file']}. 
        # Use the appropriate loader tool (e.g., excel_loader_tool for Excel files, pdf_loader_tool for PDFs) to download and process it automatically. 
        # Do not ask the user to upload files."""
    sys_msg = SystemMessage(content=f"""
You are a helpful assistant tasked with answering questions using a set of tools.

Instructions:
1. Read the question carefully.
2. Use any available tools first.
3. If tools do not help, search online.
4. Use your own knowledge if all else fails.
5. Think step-by-step and explain your reasoning.
6. Extract the final answer from your reasoning and put it in the following format:

FINAL ANSWER: <your answer here>  ← use this line exactly

Rules for FINAL ANSWER:
- If it's a number, write the digits without commas or units unless specified.
- If it's a string, do not repeat it, do not include articles or abbreviations, write digits in words if requested. If it is a string of number, use the number first.
- If it's a list, separate items with commas, and follow the above rules per item.
- DO NOT include square brackets around the answer.

Finally, in a new line, ONLY print the FINAL ANSWER, nothing else.""")

    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])], "input_file": state["input_file"]}


def catch(state: AgentState, llm):
    # System message

    file_info = ""
    # if state["input_file"]:
        # file_info = f"""\nIf the question refers to a file, it is available for download at this URL: {state['input_file']}. 
        # Use the appropriate loader tool (e.g., excel_loader_tool for Excel files, pdf_loader_tool for PDFs) to download and process it automatically. 
        # Do not ask the user to upload files."""
    sys_msg = SystemMessage(content=f"""
You are a helpful assistant tasked with finding the final answer key word from pervious message.

Instructions:
Cut and output only the a single keyword or a keyword snipnet that representing the final answer from the previous message.""")
    return {"messages": [llm.invoke([sys_msg] + state["messages"])], "input_file": state["input_file"]}
    
################ state
from langgraph.graph import START,END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

class BasicAgent:
    def __init__(self):
        print("BasicAgent initialized.")
        api_key = os.environ["OPENAI_API_KEY"]
        if not api_key:
            raise ValueError("OPENAI_API_KEY environment variable not set or loaded.")
            
        if not os.environ["GOOGLE_API_KEY"]:
            raise ValueError("GOOGLE_API_KEY environment variable not set or loaded.")

        if not os.environ["GOOGLE_CSE_ID"]:
            raise ValueError("GOOGLE_CSE_ID environment variable not set or loaded.")
            
        # self.visionLLM = ChatOpenAI(model="gpt-4o",api_key=api_key) # multi-modal LLM
        self.RVLLM = ChatOpenAI(model="gpt-4o",api_key=api_key) # Review LLM
        self.LLM = ChatOpenAI(model="gpt-4.1",api_key=api_key) # manager LLM
        self.LLM_with_tools = self.LLM.bind_tools(tools, parallel_tool_calls=False)

        assistant_with_llm = lambda state:assistant(state, self.LLM_with_tools)
        assistant_catch = lambda state:catch(state, self.RVLLM)
        # Graph
        self.Builder = StateGraph(AgentState)
        self.Builder_catch = StateGraph(AgentState)
        # Define nodes: these do the work
        self.Builder.add_node("assistant", assistant_with_llm)
        self.Builder.add_node("tools",ToolNode(tools))

        self.Builder_catch.add_node("assistant_catch", assistant_catch)
        # Define edges: these determine how the control flow moves
        self.Builder.add_edge(START,"assistant")
        self.Builder_catch.add_edge(START, "assistant_catch")
        # if state says "tools":
        #     go to tools node
        # else:
        #     go to END node
        self.Builder.add_conditional_edges("assistant",tools_condition)
        self.Builder.add_edge("tools", "assistant")
        self.Builder_catch.add_edge("assistant_catch",END)
        self.agent = self.Builder.compile()
        self.agent_catch = self.Builder_catch.compile()
    def __call__(self, question: str, file_URL: str|None ) -> str:
        print(f"Agent received question (first 50 chars): {question[:50]}...")
        messages = [HumanMessage(content=f"{question}")]

        state = self.agent.invoke({"messages": messages,"input_file": file_URL},{"recursion_limit": 100})
        answer_raw = state["messages"][-1]
        answer_cut = self.agent_catch.invoke({"messages": answer_raw,"input_file": file_URL})
        answer = answer_cut["messages"][-1].content
        print(f"Agent returning answer: {answer}")
        return 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 = 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")
        file_url = f"{api_url}/files/{task_id}"
        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,file_URL = file_url)
            print("submitted_answer: " + submitted_answer)
            time.sleep(20)
            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)