appointment receptionist usecases

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "debug streamlit",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "./.venv/Lib/site-packages/streamlit",  // /home/xx/tmp/venv/bin/streamlit",
+            "args": [
+                "run",
+                "app.py"
+            ],
+            "justMyCode": false
+        }
+    ]
+}

app.py

@@ -0,0 +1,67 @@
+import streamlit as st
+
+from configfile import Config
+from src.streamlitui.loadui import LoadStreamlitUI
+from src.LLMS.groqllm import GroqLLM
+
+from src.langgraphagent.caller_agent import Caller_Agent
+from langchain_core.messages import HumanMessage
+from src.tools.langgraphtool import APPOINTMENTS
+
+
+
+def submit_message(model):
+    obj_caller_agent = Caller_Agent(model)
+    # caller agent 
+    return obj_caller_agent.receive_message_from_caller(st.session_state["message"])
+  
+    
+
+# MAIN Function START
+
+
+if __name__ == "__main__":
+    # config
+    obj_config = Config()
+    # load ui
+    ui = LoadStreamlitUI()
+    user_input = ui.load_streamlit_ui()
+
+    
+  
+    # is_add_message_to_history = st.session_state["chat_with_history"] 
+    
+    if user_input['selected_usecase'] == "Appointment Receptionist":
+        if st.chat_input("Type message here", key="message") :
+              # Configure LLM
+            obj_llm_config = GroqLLM(user_controls_input=user_input)
+            model = obj_llm_config.get_llm_model()
+            CONVERSATION = (submit_message(model))
+            
+            col1, col2 = st.columns(2)
+            with col1:
+                for message in CONVERSATION:
+                    if type(message) == HumanMessage:
+                        with st.chat_message("user"):
+                            st.write(message.content)
+                    else:
+                        with st.chat_message("assistant"):
+                            st.write(message.content)
+                            
+            with col2:
+                st.header("Appointments")
+                st.write(APPOINTMENTS)
+                    
+                
+           
+                
+            
+        
+
+
+            
+
+        
+                
+                
+                

configfile.ini

@@ -0,0 +1,6 @@
+[DEFAULT]
+PAGE_TITLE = Langgraph IN ACTION
+LLM_OPTIONS = Groq
+USECASE_OPTIONS = Appointment Receptionist
+GROQ_MODEL_OPTIONS = mixtral-8x7b-32768, llama3-8b-8192, llama3-70b-8192, gemma-7b-i
+

configfile.py

@@ -0,0 +1,20 @@
+from configparser import ConfigParser
+
+
+class Config:
+  def __init__(self, config_file="configfile.ini"):
+    self.config = ConfigParser()
+    self.config.read(config_file)
+
+  def get_llm_options(self):
+    return self.config["DEFAULT"].get("LLM_OPTIONS").split(", ")
+
+  def get_usecase_options(self):
+    return self.config["DEFAULT"].get("USECASE_OPTIONS").split(", ")
+
+  def get_groq_model_options(self):
+    return self.config["DEFAULT"].get("GROQ_MODEL_OPTIONS").split(", ")
+
+  def get_page_title(self):
+    return self.config["DEFAULT"].get("PAGE_TITLE")
+

requirements.txt

@@ -0,0 +1,7 @@
+streamlit==1.41.1
+groq==0.13.1
+langgraph 
+langchain_community 
+langchain_openai
+langchain_core
+langchain_groq

src/LLMS/groqllm.py

@@ -0,0 +1,21 @@
+import os
+
+import streamlit as st
+from langchain_groq import ChatGroq
+
+
+class GroqLLM:
+    def __init__(self,user_controls_input):
+        self.user_controls_input = user_controls_input
+
+    def get_llm_model(self):
+        try :
+            groq_api_key = self.user_controls_input['GROQ_API_KEY']
+            selected_groq_model = self.user_controls_input['selected_groq_model']
+            if groq_api_key=='' and os.environ["GROQ_API_KEY"] =='':
+                st.error("Please Enter the Groq API KEY")
+            llm = ChatGroq(api_key =groq_api_key, model=selected_groq_model)
+            
+        except Exception as e:
+            raise ValueError(f"Error Occurred with Exception : {e}")
+        return llm

src/langgraphagent/caller_agent.py

@@ -0,0 +1,92 @@
+from langchain_core.prompts import ChatPromptTemplate
+from langgraph.graph import StateGraph, END, MessagesState
+import datetime
+from src.tools.langgraphtool import book_appointment, get_next_available_appointment, cancel_appointment
+from langchain_openai import ChatOpenAI
+from langgraph.prebuilt import ToolNode
+from langchain_core.messages import HumanMessage
+from src.LLMS.groqllm import GroqLLM
+
+CONVERSATION = []
+
+class Caller_Agent:
+    def __init__(self,model):
+        self.llm = model
+        
+        
+    
+    # Nodes
+    def call_caller_model(self,state: MessagesState):
+        state["current_time"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
+        response = self.caller_model.invoke(state)
+        return {"messages": [response]}
+    
+     # Edges
+    def should_continue_caller(self,state: MessagesState):
+        messages = state["messages"]
+        last_message = messages[-1]
+        if not last_message.tool_calls:
+            return "end"
+        else:
+            return "continue"
+
+
+    def call_tool(self):
+        caller_tools = [book_appointment, get_next_available_appointment, cancel_appointment]
+        tool_node = ToolNode(caller_tools)
+
+
+        caller_pa_prompt = """You are a personal assistant, and need to help the user to book or cancel appointments, you should check the available appointments before booking anything. Be extremely polite, so much so that it is almost rude.
+        Current time: {current_time}
+        """
+
+        caller_chat_template = ChatPromptTemplate.from_messages([
+            ("system", caller_pa_prompt),
+            ("placeholder", "{messages}"),
+        ])
+
+        self.caller_model = caller_chat_template | self.llm.bind_tools(caller_tools)
+
+
+
+        # Graph 
+        caller_workflow = StateGraph(MessagesState)
+
+        # Add Nodes
+        caller_workflow.add_node("agent", self.call_caller_model)
+        caller_workflow.add_node("action", tool_node)
+
+        # Add Edges
+        caller_workflow.add_conditional_edges(
+            "agent",
+            self.should_continue_caller,
+            {
+                "continue": "action",
+                "end": END,
+            },
+        )
+        caller_workflow.add_edge("action", "agent")
+
+        # Set Entry Point and build the graph
+        caller_workflow.set_entry_point("agent")
+
+        self.caller_app = caller_workflow.compile()
+        
+   
+    
+
+    # Invoke model
+    def receive_message_from_caller(self,message):
+        CONVERSATION.append(HumanMessage(content=message, type="human"))
+        state = {
+            "messages": CONVERSATION,
+        }
+        print(state)
+        self.call_tool()
+        new_state = self.caller_app.invoke(state)
+        CONVERSATION.extend(new_state["messages"][len(CONVERSATION):])
+        return CONVERSATION
+
+
+   
+

src/streamlitui/loadui.py

@@ -0,0 +1,55 @@
+import streamlit as st
+from configfile import Config  # Import the Config class
+import os
+
+
+class LoadStreamlitUI:
+    def __init__(self):
+        self.config = Config()  # Create a Config instance
+        self.user_controls = {}
+
+    def load_streamlit_ui(self):
+        st.set_page_config(page_title= "🤖 " + self.config.get_page_title(), layout="wide")
+        st.header("🤖 " + self.config.get_page_title())
+
+        with st.sidebar:
+            # Get options from config
+            llm_options = self.config.get_llm_options()
+            usecase_options = self.config.get_usecase_options()
+
+            # LLM selection
+            self.user_controls["selected_llm"] = st.selectbox("Select LLM", llm_options)
+
+            if self.user_controls["selected_llm"] == 'Groq':
+                # Model selection
+                model_options = self.config.get_groq_model_options()
+                self.user_controls["selected_groq_model"] = st.selectbox("Select Model", model_options)
+                # API key input
+                self.user_controls["GROQ_API_KEY"] = st.session_state["GROQ_API_KEY"] = st.text_input("API Key",
+                                                                                                      type="password")
+            # Use case selection
+            self.user_controls["selected_usecase"] = st.selectbox("Select Usecases", usecase_options)
+        
+                   
+               
+               
+               
+               
+               
+            # Use agent description about role
+            self.user_controls['agent_descriptions'] = st.text_input("Enter the description about an agent",placeholder='eg. You are a helpful assistant that always responds in a polite, upbeat and positive manner.')
+            st.session_state["chat_with_history"] = st.sidebar.toggle("Chat With History")
+            self.user_controls['num_history_responses']  = 0
+            if st.session_state["chat_with_history"]:
+                self.user_controls['num_history_responses'] = st.number_input("Enter number of history include from last chat",placeholder="eg. 1",step=1)
+                
+        if self.user_controls['selected_usecase'] == "Appointment Receptionist":
+                col1, col2 = st.columns(2)
+
+                with col1:
+                    st.subheader("Appointment Manager")
+
+                with col2:
+                    st.subheader("Appointments")
+
+        return self.user_controls

src/tools/langgraphtool.py

@@ -0,0 +1,48 @@
+from langchain_core.tools import tool
+import datetime
+
+
+APPOINTMENTS = []
+
+@tool
+def get_next_available_appointment():
+    """Returns the next available appointment"""
+    current_time = datetime.datetime.now()
+    return f"One appointment available at {current_time + datetime.timedelta(minutes=(30 - current_time.minute % 30))}"
+
+@tool
+def book_appointment(appointment_year: int, appointment_month: int, appointment_day: int, appointment_hour: int, appointment_minute: int, appointment_name: str):
+    """Book an appointment at the given time, you must know the exact time to book
+
+    Args:
+        appointment_year: The year of the appointment
+        appointment_month: The month of the appointment
+        appointment_day: The day of the appointment
+        appointment_hour: The hour of the appointment
+        appointment_minute: The minute of the appointment
+        appointment_name: The name of the person booking the appointment
+    """
+    time = datetime.datetime(appointment_year, appointment_month, appointment_day, appointment_hour, appointment_minute)
+    for appointment in APPOINTMENTS:
+        if appointment.time >= time and appointment.time < time + datetime.timedelta(minutes=30):
+            return f"Appointment at {time} is already booked"
+    APPOINTMENTS.append({"time": time, "name": appointment_name})
+    return f"Appointment booked for {time}"
+
+@tool
+def cancel_appointment(appointment_year: int, appointment_month: int, appointment_day: int, appointment_hour: int, appointment_minute: int):
+    """Cancel the appointment at the given time
+
+    Args:
+        appointment_year: The year of the appointment
+        appointment_month: The month of the appointment
+        appointment_day: The day of the appointment
+        appointment_hour: The hour of the appointment
+        appointment_minute: The minute of the appointment
+    """
+    time = datetime.datetime(appointment_year, appointment_month, appointment_day, appointment_hour, appointment_minute)
+    for appointment in APPOINTMENTS:
+        if appointment["time"] == time:
+            APPOINTMENTS.remove(appointment)
+            return f"Appointment at {time} cancelled"
+    return f"No appointment found at {time}"