Update app.py

app.py

@@ -1,64 +1,299 @@
+#pip install langchain_google_genai langgraph gradio
+import os
+import typing
+from typing import Annotated, Literal, Iterable
+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, BaseMessage, SystemMessage
+from random import randint
+
+from tkinter import messagebox
+#messagebox.showinfo("Test", "Script run successfully")
+
 import gradio as gr
-from huggingface_hub import InferenceClient
-
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
-
-
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
-
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
-
-        response += token
-        yield response
-
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
+import logging
+
+class OrderState(TypedDict):
+    """State representing the customer's order conversation."""
+    messages: Annotated[list, add_messages]
+    order: list[str]
+    finished: bool
+
+# System instruction for the BaristaBot
+BARISTABOT_SYSINT = (
+    "system",
+    "You are a BaristaBot, an interactive cafe ordering system. A human will talk to you about the "
+    "available products. Answer questions about menu items, help customers place orders, and "
+    "confirm details before finalizing. Use the provided tools to manage the order."
 )
 
+WELCOME_MSG = "Welcome to the BaristaBot cafe. Type `q` to quit. How may I serve you today?"
+
+# Initialize the Google Gemini LLM
+llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash-latest")
+
+@tool
+def get_menu() -> str:
+    """Provide the cafe menu."""
+    #messagebox.showinfo("Test", "Script run successfully")
+    with open("menu.txt", 'r', encoding = "UTF-8") as f:
+        return f.read()
+
+@tool
+def add_to_order(drink: str, modifiers: Iterable[str] = []) -> str:
+    """Adds the specified drink to the customer's order."""
+    return f"{drink} ({', '.join(modifiers) if modifiers else 'no modifiers'})"
+
+@tool
+def confirm_order() -> str:
+    """Asks the customer to confirm the order."""
+    return "Order confirmation requested"
+
+@tool
+def get_order() -> str:
+    """Returns the current order."""
+    return "Current order details requested"
+
+@tool
+def clear_order() -> str:
+    """Clears the current order."""
+    return "Order cleared"
+
+@tool
+def place_order() -> int:
+    """Sends the order to the kitchen."""
+    messagebox.showinfo("Test", "Order successful!")
+    return randint(2, 10)  # Estimated wait time
+
+def chatbot_with_tools(state: OrderState) -> OrderState:
+    """Chatbot with tool handling."""
+    logging.info(f"Messagelist sent to chatbot node: {[msg.content for msg in state.get('messages', [])]}")
+    defaults = {"order": [], "finished": False}
+
+    # Ensure we always have at least a system message
+    if not state.get("messages", []):
+        new_output = AIMessage(content=WELCOME_MSG)
+        return defaults | state | {"messages": [SystemMessage(content=BARISTABOT_SYSINT), new_output]}
+
+    try:
+        # Prepend system instruction if not already present
+        messages_with_system = [
+            SystemMessage(content=BARISTABOT_SYSINT)
+        ] + state.get("messages", [])
+
+        # Process messages through the LLM
+        new_output = llm_with_tools.invoke(messages_with_system)
+
+        return defaults | state | {"messages": [new_output]}
+    except Exception as e:
+        # Fallback if LLM processing fails
+        return defaults | state | {"messages": [AIMessage(content=f"I'm having trouble processing that. {str(e)}")]}
+
+def order_node(state: OrderState) -> OrderState:
+    """Handles order-related tool calls."""
+    logging.info("order node")
+    tool_msg = state.get("messages", [])[-1]
+    order = state.get("order", [])
+    outbound_msgs = []
+    order_placed = False
+
+    for tool_call in tool_msg.tool_calls:
+        tool_name = tool_call["name"]
+        tool_args = tool_call["args"]
+
+        if tool_name == "add_to_order":
+            modifiers = tool_args.get("modifiers", [])
+            modifier_str = ", ".join(modifiers) if modifiers else "no modifiers"
+            order.append(f'{tool_args["drink"]} ({modifier_str})')
+            response = "\n".join(order)
+
+        elif tool_name == "confirm_order":
+            response = "Your current order:\n" + "\n".join(order) + "\nIs this correct?"
+
+        elif tool_name == "get_order":
+            response = "\n".join(order) if order else "(no order)"
+
+        elif tool_name == "clear_order":
+            order.clear()
+            response = "Order cleared"
+
+        elif tool_name == "place_order":
+            order_text = "\n".join(order)
+            order_placed = True
+            response = f"Order placed successfully!\nYour order:\n{order_text}\nEstimated wait: {randint(2, 10)} minutes"
+
+        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": order, "finished": order_placed}
+
+def maybe_route_to_tools(state: OrderState) -> str:
+    """Route between chat and tool nodes."""
+    if not (msgs := state.get("messages", [])):
+        raise ValueError(f"No messages found when parsing state: {state}")
+
+    msg = msgs[-1]
+
+    if state.get("finished", False):
+        logging.info("from chatbot GOTO End node")
+        return END
+
+    elif hasattr(msg, "tool_calls") and len(msg.tool_calls) > 0:
+        if any(tool["name"] in tool_node.tools_by_name.keys() for tool in msg.tool_calls):
+            logging.info("from chatbot GOTO tools node")
+            return "tools"
+        else:
+            logging.info("from chatbot GOTO order node")
+            return "ordering"
+
+    else:
+        logging.info("from chatbot GOTO human node")
+        return "human"
+
+def human_node(state: OrderState) -> OrderState:
+    """Handle user input."""
+    logging.info(f"Messagelist sent to human node: {[msg.content for msg in state.get('messages', [])]}")
+    last_msg = state["messages"][-1]
+
+    if last_msg.content.lower() in {"q", "quit", "exit", "goodbye"}:
+        state["finished"] = True
+
+    return state
+
+def maybe_exit_human_node(state: OrderState) -> Literal["chatbot", "__end__"]:
+    """Determine if conversation should continue."""
+    if state.get("finished", False):
+        logging.info("from human GOTO End node")
+        return END
+    last_msg = state["messages"][-1]
+    if isinstance(last_msg, AIMessage):
+        logging.info("Chatbot response obtained, ending conversation")
+        return END
+    else:
+        logging.info("from human GOTO chatbot node")
+        return "chatbot"
+
+# Prepare tools
+auto_tools = [get_menu]
+tool_node = ToolNode(auto_tools)
+
+order_tools = [add_to_order, confirm_order, get_order, clear_order, place_order]
+
+# Bind all tools to the LLM
+llm_with_tools = llm.bind_tools(auto_tools + order_tools)
+
+# Build the graph
+graph_builder = StateGraph(OrderState)
+
+# Add nodes
+graph_builder.add_node("chatbot", chatbot_with_tools)
+graph_builder.add_node("human", human_node)
+graph_builder.add_node("tools", tool_node)
+graph_builder.add_node("ordering", order_node)
+
+# Add edges and routing
+graph_builder.add_conditional_edges("chatbot", maybe_route_to_tools)
+graph_builder.add_conditional_edges("human", maybe_exit_human_node)
+graph_builder.add_edge("tools", "chatbot")
+graph_builder.add_edge("ordering", "chatbot")
+graph_builder.add_edge(START, "human")
+
+# Compile the graph
+chat_graph = graph_builder.compile()
+
+def convert_history_to_messages(history: list) -> list[BaseMessage]:
+    """
+    Convert Gradio chat history to a list of Langchain messages.
+    
+    Args:
+    - history: Gradio's chat history format
+    
+    Returns:
+    - List of Langchain BaseMessage objects
+    """
+    messages = []
+    for human, ai in history:
+        if human:
+            messages.append(HumanMessage(content=human))
+        if ai:
+            messages.append(AIMessage(content=ai))
+    return messages
+
+def gradio_chat(message: str, history: list) -> str:
+    """
+    Gradio-compatible chat function that manages the conversation state.
+    
+    Args:
+    - message: User's input message
+    - history: Gradio's chat history
+    
+    Returns:
+    - Bot's response as a string
+    """
+    logging.info(f"{len(history)} history so far: {history}")
+    # Ensure non-empty message
+    if not message or message.strip() == "":
+        message = "Hello, how can I help you today?"
+
+    # Convert history to Langchain messages
+    conversation_messages = []
+    for old_message in history:
+        if old_message["content"].strip():
+            if old_message["role"] == "user":
+                conversation_messages.append(HumanMessage(content=old_message["content"]))
+            if old_message["role"] == "assistant":
+                conversation_messages.append(AIMessage(content=old_message["content"]))
+    
+    # Add current message
+    conversation_messages.append(HumanMessage(content=message))
+    
+    # Create initial state with conversation history
+    conversation_state = {
+        "messages": conversation_messages, 
+        "order": [], 
+        "finished": False
+    }
+    logging.info(f"Conversation so far: {str(conversation_state)}")
+    try:
+        # Process the conversation through the graph
+        conversation_state = chat_graph.invoke(conversation_state, {"recursion_limit": 10})
+        
+        # Extract the latest bot message
+        latest_message = conversation_state["messages"][-1]
+        
+        # Return the bot's response content
+        logging.info(f"return: {latest_message.content}")
+        return latest_message.content
+    
+    except Exception as e:
+        return f"An error occurred: {str(e)}"
+
+# Gradio interface
+def launch_baristabot():
+    gr.ChatInterface(
+        gradio_chat,
+        type="messages",
+        title="BaristaBot",
+        description="Your friendly AI cafe assistant",
+        theme="ocean"
+    ).launch()
 
 if __name__ == "__main__":
-    demo.launch()
+    # initiate logging tool
+    logging.basicConfig(filename='log.log',
+                    level=logging.INFO,
+                    format='%(asctime)s - %(levelname)s - %(message)s')
+    launch_baristabot()