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.

Das Eskalations-Dilemma

Zu wenig Eskalation

Wenn der Bot weitermacht, obwohl er übergeben sollte:

Zu viel Eskalation

Wenn der Bot zu schnell übergibt:

Der Sweet Spot

Ziel ist es, das richtige Gleichgewicht zu finden:

• 70-80% automatische Lösung
• 95%+ CSAT bei Eskalationen
• < 2 Minuten zwischen Eskalationsentscheidung und menschlichem Agenten
• Vollständiger Kontext an den Agenten übermittelt

Erkennungssignale für Eskalation

1. Explizite Anfrage

Der Nutzer fragt eindeutig nach einem Menschen:

DEVELOPERpython
class ExplicitEscalationDetector:
    """
    Erkennt explizite Anfragen, mit einem Menschen zu sprechen.
    """

    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]:
        """
        Erkennt eine explizite Eskalationsanfrage.
        """
        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. Unzureichendes RAG-Vertrauen

Das System hat nicht genug Sicherheit, um zu antworten:

DEVELOPERpython
class ConfidenceBasedEscalation:
    """
    Eskalation basierend auf dem Vertrauen des RAG-Systems.
    """

    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]:
        """
        Bestimmt, ob eine Eskalation basierend auf dem Vertrauen erforderlich ist.
        """
        # Retrieval-Score (bestes Dokument)
        best_retrieval = max(retrieval_scores) if retrieval_scores else 0

        # Gewichteter kombinierter Score
        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. Frustrationserkennung

Analyse von Sentiment und Frustrationsmustern:

DEVELOPERpython
class FrustrationDetector:
    """
    Erkennt Nutzerfrustration für präventive Eskalation.
    """

    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]:
        """
        Erkennt Frustration basierend auf Nachricht und Verlauf.
        """
        frustration_signals = []

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

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

        # 3. Wiederholungsmuster
        if self._detect_repetition(message, conversation_history):
            frustration_signals.append("user_repeating")

        # 4. Zunehmend gereizter 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. Immer längere Nachrichten (Zeichen von 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:
        """
        Erkennt, ob der Nutzer seine Frage wiederholt.
        """
        user_messages = [
            h["content"] for h in history
            if h["role"] == "user"
        ][-3:]

        if not user_messages:
            return False

        # Ähnlichkeit mit vorherigen Nachrichten
        for prev in user_messages:
            similarity = self._calculate_similarity(message, prev)
            if similarity > 0.7:
                return True

        return False

4. Übermäßige Komplexität

Die Frage übersteigt die Fähigkeiten des Bots:

DEVELOPERpython
class ComplexityAnalyzer:
    """
    Analysiert die Komplexität der Nutzeranfrage.
    """

    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]:
        """
        Analysiert die Komplexität und empfiehlt eine Aktion.
        """
        complexity_score = 0
        indicators_found = []

        # 1. Lexikalische Indikatoren
        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. Nachrichtenlänge
        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. Mehrere Fragen
        question_marks = message.count("?")
        if question_marks > 2:
            complexity_score += 0.2
            indicators_found.append(f"multiple_questions:{question_marks}")

        # 4. Geringe KB-Abdeckung = komplexes oder undokumentiertes Thema
        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. Zu lange Konversation

Die Konversation kommt ohne Lösung nicht voran:

DEVELOPERpython
class ConversationLengthMonitor:
    """
    Überwacht die Konversationslänge für Eskalation.
    """

    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]:
        """
        Prüft, ob die Konversation zu lange dauert.
        """
        total_turns = len([m for m in conversation if m["role"] == "user"])

        # Zu viele Runden insgesamt
        if total_turns >= self.max_total:
            return True, f"max_turns_exceeded:{total_turns}"

        # Keine Lösung nach X Runden
        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:
        """
        Zählt die Runden seit dem letzten Lösungsindikator.
        """
        # Indikatoren: Danksagung, Verständnisbestätigung usw.
        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

Kombiniertes Eskalationssystem

Eskalations-Orchestrator

DEVELOPERpython
class EscalationOrchestrator:
    """
    Kombiniert alle Signale, um über die Eskalation zu entscheiden.
    """

    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

        # Gewichtung der verschiedenen Signale
        self.weights = {
            "explicit": 1.0,       # Bei Anfrage immer eskalieren
            "frustration": 0.9,   # Hohe Priorität
            "confidence": 0.7,    # Wichtig
            "complexity": 0.6,    # Signifikant
            "length": 0.5         # Beitragend
        }

    async def evaluate(
        self,
        message: str,
        conversation: list,
        rag_result: dict,
        user_context: dict
    ) -> dict:
        """
        Wertet alle Signale aus und entscheidet über die Eskalation.
        """
        signals = {}

        # 1. Explizite Anfrage (absolute Priorität)
        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. RAG-Vertrauen
        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. Komplexität
        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. Konversationslänge
        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"]
        }

        # Berechnung des Eskalationsscores
        escalation_score = sum(
            signal["weight"] if signal["triggered"] else 0
            for signal in signals.values()
        )

        # Adaptiver Schwellenwert basierend auf dem Nutzerkontext
        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:
        """
        Passt den Schwellenwert je nach Nutzerprofil an.
        """
        base_threshold = 0.6

        # VIP-Kunden: niedrigerer Schwellenwert (leichtere Eskalation)
        if user_context.get("tier") == "enterprise":
            return base_threshold - 0.15

        # Neukunden: leicht niedrigerer Schwellenwert
        if user_context.get("is_new_customer"):
            return base_threshold - 0.1

        return base_threshold

Kontextueller Handoff

Kontextaufbereitung für den Agenten

DEVELOPERpython
class HandoffContextBuilder:
    """
    Erstellt den vollständigen Kontext für die Übergabe an einen Agenten.
    """

    async def build(
        self,
        conversation: list,
        user: dict,
        rag_results: list,
        escalation_data: dict
    ) -> dict:
        """
        Bereitet den gesamten Kontext für den menschlichen Agenten vor.
        """
        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:
        """
        Erstellt eine prägnante Zusammenfassung der Konversation.
        """
        prompt = f"""
        Fasse dieses Support-Gespräch in 2-3 Sätzen für einen menschlichen Agenten zusammen.
        Enthalte: das Hauptproblem, was bereits versucht wurde, den aktuellen Stand.

        Konversation:
        {self._format_conversation(conversation)}

        Zusammenfassung:
        """

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

    async def _extract_intent(self, conversation: list) -> dict:
        """
        Extrahiert die Hauptabsicht des Nutzers.
        """
        user_messages = [
            m["content"] for m in conversation
            if m["role"] == "user"
        ]

        prompt = f"""
        Analysiere diese Nutzernachrichten und identifiziere:
        1. Die Hauptabsicht
        2. Etwaige Nebenanfragen
        3. Die wahrgenommene Dringlichkeit

        Nachrichten: {json.dumps(user_messages)}

        Antworte im JSON-Format:
        {{
            "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:
        """
        Erstellt eine Timeline des Nutzer-Sentiments.
        """
        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

Übergangsnachricht

DEVELOPERpython
class TransitionMessageGenerator:
    """
    Generiert die Übergangsnachricht zum menschlichen Agenten.
    """

    TEMPLATES = {
        "explicit_request": """
Ich verbinde Sie mit einem unserer Berater.
Geschätzte Wartezeit: {wait_time}.

In der Zwischenzeit hier eine Zusammenfassung, die ich an den Agenten weitergebe:
{summary}
        """,

        "frustration": """
Ich verstehe Ihre Frustration und verbinde Sie
sofort mit einem unserer Experten, der Ihnen helfen kann.

Der Agent hat Zugriff auf unser Gespräch und kann
dort weitermachen, wo wir stehen geblieben sind.
        """,

        "complexity": """
Diese Frage erfordert die Expertise eines unserer
spezialisierten Berater. Ich verbinde Sie mit {team}.

Ich habe eine Zusammenfassung unseres Austauschs für den Agenten vorbereitet.
        """,

        "low_confidence": """
Um Ihnen eine präzise Antwort zu garantieren, verbinde
ich Sie lieber mit einem Berater.

Geschätzte Wartezeit: {wait_time}.
        """
    }

    def generate(
        self,
        reason: str,
        context: dict,
        wait_time: str
    ) -> str:
        """
        Generiert eine passende Übergangsnachricht.
        """
        template = self.TEMPLATES.get(reason, self.TEMPLATES["low_confidence"])

        return template.format(
            wait_time=wait_time,
            summary=context.get("summary", ""),
            team=context.get("team", "unser Support-Team")
        )

Eskalationsmetriken

Eskalations-Dashboard

DEVELOPERpython
class EscalationMetrics:
    """
    Erfasst und analysiert Eskalationsmetriken.
    """

    async def get_dashboard(self, period_days: int = 30) -> dict:
        """
        Generiert das Dashboard der Eskalationsmetriken.
        """
        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:
        """
        Berechnet die Rate unnötiger Eskalationen.
        (Eskalationen, bei denen der Agent in < 1 Nachricht gelöst hat)
        """
        unnecessary = [
            e for e in escalations
            if e.get("agent_messages_to_resolve", 0) <= 1
        ]
        return len(unnecessary) / len(escalations) if escalations else 0

Integration mit Ailog

DEVELOPERpython
from ailog import AilogClient

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

# Konfiguration der Eskalation
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
    }
)

# Das System steuert die Eskalation automatisch
response = client.chat(
    message="Ich möchte mit jemandem sprechen!",
    session_id="session_123"
)

if response.escalated:
    print(f"Übergabe an: {response.assigned_team}")
    print(f"Grund: {response.escalation_reason}")

Fazit

Intelligente Eskalation ist die geheime Zutat eines leistungsstarken Hybrid-Supports. Durch die Kombination von Multi-Signal-Erkennung, adaptiven Schwellenwerten und kontextuellem Handoff maximieren Sie die Automatisierung und garantieren gleichzeitig ein menschliches Erlebnis, wenn es nötig ist.

Weiterführende Ressourcen

• Automatische Ticket-Klassifizierung - Intelligentes Routing
• Zendesk + RAG - Zendesk-Integration
• Intercom + RAG - Intercom-Integration
• RAG für den Kundensupport - Pillar-Guide