Create app.py

app.py

@@ -0,0 +1,111 @@
+import os
+from langgraph.graph import StateGraph, START
+from langgraph.types import Command
+from dotenv import load_dotenv
+import gradio as gr
+from langchain_huggingface import HuggingFaceEndpoint
+
+# Load environment variables
+load_dotenv()
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+# Define HuggingFaceEndpoint for urgency classification
+classifier = HuggingFaceEndpoint(
+    repo_id="bert-base-uncased",  # You can use any model fine-tuned for text classification
+    huggingfacehub_api_token=HF_TOKEN.strip(),
+    temperature=0.5,
+    max_new_tokens=10,  # For short classification outputs
+)
+
+# Define HuggingFaceEndpoint for response generation
+llm = HuggingFaceEndpoint(
+    repo_id="mistralai/Mistral-7B-Instruct-v0.3",
+    huggingfacehub_api_token=HF_TOKEN.strip(),
+    temperature=0.7,
+    max_new_tokens=200,
+)
+
+# Define state
+class State(dict):
+    issue: str
+    priority: str
+    response: str
+    escalation_needed: bool
+
+# Create the graph
+builder = StateGraph(State)
+
+# Define nodes (agents)
+def ticket_creation_agent(state: State) -> Command[Literal["priority_classification_agent"]]:
+    """Create a ticket based on customer issue."""
+    return Command(update={"issue": state["issue"]}, goto="priority_classification_agent")
+
+def priority_classification_agent(state: State) -> Command[Literal["response_generation_agent"]]:
+    """Classify ticket priority based on NLP analysis of the issue."""
+    issue = state["issue"]
+    
+    # Use the model to classify urgency level
+    prompt = f"Classify the urgency of this issue: {issue}"
+    priority = classifier(prompt)
+    
+    # Use the classification result to assign priority
+    if "urgent" in priority.lower():
+        priority = "High"
+    elif "critical" in priority.lower():
+        priority = "Critical"
+    else:
+        priority = "Normal"
+    
+    return Command(update={"priority": priority}, goto="response_generation_agent")
+
+def response_generation_agent(state: State) -> Command[Literal["escalation_agent"]]:
+    """Generate response based on issue and priority."""
+    prompt = f"""
+    Customer Issue: {state['issue']}
+    Priority: {state['priority']}
+    
+    You are a customer service representative. Provide a detailed response to this issue based on its priority.
+    """
+    response = llm(prompt)
+    return Command(update={"response": response, "escalation_needed": state['priority'] == "High"}, goto="escalation_agent")
+
+def escalation_agent(state: State):
+    """Decide if the ticket needs to be escalated based on priority."""
+    escalation = "Yes" if state["escalation_needed"] else "No"
+    return {"response": state["response"], "escalation": escalation}
+
+# Add nodes to the graph
+builder.add_edge(START, "ticket_creation_agent")
+builder.add_node("ticket_creation_agent", ticket_creation_agent)
+builder.add_node("priority_classification_agent", priority_classification_agent)
+builder.add_node("response_generation_agent", response_generation_agent)
+builder.add_node("escalation_agent", escalation_agent)
+
+# Compile the graph
+graph = builder.compile()
+
+# Gradio Interface
+def process_ticket(issue: str):
+    """Run the multi-agent customer support flow with user input."""
+    state = {"issue": issue}
+    result = graph.invoke(state)
+
+    # Return the customer service response and escalation info
+    return result["response"], result["escalation"]
+
+iface = gr.Interface(
+    fn=process_ticket,
+    inputs=[
+        gr.Textbox(
+            label="Enter Customer Issue",
+            placeholder="Describe the issue that needs to be addressed...",
+        ),
+    ],
+    outputs=[
+        gr.Textbox(label="Generated Customer Service Response"),
+        gr.Textbox(label="Escalation Status"),
+    ],
+    title="Customer Support Multi-Agent System",
+)
+
+iface.launch()