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
-
Classification Approaches
- Rule-based systems
- Machine learning methods
- Deep learning approaches
- Hybrid systems
-
Key Concepts
- Feature extraction
- Document representation
- Classification algorithms
- Model evaluation metrics
-
Common Challenges
- Imbalanced classes
- Short text classification
- Multi-label classification
- Domain adaptation
NLP Architecture
-
Transformer Architecture
- Self-attention mechanism
- Multi-head attention
- Position encodings
- Layer normalization
-
BERT and Variants
- Bidirectional encoding
- Masked language modeling
- Next sentence prediction
- Fine-tuning approaches
-
Key Components
Input Text -> Tokenization -> Embedding -> Transformer Layers -> Classification Head -> Prediction
Transfer Learning
-
Pre-training
- Language modeling
- Masked token prediction
- Large-scale training
- Domain-specific pre-training
-
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' } -
Adaptation Techniques
- Domain adaptation
- Task-specific heads
- Prompt tuning
- Few-shot learning
Deployment considerations
-
System Architecture
Client Request -> Load Balancer -> API Server -> Model Server -> Cache -> Database -
Performance Optimization
- Model quantization
- Batch processing
- Caching strategies
- Async processing
-
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
-
Model Optimization:
- Use model quantization for faster inference
- Implement batch prediction
- Cache frequent predictions
- Use GPU acceleration where available
-
Error Handling:
- Implement robust input validation
- Add retry logic for failed predictions
- Monitor and log edge cases
- Set up alerting for critical failures
-
Performance Tips:
- Use appropriate batch sizes
- Implement proper text cleaning
- Optimize model loading
- Use caching strategically
-
Deployment Considerations:
- Use containerization (Docker)
- Implement load balancing
- Set up monitoring and logging
- Implement proper security measures
-
Maintenance Best Practices:
- Regular model retraining
- Performance monitoring
- Data drift detection
- A/B testing new models