Add Tools

agent.py

@@ -0,0 +1,281 @@
+import base64
+import os
+import re
+import tempfile
+from pathlib import Path
+from typing import TypedDict, Annotated, Optional
+import pandas as pd
+import requests
+from langchain_community.utilities.wikipedia import WikipediaAPIWrapper
+from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, SystemMessage
+from langchain_core.tools import tool
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_tavily import TavilySearch
+from langgraph.graph import START, StateGraph
+from langgraph.graph.message import add_messages
+from langgraph.prebuilt import ToolNode
+from langgraph.prebuilt import tools_condition
+from mediawikiapi import MediaWikiAPI
+from transformers import pipeline
+from wikipedia_tool import WikipediaTool
+
+@tool
+def read_xlsx_file(file_path: str) -> str:
+    """
+    Read a XLSX file using pandas and returns its content.
+
+    Args:
+        file_path: Path to the XLSX file
+
+    Returns:
+        Content of XLSX file as markdown or error message
+    """
+    try:
+        # Read the CSV file
+        df = pd.read_excel(file_path)
+        return df.to_markdown()
+
+    except ImportError:
+        return "Error: pandas is not installed. Please install it with 'pip install pandas'."
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+@tool
+def addition(a: int, b: int) -> int:
+    """
+        Add two int numbers.
+
+        Args:
+            a: int
+            b int
+
+        Returns:
+            a + b
+        """
+    return a + b
+
+@tool
+def multiple(a: int, b: int) -> float:
+    """
+        Multiple two float numbers.
+
+        Args:
+            a: int
+            b int
+
+        Returns:
+            a * b
+        """
+    return a * b
+
+class Agent:
+    def __init__(self):
+
+        llm = ChatGoogleGenerativeAI(
+            model="gemini-2.5-flash-preview-04-17",
+            # model="gemini-2.0-flash",
+            # model="gemini-1.5-pro",
+            temperature=0
+        )
+
+        self.tools = [
+            WikipediaTool(api_wrapper=WikipediaAPIWrapper(wiki_client=MediaWikiAPI())),
+            TavilySearch(),
+            read_xlsx_file,
+            addition,
+            multiple
+        ]
+
+        self.llm_with_tools = llm.bind_tools(self.tools)
+
+        self.graph = self.build_graph()
+
+    def build_graph(self):
+
+        class AgentState(TypedDict):
+            messages: Annotated[list[AnyMessage], add_messages]
+            task_id: str
+            file_name: Optional[str]
+
+        def assistant(state: AgentState):
+            try:
+                messages = state.get("messages")
+
+                # Invoke the LLM with tools
+                response = self.llm_with_tools.invoke(messages)
+
+                # Ensure we return the response in the correct format
+                return {
+                    "messages": [response]
+                }
+            except Exception as e:
+                # Create an error message if something goes wrong
+                error_msg = AIMessage(content=f"Sorry, I encountered an error: {str(e)}")
+                return {
+                    "messages": [error_msg]
+                }
+
+        def download_file_if_any(state: AgentState) -> str:
+            if state.get("file_name"):
+                return "download_file"
+            else:
+                return "assistant"
+
+        def download_file(state: AgentState):
+            filename = state.get("file_name")
+            task_id = state.get("task_id")
+            url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
+
+            try:
+                # Send a GET request to the URL
+                response = requests.get(url, stream=True)
+                # Ensure the request was successful
+                response.raise_for_status()
+
+                # Create a temporary file
+                temp_dir = tempfile.gettempdir()  # Get the temporary directory path
+                temp_file_path = os.path.join(temp_dir, os.path.basename(filename))
+
+                # Open a local file in binary write mode
+                with open(temp_file_path, 'wb') as file:
+                    # Write the content of the response to the file
+                    for chunk in response.iter_content(chunk_size=8192):
+                        file.write(chunk)
+
+                return {}
+
+            except requests.exceptions.RequestException as e:
+                error_msg = AIMessage(content=f"Sorry, I encountered an error: {str(e)}")
+                return {
+                    "messages": [error_msg]
+                }
+
+        def file_condition(state: AgentState) -> str:
+             filename = state.get("file_name")
+             suffix = Path(filename).suffix
+             if suffix in [".png", ".jpeg"]:
+                 return "add_image_message"
+             elif suffix in [".xlsx"]:
+                 return "add_xlsx_message"
+             elif suffix in [".mp3"]:
+                 return "add_audio_message"
+             elif suffix in [".py"]:
+                 return "add_py_message"
+             else:
+                 return "assistant"
+
+
+        def add_image_message(state: AgentState):
+            filename = state.get("file_name")
+            temp_dir = tempfile.gettempdir()  # Get the temporary directory path
+            image_path = os.path.join(temp_dir, os.path.basename(filename))
+            # Load the image and convert it to base64
+            with open(image_path, "rb") as img_file:
+                base64_image = base64.b64encode(img_file.read()).decode("utf-8")
+
+            # Construct the image message
+            image_message = HumanMessage(content=[{
+                "type": "image_url",
+                "image_url": {
+                    "url": f"data:image/jpeg;base64,{base64_image}"
+                }
+            }])
+
+            return { "messages" : state.get("messages") + [image_message] }
+
+        def add_xlsx_message(state: AgentState):
+            filename = state.get("file_name")
+            temp_dir = tempfile.gettempdir()  # Get the temporary directory path
+            xlsx_path = os.path.join(temp_dir, os.path.basename(filename))
+
+            # Construct the message
+            xlsx_message = HumanMessage(content=f"xlsx file is at {xlsx_path}")
+
+            return { "messages" : state.get("messages") + [xlsx_message] }
+
+        def add_audio_message(state: AgentState):
+            filename = state.get("file_name")
+            temp_dir = tempfile.gettempdir()  # Get the temporary directory path
+            audio_path = os.path.join(temp_dir, os.path.basename(filename))
+
+            pipe = pipeline(
+                task="automatic-speech-recognition",
+                model="openai/whisper-large-v3"
+            )
+
+            result = pipe(audio_path)
+
+            audio_message = HumanMessage(result["text"])
+
+            return {"messages": state.get("messages") + [audio_message]}
+
+        def add_py_message(state: AgentState):
+            filename = state.get("file_name")
+            temp_dir = tempfile.gettempdir()  # Get the temporary directory path
+            file_path = os.path.join(temp_dir, os.path.basename(filename))
+
+            with open(file_path, 'r') as file:
+                content = file.read()
+
+            py_message = HumanMessage(content=[{
+                "type": "text",
+                "text": content
+            }])
+            return {"messages": state.get("messages") + [py_message]}
+
+        ## The graph
+        builder = StateGraph(AgentState)
+
+        # Define nodes: these do the work
+        builder.add_node("assistant", assistant)
+        builder.add_node("tools", ToolNode(self.tools))
+        builder.add_node("download_file", download_file)
+        builder.add_node("add_image_message", add_image_message)
+        builder.add_node("add_xlsx_message", add_xlsx_message)
+        builder.add_node("add_py_message", add_py_message)
+        builder.add_node("add_audio_message", add_audio_message)
+
+        # Define edges: these determine how the control flow moves
+        builder.add_conditional_edges(
+            START,
+            download_file_if_any
+        )
+        # builder.add_edge("download_file", "assistant")
+        builder.add_conditional_edges(
+            "download_file",
+            file_condition
+        )
+        builder.add_edge("add_image_message", "assistant")
+        builder.add_edge("add_xlsx_message", "assistant")
+        builder.add_edge("add_py_message", "assistant")
+        builder.add_edge("add_audio_message", "assistant")
+        builder.add_conditional_edges(
+            "assistant",
+            # If the latest message requires a tool, route to tools
+            # Otherwise, provide a direct response
+            tools_condition
+        )
+        builder.add_edge("tools", "assistant")
+        return builder.compile()
+
+    def run(self, question: str, task_id: str, file_name: str | None):
+        system_prompt = SystemMessage(content="You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, use digit not letter, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.")
+
+        messages = [system_prompt, HumanMessage(content=question)]
+
+        response = self.graph.invoke({"messages": messages, "task_id": task_id, "file_name": file_name}, debug=True)
+
+        answer = response['messages'][-1].content
+
+        for m in response['messages']:
+            m.pretty_print()
+
+        # Regex to capture text after "FINAL ANSWER: "
+        match = re.search(r'FINAL ANSWER:\s*(.*)', answer)
+
+        if match:
+            final_answer = match.group(1)
+            print(final_answer)
+            return final_answer
+
+        return answer

app.py

@@ -1,8 +1,9 @@
 import os
+import time
 import gradio as gr
 import requests
-import inspect
 import pandas as pd
+from agent import Agent
 
 # (Keep Constants as is)
 # --- Constants ---
@@ -12,12 +13,29 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
     def __init__(self):
-        print("BasicAgent initialized.")
-    def __call__(self, question: str) -> str:
+        # Initialize Agent
+        self.agent = Agent()
+
+        print("Agent initialized successfully")
+    def __call__(self, question: str, task_id: str, file_name: str | None = None) -> str:
         print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = "This is a default answer."
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
+
+        max_retries = 10
+        base_sleep = 60
+
+        for attempt in range(max_retries):
+            try:
+                final_answer = self.agent.run(question=question, task_id=task_id, file_name=file_name)
+                print(f"Agent returning final answer: {final_answer}")
+                return final_answer
+            except Exception as e:
+                print(f"{str(e)}")
+                sleep_time = base_sleep * (attempt + 1)  # Incremental sleep: 1s, 2s, 3s
+                if attempt < max_retries - 1:
+                    print(f"Attempt {attempt + 1} failed. Retrying in {sleep_time} seconds...")
+                    time.sleep(sleep_time)
+                    continue
+                return f"Error processing query after {max_retries} attempts: {str(e)}"
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -44,7 +62,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     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)
+    # In the case of an app running as a hugging Face space, this link points toward your codebase (useful for others so please keep it public)
     agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
     print(agent_code)
 
@@ -54,6 +72,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
         response = requests.get(questions_url, timeout=15)
         response.raise_for_status()
         questions_data = response.json()
+        print(f"{questions_data}")
         if not questions_data:
              print("Fetched questions list is empty.")
              return "Fetched questions list is empty or invalid format.", None
@@ -76,11 +95,12 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     for item in questions_data:
         task_id = item.get("task_id")
         question_text = item.get("question")
+        file_name = item.get("file_name")
         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)
+            submitted_answer = agent(question=question_text, task_id=task_id, file_name=file_name)
             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:
@@ -171,6 +191,8 @@ with gr.Blocks() as demo:
         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
@@ -193,4 +215,4 @@ if __name__ == "__main__":
     print("-"*(60 + len(" App Starting ")) + "\n")
 
     print("Launching Gradio Interface for Basic Agent Evaluation...")
-    demo.launch(debug=True, share=False)
+    demo.launch(debug=True, share=False)

requirements.txt

@@ -1,2 +1,19 @@
-gradio
-requests
+gradio~=5.33.2
+requests~=2.32.4
+itsdangerous
+langchain~=0.3.24
+langgraph~=0.3.34
+pandas~=2.2.3
+langchain-core~=0.3.56
+langchain-google-genai~=2.1.3
+langchain-community~=0.3.22
+langchain-tavily
+mediawikiapi~=1.3
+wikipedia
+pydantic~=2.11.3
+beautifulsoup4~=4.13.4
+openpyxl
+protobuf~=5.29.4
+genai~=2.1.0
+transformers~=4.52.4
+torch

wikipedia_tool.py

@@ -0,0 +1,52 @@
+"""Tool for the Wikipedia API."""
+
+from typing import Optional, Type
+from langchain_core.callbacks import CallbackManagerForToolRun
+from langchain_core.tools import BaseTool
+from pydantic import BaseModel, Field
+import pandas as pd
+from langchain_community.utilities.wikipedia import WikipediaAPIWrapper
+
+class WikipediaQueryInput(BaseModel):
+    """Input for the WikipediaQuery tool."""
+
+    query: str = Field(description="query to look up on wikipedia")
+
+class WikipediaTool(BaseTool):  # type: ignore[override, override]
+    """Tool that searches the Wikipedia API."""
+
+    name: str = "wikipedia"
+    description: str = (
+        "A wrapper around Wikipedia. "
+        "Useful for when you need to answer general questions about "
+        "people, places, companies, facts, historical events, or other subjects. "
+        "Input should be a search query."
+    )
+    api_wrapper: WikipediaAPIWrapper
+
+    args_schema: Type[BaseModel] = WikipediaQueryInput
+
+    def _run(
+        self,
+        query: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        """Use the Wikipedia tool."""
+        pages = self.api_wrapper.load(query)
+
+        for page in pages:
+            try:
+                wikitables = pd.read_html(page.metadata["source"], attrs={"class": "wikitable"})
+                page.metadata["wikitable"] = "\n---\n".join(
+                    f'{table}'
+                    for table in wikitables
+                )
+            except:
+                continue
+
+        res = "\n---\n".join(
+            f'{page}'
+            for page in pages
+        )
+
+        return res