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.

Pourquoi automatiser la classification ?

Le coût du tri manuel

Das manuelle Triage von Tickets erzeugt Ineffizienzen:

ROI de la classification automatique

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

Architecture de classification RAG

Vue d'ensemble du système

┌─────────────────────────────────────────────────────────────┐
│                    Ticket entrant                            │
│  (sujet, description, email, metadata)                       │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                 Extraction de features                       │
│  • Texte normalisé                                          │
│  • Entités détectées (produit, version, etc.)               │
│  • Sentiment analysé                                         │
│  • Urgence lexicale                                          │
└─────────────────────────┬───────────────────────────────────┘
                          │
           ┌──────────────┴──────────────┐
           ▼                              ▼
┌─────────────────────┐      ┌─────────────────────────────────┐
│  Similarité tickets │      │    Classification RAG           │
│  résolus            │      │    • Recherche KB               │
└──────────┬──────────┘      │    • Mapping catégorie          │
           │                  └──────────────┬──────────────────┘
           └──────────────┬───────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    Décision finale                           │
│  • Catégorie                                                 │
│  • Priorité                                                  │
│  • Équipe assignée                                           │
│  • Confiance                                                 │
└─────────────────────────────────────────────────────────────┘

Classification multi-dimensionnelle

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

Implémentation du classifieur RAG

Classifieur principal

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', '')}"

        # Détection d'entités
        entities = await self._extract_entities(text)

        # Analyse de sentiment
        sentiment = await self._analyze_sentiment(text)

        # Détection d'urgence lexicale
        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.
        """
        # Recherche vectorielle dans l'historique
        query = f"{ticket.get('subject', '')} {ticket.get('description', '')}"

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

        # Enrichir avec les métadonnées de résolution
        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

Classification par LLM

DEVELOPERpython
CLASSIFICATION_PROMPT = """Tu es un expert en classification de tickets support.

TICKET À CLASSIFIER:
Sujet: {subject}
Description: {description}

FEATURES EXTRAITES:
- Entités détectées: {entities}
- Sentiment: {sentiment}
- Score urgence: {urgency_score}

TICKETS SIMILAIRES RÉSOLUS:
{similar_tickets}

ARTICLES KB PERTINENTS:
{kb_context}

CATÉGORIES DISPONIBLES:
- billing: Facturation, paiements, abonnements
- technical: Problèmes techniques, bugs, erreurs
- account: Gestion de compte, accès, permissions
- feature_request: Demandes de fonctionnalités
- bug_report: Signalement de bugs
- sales: Questions commerciales, devis
- other: Autre

CLASSIFIE CE TICKET:
1. Détermine la catégorie principale
2. Identifie une sous-catégorie si pertinent
3. Suggère des tags
4. Évalue ta confiance (0-1)

Réponds en 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)

Détermination de la priorité

Scoring multi-facteurs

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

Routage intelligent

Assignment basé sur les compétences

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 = [
            # Clients enterprise vers équipe dédiée
            {
                "condition": lambda c, cls: c.get("tier") == "enterprise",
                "team": "enterprise_support"
            },
            # Tickets critiques vers L2
            {
                "condition": lambda c, cls: cls.priority == Priority.CRITICAL,
                "team": "tier2_support"
            },
            # Feature requests vers product
            {
                "condition": lambda c, cls: cls.category == TicketCategory.FEATURE_REQUEST,
                "team": "product_team"
            },
            # Sales vers l'équipe commerciale
            {
                "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.
        """
        # Récupérer les agents de l'équipe
        agents = await self._get_available_agents(team)

        if not agents:
            return None

        # Scorer chaque agent
        scored_agents = []
        for agent in agents:
            score = self._score_agent(agent, classification)
            scored_agents.append((agent, score))

        # Trier par score et charge de travail
        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

        # Match de compétences
        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

        # Expérience sur la catégorie
        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

        # Taux de résolution
        resolution_rate = agent.get("first_contact_resolution_rate", 0.5)
        score += resolution_rate * 0.3

        return score

Indexation des tickets résolus

Création de l'index d'apprentissage

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.
        """
        # Construire le contenu indexable
        content = f"""
        Sujet: {ticket['subject']}

        Description: {ticket['description']}

        Résolution: {ticket.get('resolution_note', '')}
        """

        # Métadonnées pour filtrage et 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

Amélioration continue

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)

        # Si pattern récurrent, créer une règle
        await self._check_for_patterns(feedback)

    async def _check_for_patterns(self, feedback: dict):
        """
        Erkennt Korrektur-Patterns, um Regeln zu erstellen.
        """
        # Chercher des corrections similaires récentes
        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:
            # Pattern détecté, créer une règle
            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)

        # Accuracy par catégorie
        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
        }

Intégration avec Ailog

DEVELOPERpython
from ailog import AilogClient

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

# Configuration de la classification
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  # Apprend des corrections
)

# Classifier un ticket
result = client.classification.classify(
    subject="Erreur lors du paiement",
    description="Je n'arrive pas à finaliser mon achat...",
    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.

Ressources complémentaires

• Escalade intelligente - Quand transférer à un humain
• Zendesk + RAG - Intégration Zendesk
• Freshdesk + RAG - Intégration Freshdesk
• RAG pour le support client - Guide pilier