Automatische Klassifizierung von Tickets mit RAG

TL;DR

Die automatische Klassifizierung von Tickets per RAG eliminiert manuelles Sorting und reduziert die Zeit bis zur ersten Antwort um 60 %. Das System analysiert den Ticketinhalt, vergleicht ihn mit Ihrer Wissensdatenbank und dem Verlauf gelöster Tickets, um Kategorie, Priorität und optimales Team zu bestimmen. Dieser Leitfaden behandelt Klassifizierungsmodelle, Implementierung und Strategien zur kontinuierlichen Verbesserung.

Warum die Klassifizierung automatisieren?

Die Kosten des manuellen Sortierens

Das manuelle Triage von Tickets erzeugt Ineffizienzen:

ROI der automatischen Klassifizierung

Unternehmen, die die RAG-Klassifizierung eingeführt haben, berichten:

• 60% de réduction du temps de première réponse
• 80% de précision de classification (vs 75% humain)
• 90% de réduction du temps de tri
• 35% d'amélioration du taux de résolution premier contact

Architektur der RAG-Klassifizierung

Systemübersicht

┌─────────────────────────────────────────────────────────────┐
│                 Eingehendes Ticket                            │
│  (Betreff, Beschreibung, E-Mail, Metadaten)                  │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                 Feature-Extraktion                           │
│  • Normalisierter Text                                       │
│  • Erkannte Entitäten (Produkt, Version, etc.)               │
│  • Analysiertes Sentiment                                    │
│  • Lexikalische Dringlichkeit                                │
└─────────────────────────┬───────────────────────────────────┘
                          │
           ┌──────────────┴──────────────┐
           ▼                              ▼
┌─────────────────────┐      ┌─────────────────────────────────┐
│  Ähnlichkeit gelöste│      │    RAG-Klassifizierung          │
│  Tickets            │      │    • KB-Suche                   │
└──────────┬──────────┘      │    • Kategorie-Mapping          │
           │                  └──────────────┬──────────────────┘
           └──────────────┬───────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    Endentscheidung                           │
│  • Kategorie                                                 │
│  • Priorität                                                 │
│  • Zugewiesenes Team                                         │
│  • Konfidenz                                                 │
└─────────────────────────────────────────────────────────────┘

Mehrdimensionale Klassifizierung

DEVELOPERpython
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class TicketCategory(Enum):
    BILLING = "billing"
    TECHNICAL = "technical"
    ACCOUNT = "account"
    FEATURE_REQUEST = "feature_request"
    BUG_REPORT = "bug_report"
    SALES = "sales"
    OTHER = "other"

class Priority(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

@dataclass
class ClassificationResult:
    category: TicketCategory
    category_confidence: float
    subcategory: Optional[str]
    priority: Priority
    priority_confidence: float
    assigned_team: str
    assigned_agent: Optional[str]
    suggested_tags: list[str]
    similar_tickets: list[dict]
    estimated_resolution_time: Optional[int]  # minutes

Implementierung des RAG-Klassifizierers

Hauptklassifizierer

DEVELOPERpython
class RAGTicketClassifier:
    """
    Klassifiziert Tickets durch die Kombination von RAG und ML.
    """

    def __init__(
        self,
        rag_client,
        llm_client,
        ticket_history_index
    ):
        self.rag = rag_client
        self.llm = llm_client
        self.history = ticket_history_index

    async def classify(self, ticket: dict) -> ClassificationResult:
        """
        Vollständige Klassifizierung eines Tickets.
        """
        # 1. Extraktion der Features
        features = await self._extract_features(ticket)

        # 2. Suche nach ähnlichen, bereits gelösten Tickets
        similar = await self._find_similar_resolved(ticket, features)

        # 3. Suche in der KB für Kontext
        kb_context = await self._search_kb(ticket)

        # 4. Klassifizierung durch LLM mit Kontext
        classification = await self._llm_classify(
            ticket=ticket,
            features=features,
            similar_tickets=similar,
            kb_context=kb_context
        )

        # 5. Bestimmen der Priorität
        priority = await self._determine_priority(
            ticket=ticket,
            features=features,
            classification=classification
        )

        # 6. An Routing (Team)
        team, agent = await self._route_ticket(
            classification=classification,
            priority=priority,
            ticket=ticket
        )

        return ClassificationResult(
            category=classification["category"],
            category_confidence=classification["confidence"],
            subcategory=classification.get("subcategory"),
            priority=priority["level"],
            priority_confidence=priority["confidence"],
            assigned_team=team,
            assigned_agent=agent,
            suggested_tags=classification.get("tags", []),
            similar_tickets=similar[:3],
            estimated_resolution_time=self._estimate_resolution_time(
                similar, classification
            )
        )

    async def _extract_features(self, ticket: dict) -> dict:
        """
        Extrahiert die Features des Tickets für die Klassifizierung.
        """
        text = f"{ticket.get('subject', '')} {ticket.get('description', '')}"

        # Erkennung von Entitäten
        entities = await self._extract_entities(text)

        # Sentiment-Analyse
        sentiment = await self._analyze_sentiment(text)

        # Erkennung lexikalischer Dringlichkeit
        urgency_words = [
            "urgent", "critique", "bloqué", "production",
            "down", "erreur", "impossible", "immédiat"
        ]
        urgency_score = sum(
            1 for word in urgency_words
            if word.lower() in text.lower()
        ) / len(urgency_words)

        return {
            "text": text,
            "entities": entities,
            "sentiment": sentiment,
            "urgency_score": urgency_score,
            "word_count": len(text.split()),
            "has_attachment": bool(ticket.get("attachments")),
            "is_reply": ticket.get("is_reply", False)
        }

    async def _find_similar_resolved(
        self,
        ticket: dict,
        features: dict
    ) -> list:
        """
        Findet ähnliche, bereits gelöste Tickets.
        """
        # Vektorsuche im Verlauf
        query = f"{ticket.get('subject', '')} {ticket.get('description', '')}"

        results = await self.history.search(
            query=query,
            filter={"status": "resolved"},
            top_k=10
        )

        # Anreicherung mit Auflösungsmetadaten
        similar = []
        for result in results:
            similar.append({
                "id": result.metadata["ticket_id"],
                "subject": result.metadata["subject"],
                "category": result.metadata["category"],
                "resolution_time": result.metadata["resolution_time"],
                "assigned_team": result.metadata["team"],
                "score": result.score
            })

        return similar

Klassifizierung durch LLM

DEVELOPERpython
CLASSIFICATION_PROMPT = """Du bist ein Experte für die Klassifizierung von Support-Tickets.

ZU KLASSIFIZIERENDES TICKET:
Betreff: {subject}
Beschreibung: {description}

EXTRAHIERTE FEATURES:
- Erkannte Entitäten: {entities}
- Sentiment: {sentiment}
- Dringlichkeits-Score: {urgency_score}

ÄHNLICHE GELÖSTE TICKETS:
{similar_tickets}

RELEVANTE KB-ARTIKEL:
{kb_context}

VERFÜGBARE KATEGORIEN:
- billing: Abrechnung, Zahlungen, Abonnements
- technical: Technische Probleme, Bugs, Fehler
- account: Kontoverwaltung, Zugriff, Berechtigungen
- feature_request: Feature-Anfragen
- bug_report: Bug-Meldungen
- sales: Vertriebsfragen, Angebote
- other: Sonstiges

KLASSIFIZIERE DIESES TICKET:
1. Bestimme die Hauptkategorie
2. Identifiziere gegebenenfalls eine Unterkategorie
3. Schlage Tags vor
4. Bewerte deine Konfidenz (0-1)

Antworte in JSON:
{{
    "category": "...",
    "subcategory": "...",
    "tags": ["...", "..."],
    "confidence": 0.X,
    "reasoning": "..."
}}
"""

async def _llm_classify(
    self,
    ticket: dict,
    features: dict,
    similar_tickets: list,
    kb_context: list
) -> dict:
    """
    Klassifizierung durch LLM mit RAG-Kontext.
    """
    prompt = CLASSIFICATION_PROMPT.format(
        subject=ticket.get("subject", ""),
        description=ticket.get("description", ""),
        entities=json.dumps(features["entities"]),
        sentiment=features["sentiment"],
        urgency_score=features["urgency_score"],
        similar_tickets=self._format_similar(similar_tickets),
        kb_context=self._format_kb_context(kb_context)
    )

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

Bestimmung der Priorität

Multi-Faktor-Scoring

DEVELOPERpython
class PriorityScorer:
    """
    Bestimmt die Priorität basierend auf mehreren Faktoren.
    """

    def __init__(self, config: dict):
        self.weights = config.get("weights", {
            "urgency_lexical": 0.2,
            "sentiment": 0.15,
            "customer_tier": 0.25,
            "category_severity": 0.2,
            "similar_tickets_priority": 0.1,
            "sla_risk": 0.1
        })

        self.category_severity = config.get("category_severity", {
            "technical": 0.8,
            "billing": 0.6,
            "bug_report": 0.7,
            "account": 0.5,
            "feature_request": 0.2,
            "sales": 0.4,
            "other": 0.3
        })

    async def score(
        self,
        ticket: dict,
        features: dict,
        classification: dict,
        customer: dict
    ) -> dict:
        """
        Berechnet den Prioritäts-Score.
        """
        scores = {}

        # 1. Lexikalische Dringlichkeit
        scores["urgency_lexical"] = features["urgency_score"]

        # 2. Negatives Sentiment
        sentiment_score = 0
        if features["sentiment"] == "negative":
            sentiment_score = 0.8
        elif features["sentiment"] == "very_negative":
            sentiment_score = 1.0
        scores["sentiment"] = sentiment_score

        # 3. Kundentier
        customer_tier = customer.get("tier", "standard")
        tier_scores = {"enterprise": 1.0, "business": 0.7, "standard": 0.3}
        scores["customer_tier"] = tier_scores.get(customer_tier, 0.3)

        # 4. Schwere der Kategorie
        category = classification["category"]
        scores["category_severity"] = self.category_severity.get(category, 0.5)

        # 5. Priorität ähnlicher Tickets
        if classification.get("similar_tickets"):
            avg_priority = self._average_similar_priority(
                classification["similar_tickets"]
            )
            scores["similar_tickets_priority"] = avg_priority
        else:
            scores["similar_tickets_priority"] = 0.5

        # 6. SLA-Risiko
        scores["sla_risk"] = await self._calculate_sla_risk(customer)

        # Gewichteter Endscore
        final_score = sum(
            scores[key] * self.weights[key]
            for key in self.weights
        )

        # Mapping zu einer Priorität
        priority = self._score_to_priority(final_score)

        return {
            "level": priority,
            "score": final_score,
            "components": scores,
            "confidence": self._calculate_confidence(scores)
        }

    def _score_to_priority(self, score: float) -> Priority:
        if score >= 0.8:
            return Priority.CRITICAL
        elif score >= 0.6:
            return Priority.HIGH
        elif score >= 0.4:
            return Priority.MEDIUM
        else:
            return Priority.LOW

Intelligentes Routing

Kompetenzbasierte Zuweisung

DEVELOPERpython
class SkillBasedRouter:
    """
    Leitet Tickets an das optimale Team und den optimalen Agenten weiter.
    """

    def __init__(self, team_config: dict, agent_skills: dict):
        self.teams = team_config
        self.skills = agent_skills

    async def route(
        self,
        classification: ClassificationResult,
        ticket: dict,
        customer: dict
    ) -> tuple[str, Optional[str]]:
        """
        Bestimmt das Team und den optimalen Agenten.
        """
        # 1. Bestimme das Team anhand der Kategorie
        category = classification.category.value
        primary_team = self.teams.get(category, {}).get("primary_team")

        # 2. Prüfe spezielle Routing-Regeln
        team = await self._apply_routing_rules(
            primary_team, classification, customer
        )

        # 3. Wähle den optimalen Agenten
        agent = await self._select_best_agent(team, classification, ticket)

        return team, agent

    async def _apply_routing_rules(
        self,
        primary_team: str,
        classification: ClassificationResult,
        customer: dict
    ) -> str:
        """
        Wendet spezielle Routing-Regeln an.
        """
        rules = [
            # Enterprise-Kunden an dediziertes Team
            {
                "condition": lambda c, cls: c.get("tier") == "enterprise",
                "team": "enterprise_support"
            },
            # Kritische Tickets an L2
            {
                "condition": lambda c, cls: cls.priority == Priority.CRITICAL,
                "team": "tier2_support"
            },
            # Feature-Anfragen an das Produktteam
            {
                "condition": lambda c, cls: cls.category == TicketCategory.FEATURE_REQUEST,
                "team": "product_team"
            },
            # Sales an das Vertriebsteam
            {
                "condition": lambda c, cls: cls.category == TicketCategory.SALES,
                "team": "sales_team"
            }
        ]

        for rule in rules:
            if rule["condition"](customer, classification):
                return rule["team"]

        return primary_team

    async def _select_best_agent(
        self,
        team: str,
        classification: ClassificationResult,
        ticket: dict
    ) -> Optional[str]:
        """
        Wählt den qualifiziertesten verfügbaren Agenten aus.
        """
        # Team-Agenten abrufen
        agents = await self._get_available_agents(team)

        if not agents:
            return None

        # Jeden Agenten bewerten
        scored_agents = []
        for agent in agents:
            score = self._score_agent(agent, classification)
            scored_agents.append((agent, score))

        # Nach Score und Arbeitsbelastung sortieren
        scored_agents.sort(
            key=lambda x: (x[1], -x[0]["current_tickets"]),
            reverse=True
        )

        return scored_agents[0][0]["id"] if scored_agents else None

    def _score_agent(
        self,
        agent: dict,
        classification: ClassificationResult
    ) -> float:
        """
        Bewertet einen Agenten basierend auf seinen Fähigkeiten.
        """
        score = 0.0

        # Kompetenz-Abgleich
        agent_skills = set(agent.get("skills", []))
        required_skills = set(classification.suggested_tags)
        if required_skills:
            skill_match = len(agent_skills & required_skills) / len(required_skills)
            score += skill_match * 0.4

        # Erfahrung in der Kategorie
        category_exp = agent.get("category_experience", {})
        category = classification.category.value
        if category in category_exp:
            score += min(category_exp[category] / 100, 1.0) * 0.3

        # Lösungsrate
        resolution_rate = agent.get("first_contact_resolution_rate", 0.5)
        score += resolution_rate * 0.3

        return score

Indexierung gelöster Tickets

Erstellung des Lernindex

DEVELOPERpython
class TicketHistoryIndexer:
    """
    Indiziert gelöste Tickets für das Lernen.
    """

    def __init__(self, rag_client):
        self.rag = rag_client

    async def index_resolved_ticket(self, ticket: dict):
        """
        Indiziert ein gelöstes Ticket, um zukünftige Klassifizierungen zu verbessern.
        """
        # Indexierbaren Inhalt erstellen
        content = f"""
        Betreff: {ticket['subject']}

        Beschreibung: {ticket['description']}

        Lösung: {ticket.get('resolution_note', '')}
        """

        # Metadaten für Filterung und Analytics
        metadata = {
            "ticket_id": ticket["id"],
            "subject": ticket["subject"],
            "category": ticket["category"],
            "subcategory": ticket.get("subcategory"),
            "priority": ticket["priority"],
            "team": ticket["assigned_team"],
            "agent": ticket["assigned_agent"],
            "resolution_time": ticket["resolution_time_minutes"],
            "first_contact_resolution": ticket.get("fcr", False),
            "customer_satisfaction": ticket.get("csat"),
            "created_at": ticket["created_at"],
            "resolved_at": ticket["resolved_at"],
            "tags": ticket.get("tags", [])
        }

        await self.rag.index_document(
            content=content,
            metadata=metadata,
            doc_id=f"ticket_{ticket['id']}",
            collection="resolved_tickets"
        )

    async def batch_index(self, tickets: list[dict]) -> dict:
        """
        Batch-Indexierung historischer Tickets.
        """
        stats = {"indexed": 0, "errors": 0}

        for ticket in tickets:
            try:
                await self.index_resolved_ticket(ticket)
                stats["indexed"] += 1
            except Exception as e:
                stats["errors"] += 1
                print(f"Error indexing ticket {ticket['id']}: {e}")

        return stats

Kontinuierliche Verbesserung

Feedback-Loop

DEVELOPERpython
class ClassificationFeedback:
    """
    Sammelt und nutzt Feedback zur Verbesserung der Klassifizierung.
    """

    async def record_correction(
        self,
        ticket_id: str,
        original_classification: ClassificationResult,
        corrected_category: str,
        corrected_priority: str,
        corrected_team: str
    ):
        """
        Speichert eine Korrektur der Klassifizierung.
        """
        feedback = {
            "ticket_id": ticket_id,
            "original": {
                "category": original_classification.category.value,
                "priority": original_classification.priority.value,
                "team": original_classification.assigned_team,
                "confidence": original_classification.category_confidence
            },
            "corrected": {
                "category": corrected_category,
                "priority": corrected_priority,
                "team": corrected_team
            },
            "timestamp": datetime.utcnow()
        }

        await self.db.insert("classification_feedback", feedback)

        # Bei wiederkehrendem Muster eine Regel erstellen
        await self._check_for_patterns(feedback)

    async def _check_for_patterns(self, feedback: dict):
        """
        Erkennt Korrektur-Patterns, um Regeln zu erstellen.
        """
        # Nach ähnlichen, kürzlichen Korrekturen suchen
        similar = await self.db.query(
            "classification_feedback",
            filter={
                "original.category": feedback["original"]["category"],
                "corrected.category": feedback["corrected"]["category"],
                "timestamp": {"$gte": datetime.utcnow() - timedelta(days=7)}
            }
        )

        if len(similar) >= 5:
            # Muster erkannt, Regel erstellen
            await self._create_rule_from_pattern(similar, feedback)

    async def calculate_accuracy(self, period_days: int = 30) -> dict:
        """
        Berechnet die Klassifizierungsgenauigkeit über einen Zeitraum.
        """
        all_classifications = await self._get_classifications(period_days)
        corrections = await self._get_corrections(period_days)

        total = len(all_classifications)
        corrected = len(corrections)

        # Genauigkeit pro Kategorie
        category_accuracy = {}
        for cat in TicketCategory:
            cat_total = len([c for c in all_classifications if c["category"] == cat.value])
            cat_corrected = len([c for c in corrections if c["original"]["category"] == cat.value])
            if cat_total > 0:
                category_accuracy[cat.value] = 1 - (cat_corrected / cat_total)

        return {
            "overall_accuracy": 1 - (corrected / total) if total > 0 else 0,
            "total_classifications": total,
            "corrections": corrected,
            "by_category": category_accuracy,
            "period_days": period_days
        }

Integration mit Ailog

DEVELOPERpython
from ailog import AilogClient

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

# Konfiguration der Klassifizierung
client.classification.configure(
    categories=["billing", "technical", "account", "sales"],
    priority_weights={
        "customer_tier": 0.3,
        "urgency": 0.2,
        "sentiment": 0.2,
        "category": 0.3
    },
    auto_learn=True  # Lernt aus Korrekturen
)

# Ein Ticket klassifizieren
result = client.classification.classify(
    subject="Fehler bei der Zahlung",
    description="Ich kann meinen Kauf nicht abschließen...",
    customer_id="cust_123"
)

print(result.category)  # "billing"
print(result.priority)  # "high"
print(result.assigned_team)  # "billing_support"

Conclusion

Die automatische Klassifizierung per RAG verwandelt das Ticket-Triage von einem Engpass in einen flüssigen und präzisen Prozess. Die Kombination von Vektorsuche nach ähnlichen Tickets und kontextueller LLM-Klassifizierung ermöglicht eine höhere Genauigkeit als manuelle Sortierung und eliminiert Routing-Verzögerungen.

Weiterführende Ressourcen

• Intelligente Eskalation - Wann an einen Menschen weiterleiten
• Zendesk + RAG - Zendesk-Integration
• Freshdesk + RAG - Freshdesk-Integration
• RAG für den Kundensupport - Leitfaden-Übersicht