Create app.py

app.py

@@ -0,0 +1,197 @@
+import os
+import uuid
+import logging
+from dotenv import load_dotenv
+import json
+import gradio as gr
+from langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_core.messages import AIMessage, HumanMessage
+from typing_extensions import TypedDict
+from typing import Annotated
+from langchain_core.messages import ToolMessage
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+from langchain_openai import ChatOpenAI as Chat
+
+from uuid import uuid4
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# Load environment variables
+load_dotenv()
+
+# LangGraph setup
+openai_api_key = os.getenv("OPENAI_API_KEY")
+model = os.getenv("OPENAI_MODEL", "gpt-4")
+temperature = float(os.getenv("OPENAI_TEMPERATURE", 0))
+
+web_search = TavilySearchResults(max_results=2)
+tools = [web_search]
+
+
+class State(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+graph_builder = StateGraph(State)
+
+
+llm = Chat(
+    openai_api_key=openai_api_key,
+    model=model,
+    temperature=temperature
+)
+llm_with_tools = llm.bind_tools(tools)
+
+
+def chatbot(state: State):
+    return {"messages": [llm_with_tools.invoke(state["messages"])]}
+
+
+graph_builder.add_node("chatbot", chatbot)
+
+
+class BasicToolNode:
+    """A node that runs the tools requested in the last AIMessage."""
+
+    def __init__(self, tools: list) -> None:
+        self.tools_by_name = {tool.name: tool for tool in tools}
+
+    def __call__(self, inputs: dict):
+        if messages := inputs.get("messages", []):
+            message = messages[-1]
+        else:
+            raise ValueError("No message found in input")
+        outputs = []
+        for tool_call in message.tool_calls:
+            tool_result = self.tools_by_name[tool_call["name"]].invoke(
+                tool_call["args"]
+            )
+            outputs.append(
+                ToolMessage(
+                    content=json.dumps(tool_result),
+                    name=tool_call["name"],
+                    tool_call_id=tool_call["id"],
+                )
+            )
+        return {"messages": outputs}
+
+
+def route_tools(
+    state: State,
+):
+    """
+    Use in the conditional_edge to route to the ToolNode if the last message
+    has tool calls. Otherwise, route to the end.
+    """
+    if isinstance(state, list):
+        ai_message = state[-1]
+    elif messages := state.get("messages", []):
+        ai_message = messages[-1]
+    else:
+        raise ValueError(
+            f"No messages found in input state to tool_edge: {state}")
+    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
+        return "tools"
+    return END
+
+
+tool_node = BasicToolNode(tools=[web_search])
+graph_builder.add_node("tools", tool_node)
+graph_builder.add_conditional_edges(
+    "chatbot",
+    route_tools,
+    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
+    # It defaults to the identity function, but if you
+    # want to use a node named something else apart from "tools",
+    # You can update the value of the dictionary to something else
+    # e.g., "tools": "my_tools"
+    {"tools": "tools", END: END},
+)
+# Any time a tool is called, we return to the chatbot to decide the next step
+graph_builder.add_edge("tools", "chatbot")
+graph_builder.add_edge(START, "chatbot")
+
+
+def chatbot(state: State):
+    if not state["messages"]:
+        logger.info(
+            "Received an empty message list. Returning default response.")
+        return {"messages": [AIMessage(content="Hello! How can I assist you today?")]}
+
+    # Check for tool call in the last message
+    last_message = state["messages"][-1]
+    if not getattr(last_message, "tool_calls", None):
+        logger.info(
+            "No tool call in the last message. Proceeding without tool invocation.")
+        response = llm.invoke(state["messages"])
+    else:
+        logger.info(
+            "Tool call detected in the last message. Invoking tool response.")
+        response = llm_with_tools.invoke(state["messages"])
+
+    # Ensure the response is wrapped as AIMessage if it's not already
+    if not isinstance(response, AIMessage):
+        response = AIMessage(content=response.content)
+
+    return {"messages": [response]}
+
+
+graph = graph_builder.compile()
+
+
+def gradio_chat(message, history):
+    try:
+        if not isinstance(message, str):
+            message = str(message)
+
+        config = {
+            "configurable": {"thread_id": "1"},
+            "checkpoint_id": str(uuid4()),
+            "recursion_limit": 300
+        }
+
+        # Format the user message correctly as a HumanMessage
+        formatted_message = [HumanMessage(content=message)]
+        response = graph.invoke(
+            {
+                "messages": formatted_message
+            },
+            config=config,
+            stream_mode="values"
+        )
+
+        # Extract assistant messages and ensure they are AIMessage type
+        assistant_messages = [
+            msg for msg in response["messages"] if isinstance(msg, AIMessage)
+        ]
+        last_message = assistant_messages[-1] if assistant_messages else AIMessage(
+            content="No response generated.")
+
+        logger.info("Sending response back to Gradio interface.")
+        return last_message.content
+    except Exception as e:
+        logger.error(f"Error encountered in gradio_chat: {e}")
+        return "Sorry, I encountered an error. Please try again."
+
+
+with gr.Blocks(theme=gr.themes.Default()) as demo:
+    chatbot = gr.ChatInterface(
+        chatbot=gr.Chatbot(height=800, render=False),
+        fn=gradio_chat,
+        multimodal=False,
+        title="LangGraph Agentic Chatbot",
+        examples=[
+            "What's the weather like today?",
+            "Show me the Movie Trailer for Doctor Strange.",
+            "Give me the latest news on the COVID-19 pandemic.",
+            "What are the latest updtaes on NVIDIA's new GPU?",
+        ],
+    )
+
+if __name__ == "__main__":
+    demo.launch()