Capstone: Building an AI-Powered Help Desk with Ticket Management and Escalation

Help Desk Architecture

A modern AI-powered help desk goes beyond simple ticket tracking. It classifies incoming tickets by category and priority, suggests solutions from historical data, assigns tickets to the right team member, enforces SLA deadlines, and escalates automatically when SLAs are about to breach. This capstone builds all of these capabilities into a single, deployable system.

The system has six components: ticket ingestion (email, web form, API), AI classification (category, priority, and suggested resolution), assignment engine (skill-based routing to agents), SLA tracker (deadline enforcement with escalation), resolution workflow (agent workspace with AI-suggested responses), and reporting dashboard (team performance and SLA compliance metrics).

Data Model

# models.py
from sqlalchemy import Column, String, Text, Integer, Float, DateTime, ForeignKey, Enum
from sqlalchemy.dialects.postgresql import UUID, JSONB, ARRAY
import uuid, enum

class Priority(str, enum.Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"

class TicketStatus(str, enum.Enum):
    NEW = "new"
    ASSIGNED = "assigned"
    IN_PROGRESS = "in_progress"
    WAITING_CUSTOMER = "waiting_customer"
    RESOLVED = "resolved"
    CLOSED = "closed"
    ESCALATED = "escalated"

class SupportAgent(Base):
    __tablename__ = "support_agents"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String(200))
    email = Column(String(255), unique=True)
    skills = Column(ARRAY(String))  # ["billing", "technical", "account"]
    max_tickets = Column(Integer, default=10)
    is_available = Column(String(10), default="true")

class SupportTicket(Base):
    __tablename__ = "support_tickets"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    subject = Column(String(500))
    description = Column(Text)
    customer_email = Column(String(255), index=True)
    category = Column(String(100))  # billing, technical, account, feature_request
    priority = Column(Enum(Priority), default=Priority.MEDIUM)
    status = Column(Enum(TicketStatus), default=TicketStatus.NEW)
    assigned_to = Column(UUID(as_uuid=True), ForeignKey("support_agents.id"), nullable=True)
    sla_deadline = Column(DateTime, nullable=True)
    escalation_level = Column(Integer, default=0)
    ai_suggested_response = Column(Text, nullable=True)
    source = Column(String(50))  # "email", "web", "api"
    tags = Column(ARRAY(String), default=[])
    created_at = Column(DateTime, server_default="now()")
    resolved_at = Column(DateTime, nullable=True)

class TicketComment(Base):
    __tablename__ = "ticket_comments"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    ticket_id = Column(UUID(as_uuid=True), ForeignKey("support_tickets.id"))
    author_type = Column(String(20))  # "customer", "agent", "system"
    author_email = Column(String(255))
    content = Column(Text)
    is_internal = Column(String(10), default="false")  # internal notes
    created_at = Column(DateTime, server_default="now()")

class SLAPolicy(Base):
    __tablename__ = "sla_policies"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    priority = Column(Enum(Priority), unique=True)
    first_response_minutes = Column(Integer)
    resolution_minutes = Column(Integer)
    escalation_after_minutes = Column(Integer)

AI Ticket Classification

When a ticket arrives, classify it by category and priority, and generate a suggested response.

flowchart TD
    START["Capstone: Building an AI-Powered Help Desk with T…"] --> A
    A["Help Desk Architecture"]
    A --> B
    B["Data Model"]
    B --> C
    C["AI Ticket Classification"]
    C --> D
    D["Skill-Based Assignment Engine"]
    D --> E
    E["SLA Monitoring and Escalation"]
    E --> F
    F["Ticket CRUD API"]
    F --> G
    G["Reporting Dashboard API"]
    G --> H
    H["FAQ"]
    H --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

# services/classifier.py
import openai, json

SLA_DEFAULTS = {
    Priority.URGENT: {"response": 30, "resolution": 240},
    Priority.HIGH: {"response": 60, "resolution": 480},
    Priority.MEDIUM: {"response": 240, "resolution": 1440},
    Priority.LOW: {"response": 480, "resolution": 2880},
}

async def classify_ticket(ticket_id: str, db):
    ticket = db.query(SupportTicket).get(ticket_id)

    # Search for similar resolved tickets
    similar = await find_similar_tickets(ticket.description, db, limit=3)
    similar_context = "\n".join(
        [f"[{t.category}] {t.subject}: {t.ai_suggested_response}" for t in similar]
    )

    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"""Classify this support ticket.

Similar resolved tickets for context:
{similar_context}

Return JSON with:
- category: one of [billing, technical, account, feature_request, bug_report]
- priority: one of [low, medium, high, urgent]
- tags: list of relevant tags
- suggested_response: a draft response the agent can send
- confidence: 0-1"""},
            {"role": "user", "content": f"Subject: {ticket.subject}\n\n{ticket.description}"},
        ],
        response_format={"type": "json_object"},
    )

    result = json.loads(response.choices[0].message.content)

    ticket.category = result["category"]
    ticket.priority = Priority(result["priority"])
    ticket.tags = result.get("tags", [])
    ticket.ai_suggested_response = result.get("suggested_response")

    # Set SLA deadline
    sla = SLA_DEFAULTS[ticket.priority]
    ticket.sla_deadline = datetime.utcnow() + timedelta(minutes=sla["resolution"])

    db.commit()
    return result

Skill-Based Assignment Engine

Assign tickets to the agent best suited for the category, with the lowest current workload.

# services/assignment.py
from sqlalchemy import func

async def assign_ticket(ticket_id: str, db):
    ticket = db.query(SupportTicket).get(ticket_id)

    # Map categories to required skills
    skill_map = {
        "billing": "billing",
        "technical": "technical",
        "account": "account",
        "bug_report": "technical",
        "feature_request": "account",
    }
    required_skill = skill_map.get(ticket.category, "general")

    # Find available agents with the required skill and lowest workload
    agents = db.query(
        SupportAgent,
        func.count(SupportTicket.id).label("current_tickets"),
    ).outerjoin(
        SupportTicket,
        (SupportTicket.assigned_to == SupportAgent.id) &
        (SupportTicket.status.in_([TicketStatus.ASSIGNED, TicketStatus.IN_PROGRESS]))
    ).filter(
        SupportAgent.skills.contains([required_skill]),
        SupportAgent.is_available == "true",
    ).group_by(SupportAgent.id).having(
        func.count(SupportTicket.id) < SupportAgent.max_tickets
    ).order_by("current_tickets").all()

    if agents:
        best_agent = agents[0][0]
        ticket.assigned_to = best_agent.id
        ticket.status = TicketStatus.ASSIGNED
        db.commit()
        await notify_agent(best_agent.email, ticket)
        return best_agent
    else:
        # No available agents — auto-escalate
        ticket.escalation_level = 1
        ticket.status = TicketStatus.ESCALATED
        db.commit()
        await notify_managers(ticket)
        return None

SLA Monitoring and Escalation

A background task checks for SLA breaches and escalates tickets automatically.

# services/sla_monitor.py
from datetime import datetime, timedelta

async def check_sla_compliance():
    """Run every 5 minutes to check for SLA breaches."""
    now = datetime.utcnow()

    # Find tickets approaching or past SLA deadline
    at_risk = db.query(SupportTicket).filter(
        SupportTicket.status.in_([
            TicketStatus.NEW, TicketStatus.ASSIGNED, TicketStatus.IN_PROGRESS
        ]),
        SupportTicket.sla_deadline.isnot(None),
        SupportTicket.sla_deadline <= now + timedelta(minutes=30),
    ).all()

    for ticket in at_risk:
        minutes_remaining = (ticket.sla_deadline - now).total_seconds() / 60

        if minutes_remaining <= 0:
            # SLA breached
            ticket.escalation_level = max(ticket.escalation_level, 2)
            ticket.status = TicketStatus.ESCALATED
            await notify_managers(ticket, breach=True)
            add_system_comment(ticket.id, "SLA BREACHED. Auto-escalated to management.")

        elif minutes_remaining <= 30 and ticket.escalation_level == 0:
            # SLA at risk — first escalation
            ticket.escalation_level = 1
            await notify_agent_urgent(ticket)
            add_system_comment(
                ticket.id,
                f"SLA at risk. {int(minutes_remaining)} minutes remaining."
            )

    db.commit()

Ticket CRUD API

# routes/tickets.py
from fastapi import APIRouter

router = APIRouter(prefix="/tickets")

@router.post("/")
async def create_ticket(body: TicketCreate, db=Depends(get_db)):
    ticket = SupportTicket(
        subject=body.subject,
        description=body.description,
        customer_email=body.customer_email,
        source=body.source,
    )
    db.add(ticket)
    db.commit()

    # Async classification and assignment
    classification = await classify_ticket(str(ticket.id), db)
    agent = await assign_ticket(str(ticket.id), db)

    db.refresh(ticket)
    return {"ticket": ticket, "classification": classification}

@router.get("/{ticket_id}")
async def get_ticket(ticket_id: str, db=Depends(get_db)):
    ticket = db.query(SupportTicket).get(ticket_id)
    comments = db.query(TicketComment).filter(
        TicketComment.ticket_id == ticket_id
    ).order_by(TicketComment.created_at).all()
    return {"ticket": ticket, "comments": comments}

@router.patch("/{ticket_id}/resolve")
async def resolve_ticket(ticket_id: str, body: ResolveRequest, db=Depends(get_db)):
    ticket = db.query(SupportTicket).get(ticket_id)
    ticket.status = TicketStatus.RESOLVED
    ticket.resolved_at = datetime.utcnow()
    add_system_comment(ticket_id, f"Resolved by {body.agent_email}: {body.resolution_note}")
    db.commit()
    return {"status": "resolved"}

Reporting Dashboard API

# routes/reports.py
@router.get("/reports/overview")
async def reports_overview(days: int = 30, db=Depends(get_db)):
    since = datetime.utcnow() - timedelta(days=days)
    tickets = db.query(SupportTicket).filter(
        SupportTicket.created_at >= since
    ).all()

    resolved = [t for t in tickets if t.resolved_at]
    breached = [t for t in tickets if t.escalation_level >= 2]

    avg_resolution = None
    if resolved:
        deltas = [(t.resolved_at - t.created_at).total_seconds() / 3600 for t in resolved]
        avg_resolution = sum(deltas) / len(deltas)

    return {
        "total_tickets": len(tickets),
        "resolved": len(resolved),
        "open": len(tickets) - len(resolved),
        "sla_breach_count": len(breached),
        "sla_compliance_pct": round(
            (1 - len(breached) / max(len(tickets), 1)) * 100, 1
        ),
        "avg_resolution_hours": round(avg_resolution, 1) if avg_resolution else None,
        "by_category": count_by_field(tickets, "category"),
        "by_priority": count_by_field(tickets, "priority"),
    }

The complete help desk system demonstrates end-to-end AI integration in a business-critical application: from automatic classification and assignment through SLA enforcement to executive reporting. Each component is independently deployable and testable, and the architecture supports scaling by adding more support agents and increasing the background task frequency.

FAQ

How do I handle tickets that arrive via email?

Set up an inbound email webhook using SendGrid or Mailgun. When an email arrives at [email protected], the webhook sends the sender, subject, and body to your /tickets endpoint. Parse the email body to extract the description, use the sender address as customer_email, and set the source to "email". Reply notifications are sent back via the same email service.

How do I prevent the AI from misclassifying urgent tickets as low priority?

Use keyword-based priority overrides as a safety net. If the ticket contains phrases like "system down", "data loss", "cannot login", or "security breach", force the priority to URGENT regardless of the AI classification. Log every override so you can tune the classifier to handle these cases natively over time.

How do I measure individual agent performance fairly?

Track metrics that the agent can control: average first response time, customer satisfaction rating, and resolution rate. Do not penalize agents for SLA breaches caused by assignment delays or ticket volume spikes. Compare each agent's metrics against tickets of similar category and priority to normalize for workload difficulty.

---

#CapstoneProject #HelpDesk #TicketManagement #SLATracking #Escalation #FullStackAI #AgenticAI #LearnAI #AIEngineering