Escalade intelligente : Quand transférer à un humain

TL;DR

L'escalade intelligente est la clé d'un support hybride réussi. Un chatbot RAG doit savoir quand il atteint ses limites et transférer élégamment vers un humain. Ce guide couvre les signaux de détection, les seuils de confiance, le handoff contextuel et les métriques pour optimiser l'équilibre automatisation/humain. Objectif : 70% de résolution automatique avec 95% de satisfaction sur les escalades.

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