Create Customer_Support/handlers.py

Customer_Support/handlers.py

@@ -0,0 +1,105 @@
+"""
+Handler functions for the customer support workflow nodes.
+"""
+from .models import State
+from config.app_config import setup_api
+
+# Initialize the model
+model = setup_api()
+
+
+def categorize(state: State) -> State:
+    """Categorize the customer query into Technical, Billing, or General."""
+    prompt = (
+        "Categorize the following customer query into one of these categories: "
+        "Technical, Billing, or General.\n"
+        f"Query: {state['query']}"
+    )
+    response = model.generate_content(prompt)
+    category = response.text.strip()
+
+    return {
+        **state,
+        "category": category
+    }
+
+
+def analyze_sentiment(state: State) -> State:
+    """Analyze the sentiment of the customer query as Positive, Neutral, or Negative."""
+    prompt = (
+        "Analyze the sentiment of the following customer query. "
+        "Respond with either 'Positive', 'Neutral', or 'Negative'.\n"
+        f"Query: {state['query']}"
+    )
+    response = model.generate_content(prompt)
+    sentiment = response.text.strip()
+
+    return {
+        **state,
+        "sentiment": sentiment
+    }
+
+
+def handle_technical(state: State) -> State:
+    """Provide a technical support response to the query."""
+    prompt = (
+        "Provide a technical support response to the following query: "
+        f"{state['query']}"
+    )
+    response = model.generate_content(prompt)
+    technical_response = response.text.strip()
+
+    return {
+        **state,
+        "response": technical_response
+    }
+
+
+def handle_billing(state: State) -> State:
+    """Provide a billing support response to the query."""
+    prompt = (
+        "Provide a billing support response to the following query: "
+        f"{state['query']}"
+    )
+    response = model.generate_content(prompt)
+    billing_response = response.text.strip()
+
+    return {
+        **state,
+        "response": billing_response
+    }
+
+
+def handle_general(state: State) -> State:
+    """Provide a general support response to the query."""
+    prompt = (
+        "Provide a general support response to the following query: "
+        f"{state['query']}"
+    )
+    response = model.generate_content(prompt)
+    general_response = response.text.strip()
+
+    return {
+        **state,
+        "response": general_response
+    }
+
+
+def escalate(state: State) -> State:
+    """Escalate the query to a human agent due to negative sentiment."""
+    return {
+        **state,
+        "response": "This query has been escalated to a human agent due to its negative sentiment."
+    }
+
+
+def route_query(state: State) -> str:
+    """Route the query based on its sentiment and category."""
+    if state["sentiment"] == "Negative":
+        return "escalate"
+    elif state["category"] == "Technical":
+        return "handle_technical"
+    elif state["category"] == "Billing":
+        return "handle_billing"
+    else:
+        return "handle_general"