Classification automatique des tickets avec RAG

TL;DR

La classification automatique des tickets par RAG élimine le tri manuel et réduit le temps de première réponse de 60%. Le système analyse le contenu du ticket, le compare à votre base de connaissances et à l'historique des tickets résolus pour déterminer la catégorie, la priorité et l'équipe optimale. Ce guide couvre les modèles de classification, l'implémentation et les stratégies d'amélioration continue.

Pourquoi automatiser la classification ?

Le coût du tri manuel

Le triage manuel des tickets génère des inefficacités :

ROI de la classification automatique

Les entreprises ayant déployé la classification RAG rapportent :

• 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:
    """
    Classifie les tickets en combinant RAG et 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:
        """
        Classification complète d'un ticket.
        """
        # 1. Extraction des features
        features = await self._extract_features(ticket)

        # 2. Recherche de tickets similaires résolus
        similar = await self._find_similar_resolved(ticket, features)

        # 3. Recherche dans la KB pour contexte
        kb_context = await self._search_kb(ticket)

        # 4. Classification par LLM avec contexte
        classification = await self._llm_classify(
            ticket=ticket,
            features=features,
            similar_tickets=similar,
            kb_context=kb_context
        )

        # 5. Déterminer la priorité
        priority = await self._determine_priority(
            ticket=ticket,
            features=features,
            classification=classification
        )

        # 6. Router vers l'équipe
        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:
        """
        Extrait les features du ticket pour la classification.
        """
        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:
        """
        Trouve des tickets similaires déjà résolus.
        """
        # 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:
    """
    Classification par LLM avec contexte RAG.
    """
    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:
    """
    Détermine la priorité basée sur plusieurs facteurs.
    """

    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:
        """
        Calcule le score de priorité.
        """
        scores = {}

        # 1. Urgence lexicale
        scores["urgency_lexical"] = features["urgency_score"]

        # 2. Sentiment négatif
        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. Tier client
        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. Sévérité de la catégorie
        category = classification["category"]
        scores["category_severity"] = self.category_severity.get(category, 0.5)

        # 5. Priorité des tickets similaires
        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. Risque SLA
        scores["sla_risk"] = await self._calculate_sla_risk(customer)

        # Score final pondéré
        final_score = sum(
            scores[key] * self.weights[key]
            for key in self.weights
        )

        # Mapper vers une priorité
        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:
    """
    Route les tickets vers l'équipe et l'agent optimal.
    """

    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]]:
        """
        Détermine l'équipe et l'agent optimal.
        """
        # 1. Déterminer l'équipe par catégorie
        category = classification.category.value
        primary_team = self.teams.get(category, {}).get("primary_team")

        # 2. Vérifier les règles de routage spéciales
        team = await self._apply_routing_rules(
            primary_team, classification, customer
        )

        # 3. Sélectionner l'agent optimal
        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:
        """
        Applique les règles de routage spéciales.
        """
        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]:
        """
        Sélectionne l'agent le plus qualifié disponible.
        """
        # 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:
        """
        Score un agent basé sur ses compétences.
        """
        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:
    """
    Indexe les tickets résolus pour l'apprentissage.
    """

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

    async def index_resolved_ticket(self, ticket: dict):
        """
        Indexe un ticket résolu pour améliorer les futures classifications.
        """
        # 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:
        """
        Indexation en batch des tickets historiques.
        """
        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:
    """
    Collecte et utilise le feedback pour améliorer la classification.
    """

    async def record_correction(
        self,
        ticket_id: str,
        original_classification: ClassificationResult,
        corrected_category: str,
        corrected_priority: str,
        corrected_team: str
    ):
        """
        Enregistre une correction de classification.
        """
        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):
        """
        Détecte les patterns de correction pour créer des règles.
        """
        # 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:
        """
        Calcule la précision de classification sur une période.
        """
        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

La classification automatique par RAG transforme le triage des tickets d'un goulot d'étranglement en processus fluide et précis. Combiner la recherche vectorielle de tickets similaires avec la classification LLM contextuelle permet d'atteindre une précision supérieure au tri manuel, tout en éliminant le délai de routage.

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