Intelligente Eskalation: Wann an einen Menschen übergeben

TL;DR

Intelligente Eskalation ist der Schlüssel zu erfolgreichem Hybrid-Support. Ein RAG-Chatbot muss erkennen, wann er an seine Grenzen stößt, und elegant an einen Menschen übergeben. Dieser Leitfaden behandelt Erkennungs-Signale, Vertrauensschwellen, kontextuelle Übergabe und Metriken zur Optimierung des Verhältnisses Automatisierung/Mensch. Ziel: 70% automatische Lösung bei 95% Zufriedenheit für Eskalationen.

Le dilemme de l'escalade

Trop peu d'escalade

Quand le bot persiste alors qu'il devrait transférer :

Trop d'escalade

Quand le bot transfert trop facilement :

Le sweet spot

L'objectif est de trouver le bon équilibre :

• 70-80% de résolution automatique
• 95%+ de CSAT sur les escalades
• < 2 minutes entre décision d'escalade et agent humain
• Contexte complet transmis à l'agent

Signaux de détection d'escalade

1. Demande explicite

L'utilisateur demande clairement un humain :

DEVELOPERpython
class ExplicitEscalationDetector:
    """
    Détecte les demandes explicites de parler à un humain.
    """

    TRIGGERS_FR = [
        "parler à quelqu'un",
        "agent humain",
        "vraie personne",
        "parler à un conseiller",
        "contacter le support",
        "joindre un agent",
        "appeler quelqu'un",
        "être rappelé",
        "service client"
    ]

    TRIGGERS_EN = [
        "talk to someone",
        "human agent",
        "real person",
        "speak to a representative",
        "contact support",
        "reach an agent",
        "call someone",
        "callback",
        "customer service"
    ]

    def detect(self, message: str, language: str = "fr") -> tuple[bool, str]:
        """
        Détecte une demande explicite d'escalade.
        """
        triggers = self.TRIGGERS_FR if language == "fr" else self.TRIGGERS_EN
        message_lower = message.lower()

        for trigger in triggers:
            if trigger in message_lower:
                return True, f"explicit_request: {trigger}"

        return False, None

2. Confiance RAG insuffisante

Le système n'a pas assez de certitude pour répondre :

DEVELOPERpython
class ConfidenceBasedEscalation:
    """
    Escalade basée sur la confiance du système RAG.
    """

    def __init__(
        self,
        retrieval_threshold: float = 0.6,
        generation_threshold: float = 0.7,
        combined_threshold: float = 0.65
    ):
        self.retrieval_threshold = retrieval_threshold
        self.generation_threshold = generation_threshold
        self.combined_threshold = combined_threshold

    def should_escalate(
        self,
        retrieval_scores: list[float],
        generation_confidence: float
    ) -> tuple[bool, dict]:
        """
        Détermine si l'escalade est nécessaire basée sur la confiance.
        """
        # Score de retrieval (meilleur document)
        best_retrieval = max(retrieval_scores) if retrieval_scores else 0

        # Score combiné pondéré
        combined = (best_retrieval * 0.4) + (generation_confidence * 0.6)

        reasons = []

        if best_retrieval < self.retrieval_threshold:
            reasons.append(f"low_retrieval: {best_retrieval:.2f}")

        if generation_confidence < self.generation_threshold:
            reasons.append(f"low_generation: {generation_confidence:.2f}")

        should_escalate = combined < self.combined_threshold

        return should_escalate, {
            "retrieval_score": best_retrieval,
            "generation_confidence": generation_confidence,
            "combined_score": combined,
            "reasons": reasons
        }

3. Détection de frustration

Analyse du sentiment et des patterns de frustration :

DEVELOPERpython
class FrustrationDetector:
    """
    Détecte la frustration utilisateur pour escalade préventive.
    """

    FRUSTRATION_PATTERNS = [
        r"ça (ne )?marche (toujours )?pas",
        r"j'ai déjà (dit|expliqué|essayé)",
        r"vous (ne )?(comprenez|écoutez) (pas|rien)",
        r"c'est (nul|inutile|incompétent)",
        r"je (vais|dois) (annuler|résilier|partir)",
        r"(depuis|ça fait) (des heures|longtemps|trop longtemps)",
        r"(service|support) (pourri|nul|incompétent)"
    ]

    def __init__(self, sentiment_analyzer):
        self.sentiment = sentiment_analyzer
        self.patterns = [re.compile(p, re.IGNORECASE) for p in self.FRUSTRATION_PATTERNS]

    async def detect(
        self,
        message: str,
        conversation_history: list
    ) -> tuple[bool, dict]:
        """
        Détecte la frustration basée sur le message et l'historique.
        """
        frustration_signals = []

        # 1. Pattern matching
        for pattern in self.patterns:
            if pattern.search(message):
                frustration_signals.append("frustration_pattern")
                break

        # 2. Analyse de sentiment
        sentiment = await self.sentiment.analyze(message)
        if sentiment["score"] < -0.6:
            frustration_signals.append(f"negative_sentiment: {sentiment['score']:.2f}")

        # 3. Pattern de répétition
        if self._detect_repetition(message, conversation_history):
            frustration_signals.append("user_repeating")

        # 4. Escalade progressive du ton
        if len(conversation_history) >= 3:
            tone_trend = await self._analyze_tone_trend(conversation_history)
            if tone_trend["degrading"]:
                frustration_signals.append("degrading_tone")

        # 5. Messages de plus en plus longs (signe de frustration)
        if self._detect_increasing_length(conversation_history):
            frustration_signals.append("increasing_length")

        is_frustrated = len(frustration_signals) >= 2

        return is_frustrated, {
            "signals": frustration_signals,
            "sentiment_score": sentiment["score"],
            "recommendation": "escalate_immediately" if is_frustrated else "continue"
        }

    def _detect_repetition(self, message: str, history: list) -> bool:
        """
        Détecte si l'utilisateur répète sa question.
        """
        user_messages = [
            h["content"] for h in history
            if h["role"] == "user"
        ][-3:]

        if not user_messages:
            return False

        # Similarité avec messages précédents
        for prev in user_messages:
            similarity = self._calculate_similarity(message, prev)
            if similarity > 0.7:
                return True

        return False

4. Complexité excessive

La question dépasse les capacités du bot :

DEVELOPERpython
class ComplexityAnalyzer:
    """
    Analyse la complexité de la demande utilisateur.
    """

    COMPLEX_INDICATORS = {
        "multi_step": [
            "d'abord", "ensuite", "puis", "enfin",
            "premièrement", "deuxièmement",
            "first", "then", "after that", "finally"
        ],
        "conditional": [
            "si", "dans le cas où", "à condition que",
            "if", "in case", "provided that", "unless"
        ],
        "comparison": [
            "comparer", "différence entre", "versus", "vs",
            "compare", "difference between", "vs"
        ],
        "exception": [
            "sauf si", "à l'exception", "mais pas",
            "except", "unless", "but not"
        ]
    }

    async def analyze(
        self,
        message: str,
        kb_coverage: float
    ) -> tuple[bool, dict]:
        """
        Analyse la complexité et recommande une action.
        """
        complexity_score = 0
        indicators_found = []

        # 1. Indicateurs lexicaux
        for category, keywords in self.COMPLEX_INDICATORS.items():
            for keyword in keywords:
                if keyword.lower() in message.lower():
                    complexity_score += 0.15
                    indicators_found.append(f"{category}:{keyword}")
                    break

        # 2. Longueur du message
        word_count = len(message.split())
        if word_count > 100:
            complexity_score += 0.2
            indicators_found.append(f"long_message:{word_count}")
        elif word_count > 50:
            complexity_score += 0.1

        # 3. Questions multiples
        question_marks = message.count("?")
        if question_marks > 2:
            complexity_score += 0.2
            indicators_found.append(f"multiple_questions:{question_marks}")

        # 4. Couverture KB faible = sujet complexe ou non documenté
        if kb_coverage < 0.5:
            complexity_score += 0.25
            indicators_found.append(f"low_kb_coverage:{kb_coverage:.2f}")

        is_complex = complexity_score > 0.5

        return is_complex, {
            "score": complexity_score,
            "indicators": indicators_found,
            "recommendation": "escalate" if is_complex else "attempt_answer"
        }

5. Conversation trop longue

La conversation s'enlise sans résolution :

DEVELOPERpython
class ConversationLengthMonitor:
    """
    Monitore la longueur de conversation pour escalade.
    """

    def __init__(
        self,
        max_turns_no_resolution: int = 8,
        max_total_turns: int = 15
    ):
        self.max_no_resolution = max_turns_no_resolution
        self.max_total = max_total_turns

    def should_escalate(
        self,
        conversation: list,
        resolution_indicators: list
    ) -> tuple[bool, str]:
        """
        Vérifie si la conversation est trop longue.
        """
        total_turns = len([m for m in conversation if m["role"] == "user"])

        # Trop de tours au total
        if total_turns >= self.max_total:
            return True, f"max_turns_exceeded:{total_turns}"

        # Pas de résolution après X tours
        turns_since_last_indicator = self._turns_since_indicator(
            conversation, resolution_indicators
        )

        if turns_since_last_indicator >= self.max_no_resolution:
            return True, f"no_progress:{turns_since_last_indicator}_turns"

        return False, None

    def _turns_since_indicator(
        self,
        conversation: list,
        indicators: list
    ) -> int:
        """
        Compte les tours depuis le dernier indicateur de résolution.
        """
        # Indicateurs : remerciement, confirmation de compréhension, etc.
        resolution_phrases = [
            "merci", "parfait", "c'est bon", "ça marche",
            "thank you", "perfect", "got it", "that works"
        ]

        turns = 0
        for msg in reversed(conversation):
            if msg["role"] == "user":
                turns += 1
                message_lower = msg["content"].lower()
                if any(phrase in message_lower for phrase in resolution_phrases):
                    return turns

        return turns

Système d'escalade combiné

Orchestrateur d'escalade

DEVELOPERpython
class EscalationOrchestrator:
    """
    Combine tous les signaux pour décider de l'escalade.
    """

    def __init__(
        self,
        explicit_detector,
        confidence_checker,
        frustration_detector,
        complexity_analyzer,
        length_monitor
    ):
        self.explicit = explicit_detector
        self.confidence = confidence_checker
        self.frustration = frustration_detector
        self.complexity = complexity_analyzer
        self.length = length_monitor

        # Poids des différents signaux
        self.weights = {
            "explicit": 1.0,       # Toujours escalader si demandé
            "frustration": 0.9,   # Priorité haute
            "confidence": 0.7,    # Important
            "complexity": 0.6,    # Significatif
            "length": 0.5         # Contributif
        }

    async def evaluate(
        self,
        message: str,
        conversation: list,
        rag_result: dict,
        user_context: dict
    ) -> dict:
        """
        Évalue tous les signaux et décide de l'escalade.
        """
        signals = {}

        # 1. Demande explicite (priorité absolue)
        is_explicit, explicit_reason = self.explicit.detect(message)
        if is_explicit:
            return {
                "escalate": True,
                "reason": "explicit_request",
                "details": explicit_reason,
                "priority": "immediate",
                "confidence": 1.0
            }

        # 2. Frustration
        is_frustrated, frustration_data = await self.frustration.detect(
            message, conversation
        )
        signals["frustration"] = {
            "triggered": is_frustrated,
            "data": frustration_data,
            "weight": self.weights["frustration"]
        }

        # 3. Confiance RAG
        should_escalate_conf, conf_data = self.confidence.should_escalate(
            rag_result.get("retrieval_scores", []),
            rag_result.get("generation_confidence", 0)
        )
        signals["confidence"] = {
            "triggered": should_escalate_conf,
            "data": conf_data,
            "weight": self.weights["confidence"]
        }

        # 4. Complexité
        is_complex, complexity_data = await self.complexity.analyze(
            message,
            rag_result.get("kb_coverage", 0)
        )
        signals["complexity"] = {
            "triggered": is_complex,
            "data": complexity_data,
            "weight": self.weights["complexity"]
        }

        # 5. Longueur conversation
        too_long, length_reason = self.length.should_escalate(
            conversation,
            rag_result.get("resolution_indicators", [])
        )
        signals["length"] = {
            "triggered": too_long,
            "data": {"reason": length_reason},
            "weight": self.weights["length"]
        }

        # Calcul du score d'escalade
        escalation_score = sum(
            signal["weight"] if signal["triggered"] else 0
            for signal in signals.values()
        )

        # Seuil adaptatif basé sur le contexte utilisateur
        threshold = self._get_adaptive_threshold(user_context)

        should_escalate = escalation_score >= threshold

        return {
            "escalate": should_escalate,
            "score": escalation_score,
            "threshold": threshold,
            "signals": signals,
            "priority": self._determine_priority(signals),
            "recommended_team": self._recommend_team(signals, user_context)
        }

    def _get_adaptive_threshold(self, user_context: dict) -> float:
        """
        Ajuste le seuil selon le profil utilisateur.
        """
        base_threshold = 0.6

        # VIP clients : seuil plus bas (escalade plus facile)
        if user_context.get("tier") == "enterprise":
            return base_threshold - 0.15

        # Nouveaux clients : seuil légèrement plus bas
        if user_context.get("is_new_customer"):
            return base_threshold - 0.1

        return base_threshold

Handoff contextuel

Préparation du contexte pour l'agent

DEVELOPERpython
class HandoffContextBuilder:
    """
    Construit le contexte complet pour le transfert à un agent.
    """

    async def build(
        self,
        conversation: list,
        user: dict,
        rag_results: list,
        escalation_data: dict
    ) -> dict:
        """
        Prépare tout le contexte pour l'agent humain.
        """
        return {
            "summary": await self._generate_summary(conversation),
            "user_intent": await self._extract_intent(conversation),
            "attempted_solutions": self._extract_bot_responses(conversation),
            "relevant_documentation": self._format_rag_results(rag_results),
            "user_profile": self._format_user_profile(user),
            "escalation_reason": escalation_data,
            "suggested_actions": await self._suggest_actions(
                conversation, escalation_data
            ),
            "sentiment_timeline": self._build_sentiment_timeline(conversation)
        }

    async def _generate_summary(self, conversation: list) -> str:
        """
        Génère un résumé concis de la conversation.
        """
        prompt = f"""
        Résume cette conversation support en 2-3 phrases pour un agent humain.
        Inclus: le problème principal, ce qui a été essayé, où on en est.

        Conversation:
        {self._format_conversation(conversation)}

        Résumé:
        """

        return await self.llm.generate(prompt, temperature=0.2)

    async def _extract_intent(self, conversation: list) -> dict:
        """
        Extrait l'intention principale de l'utilisateur.
        """
        user_messages = [
            m["content"] for m in conversation
            if m["role"] == "user"
        ]

        prompt = f"""
        Analyse ces messages utilisateur et identifie:
        1. L'intention principale
        2. Les sous-demandes éventuelles
        3. L'urgence perçue

        Messages: {json.dumps(user_messages)}

        Réponds en JSON:
        {{
            "primary_intent": "...",
            "secondary_intents": ["..."],
            "perceived_urgency": "low/medium/high",
            "key_entities": ["..."]
        }}
        """

        result = await self.llm.generate(prompt, temperature=0.1)
        return json.loads(result)

    def _build_sentiment_timeline(self, conversation: list) -> list:
        """
        Construit une timeline du sentiment utilisateur.
        """
        timeline = []
        for i, msg in enumerate(conversation):
            if msg["role"] == "user":
                timeline.append({
                    "turn": i,
                    "sentiment": msg.get("sentiment", "neutral"),
                    "excerpt": msg["content"][:50] + "..."
                })
        return timeline

Message de transition

DEVELOPERpython
class TransitionMessageGenerator:
    """
    Génère le message de transition vers l'agent humain.
    """

    TEMPLATES = {
        "explicit_request": """
Je vous mets en relation avec un de nos conseillers.
Temps d'attente estimé : {wait_time}.

En attendant, voici un résumé que je transmets à l'agent :
{summary}
        """,

        "frustration": """
Je comprends votre frustration et je vous transfère
immédiatement vers un de nos experts qui pourra vous aider.

L'agent aura accès à notre conversation et pourra
reprendre là où nous en sommes.
        """,

        "complexity": """
Cette question nécessite l'expertise d'un de nos conseillers
spécialisés. Je vous transfère vers {team}.

J'ai préparé un résumé de notre échange pour l'agent.
        """,

        "low_confidence": """
Pour vous garantir une réponse précise, je préfère
vous mettre en relation avec un conseiller.

Temps d'attente estimé : {wait_time}.
        """
    }

    def generate(
        self,
        reason: str,
        context: dict,
        wait_time: str
    ) -> str:
        """
        Génère un message de transition adapté.
        """
        template = self.TEMPLATES.get(reason, self.TEMPLATES["low_confidence"])

        return template.format(
            wait_time=wait_time,
            summary=context.get("summary", ""),
            team=context.get("team", "notre équipe support")
        )

Métriques d'escalade

Dashboard d'escalade

DEVELOPERpython
class EscalationMetrics:
    """
    Collecte et analyse les métriques d'escalade.
    """

    async def get_dashboard(self, period_days: int = 30) -> dict:
        """
        Génère le dashboard des métriques d'escalade.
        """
        escalations = await self._get_escalations(period_days)
        total_conversations = await self._get_total_conversations(period_days)

        metrics = {
            "overview": {
                "total_conversations": total_conversations,
                "escalated": len(escalations),
                "escalation_rate": len(escalations) / total_conversations,
                "auto_resolution_rate": 1 - (len(escalations) / total_conversations)
            },
            "by_reason": self._group_by_reason(escalations),
            "by_team": self._group_by_team(escalations),
            "timing": {
                "avg_turns_before_escalation": self._avg_turns(escalations),
                "avg_time_to_escalate": self._avg_time(escalations),
                "avg_wait_time_after": self._avg_wait_after(escalations)
            },
            "satisfaction": {
                "csat_after_escalation": self._csat_after(escalations),
                "csat_no_escalation": await self._csat_no_escalation(period_days),
                "resolution_rate_after": self._resolution_rate(escalations)
            },
            "quality": {
                "unnecessary_escalations": self._unnecessary_rate(escalations),
                "missed_escalations": await self._missed_rate(period_days),
                "avg_handoff_quality": self._handoff_quality(escalations)
            }
        }

        return metrics

    def _unnecessary_rate(self, escalations: list) -> float:
        """
        Calcule le taux d'escalades inutiles.
        (Escalades où l'agent a résolu en < 1 message)
        """
        unnecessary = [
            e for e in escalations
            if e.get("agent_messages_to_resolve", 0) <= 1
        ]
        return len(unnecessary) / len(escalations) if escalations else 0

Intégration avec Ailog

DEVELOPERpython
from ailog import AilogClient

client = AilogClient(api_key="your-key")

# Configuration de l'escalade
client.escalation.configure(
    thresholds={
        "confidence_min": 0.6,
        "frustration_sensitivity": 0.7,
        "max_turns": 10
    },
    vip_override={
        "enterprise": {"threshold_modifier": -0.15}
    },
    handoff={
        "include_summary": True,
        "include_sentiment_timeline": True,
        "suggested_actions": True
    }
)

# Le système gère automatiquement l'escalade
response = client.chat(
    message="Je veux parler à quelqu'un !",
    session_id="session_123"
)

if response.escalated:
    print(f"Transfert vers: {response.assigned_team}")
    print(f"Raison: {response.escalation_reason}")

Conclusion

L'escalade intelligente est l'ingrédient secret d'un support hybride performant. En combinant détection multi-signal, seuils adaptatifs et handoff contextuel, vous maximisez l'automatisation tout en garantissant une expérience humaine quand c'est nécessaire.

Ressources complémentaires

• Classification automatique des tickets - Routage intelligent
• Zendesk + RAG - Intégration Zendesk
• Intercom + RAG - Intégration Intercom
• RAG pour le support client - Guide pilier