Escalation & Handoff

Detect when the AI cannot adequately help a customer and seamlessly hand off to a human agent with full conversation context, customer information, and suggested resolution.

Escalation Detector

# app/escalation/detector.py
class EscalationDetector:
    ESCALATION_PHRASES = [
        "speak to a human", "talk to agent", "real person",
        "manager", "supervisor", "complaint", "not helpful",
        "this is urgent", "cancel my account", "legal",
    ]

    def should_escalate(self, message: str, confidence: float,
                        turn_count: int, sentiment: float = 0) -> dict:
        reasons = []

        # Low confidence from retrieval
        if confidence < 0.3:
            reasons.append(f"Low retrieval confidence: {confidence:.2f}")

        # Explicit escalation request
        msg_lower = message.lower()
        for phrase in self.ESCALATION_PHRASES:
            if phrase in msg_lower:
                reasons.append(f"Escalation phrase: '{phrase}'")
                break

        # Too many turns without resolution
        if turn_count > 5:
            reasons.append(f"Extended conversation: {turn_count} turns")

        # Negative sentiment
        if sentiment < -0.5:
            reasons.append(f"Negative sentiment: {sentiment:.2f}")

        return {
            "should_escalate": len(reasons) > 0,
            "reasons": reasons,
            "priority": "high" if len(reasons) > 1 else "normal",
        }

Agent Router

# app/escalation/router.py
import uuid
from datetime import datetime
from dataclasses import dataclass, field

@dataclass
class Ticket:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    session_id: str = ""
    customer_context: dict = field(default_factory=dict)
    conversation_history: list = field(default_factory=list)
    escalation_reasons: list = field(default_factory=list)
    priority: str = "normal"
    assigned_agent: str = ""
    status: str = "open"
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())


class AgentRouter:
    def __init__(self):
        self.tickets: dict[str, Ticket] = {}
        self.available_agents: list[dict] = []

    def create_ticket(self, session_id, context, history, reasons, priority):
        ticket = Ticket(
            session_id=session_id, customer_context=context,
            conversation_history=history, escalation_reasons=reasons,
            priority=priority,
        )
        # Auto-assign based on priority and availability
        if self.available_agents:
            ticket.assigned_agent = self.available_agents[0]["id"]
            ticket.status = "assigned"
        self.tickets[ticket.id] = ticket
        return ticket

    def get_open_tickets(self):
        return [t for t in self.tickets.values() if t.status in ("open", "assigned")]

Key Takeaways

• Multiple escalation signals (low confidence, explicit request, sentiment, turn count) catch different failure modes.
• Priority scoring ensures urgent issues get human attention faster.
• Full context transfer prevents customers from repeating information to the agent.