Text Classification for Customer Support Ticket Routing

Building a Customer Support Ticket Classification System

Objective

Build an end-to-end text classification system for customer support tickets using modern NLP techniques. This system will automatically categorize support tickets to ensure efficient routing and faster response times.


Learning Outcomes

By completing this project, you will:

  • Master text preprocessing and classification techniques
  • Implement transformer-based models for NLP tasks
  • Build production-ready ML pipelines
  • Deploy ML models as web services
  • Evaluate and optimize model performance
  • Handle real-world text data challenges

Skills Gained

  • Building text classification pipelines
  • Using modern NLP libraries and transformers
  • Implementing web services with Flask
  • Processing and analyzing customer support data
  • Deploying machine learning models
  • Working with REST APIs

Tools Required

# Core libraries
pip install transformers
pip install torch
pip install datasets
pip install scikit-learn
pip install flask

# Additional utilities
pip install pandas numpy
pip install nltk
pip install optuna  # For hyperparameter tuning

Project Structure

ticket_classifier/
│
├── data/
│   ├── raw/
│   │   └── support_tickets.csv
│   └── processed/
│       └── cleaned_tickets.csv
│
├── src/
│   ├── preprocessing.py
│   ├── model.py
│   ├── training.py
│   └── api.py
│
├── models/
│   └── saved_models/
│
└── app/
    ├── server.py
    └── templates/

Prerequisites and Theoretical Foundations

1. Python Programming Foundations

  • OOP concepts
  • API development
  • Error handling
  • Web services basics
Click to view Python prerequisites code examples
# Class implementation example
class TicketProcessor:
    def __init__(self, model_type):
        self.model_type = model_type
        
    def process_text(self, text):
        """Process incoming support ticket text"""
        return text.lower().strip()
    
    def handle_error(self, error_type):
        """Error handling example"""
        error_map = {
            'invalid_input': 'Text input is invalid',
            'processing_error': 'Error processing ticket'
        }
        return error_map.get(error_type, 'Unknown error')

# API endpoint example
from flask import Flask, request
app = Flask(__name__)

@app.route('/api/ticket', methods=['POST'])
def process_ticket():
    try:
        ticket_data = request.json
        return {'status': 'processed'}
    except Exception as e:
        return {'error': str(e)}, 400

2. NLP Fundamentals

  • Text preprocessing
  • Tokenization
  • Word embeddings
  • Text normalization
Click to view NLP fundamentals code
import nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
import re

class TextPreprocessor:
    def __init__(self):
        nltk.download('punkt')
        nltk.download('stopwords')
        self.stop_words = set(stopwords.words('english'))
    
    def clean_text(self, text):
        """Basic text cleaning"""
        # Lowercase
        text = text.lower()
        
        # Remove special characters
        text = re.sub(r'[^\w\s]', '', text)
        
        # Tokenize
        tokens = word_tokenize(text)
        
        # Remove stopwords
        tokens = [t for t in tokens if t not in self.stop_words]
        
        return ' '.join(tokens)

# Example usage
preprocessor = TextPreprocessor()
clean_text = preprocessor.clean_text("The printer isn't working!")

3. Deep Learning Concepts

  • Neural network basics
  • Transformer architecture
  • Transfer learning
  • Fine-tuning models
Click to view deep learning concepts code
from transformers import AutoModelForSequenceClassification
import torch.nn as nn

# Simple classifier using transformers
class TicketClassifier(nn.Module):
    def __init__(self, pretrained_model="bert-base-uncased", num_labels=4):
        super().__init__()
        self.transformer = AutoModelForSequenceClassification.from_pretrained(
            pretrained_model,
            num_labels=num_labels
        )
    
    def forward(self, input_ids, attention_mask):
        outputs = self.transformer(
            input_ids=input_ids,
            attention_mask=attention_mask
        )
        return outputs.logits

# Fine-tuning example
def fine_tune_model(model, train_loader, num_epochs=3):
    optimizer = torch.optim.AdamW(model.parameters())
    for epoch in range(num_epochs):
        for batch in train_loader:
            outputs = model(batch['input_ids'], batch['attention_mask'])
            loss = outputs.loss
            loss.backward()
            optimizer.step()

List of Theoretical Concepts

Text Classification
  1. Classification Approaches

    • Rule-based systems
    • Machine learning methods
    • Deep learning approaches
    • Hybrid systems
  2. Key Concepts

    • Feature extraction
    • Document representation
    • Classification algorithms
    • Model evaluation metrics
  3. Common Challenges

    • Imbalanced classes
    • Short text classification
    • Multi-label classification
    • Domain adaptation
NLP Architecture
  1. Transformer Architecture

    • Self-attention mechanism
    • Multi-head attention
    • Position encodings
    • Layer normalization
  2. BERT and Variants

    • Bidirectional encoding
    • Masked language modeling
    • Next sentence prediction
    • Fine-tuning approaches
  3. Key Components

    Input Text -> Tokenization -> Embedding -> 
    Transformer Layers -> Classification Head -> Prediction
    
Transfer Learning
  1. Pre-training

    • Language modeling
    • Masked token prediction
    • Large-scale training
    • Domain-specific pre-training
  2. Fine-tuning Strategies

    # Different fine-tuning approaches
    strategies = {
        'full_fine_tuning': 'Update all model parameters',
        'feature_extraction': 'Freeze base model',
        'gradual_unfreezing': 'Progressively unfreeze layers',
        'layer_wise': 'Different learning rates per layer'
    }
    
  3. Adaptation Techniques

    • Domain adaptation
    • Task-specific heads
    • Prompt tuning
    • Few-shot learning
Deployment considerations
  1. System Architecture

    Client Request -> Load Balancer -> API Server -> 
    Model Server -> Cache -> Database
    
  2. Performance Optimization

    • Model quantization
    • Batch processing
    • Caching strategies
    • Async processing
  3. Monitoring and Maintenance

    • Performance metrics
    • Error tracking
    • Data drift detection
    • Model retraining

Steps and Tasks

1. Data Preprocessing

Set up data preprocessing pipeline:

import pandas as pd
import numpy as np
from transformers import AutoTokenizer
import nltk
from nltk.corpus import stopwords

class TicketPreprocessor:
    def __init__(self, model_name="bert-base-uncased"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        nltk.download('stopwords')
        self.stop_words = set(stopwords.words('english'))
        
    def clean_text(self, text):
        """Basic text cleaning"""
        # Convert to lowercase
        text = text.lower()
        
        # Remove special characters
        text = re.sub(r'[^a-zA-Z\s]', '', text)
        
        # Remove stopwords
        words = text.split()
        words = [w for w in words if w not in self.stop_words]
        
        return ' '.join(words)
    
    def prepare_features(self, texts, max_length=128):
        """Convert texts to model inputs"""
        return self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=max_length,
            return_tensors="pt"
        )
Click to view advanced preprocessing code
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder

class AdvancedTicketPreprocessor:
    def __init__(self, model_name="bert-base-uncased"):
        self.basic_preprocessor = TicketPreprocessor(model_name)
        self.label_encoder = LabelEncoder()
        
    def create_dataset(self, df):
        """Create dataset from DataFrame"""
        # Clean texts
        df['cleaned_text'] = df['text'].apply(self.basic_preprocessor.clean_text)
        
        # Encode labels
        df['encoded_label'] = self.label_encoder.fit_transform(df['category'])
        
        # Split dataset
        train_df, val_df = train_test_split(
            df, test_size=0.2, stratify=df['category']
        )
        
        return {
            'train': self.create_pytorch_dataset(train_df),
            'val': self.create_pytorch_dataset(val_df)
        }
    
    def create_pytorch_dataset(self, df):
        """Convert DataFrame to PyTorch dataset"""
        features = self.basic_preprocessor.prepare_features(df['cleaned_text'])
        
        return TensorDataset(
            features['input_ids'],
            features['attention_mask'],
            torch.tensor(df['encoded_label'].values)
        )

2. Model Implementation

Create the classification model:

from transformers import AutoModelForSequenceClassification
import torch.nn as nn

class TicketClassifier:
    def __init__(self, model_name="bert-base-uncased", num_labels=4):
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name,
            num_labels=num_labels
        )
        
    def train_step(self, batch):
        """Single training step"""
        self.model.train()
        inputs = {
            'input_ids': batch[0],
            'attention_mask': batch[1],
            'labels': batch[2]
        }
        
        outputs = self.model(**inputs)
        loss = outputs.loss
        
        return loss
    
    def eval_step(self, batch):
        """Single evaluation step"""
        self.model.eval()
        with torch.no_grad():
            inputs = {
                'input_ids': batch[0],
                'attention_mask': batch[1]
            }
            outputs = self.model(**inputs)
            predictions = outputs.logits.argmax(dim=-1)
            
        return predictions
Click to view advanced model implementation
class AdvancedTicketClassifier:
    def __init__(
        self,
        model_name="bert-base-uncased",
        num_labels=4,
        learning_rate=2e-5
    ):
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name,
            num_labels=num_labels
        )
        self.optimizer = AdamW(self.model.parameters(), lr=learning_rate)
        self.scheduler = get_linear_schedule_with_warmup(
            self.optimizer,
            num_warmup_steps=0,
            num_training_steps=1000  # Will be updated during training
        )
        
    def train(self, train_loader, val_loader, num_epochs=3):
        """Train the model"""
        best_val_loss = float('inf')
        
        for epoch in range(num_epochs):
            # Training
            self.model.train()
            train_loss = 0
            
            for batch in train_loader:
                loss = self.train_step(batch)
                train_loss += loss.item()
                
                # Backward pass
                loss.backward()
                
                # Gradient clipping
                torch.nn.utils.clip_grad_norm_(
                    self.model.parameters(),
                    max_norm=1.0
                )
                
                self.optimizer.step()
                self.scheduler.step()
                self.optimizer.zero_grad()
            
            # Validation
            val_loss, val_acc = self.evaluate(val_loader)
            
            print(f"Epoch {epoch+1}:")
            print(f"Train Loss: {train_loss/len(train_loader):.4f}")
            print(f"Val Loss: {val_loss:.4f}")
            print(f"Val Accuracy: {val_acc:.4f}")
            
            # Save best model
            if val_loss < best_val_loss:
                best_val_loss = val_loss
                torch.save(
                    self.model.state_dict(),
                    'models/best_model.pt'
                )
    
    def predict(self, text, preprocessor):
        """Make prediction for single text"""
        self.model.eval()
        
        # Preprocess text
        features = preprocessor.prepare_features([text])
        
        # Make prediction
        with torch.no_grad():
            outputs = self.model(**features)
            prediction = outputs.logits.argmax(dim=-1)
            
        return prediction.item()

3. Training Pipeline

Implement the training pipeline:

from torch.utils.data import DataLoader
from tqdm import tqdm

def train_model(model, train_dataset, val_dataset, num_epochs=3):
    """Train the classifier"""
    train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=64)
    
    for epoch in range(num_epochs):
        # Training
        model.train()
        train_loss = 0
        
        for batch in tqdm(train_loader):
            loss = model.train_step(batch)
            train_loss += loss.item()
            
        # Validation
        model.eval()
        val_correct = 0
        val_total = 0
        
        for batch in val_loader:
            predictions = model.eval_step(batch)
            val_correct += (predictions == batch[2]).sum().item()
            val_total += len(batch[2])
            
        print(f"Epoch {epoch+1}:")
        print(f"Train Loss: {train_loss/len(train_loader):.4f}")
        print(f"Val Accuracy: {val_correct/val_total:.4f}")
Click to view advanced training implementation
class TrainingManager:
    def __init__(
        self,
        model,
        train_dataset,
        val_dataset,
        batch_size=32,
        num_epochs=3
    ):
        self.model = model
        self.train_loader = DataLoader(
            train_dataset,
            batch_size=batch_size,
            shuffle=True
        )
        self.val_loader = DataLoader(
            val_dataset,
            batch_size=batch_size * 2
        )
        self.num_epochs = num_epochs
        
        # Initialize tracking
        self.train_losses = []
        self.val_losses = []
        self.val_accuracies = []
        
    def train(self):
        """Complete training pipeline"""
        best_val_acc = 0
        
        for epoch in range(self.num_epochs):
            # Training phase
            train_loss = self._train_epoch()
            self.train_losses.append(train_loss)
            
            # Validation phase
            val_loss, val_acc = self._validate()
            self.val_losses.append(val_loss)
            self.val_accuracies.append(val_acc)
            
            # Save best model
            if val_acc > best_val_acc:
                best_val_acc = val_acc
                self._save_checkpoint(epoch, val_acc)
                
            self._log_epoch(epoch, train_loss, val_loss, val_acc)
    
    def _train_epoch(self):
        """Train for one epoch"""
        self.model.train()
        total_loss = 0
        
        with tqdm(self.train_loader, desc="Training") as pbar:
            for batch in pbar:
                loss = self.model.train_step(batch)
                total_loss += loss.item()
                pbar.set_postfix({'loss': loss.item()})
                
        return total_loss / len(self.train_loader)
    
    def _validate(self):
        """Validate the model"""
        self.model.eval()
        total_loss = 0
        correct = 0
        total = 0
        
        with torch.no_grad():
            for batch in tqdm(self.val_loader, desc="Validation"):
                loss = self.model.eval_step(batch)
                total_loss += loss.item()
                
                predictions = self.model.predict_step(batch)
                correct += (predictions == batch[2]).sum().item()
                total += len(batch[2])
                
        return total_loss / len(self.val_loader), correct / total
    
    def _save_checkpoint(self, epoch, val_acc):
        """Save model checkpoint"""
        checkpoint = {
            'epoch': epoch,
            'model_state_dict': self.model.state_dict(),
            'val_acc': val_acc
        }
        torch.save(
            checkpoint,
            f'models/checkpoint_epoch_{epoch}_acc_{val_acc:.4f}.pt'
        )
    
    def _log_epoch(self, epoch, train_loss, val_loss, val_acc):
        """Log epoch results"""
        print(f"\nEpoch {epoch+1}/{self.num_epochs}")
        print(f"Train Loss: {train_loss:.4f}")
        print(f"Val Loss: {val_loss:.4f}")
        print(f"Val Accuracy: {val_acc:.4f}")

4. Web Service Implementation

Create Flask API for the classifier:

from flask import Flask, request, jsonify

app = Flask(__name__)

class TicketClassifierAPI:
    def __init__(self, model_path, preprocessor):
        self.model = TicketClassifier.load(model_path)
        self.preprocessor = preprocessor
        
    @app.route('/classify', methods=['POST'])
    def classify_ticket(self):
        """Classify a support ticket"""
        data = request.json
        
        if 'text' not in data:
            return jsonify({'error': 'No text provided'}), 400
            
        text = data['text']
        
        # Preprocess
        processed_text = self.preprocessor.clean_text(text)
        features = self.preprocessor.prepare_features([processed_text])
        
        # Predict
        prediction = self.model.predict(features)
        
        return jsonify({
            'category': self.preprocessor.label_encoder.inverse_transform([prediction])[0],
            'confidence': float(prediction_proba.max())
        })
Click to view advanced API implementation
from flask import Flask, request, jsonify
from flask_cors import CORS
import redis
import json

class AdvancedTicketClassifierAPI:
    def __init__(self, model_path, preprocessor):
        self.app = Flask(__name__)
        CORS(self.app)
        
        # Initialize model and preprocessor
        self.model = TicketClassifier.load(model_path)
        self.preprocessor = preprocessor
        
        # Setup Redis for caching
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        
        # Setup routes
        self._setup_routes()
        
    def _setup_routes(self):
        @self.app.route('/health', methods=['GET'])
        def health_check():
            return jsonify({'status': 'healthy'})
        
        @self.app.route('/classify', methods=['POST'])
        def classify_ticket():
            try:
                data = request.json
                
                if 'text' not in data:
                    return jsonify({
                        'error': 'No text provided'
                    }), 400
                    
                text = data['text']
                
                # Check cache
                cache_key = f"prediction:{hash(text)}"
                cached_result = self.redis_client.get(cache_key)
                
                if cached_result:
                    return jsonify(json.loads(cached_result))
                
                # Preprocess
                processed_text = self.preprocessor.clean_text(text)
                features = self.preprocessor.prepare_features([processed_text])
                
                # Predict
                prediction, confidence = self.model.predict_with_confidence(
                    features
                )
                
                # Prepare response
                result = {
                    'category': self.preprocessor.label_encoder.inverse_transform(
                        [prediction]
                    )[0],
                    'confidence': float(confidence),
                    'processed_text': processed_text
                }
                
  [Continuing the advanced API implementation...]

```python
                # Cache result
                self.redis_client.setex(
                    cache_key,
                    300,  # Cache for 5 minutes
                    json.dumps(result)
                )
                
                return jsonify(result)
                
            except Exception as e:
                return jsonify({
                    'error': str(e),
                    'type': 'Internal server error'
                }), 500
        
        @self.app.route('/batch-classify', methods=['POST'])
        def batch_classify():
            try:
                data = request.json
                
                if 'texts' not in data:
                    return jsonify({
                        'error': 'No texts provided'
                    }), 400
                    
                texts = data['texts']
                
                # Process batch
                processed_texts = [
                    self.preprocessor.clean_text(text)
                    for text in texts
                ]
                features = self.preprocessor.prepare_features(processed_texts)
                
                # Predict
                predictions, confidences = self.model.predict_batch_with_confidence(
                    features
                )
                
                # Prepare response
                results = [{
                    'category': self.preprocessor.label_encoder.inverse_transform(
                        [pred]
                    )[0],
                    'confidence': float(conf),
                    'processed_text': proc_text
                } for pred, conf, proc_text in zip(
                    predictions, confidences, processed_texts
                )]
                
                return jsonify({'results': results})
                
            except Exception as e:
                return jsonify({
                    'error': str(e),
                    'type': 'Internal server error'
                }), 500
    
    def run(self, host='0.0.0.0', port=5000):
        self.app.run(host=host, port=port)

5. Performance Monitoring

Implement monitoring for the deployed model:

import logging
from datetime import datetime

class ModelMonitor:
    def __init__(self, log_path='logs'):
        self.setup_logging(log_path)
        self.metrics = {
            'predictions': 0,
            'errors': 0,
            'avg_confidence': 0.0
        }
    
    def setup_logging(self, log_path):
        logging.basicConfig(
            filename=f'{log_path}/model_{datetime.now():%Y%m%d}.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
    
    def log_prediction(self, text, prediction, confidence):
        """Log prediction details"""
        self.metrics['predictions'] += 1
        self.metrics['avg_confidence'] = (
            (self.metrics['avg_confidence'] * (self.metrics['predictions'] - 1) +
             confidence) / self.metrics['predictions']
        )
        
        logging.info(
            f"Prediction made:\n"
            f"Text length: {len(text)}\n"
            f"Category: {prediction}\n"
            f"Confidence: {confidence:.4f}"
        )
Click to view advanced monitoring implementation
import pandas as pd
from sklearn.metrics import classification_report
import plotly.graph_objects as go

class AdvancedModelMonitor:
    def __init__(self, log_path='logs'):
        self.basic_monitor = ModelMonitor(log_path)
        self.predictions_log = []
        self.performance_metrics = {}
        
    def log_prediction_with_metadata(self, text, prediction, confidence, metadata=None):
        """Log detailed prediction information"""
        log_entry = {
            'timestamp': datetime.now(),
            'text_length': len(text),
            'prediction': prediction,
            'confidence': confidence,
            'metadata': metadata or {}
        }
        self.predictions_log.append(log_entry)
        
    def calculate_metrics(self, window_size=1000):
        """Calculate rolling performance metrics"""
        if len(self.predictions_log) < window_size:
            return
            
        recent_predictions = pd.DataFrame(
            self.predictions_log[-window_size:]
        )
        
        # Calculate accuracy if actual labels are available
        if 'actual' in recent_predictions.columns:
            self.performance_metrics['accuracy'] = (
                (recent_predictions['actual'] == 
                 recent_predictions['prediction']).mean()
            )
        
        # Calculate average confidence
        self.performance_metrics['avg_confidence'] = (
            recent_predictions['confidence'].mean()
        )
        
        # Calculate prediction distribution
        pred_dist = recent_predictions['prediction'].value_counts(normalize=True)
        self.performance_metrics['prediction_distribution'] = pred_dist.to_dict()
        
    def plot_metrics(self):
        """Create monitoring dashboard"""
        df = pd.DataFrame(self.predictions_log)
        
        # Create confidence trend plot
        fig = go.Figure()
        fig.add_trace(go.Scatter(
            x=df['timestamp'],
            y=df['confidence'].rolling(100).mean(),
            mode='lines',
            name='Average Confidence'
        ))
        
        # Add prediction distribution
        fig.add_trace(go.Bar(
            x=df['prediction'].value_counts().index,
            y=df['prediction'].value_counts().values,
            name='Prediction Distribution'
        ))
        
        fig.update_layout(
            title='Model Performance Metrics',
            xaxis_title='Time',
            yaxis_title='Value',
            hovermode='x unified'
        )
        
        return fig
    
    def generate_report(self):
        """Generate comprehensive performance report"""
        report = {
            'timestamp': datetime.now(),
            'total_predictions': len(self.predictions_log),
            'recent_metrics': self.performance_metrics,
            'prediction_volume': len(self.predictions_log[-1000:]),
            'average_confidence': np.mean([
                log['confidence'] for log in self.predictions_log[-1000:]
            ])
        }
        
        if 'actual' in pd.DataFrame(self.predictions_log).columns:
            report['classification_report'] = classification_report(
                [log['actual'] for log in self.predictions_log[-1000:]],
                [log['prediction'] for log in self.predictions_log[-1000:]],
                output_dict=True
            )
        
        return report

6. Best Practices and Optimization

  1. Model Optimization:

    • Use model quantization for faster inference
    • Implement batch prediction
    • Cache frequent predictions
    • Use GPU acceleration where available
  2. Error Handling:

    • Implement robust input validation
    • Add retry logic for failed predictions
    • Monitor and log edge cases
    • Set up alerting for critical failures
  3. Performance Tips:

    • Use appropriate batch sizes
    • Implement proper text cleaning
    • Optimize model loading
    • Use caching strategically
  4. Deployment Considerations:

    • Use containerization (Docker)
    • Implement load balancing
    • Set up monitoring and logging
    • Implement proper security measures
  5. Maintenance Best Practices:

    • Regular model retraining
    • Performance monitoring
    • Data drift detection
    • A/B testing new models