APRG/BaristaBot
0
1#pip install langchain_google_genai langgraph gradio2import os3import sys4import typing5from typing import Annotated, Literal, Iterable6from typing_extensions import TypedDict7 8from langchain_google_genai import ChatGoogleGenerativeAI9from langgraph.graph import StateGraph, START, END10from langgraph.graph.message import add_messages11from langgraph.prebuilt import ToolNode12from langchain_core.tools import tool13from langchain_core.messages import AIMessage, ToolMessage, HumanMessage, BaseMessage, SystemMessage14from random import randint15 16from tkinter import messagebox17#messagebox.showinfo("Test", "Script run successfully")18 19import gradio as gr20import logging21 22class OrderState(TypedDict):23 """State representing the customer's order conversation."""24 messages: Annotated[list, add_messages]25 order: list[str]26 finished: bool27 28# System instruction for the BaristaBot29BARISTABOT_SYSINT = (30 "system",31 "You are a BaristaBot, an interactive cafe ordering system. A human will talk to you about the "32 "available products. Answer questions about menu items, help customers place orders, and "33 "confirm details before finalizing. Use the provided tools to manage the order."34)35 36WELCOME_MSG = "Welcome to the BaristaBot cafe. Type `q` to quit. How may I serve you today?"37 38# Initialize the Google Gemini LLM39llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash-latest")40 41@tool42def get_menu() -> str:43 """Provide the cafe menu."""44 #messagebox.showinfo("Test", "Script run successfully")45 with open("menu.txt", 'r', encoding = "UTF-8") as f:46 return f.read()47 48@tool49def add_to_order(drink: str, modifiers: Iterable[str] = []) -> str:50 """Adds the specified drink to the customer's order."""51 return f"{drink} ({', '.join(modifiers) if modifiers else 'no modifiers'})"52 53@tool54def confirm_order() -> str:55 """Asks the customer to confirm the order."""56 return "Order confirmation requested"57 58@tool59def get_order() -> str:60 """Returns the current order."""61 return "Current order details requested"62 63@tool64def clear_order() -> str:65 """Clears the current order."""66 return "Order cleared"67 68@tool69def place_order() -> int:70 """Sends the order to the kitchen."""71 #messagebox.showinfo("Test", "Order successful!")72 return randint(2, 10) # Estimated wait time73 74def chatbot_with_tools(state: OrderState) -> OrderState:75 """Chatbot with tool handling."""76 logging.info(f"Messagelist sent to chatbot node: {[msg.content for msg in state.get('messages', [])]}")77 defaults = {"order": [], "finished": False}78 79 # Ensure we always have at least a system message80 if not state.get("messages", []):81 new_output = AIMessage(content=WELCOME_MSG)82 return defaults | state | {"messages": [SystemMessage(content=BARISTABOT_SYSINT), new_output]}83 84 try:85 # Prepend system instruction if not already present86 messages_with_system = [87 SystemMessage(content=BARISTABOT_SYSINT)88 ] + state.get("messages", [])89 90 # Process messages through the LLM91 new_output = llm_with_tools.invoke(messages_with_system)92 93 return defaults | state | {"messages": [new_output]}94 except Exception as e:95 # Fallback if LLM processing fails96 return defaults | state | {"messages": [AIMessage(content=f"I'm having trouble processing that. {str(e)}")]}97 98def order_node(state: OrderState) -> OrderState:99 """Handles order-related tool calls."""100 logging.info("order node")101 tool_msg = state.get("messages", [])[-1]102 order = state.get("order", [])103 outbound_msgs = []104 order_placed = False105 106 for tool_call in tool_msg.tool_calls:107 tool_name = tool_call["name"]108 tool_args = tool_call["args"]109 110 if tool_name == "add_to_order":111 modifiers = tool_args.get("modifiers", [])112 modifier_str = ", ".join(modifiers) if modifiers else "no modifiers"113 order.append(f'{tool_args["drink"]} ({modifier_str})')114 response = "\n".join(order)115 116 elif tool_name == "confirm_order":117 response = "Your current order:\n" + "\n".join(order) + "\nIs this correct?"118 119 elif tool_name == "get_order":120 response = "\n".join(order) if order else "(no order)"121 122 elif tool_name == "clear_order":123 order.clear()124 response = "Order cleared"125 126 elif tool_name == "place_order":127 order_text = "\n".join(order)128 order_placed = True129 response = f"Order placed successfully!\nYour order:\n{order_text}\nEstimated wait: {randint(2, 10)} minutes"130 131 else:132 raise NotImplementedError(f'Unknown tool call: {tool_name}')133 134 outbound_msgs.append(135 ToolMessage(136 content=response,137 name=tool_name,138 tool_call_id=tool_call["id"],139 )140 )141 142 return {"messages": outbound_msgs, "order": order, "finished": order_placed}143 144def maybe_route_to_tools(state: OrderState) -> str:145 """Route between chat and tool nodes."""146 if not (msgs := state.get("messages", [])):147 raise ValueError(f"No messages found when parsing state: {state}")148 149 msg = msgs[-1]150 151 if state.get("finished", False):152 logging.info("from chatbot GOTO End node")153 return END154 155 elif hasattr(msg, "tool_calls") and len(msg.tool_calls) > 0:156 if any(tool["name"] in tool_node.tools_by_name.keys() for tool in msg.tool_calls):157 logging.info("from chatbot GOTO tools node")158 return "tools"159 else:160 logging.info("from chatbot GOTO order node")161 return "ordering"162 163 else:164 logging.info("from chatbot GOTO human node")165 return "human"166 167def human_node(state: OrderState) -> OrderState:168 """Handle user input."""169 logging.info(f"Messagelist sent to human node: {[msg.content for msg in state.get('messages', [])]}")170 last_msg = state["messages"][-1]171 172 if last_msg.content.lower() in {"q", "quit", "exit", "goodbye"}:173 state["finished"] = True174 175 return state176 177def maybe_exit_human_node(state: OrderState) -> Literal["chatbot", "__end__"]:178 """Determine if conversation should continue."""179 if state.get("finished", False):180 logging.info("from human GOTO End node")181 return END182 last_msg = state["messages"][-1]183 if isinstance(last_msg, AIMessage):184 logging.info("Chatbot response obtained, ending conversation")185 return END186 else:187 logging.info("from human GOTO chatbot node")188 return "chatbot"189 190# Prepare tools191auto_tools = [get_menu]192tool_node = ToolNode(auto_tools)193 194order_tools = [add_to_order, confirm_order, get_order, clear_order, place_order]195 196# Bind all tools to the LLM197llm_with_tools = llm.bind_tools(auto_tools + order_tools)198 199# Build the graph200graph_builder = StateGraph(OrderState)201 202# Add nodes203graph_builder.add_node("chatbot", chatbot_with_tools)204graph_builder.add_node("human", human_node)205graph_builder.add_node("tools", tool_node)206graph_builder.add_node("ordering", order_node)207 208# Add edges and routing209graph_builder.add_conditional_edges("chatbot", maybe_route_to_tools)210graph_builder.add_conditional_edges("human", maybe_exit_human_node)211graph_builder.add_edge("tools", "chatbot")212graph_builder.add_edge("ordering", "chatbot")213graph_builder.add_edge(START, "human")214 215# Compile the graph216chat_graph = graph_builder.compile()217 218def convert_history_to_messages(history: list) -> list[BaseMessage]:219 """220 Convert Gradio chat history to a list of Langchain messages.221 222 Args:223 - history: Gradio's chat history format224 225 Returns:226 - List of Langchain BaseMessage objects227 """228 messages = []229 for human, ai in history:230 if human:231 messages.append(HumanMessage(content=human))232 if ai:233 messages.append(AIMessage(content=ai))234 return messages235 236def gradio_chat(message: str, history: list) -> str:237 """238 Gradio-compatible chat function that manages the conversation state.239 240 Args:241 - message: User's input message242 - history: Gradio's chat history243 244 Returns:245 - Bot's response as a string246 """247 logging.info(f"{len(history)} history so far: {history}")248 # Ensure non-empty message249 if not message or message.strip() == "":250 message = "Hello, how can I help you today?"251 252 # Convert history to Langchain messages253 conversation_messages = []254 for old_message in history:255 if old_message["content"].strip():256 if old_message["role"] == "user":257 conversation_messages.append(HumanMessage(content=old_message["content"]))258 if old_message["role"] == "assistant":259 conversation_messages.append(AIMessage(content=old_message["content"]))260 261 # Add current message262 conversation_messages.append(HumanMessage(content=message))263 264 # Create initial state with conversation history265 conversation_state = {266 "messages": conversation_messages, 267 "order": [], 268 "finished": False269 }270 logging.info(f"Conversation so far: {str(conversation_state)}")271 try:272 # Process the conversation through the graph273 conversation_state = chat_graph.invoke(conversation_state, {"recursion_limit": 10})274 275 # Extract the latest bot message276 latest_message = conversation_state["messages"][-1]277 278 # Return the bot's response content279 logging.info(f"return: {latest_message.content}")280 return latest_message.content281 282 except Exception as e:283 return f"An error occurred: {str(e)}"284 285# Gradio interface286def launch_baristabot():287 gr.ChatInterface(288 gradio_chat,289 type="messages",290 title="BaristaBot",291 description="Your friendly AI cafe assistant",292 theme="ocean"293 ).launch()294 295if __name__ == "__main__":296 # initiate logging tool297 logging.basicConfig(298 stream=sys.stdout,299 level=logging.INFO,300 format='%(asctime)s - %(levelname)s - %(message)s')301 launch_baristabot()