📝 INTRODUCTION
This is an enterprise-grade sentiment analysis system that leverages state-of-the-art transformer models (BERT, RoBERTa, XLNet) alongside traditional ML approaches. It includes advanced features like aspect-based sentiment analysis, emotion detection, sarcasm detection, and multi-language support.
✨ FEATURES
- Multi-Model Architecture: BERT, RoBERTa, XLNet, DistilBERT, ALBERT
- Aspect-Based Sentiment Analysis: Extract sentiment for specific aspects (price, quality, service)
- Emotion Detection: Detect 7 emotions (joy, sadness, anger, fear, surprise, disgust, love)
- Sarcasm Detection: Identify sarcastic reviews
- Multi-Language Support: 50+ languages using XLM-RoBERTa
- Explainable AI: SHAP and LIME explanations for predictions
- Active Learning: Continuously improve with human feedback
- Real-time Streaming: Kafka integration for real-time analysis
- Model Versioning: MLflow integration for experiment tracking
- Automated Retraining: CI/CD pipeline for model updates
📁 PROJECT STRUCTURE
advanced-sentiment-analyzer/ │ ├── src/ │ ├── models/ │ │ ├── base_model.py │ │ ├── bert_model.py │ │ ├── roberta_model.py │ │ ├── xlnet_model.py │ │ ├── ensemble_model.py │ │ ├── aspect_model.py │ │ ├── emotion_model.py │ │ └── sarcasm_model.py │ │ │ ├── training/ │ │ ├── trainer.py │ │ ├── data_preprocessor.py │ │ ├── active_learner.py │ │ └── hyperparameter_tuner.py │ │ │ ├── inference/ │ │ ├── predictor.py │ │ ├── batch_processor.py │ │ └── streaming_processor.py │ │ │ ├── explainability/ │ │ ├── shap_explainer.py │ │ └── lime_explainer.py │ │ │ └── utils/ │ ├── text_processor.py │ ├── language_detector.py │ └── metrics_calculator.py │ ├── api/ │ ├── rest_api.py │ ├── grpc_api.py │ └── websocket_api.py │ ├── data/ │ ├── raw/ │ ├── processed/ │ └── augmented/ │ ├── config/ │ ├── model_config.yaml │ └── training_config.yaml │ ├── experiments/ │ └── mlflow/ │ ├── docker/ │ ├── Dockerfile.api │ ├── Dockerfile.training │ └── docker-compose.yml │ ├── kafka/ │ ├── producer.py │ └── consumer.py │ ├── monitoring/ │ ├── prometheus_config.yml │ └── grafana_dashboards/ │ ├── tests/ │ ├── unit/ │ ├── integration/ │ └── performance/ │ ├── requirements.txt ├── requirements-dev.txt └── README.md
🚀 COMPLETE CODE
1. requirements.txt
# Core ML/DL torch==2.0.1 transformers==4.35.0 datasets==2.14.0 accelerate==0.24.0 sentencepiece==0.1.99 protobuf==3.20.3 # Traditional ML scikit-learn==1.3.0 xgboost==1.7.6 lightgbm==4.0.0 # Data Processing pandas==2.0.3 numpy==1.24.3 polars==0.19.0 dask==2023.9.0 # Explainability shap==0.42.1 lime==0.2.0.1 eli5==0.13.0 # API & Services fastapi==0.104.0 uvicorn==0.24.0 grpcio==1.59.0 grpcio-tools==1.59.0 websockets==12.0 kafka-python==2.0.2 # Database pymongo==4.5.0 redis==5.0.1 elasticsearch==8.11.0 # Monitoring prometheus-client==0.18.0 mlflow==2.6.0 wandb==0.15.0 # Visualization plotly==5.17.0 dash==2.14.0 streamlit==1.27.0 # Testing & Quality pytest==7.4.0 pytest-cov==4.1.0 black==23.9.0 flake8==6.1.0 mypy==1.5.0 # Utils tqdm==4.66.0 python-dotenv==1.0.0 pyyaml==6.0.1 joblib==1.3.2
2. src/models/base_model.py
import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer, AutoConfig
from abc import ABC, abstractmethod
import logging
from typing import Dict, List, Optional, Union
import numpy as np
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BaseSentimentModel(ABC, nn.Module):
"""Base class for all sentiment models"""
def __init__(self, model_name: str, num_labels: int = 3, device: str = None):
super().__init__()
self.model_name = model_name
self.num_labels = num_labels
self.device = device or ('cuda' if torch.cuda.is_available() else 'cpu')
# Load model and tokenizer
self.config = AutoConfig.from_pretrained(
model_name,
num_labels=num_labels,
output_hidden_states=True
)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModel.from_pretrained(model_name, config=self.config)
# Classification head
self.classifier = nn.Sequential(
nn.Linear(self.config.hidden_size, 512),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(512, 256),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(256, num_labels)
)
self.to(self.device)
logger.info(f"Initialized {model_name} on {self.device}")
@abstractmethod
def forward(self, input_ids, attention_mask, labels=None):
"""Forward pass"""
pass
def predict(self, texts: Union[str, List[str]], return_proba: bool = False):
"""Make predictions on input texts"""
self.eval()
if isinstance(texts, str):
texts = [texts]
# Tokenize
inputs = self.tokenizer(
texts,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
outputs = self.forward(**inputs)
if return_proba:
probabilities = torch.softmax(outputs, dim=-1)
return probabilities.cpu().numpy()
else:
predictions = torch.argmax(outputs, dim=-1)
return predictions.cpu().numpy()
def save_model(self, path: str):
"""Save model to disk"""
torch.save({
'model_state_dict': self.state_dict(),
'model_name': self.model_name,
'num_labels': self.num_labels,
'config': self.config
}, path)
logger.info(f"Model saved to {path}")
def load_model(self, path: str):
"""Load model from disk"""
checkpoint = torch.load(path, map_location=self.device)
self.model_name = checkpoint['model_name']
self.num_labels = checkpoint['num_labels']
self.config = checkpoint['config']
self.load_state_dict(checkpoint['model_state_dict'])
logger.info(f"Model loaded from {path}")
class TransformerSentimentModel(BaseSentimentModel):
"""Transformer-based sentiment model"""
def forward(self, input_ids, attention_mask, labels=None):
outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
pooled_output = outputs.last_hidden_state[:, 0, :] # Use [CLS] token
logits = self.classifier(pooled_output)
if labels is not None:
loss_fn = nn.CrossEntropyLoss()
loss = loss_fn(logits, labels)
return loss, logits
return logits
class EnsembleSentimentModel(nn.Module):
"""Ensemble of multiple models"""
def __init__(self, models: List[BaseSentimentModel], weights: Optional[List[float]] = None):
super().__init__()
self.models = nn.ModuleList(models)
self.weights = weights or [1.0/len(models)] * len(models)
def forward(self, input_ids, attention_mask):
predictions = []
for model in self.models:
output = model(input_ids, attention_mask)
predictions.append(output)
# Weighted average
ensemble_output = sum(w * pred for w, pred in zip(self.weights, predictions))
return ensemble_output
def predict(self, texts: Union[str, List[str]], return_proba: bool = False):
"""Ensemble prediction"""
all_probs = []
for model in self.models:
probs = model.predict(texts, return_proba=True)
all_probs.append(probs)
# Weighted average of probabilities
ensemble_probs = np.average(all_probs, axis=0, weights=self.weights)
if return_proba:
return ensemble_probs
else:
return np.argmax(ensemble_probs, axis=-1)
3. src/models/bert_model.py
from src.models.base_model import TransformerSentimentModel import torch import torch.nn as nn class BERTSentimentModel(TransformerSentimentModel): """BERT-based sentiment model with advanced features""" def __init__(self, model_name='bert-base-uncased', num_labels=3, use_pooler=True): super().__init__(model_name, num_labels) self.use_pooler = use_pooler # Additional layers for better performance self.attention = nn.MultiheadAttention( embed_dim=self.config.hidden_size, num_heads=8, dropout=0.1, batch_first=True ) self.layer_norm = nn.LayerNorm(self.config.hidden_size) def forward(self, input_ids, attention_mask, token_type_ids=None, labels=None): outputs = self.model( input_ids=input_ids, attention_mask=attention_mask, token_type_ids=token_type_ids ) if self.use_pooler: # Use pooler output pooled_output = outputs.pooler_output else: # Use mean of last hidden states last_hidden = outputs.last_hidden_state masked_output = last_hidden * attention_mask.unsqueeze(-1) pooled_output = masked_output.sum(dim=1) / attention_mask.sum(dim=1, keepdim=True) # Apply attention attended_output, _ = self.attention(pooled_output.unsqueeze(1), pooled_output.unsqueeze(1), pooled_output.unsqueeze(1)) attended_output = attended_output.squeeze(1) # Residual connection and layer norm attended_output = self.layer_norm(pooled_output + attended_output) # Classification logits = self.classifier(attended_output) if labels is not None: loss_fn = nn.CrossEntropyLoss() loss = loss_fn(logits, labels) return loss, logits return logits def get_attention_weights(self, text): """Get attention weights for explainability""" self.eval() inputs = self.tokenizer(text, return_tensors="pt").to(self.device) with torch.no_grad(): outputs = self.model(**inputs, output_attentions=True) attention = outputs.attentions[-1] # Last layer attention return attention.cpu().numpy() class DistilBERTSentimentModel(BERTSentimentModel): """DistilBERT model for faster inference""" def __init__(self, model_name='distilbert-base-uncased', num_labels=3): super().__init__(model_name, num_labels) def forward(self, input_ids, attention_mask, labels=None): outputs = self.model( input_ids=input_ids, attention_mask=attention_mask ) # DistilBERT doesn't have pooler, use mean of last hidden last_hidden = outputs.last_hidden_state masked_output = last_hidden * attention_mask.unsqueeze(-1) pooled_output = masked_output.sum(dim=1) / attention_mask.sum(dim=1, keepdim=True) logits = self.classifier(pooled_output) if labels is not None: loss_fn = nn.CrossEntropyLoss() loss = loss_fn(logits, labels) return loss, logits return logits
4. src/models/aspect_model.py
import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
from typing import List, Dict, Tuple
import numpy as np
class AspectSentimentModel(nn.Module):
"""Aspect-based sentiment analysis model"""
def __init__(self, model_name='bert-base-uncased', aspects: List[str] = None):
super().__init__()
self.model_name = model_name
self.aspects = aspects or ['price', 'quality', 'service', 'delivery', 'design']
# Base model
self.base_model = AutoModel.from_pretrained(model_name)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.config = self.base_model.config
# Aspect encoders
self.aspect_embeddings = nn.Embedding(len(self.aspects), self.config.hidden_size)
# Attention mechanism
self.aspect_attention = nn.MultiheadAttention(
embed_dim=self.config.hidden_size,
num_heads=8,
dropout=0.1,
batch_first=True
)
# Aspect-specific classifiers
self.aspect_classifiers = nn.ModuleList([
nn.Sequential(
nn.Linear(self.config.hidden_size * 2, 256),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(256, 3) # 3 sentiments per aspect
) for _ in range(len(self.aspects))
])
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.to(self.device)
def forward(self, input_ids, attention_mask, aspect_indices=None):
# Get base model outputs
outputs = self.base_model(
input_ids=input_ids,
attention_mask=attention_mask
)
# Get token embeddings
token_embeddings = outputs.last_hidden_state
# Get aspect embeddings
batch_size = input_ids.shape[0]
aspect_emb = self.aspect_embeddings(
torch.arange(len(self.aspects)).to(self.device)
).unsqueeze(0).expand(batch_size, -1, -1)
# Apply attention between tokens and aspects
attended_tokens, attention_weights = self.aspect_attention(
aspect_emb, # Query: aspects
token_embeddings, # Key: tokens
token_embeddings # Value: tokens
)
# Combine aspect and token information
combined = torch.cat([aspect_emb, attended_tokens], dim=-1)
# Get predictions for each aspect
aspect_logits = []
for i, classifier in enumerate(self.aspect_classifiers):
logits = classifier(combined[:, i, :])
aspect_logits.append(logits)
return torch.stack(aspect_logits, dim=1), attention_weights
def predict(self, text: str) -> Dict[str, Dict[str, float]]:
"""Predict sentiment for each aspect"""
self.eval()
# Tokenize
inputs = self.tokenizer(
text,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
logits, attention_weights = self.forward(**inputs)
probabilities = torch.softmax(logits, dim=-1)
# Convert to predictions
sentiment_map = {0: 'negative', 1: 'neutral', 2: 'positive'}
results = {}
for i, aspect in enumerate(self.aspects):
probs = probabilities[0, i].cpu().numpy()
sentiment = sentiment_map[np.argmax(probs)]
confidence = float(np.max(probs))
results[aspect] = {
'sentiment': sentiment,
'confidence': confidence,
'probabilities': {
'negative': float(probs[0]),
'neutral': float(probs[1]),
'positive': float(probs[2])
}
}
return results
def get_aspect_attention(self, text: str) -> Dict[str, List[Tuple[str, float]]]:
"""Get attention weights for each aspect"""
self.eval()
# Tokenize
inputs = self.tokenizer(
text,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
_, attention_weights = self.forward(**inputs)
# Decode tokens
tokens = self.tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
# Get attention for each aspect
aspect_attention = {}
attention = attention_weights[0].cpu().numpy() # [num_aspects, seq_len]
for i, aspect in enumerate(self.aspects):
token_attention = list(zip(tokens, attention[i]))
token_attention = sorted(token_attention, key=lambda x: x[1], reverse=True)
aspect_attention[aspect] = token_attention[:10] # Top 10 tokens
return aspect_attention
5. src/models/emotion_model.py
import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
from typing import Dict, List
import numpy as np
class EmotionDetectionModel(nn.Module):
"""Multi-label emotion detection model"""
def __init__(self, model_name='roberta-base'):
super().__init__()
self.model_name = model_name
self.emotions = ['joy', 'sadness', 'anger', 'fear', 'surprise', 'disgust', 'love']
self.num_emotions = len(self.emotions)
# Base model
self.base_model = AutoModel.from_pretrained(model_name)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.config = self.base_model.config
# Multi-label classification head
self.classifier = nn.Sequential(
nn.Linear(self.config.hidden_size, 512),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(512, 256),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(256, self.num_emotions)
)
# Emotion-specific attention
self.emotion_attention = nn.Parameter(
torch.randn(self.num_emotions, self.config.hidden_size)
)
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.to(self.device)
def forward(self, input_ids, attention_mask, labels=None):
# Get base model outputs
outputs = self.base_model(
input_ids=input_ids,
attention_mask=attention_mask
)
# Get token embeddings
token_embeddings = outputs.last_hidden_state
# Apply emotion-specific attention
# emotion_attention: [num_emotions, hidden]
# token_embeddings: [batch, seq_len, hidden]
attention_scores = torch.einsum('eh,bsh->bse', self.emotion_attention, token_embeddings)
attention_weights = torch.softmax(attention_scores, dim=1)
# Weighted sum of token embeddings
emotion_vectors = torch.einsum('bse,bsh->beh', attention_weights, token_embeddings)
# Classification
logits = self.classifier(emotion_vectors)
if labels is not None:
loss_fn = nn.BCEWithLogitsLoss()
loss = loss_fn(logits, labels.float())
return loss, torch.sigmoid(logits)
return torch.sigmoid(logits)
def predict(self, text: str, threshold: float = 0.5) -> Dict[str, float]:
"""Predict emotions in text"""
self.eval()
# Tokenize
inputs = self.tokenizer(
text,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
probabilities = self.forward(**inputs)[0] if len(inputs) > 1 else self.forward(**inputs)
probabilities = probabilities.cpu().numpy()[0]
# Apply threshold
results = {}
for emotion, prob in zip(self.emotions, probabilities):
results[emotion] = float(prob)
results[f"{emotion}_detected"] = bool(prob >= threshold)
# Get primary emotion
primary_idx = np.argmax(probabilities)
results['primary_emotion'] = self.emotions[primary_idx]
results['primary_confidence'] = float(probabilities[primary_idx])
return results
def predict_batch(self, texts: List[str], threshold: float = 0.5) -> List[Dict[str, float]]:
"""Predict emotions for multiple texts"""
self.eval()
# Tokenize
inputs = self.tokenizer(
texts,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
probabilities = self.forward(**inputs)
probabilities = probabilities.cpu().numpy()
results = []
for probs in probabilities:
result = {}
for emotion, prob in zip(self.emotions, probs):
result[emotion] = float(prob)
result[f"{emotion}_detected"] = bool(prob >= threshold)
primary_idx = np.argmax(probs)
result['primary_emotion'] = self.emotions[primary_idx]
result['primary_confidence'] = float(probs[primary_idx])
results.append(result)
return results
6. src/models/sarcasm_model.py
import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
import numpy as np
class SarcasmDetectionModel(nn.Module):
"""Sarcasm detection model using contrastive learning"""
def __init__(self, model_name='roberta-base'):
super().__init__()
self.model_name = model_name
# Base model
self.base_model = AutoModel.from_pretrained(model_name)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.config = self.base_model.config
# Projection heads
self.projection = nn.Sequential(
nn.Linear(self.config.hidden_size, 512),
nn.ReLU(),
nn.Linear(512, 256)
)
# Contrastive learning head
self.temperature = nn.Parameter(torch.ones(1) * 0.07)
# Classification head
self.classifier = nn.Sequential(
nn.Linear(256, 128),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(128, 2) # sarcastic or not
)
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.to(self.device)
def forward(self, input_ids, attention_mask, labels=None):
# Get base model outputs
outputs = self.base_model(
input_ids=input_ids,
attention_mask=attention_mask
)
# Use [CLS] token
pooled = outputs.last_hidden_state[:, 0, :]
# Project
projected = self.projection(pooled)
# Classify
logits = self.classifier(projected)
if labels is not None:
loss_fn = nn.CrossEntropyLoss()
loss = loss_fn(logits, labels)
return loss, logits
return logits
def contrastive_loss(self, anchor, positive, negative):
"""Compute contrastive loss for better representations"""
# Normalize embeddings
anchor = nn.functional.normalize(anchor, dim=-1)
positive = nn.functional.normalize(positive, dim=-1)
negative = nn.functional.normalize(negative, dim=-1)
# Compute similarities
pos_sim = torch.sum(anchor * positive, dim=-1) / self.temperature
neg_sim = torch.sum(anchor * negative, dim=-1) / self.temperature
# Compute loss
pos_exp = torch.exp(pos_sim)
neg_exp = torch.exp(neg_sim)
loss = -torch.log(pos_exp / (pos_exp + neg_exp))
return loss.mean()
def predict(self, text: str) -> Dict[str, any]:
"""Predict if text is sarcastic"""
self.eval()
# Tokenize
inputs = self.tokenizer(
text,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
logits = self.forward(**inputs)
probabilities = torch.softmax(logits, dim=-1)
probs = probabilities.cpu().numpy()[0]
return {
'is_sarcastic': bool(np.argmax(probs) == 1),
'sarcastic_probability': float(probs[1]),
'not_sarcastic_probability': float(probs[0]),
'confidence': float(np.max(probs))
}
def get_sarcasm_features(self, text: str) -> Dict[str, float]:
"""Extract features that indicate sarcasm"""
self.eval()
# Get embeddings
inputs = self.tokenizer(
text,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
outputs = self.base_model(**inputs)
embeddings = outputs.last_hidden_state[0] # [seq_len, hidden]
# Calculate linguistic features
tokens = self.tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
# Feature extraction
features = {
'positive_word_ratio': self._calculate_positive_ratio(tokens),
'negative_word_ratio': self._calculate_negative_ratio(tokens),
'exaggeration_score': self._calculate_exaggeration(text),
'contrast_score': self._calculate_contrast(embeddings),
'punctuation_anomaly': self._check_punctuation(text)
}
return features
def _calculate_positive_ratio(self, tokens):
# Simplified - in practice use sentiment lexicon
positive_words = {'good', 'great', 'awesome', 'fantastic', 'amazing'}
token_set = set(t.lower() for t in tokens)
return len(positive_words & token_set) / max(len(tokens), 1)
def _calculate_negative_ratio(self, tokens):
negative_words = {'bad', 'terrible', 'awful', 'horrible', 'worst'}
token_set = set(t.lower() for t in tokens)
return len(negative_words & token_set) / max(len(tokens), 1)
def _calculate_exaggeration(self, text):
# Count exaggerative markers
markers = ['!', '?', 'absolutely', 'literally', 'totally', 'completely']
count = sum(text.count(m) for m in markers)
return min(count / 10, 1.0)
def _calculate_contrast(self, embeddings):
# Measure semantic contrast using cosine similarity
if len(embeddings) < 2:
return 0.0
# Compare first and last sentence embeddings
first_sent = embeddings[:len(embeddings)//2].mean(0)
last_sent = embeddings[len(embeddings)//2:].mean(0)
similarity = nn.functional.cosine_similarity(
first_sent.unsqueeze(0),
last_sent.unsqueeze(0)
)
return 1 - similarity.item() # Contrast = 1 - similarity
def _check_punctuation(self, text):
# Check for unusual punctuation patterns
exclamation_count = text.count('!')
question_count = text.count('?')
return (exclamation_count + question_count) > 3
7. src/training/trainer.py
import torch
from torch.utils.data import DataLoader, Dataset
from transformers import get_linear_schedule_with_warmup
from tqdm import tqdm
import numpy as np
from sklearn.metrics import accuracy_score, f1_score, precision_recall_fscore_support
import wandb
import mlflow
from typing import Dict, List, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SentimentDataset(Dataset):
"""Custom dataset for sentiment analysis"""
def __init__(self, texts, labels, tokenizer, max_length=512):
self.texts = texts
self.labels = labels
self.tokenizer = tokenizer
self.max_length = max_length
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
text = str(self.texts[idx])
label = self.labels[idx]
encoding = self.tokenizer(
text,
truncation=True,
padding='max_length',
max_length=self.max_length,
return_tensors='pt'
)
return {
'input_ids': encoding['input_ids'].flatten(),
'attention_mask': encoding['attention_mask'].flatten(),
'labels': torch.tensor(label, dtype=torch.long)
}
class ModelTrainer:
"""Advanced model trainer with experiment tracking"""
def __init__(self, model, config):
self.model = model
self.config = config
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.model.to(self.device)
# Initialize experiment tracking
self._init_experiment_tracking()
def _init_experiment_tracking(self):
"""Initialize wandb and mlflow"""
# Weights & Biases
wandb.init(
project="sentiment-analyzer",
config=self.config,
name=f"run_{self.config.get('model_name', 'experiment')}"
)
# MLflow
mlflow.set_experiment("sentiment_analysis")
mlflow.start_run()
mlflow.log_params(self.config)
def train(self, train_texts, train_labels, val_texts=None, val_labels=None):
"""Train the model"""
# Create datasets
train_dataset = SentimentDataset(
train_texts, train_labels,
self.model.tokenizer,
self.config.get('max_length', 512)
)
train_loader = DataLoader(
train_dataset,
batch_size=self.config.get('batch_size', 16),
shuffle=True,
num_workers=4
)
# Validation loader
val_loader = None
if val_texts and val_labels:
val_dataset = SentimentDataset(
val_texts, val_labels,
self.model.tokenizer,
self.config.get('max_length', 512)
)
val_loader = DataLoader(
val_dataset,
batch_size=self.config.get('batch_size', 16),
shuffle=False,
num_workers=4
)
# Optimizer
optimizer = torch.optim.AdamW(
self.model.parameters(),
lr=self.config.get('learning_rate', 2e-5),
weight_decay=self.config.get('weight_decay', 0.01)
)
# Scheduler
total_steps = len(train_loader) * self.config.get('epochs', 3)
scheduler = get_linear_schedule_with_warmup(
optimizer,
num_warmup_steps=int(total_steps * 0.1),
num_training_steps=total_steps
)
# Training loop
best_val_acc = 0
train_losses = []
val_metrics = []
for epoch in range(self.config.get('epochs', 3)):
# Training
self.model.train()
epoch_loss = 0
train_progress = tqdm(train_loader, desc=f'Epoch {epoch + 1}/{self.config.get("epochs", 3)}')
for batch in train_progress:
# Move to device
input_ids = batch['input_ids'].to(self.device)
attention_mask = batch['attention_mask'].to(self.device)
labels = batch['labels'].to(self.device)
# Forward pass
loss, logits = self.model(input_ids, attention_mask, labels)
# Backward pass
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
optimizer.step()
scheduler.step()
epoch_loss += loss.item()
train_progress.set_postfix({'loss': loss.item()})
# Log to wandb
wandb.log({'train_loss': loss.item()})
avg_train_loss = epoch_loss / len(train_loader)
train_losses.append(avg_train_loss)
# Validation
if val_loader:
val_metrics_epoch = self.evaluate(val_loader)
val_metrics.append(val_metrics_epoch)
# Log validation metrics
wandb.log({
'val_loss': val_metrics_epoch['loss'],
'val_accuracy': val_metrics_epoch['accuracy'],
'val_f1': val_metrics_epoch['f1']
})
mlflow.log_metrics({
f'val_accuracy_epoch_{epoch}': val_metrics_epoch['accuracy'],
f'val_f1_epoch_{epoch}': val_metrics_epoch['f1']
})
# Save best model
if val_metrics_epoch['accuracy'] > best_val_acc:
best_val_acc = val_metrics_epoch['accuracy']
self.save_model('best_model.pt')
logger.info(f"New best model saved with accuracy: {best_val_acc:.4f}")
# Log to mlflow
mlflow.log_metric('train_loss', avg_train_loss, step=epoch)
# Final logging
wandb.finish()
mlflow.end_run()
return {
'train_losses': train_losses,
'val_metrics': val_metrics,
'best_val_accuracy': best_val_acc
}
def evaluate(self, dataloader):
"""Evaluate the model"""
self.model.eval()
total_loss = 0
all_preds = []
all_labels = []
with torch.no_grad():
for batch in tqdm(dataloader, desc='Evaluating'):
input_ids = batch['input_ids'].to(self.device)
attention_mask = batch['attention_mask'].to(self.device)
labels = batch['labels'].to(self.device)
loss, logits = self.model(input_ids, attention_mask, labels)
total_loss += loss.item()
preds = torch.argmax(logits, dim=-1)
all_preds.extend(preds.cpu().numpy())
all_labels.extend(labels.cpu().numpy())
# Calculate metrics
accuracy = accuracy_score(all_labels, all_preds)
precision, recall, f1, _ = precision_recall_fscore_support(
all_labels, all_preds, average='weighted'
)
return {
'loss': total_loss / len(dataloader),
'accuracy': accuracy,
'precision': precision,
'recall': recall,
'f1': f1
}
def save_model(self, path):
"""Save model checkpoint"""
torch.save({
'model_state_dict': self.model.state_dict(),
'config': self.config
}, path)
logger.info(f"Model saved to {path}")
def load_model(self, path):
"""Load model checkpoint"""
checkpoint = torch.load(path, map_location=self.device)
self.model.load_state_dict(checkpoint['model_state_dict'])
self.config = checkpoint['config']
logger.info(f"Model loaded from {path}")
class ActiveLearningTrainer(ModelTrainer):
"""Trainer with active learning capabilities"""
def __init__(self, model, config):
super().__init__(model, config)
self.unlabeled_pool = []
def add_unlabeled_data(self, texts):
"""Add unlabeled data to pool"""
self.unlabeled_pool.extend(texts)
def select_samples_for_labeling(self, n_samples=10, strategy='uncertainty'):
"""Select most informative samples for labeling"""
if strategy == 'uncertainty':
return self._select_by_uncertainty(n_samples)
elif strategy == 'diversity':
return self._select_by_diversity(n_samples)
elif strategy == 'hybrid':
return self._select_hybrid(n_samples)
def _select_by_uncertainty(self, n_samples):
"""Select samples with highest prediction uncertainty"""
self.model.eval()
uncertainties = []
for text in tqdm(self.unlabeled_pool, desc='Calculating uncertainty'):
inputs = self.model.tokenizer(
text,
return_tensors="pt",
truncation=True,
max_length=512
).to(self.device)
with torch.no_grad():
logits = self.model(**inputs)
probs = torch.softmax(logits, dim=-1)
# Entropy as uncertainty measure
entropy = -torch.sum(probs * torch.log(probs + 1e-10))
uncertainties.append(entropy.item())
# Select top uncertain samples
indices = np.argsort(uncertainties)[-n_samples:]
selected = [self.unlabeled_pool[i] for i in indices]
# Remove selected from pool
self.unlabeled_pool = [t for i, t in enumerate(self.unlabeled_pool)
if i not in indices]
return selected
def _select_by_diversity(self, n_samples):
"""Select diverse samples using clustering"""
# Get embeddings for all samples
embeddings = []
for text in tqdm(self.unlabeled_pool, desc='Getting embeddings'):
inputs = self.model.tokenizer(
text,
return_tensors="pt",
truncation=True,
max_length=512
).to(self.device)
with torch.no_grad():
outputs = self.model.base_model(**inputs)
emb = outputs.last_hidden_state[:, 0, :].cpu().numpy()
embeddings.append(emb[0])
embeddings = np.array(embeddings)
# Cluster using k-means
from sklearn.cluster import KMeans
kmeans = KMeans(n_clusters=min(n_samples, len(embeddings)))
clusters = kmeans.fit_predict(embeddings)
# Select one sample from each cluster
selected = []
for i in range(min(n_samples, len(np.unique(clusters)))):
cluster_indices = np.where(clusters == i)[0]
selected_idx = np.random.choice(cluster_indices)
selected.append(self.unlabeled_pool[selected_idx])
# Remove selected from pool
self.unlabeled_pool = [t for i, t in enumerate(self.unlabeled_pool)
if i not in [self.unlabeled_pool.index(s) for s in selected]]
return selected
def _select_hybrid(self, n_samples):
"""Hybrid selection: uncertainty + diversity"""
# First get uncertain samples
uncertain_samples = self._select_by_uncertainty(n_samples * 2)
# Then select diverse among them
self.unlabeled_pool = uncertain_samples + self.unlabeled_pool
return self._select_by_diversity(n_samples)
8. src/explainability/shap_explainer.py
import shap
import torch
import numpy as np
import matplotlib.pyplot as plt
from typing import List, Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SHAPExplainer:
"""SHAP-based model explainer"""
def __init__(self, model, tokenizer, background_texts: List[str]):
self.model = model
self.tokenizer = tokenizer
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# Create background dataset
self.background = self._create_background_dataset(background_texts)
# Create explainer
self.explainer = shap.Explainer(
self._model_predict,
self.background,
feature_names=self.tokenizer.convert_ids_to_tokens
)
def _create_background_dataset(self, texts: List[str]):
"""Create background dataset for SHAP"""
encodings = self.tokenizer(
texts,
padding=True,
truncation=True,
max_length=128,
return_tensors="pt"
)
return encodings['input_ids'].numpy()
def _model_predict(self, texts):
"""Wrapper for model prediction"""
self.model.eval()
inputs = self.tokenizer(
texts.tolist() if hasattr(texts, 'tolist') else texts,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
).to(self.device)
with torch.no_grad():
outputs = self.model(**inputs)
if hasattr(outputs, 'logits'):
logits = outputs.logits
else:
logits = outputs
probs = torch.softmax(logits, dim=-1)
return probs.cpu().numpy()
def explain(self, text: str, class_idx: int = None) -> Dict[str, Any]:
"""Generate SHAP explanation for text"""
# Tokenize
tokens = self.tokenizer.tokenize(text)
input_ids = self.tokenizer.encode(text, return_tensors='pt').to(self.device)
# Get SHAP values
shap_values = self.explainer([text])
# Get explanation for specific class
if class_idx is None:
# Get prediction
probs = self._model_predict([text])[0]
class_idx = np.argmax(probs)
# Extract values for the class
values = shap_values.values[0, :, class_idx]
# Create explanation
explanation = {
'text': text,
'tokens': tokens,
'class_idx': int(class_idx),
'class_name': self._get_class_name(class_idx),
'shap_values': values.tolist(),
'base_value': float(shap_values.base_values[0, class_idx]),
'prediction': float(shap_values.data[0, class_idx])
}
# Get top contributing tokens
token_contributions = list(zip(tokens, values))
token_contributions.sort(key=lambda x: abs(x[1]), reverse=True)
explanation['top_positive'] = [
{'token': t, 'contribution': float(v)}
for t, v in token_contributions if v > 0
][:5]
explanation['top_negative'] = [
{'token': t, 'contribution': float(v)}
for t, v in token_contributions if v < 0
][:5]
return explanation
def plot_explanation(self, text: str, class_idx: int = None):
"""Plot SHAP explanation"""
explanation = self.explain(text, class_idx)
# Create waterfall plot
shap.waterfall_plot(
shap.Explanation(
values=np.array(explanation['shap_values']),
base_values=explanation['base_value'],
data=explanation['tokens'],
feature_names=explanation['tokens']
),
show=False
)
plt.title(f"SHAP Explanation - Class: {explanation['class_name']}")
plt.tight_layout()
return plt
def _get_class_name(self, class_idx: int) -> str:
"""Get class name from index"""
class_names = {0: 'Negative', 1: 'Neutral', 2: 'Positive'}
return class_names.get(class_idx, f'Class_{class_idx}')
def get_feature_importance(self, texts: List[str]) -> Dict[str, float]:
"""Get global feature importance"""
all_contributions = {}
for text in texts:
explanation = self.explain(text)
for token, value in zip(explanation['tokens'], explanation['shap_values']):
if token not in all_contributions:
all_contributions[token] = []
all_contributions[token].append(abs(value))
# Average contributions
feature_importance = {
token: np.mean(values)
for token, values in all_contributions.items()
}
# Sort by importance
feature_importance = dict(
sorted(feature_importance.items(), key=lambda x: x[1], reverse=True)
)
return feature_importance
9. src/explainability/lime_explainer.py
```python
from lime.lime_text import LimeTextExplainer
import numpy as np
from typing import List, Dict, Any
import matplotlib.pyplot as plt
class LIM