Upload GeneralAgent.py

GeneralAgent.py

@@ -0,0 +1,223 @@
+import os
+import requests
+import pandas as pd
+
+from typing import Annotated
+from typing_extensions import TypedDict
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+from langgraph.prebuilt import ToolNode
+from langchain_core.tools import tool
+from langchain_core.messages import AIMessage, ToolMessage, HumanMessage, SystemMessage
+
+from smolagents import DuckDuckGoSearchTool
+import requests
+from bs4 import BeautifulSoup
+import wikipedia
+import pandas as pd
+
+# (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 ------
+class OrderState(TypedDict):
+    """State representing the customer's order conversation."""
+    messages: Annotated[list, add_messages]
+    order: list[str]
+    finished: bool
+
+# System instruction for the Agent
+SYSINT = (
+    "system",
+    "You are a general AI assistant. I will ask you a question."
+    "The question requires a tool to solve. You must attempt to use at least one of the available tools before returning an answer."
+    "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, 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."
+    "If a tool required for task completion is not functioning, return 0."
+    )
+
+WELCOME_MSG = "Welcome to my general-purpose AI agent. Type `q` to quit. How shall I fail to serve you today?"
+
+@tool
+def wikipedia_search_tool(title: str) -> str:
+    """Provides an excerpt from a Wikipedia article with the given title."""
+    page = wikipedia.page(title, auto_suggest=False)
+    return page.content[:3000]
+
+@tool
+def media_tool(file_path: str) -> str:
+    """Used for deciphering video and audio files."""
+    return "This tool hasn't been implemented yet. Please return 0 if the task cannot be solved without knowing the contents of this file."
+
+@tool
+def internet_search_tool(search_query: str) -> str:
+    """Does a google search with using the input as the search query. Returns a long batch of textual information related to the query."""
+    search_tool = DuckDuckGoSearchTool()
+    result = search_tool(search_query)
+    return result
+
+@tool
+def webscraper_tool(url: str) -> str:
+    """Returns the page's html content from the input url."""
+    response = requests.get(url, stream=True)
+    if response.status_code == 200:
+        soup = BeautifulSoup(response.content, 'html.parser')
+        html_text = soup.get_text()
+        return html_text
+    else:
+        raise Exception(f"Failed to retrieve the webpage. Status code: {response.status_code}")
+
+@tool
+def read_excel_tool(file_path: str) -> str:
+    """Returns the contents of an Excel file as a Pandas dataframe."""
+    df = pd.read_excel(file_path, engine = "openpyxl")
+    return df.to_string(index=False)
+
+class AgenticAI:
+    def __init__(self):
+        # initialize LLM
+        self.llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
+        # prepare tool list
+        self.tools = [
+            wikipedia_search_tool,
+            media_tool,
+            internet_search_tool,
+            webscraper_tool,
+            read_excel_tool,
+        ]
+        # bind tools
+        self.llm_with_tools = self.llm.bind_tools(self.tools)
+        # standalone ToolNode for any non-interactive tools (none here)
+        self.tool_node = ToolNode([])
+        # build state graph
+        self.graph = StateGraph(OrderState)
+        self.graph.add_node("agent", self._agent_node)
+        self.graph.add_node("interactive_tools", self._interactive_tools_node)
+        self.graph.add_node("human", self._human_node)
+        # routing
+        self.graph.add_conditional_edges("agent", self._maybe_route_to_tools)
+        self.graph.add_conditional_edges("human", self._maybe_exit_human_node)
+        self.graph.add_edge("interactive_tools", "agent")
+        self.graph.add_edge(START, "human")
+        self.chat_graph = self.graph.compile()
+
+    def ask(self, human_input: str) -> str:
+        """
+        Take a single human input, run through the full agent+tool graph,
+        return the AI's reply, and discard any stored human/chat history.
+        """
+        # build initial messages
+        init_msgs = [
+            SystemMessage(content=SYSINT),
+            HumanMessage(content=human_input)
+        ]
+        state = {"messages": init_msgs, "order": [], "finished": False}
+        try:
+            final_state = self.chat_graph.invoke(state, {"recursion_limit": 10})
+            # last message should be from the AI
+            ai_msg = final_state["messages"][-1]
+            return ai_msg.content
+        except Exception as e:
+            return f"Error during processing: {e}"
+
+    # --- internal node functions (mirror your original code) ---
+    def _agent_node(self, state: OrderState) -> OrderState:
+        print(f"Messagelist sent to agent node: {[msg.content for msg in state.get('messages', [])]}")
+        defaults = {"order": [], "finished": False}
+        msgs = state.get("messages", [])
+        if not msgs:
+            # no prior messages: seed with system + empty AI message
+            return {**defaults, "messages": [SystemMessage(SYSINT), AIMessage(content="")]}
+
+        try:
+            # always ensure system prompt is first
+            msgs = [SystemMessage(SYSINT)] + msgs
+            new_output = self.llm_with_tools.invoke(msgs)
+            return {**defaults, "messages": [new_output]}
+        except Exception as e:
+            return {**defaults, "messages": [AIMessage(content=f"I'm having trouble: {e}")]}
+
+    def _interactive_tools_node(self, state: OrderState) -> OrderState:
+        tool_msg = state["messages"][-1]
+        outbound_msgs = []
+        for tool_call in tool_msg.tool_calls:
+            tool_name = tool_call["name"]
+            tool_args = tool_call["args"]
+
+            if tool_name == "wikipedia_search_tool":
+                print(f"called wikipedia with {str(tool_args)}")
+                page = wikipedia.page(tool_args.get("title"), auto_suggest=False)
+                response = page.content[:3000]
+            elif tool_name == "media_tool":
+                print(f"called media with {str(tool_args)}")
+                response = "This tool hasn't been implemented yet. Please return 0 if the task cannot be solved without knowing the contents of this file."
+            elif tool_name == "internet_search_tool":
+                print(f"called internet with {str(tool_args)}")
+                question = tool_args.get("search_query")
+                search_tool = DuckDuckGoSearchTool()
+                response = search_tool(question)[:3000]
+            elif tool_name == "webscraper_tool":
+                print(f"called webscraper with {str(tool_args)}")
+                url = tool_args.get("url")
+                response = requests.get(url, stream=True)
+                if response.status_code == 200:
+                    soup = BeautifulSoup(response.content, 'html.parser')
+                    html_text = soup.get_text()
+                    response = html_text
+                else:
+                    response = Exception(f"Failed to retrieve the webpage. Status code: {response.status_code}")
+            elif tool_name == "read_excel_tool":
+                print(f"called excel with {str(tool_args)}")
+                path = tool_args.get("file_path")
+                df = pd.read_excel(path, engine = "openpyxl")
+                response = df
+
+            else:
+                raise NotImplementedError(f'Unknown tool call: {tool_name}')
+            
+            outbound_msgs.append(
+            ToolMessage(
+                content=response,
+                name=tool_name,
+                tool_call_id=tool_call["id"],
+            )
+        )
+
+        return {"messages": outbound_msgs, "order": state.get("order", []), "finished": False}
+
+    def _human_node(self, state: OrderState) -> OrderState:
+        print(f"Messagelist sent to human node: {[msg.content for msg in state.get('messages', [])]}")
+        last = state["messages"][-1]
+        if isinstance(last, HumanMessage) and last.content.strip().lower() in {"q", "quit", "exit", "goodbye"}:
+            state["finished"] = True
+        return state
+
+    def _maybe_route_to_tools(self, state: OrderState) -> str:
+        msgs = state.get("messages", [])
+        if state.get("finished"):
+            print("from agent GOTO End node")
+            return END
+
+        last = msgs[-1]
+        if hasattr(last, "tool_calls") and last.tool_calls:
+            print("from agent GOTO tools node")
+            # go run interactive tools
+            return "interactive_tools"
+        # else, end conversation
+        print("tool call failed, quitting")
+        return END
+
+    def _maybe_exit_human_node(self, state: OrderState) -> str:
+        if state.get("finished"):
+            print("from human GOTO End node")
+            return END
+        last = state["messages"][-1]
+        # if AIMessage then end after one turn
+        print("from human GOTO agent node or quit")
+        return END if isinstance(last, AIMessage) else "agent"