Create app.py

app.py

@@ -0,0 +1,156 @@
+from typing import TypedDict, Dict
+from langgraph.graph import StateGraph, END
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.runnables.graph import MermaidDrawMethod
+from IPython.display import display, Image
+
+class State(TypedDict):
+  query : str
+  category : str
+  sentiment : str
+  response : str
+
+from langchain_groq import ChatGroq
+
+llm = ChatGroq(
+    temperature = 0,
+    groq_api_key = "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",
+    model_name = "llama-3.3-70b-versatile"
+)
+result = llm.invoke("what is langchain")
+result.content
+
+def categorize(state: State) -> State:
+  "Technical, Billing, General"
+  prompt = ChatPromptTemplate.from_template(
+      "Categorize the following customer query into one of these categories: "
+      "Technical, Billing, General. Query: {query}"
+  )
+  chain = prompt | llm
+  category = chain.invoke({"query": state["query"]}).content
+  return {"category": category}
+
+def analyze_sentiment(state: State) -> State:
+  prompt = ChatPromptTemplate.from_template(
+      "Analyze the sentiment of the following customer query"
+      "Response with either 'Position', 'Neutral' , or 'Negative'. Query: {query}"
+  )
+  chain = prompt | llm
+  sentiment = chain.invoke({"query": state["query"]}).content
+  return {"sentiment": sentiment}
+
+def handle_technical(state: State)->State:
+  prompt = ChatPromptTemplate.from_template(
+      "Provide a technical support response to the following query : {query}"
+  )
+  chain = prompt | llm
+  response = chain.invoke({"query": state["query"]}).content
+  return {"response": response}
+
+def handle_billing(state: State)->State:
+  prompt = ChatPromptTemplate.from_template(
+      "Provide a billing support response to the following query : {query}"
+  )
+  chain = prompt | llm
+  response = chain.invoke({"query": state["query"]}).content
+  return {"response": response}
+
+def handle_general(state: State)->State:
+  prompt = ChatPromptTemplate.from_template(
+      "Provide a general support response to the following query : {query}"
+  )
+  chain = prompt | llm
+  response = chain.invoke({"query": state["query"]}).content
+  return {"response": response}
+
+def escalate(state: State)->State:
+  return {"response": "This query has been escalate to a human agent due to its negative sentiment"}
+
+def route_query(state: State)->State:
+  if state["sentiment"] == "Negative":
+    return "escalate"
+  elif state["category"] == "Technical":
+    return "handle_technical"
+  elif state["category"] == "Billing":
+    return "handle_billing"
+  else:
+    return "handle_general"
+
+workflow = StateGraph(State)
+
+workflow.add_node("categorize", categorize)
+workflow.add_node("analyze_sentiment", analyze_sentiment)
+workflow.add_node("handle_technical", handle_technical)
+workflow.add_node("handle_billing", handle_billing)
+workflow.add_node("handle_general", handle_general)
+workflow.add_node("escalate", escalate)
+
+workflow.add_edge("categorize", "analyze_sentiment")
+workflow.add_conditional_edges(
+    "analyze_sentiment",
+    route_query,{
+        "handle_technical" : "handle_technical",
+        "handle_billing" :  "handle_billing",
+        "handle_general" : "handle_general",
+        "escalate": "escalate"
+    }
+)
+
+workflow.add_edge("handle_technical", END)
+workflow.add_edge("handle_billing", END)
+workflow.add_edge("handle_general", END)
+workflow.add_edge("escalate", END)
+
+workflow.set_entry_point("categorize")
+
+app  = workflow.compile()
+
+
+def run_customer_support(query: str)->Dict[str, str]:
+  results = app.invoke({"query": query})
+  return {
+      "category":results['category'],
+      "sentiment":results['sentiment'],
+      "response": results['response']
+  }
+
+# query = "my laptop is not charging what should i do?"
+# result = run_customer_support(query)
+# print(f"Query: {query}")
+# print(f"Category : {result['category']}")
+# print(f"Sentiment : {result['sentiment']}")
+# print(f"Response : {result['response']}")
+
+import gradio as gr
+
+# Define the function that integrates the workflow.
+def run_customer_support(query: str) -> Dict[str, str]:
+    results = app.invoke({"query": query})
+    return {
+        "Category": results['category'],
+        "Sentiment": results['sentiment'],
+        "Response": results['response']
+    }
+
+# Create the Gradio interface
+def gradio_interface(query: str):
+    result = run_customer_support(query)
+    return (
+        f"**Category:** {result['Category']}\n\n"
+        f"**Sentiment:** {result['Sentiment']}\n\n"
+        f"**Response:** {result['Response']}"
+    )
+
+# Build the Gradio app
+gui = gr.Interface(
+    fn=gradio_interface,
+    theme='Yntec/HaleyCH_Theme_Orange_Green',
+    inputs=gr.Textbox(lines=2, placeholder="Enter your query here..."),
+    outputs=gr.Markdown(),
+    title="Customer Support Assistant",
+    description="Provide a query and receive a categorized response. The system analyzes sentiment and routes to the appropriate support channel.",
+)
+
+# Launch the app
+if __name__ == "__main__":
+    gui.launch(share=True)