Customer Support Automation: Architecture Blueprint & Production Stack

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
import httpx

class TicketState(TypedDict):
    ticket_id: str
    user_query: str
    intent_category: str
    confidence_score: float
    retrieved_docs: List[str]
    draft_response: str
    requires_human: bool

async def supervisor_router_node(state: TicketState) -> TicketState:
    """Classifies user query intent and evaluates initial risk score."""
    query = state["user_query"].lower()
    if any(k in query for k in ["refund", "billing dispute", "cancel account"]):
        state["intent_category"] = "billing_action"
        state["confidence_score"] = 0.75  # Forces human review path
    else:
        state["intent_category"] = "tech_support"
        state["confidence_score"] = 0.94
    return state

async def rag_retrieval_node(state: TicketState) -> TicketState:
    """Retrieves grounded help center context."""
    state["retrieved_docs"] = [
        "Knowledge Article #402: Reset password via account security tab.",
        "SLA Policy: Tier-1 requests resolved within 2 hours."
    ]
    state["draft_response"] = (
        "To reset your password, navigate to the Account Security tab in your dashboard "
        "and select 'Reset Credentials'. Ref: Article #402."
    )
    return state

def evaluate_escalation_node(state: TicketState) -> str:
    """Routes state based on confidence score threshold (88%)."""
    if state["confidence_score"] < 0.88 or state["intent_category"] == "billing_action":
        return "escalate_to_human"
    return "auto_resolve_ticket"

async def zendesk_escalation_node(state: TicketState) -> TicketState:
    """Posts prepared draft to Zendesk agent workspace sidebar."""
    state["requires_human"] = True
    # Async webhook payload sent to Zendesk REST API
    return state

# LangGraph Support Workflow Definition
builder = StateGraph(TicketState)
builder.add_node("supervisor_router", supervisor_router_node)
builder.add_node("rag_retrieval", rag_retrieval_node)
builder.add_node("zendesk_escalate", zendesk_escalation_node)

builder.set_entry_point("supervisor_router")
builder.add_edge("supervisor_router", "rag_retrieval")
builder.add_conditional_edges(
    "rag_retrieval",
    evaluate_escalation_node,
    {
        "escalate_to_human": "zendesk_escalate",
        "auto_resolve_ticket": END
    }
)
support_agent_app = builder.compile()