Learn to build production-ready NLP systems with proper evaluation, deployment strategies, monitoring, and ethical considerations.
Building production NLP systems requires more than accurate models. You need robust data pipelines, comprehensive evaluation, scalable deployment, continuous monitoring, and attention to ethical implications. Many promising research models fail in production due to overlooked practical considerations.
This lesson covers the full lifecycle of NLP projects, from data collection and annotation to deployment and maintenance. You'll learn best practices that distinguish reliable production systems from prototype code.
Robust data processing is fundamental to NLP success.
Data Pipeline Components:
1class NLPPipeline:
2 """Production-ready NLP data pipeline"""
3
4 def __init__(self):
5 self.preprocessors = []
6 self.validators = []
7 self.cache = {}
8
9 def add_preprocessor(self, fn, name):
10 self.preprocessors.append((name, fn))
11
12 def add_validator(self, fn, name):
13 self.validators.append((name, fn))
14
15 def process(self, text, validate=True):
16 # Validate input
17 if validate:
18 for name, validator in self.validators:
19 is_valid, error = validator(text)
20 if not is_valid:
21 raise ValueError(f"{name}: {error}")
22
23 # Apply preprocessing chain
24 result = text
25 for name, preprocessor in self.preprocessors:
26 try:
27 result = preprocessor(result)
28 except Exception as e:
29 logging.error(f"Preprocessor {name} failed: {e}")
30 raise
31
32 return result
33
34# Standard validators
35def validate_length(text, min_len=1, max_len=10000):
36 if len(text) < min_len:
37 return False, "Text too short"
38 if len(text) > max_len:
39 return False, "Text too long"
40 return True, None
41
42def validate_encoding(text):
43 try:
44 text.encode('utf-8').decode('utf-8')
45 return True, None
46 except:
47 return False, "Invalid encoding"Proper evaluation prevents overconfident deployment.
1class NLPEvaluator:
2 """Comprehensive NLP model evaluation"""
3
4 def __init__(self, model):
5 self.model = model
6
7 def evaluate(self, test_data, metrics=['accuracy', 'f1', 'precision', 'recall']):
8 results = {}
9
10 # Standard metrics
11 y_true = [d['label'] for d in test_data]
12 y_pred = [self.model.predict(d['text']) for d in test_data]
13
14 from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
15 results['accuracy'] = accuracy_score(y_true, y_pred)
16 results['f1_macro'] = f1_score(y_true, y_pred, average='macro')
17 results['f1_weighted'] = f1_score(y_true, y_pred, average='weighted')
18
19 return results
20
21 def evaluate_subgroups(self, test_data, subgroup_key):
22 """Evaluate on different data subgroups"""
23 subgroups = {}
24 for d in test_data:
25 sg = d.get(subgroup_key, 'unknown')
26 if sg not in subgroups:
27 subgroups[sg] = []
28 subgroups[sg].append(d)
29
30 results = {}
31 for sg_name, sg_data in subgroups.items():
32 results[sg_name] = self.evaluate(sg_data)
33
34 return results
35
36 def cross_validate(self, data, n_folds=5):
37 """K-fold cross validation"""
38 from sklearn.model_selection import KFold
39 kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)
40
41 fold_results = []
42 for train_idx, val_idx in kf.split(data):
43 train_data = [data[i] for i in train_idx]
44 val_data = [data[i] for i in val_idx]
45
46 self.model.train(train_data)
47 fold_results.append(self.evaluate(val_data))
48
49 return fold_resultsCommon patterns for deploying NLP models:
Batch Processing:
1class BatchProcessor:
2 """Process large datasets efficiently"""
3
4 def __init__(self, model, batch_size=32):
5 self.model = model
6 self.batch_size = batch_size
7
8 def process(self, texts):
9 results = []
10 for i in range(0, len(texts), self.batch_size):
11 batch = texts[i:i + self.batch_size]
12 batch_results = self.model.predict_batch(batch)
13 results.extend(batch_results)
14 return resultsReal-time API:
1from fastapi import FastAPI, HTTPException
2from pydantic import BaseModel
3
4app = FastAPI()
5
6class PredictionRequest(BaseModel):
7 text: str
8 options: dict = {}
9
10class PredictionResponse(BaseModel):
11 prediction: str
12 confidence: float
13 latency_ms: float
14
15@app.post("/predict", response_model=PredictionResponse)
16async def predict(request: PredictionRequest):
17 start = time.time()
18
19 try:
20 result = model.predict(request.text)
21 latency = (time.time() - start) * 1000
22
23 return PredictionResponse(
24 prediction=result['label'],
25 confidence=result['confidence'],
26 latency_ms=latency
27 )
28 except Exception as e:
29 raise HTTPException(status_code=500, detail=str(e))Production systems require continuous monitoring.
1class NLPMonitor:
2 """Monitor NLP model in production"""
3
4 def __init__(self, model, alert_threshold=0.1):
5 self.model = model
6 self.alert_threshold = alert_threshold
7 self.baseline_metrics = None
8 self.recent_predictions = []
9
10 def set_baseline(self, test_data):
11 """Establish performance baseline"""
12 self.baseline_metrics = self.evaluate(test_data)
13
14 def log_prediction(self, text, prediction, confidence):
15 """Log predictions for analysis"""
16 self.recent_predictions.append({
17 'timestamp': datetime.now(),
18 'text_length': len(text),
19 'prediction': prediction,
20 'confidence': confidence
21 })
22
23 def check_data_drift(self, recent_texts, reference_texts):
24 """Detect distribution shift in input data"""
25 # Compare text length distributions
26 recent_lengths = [len(t) for t in recent_texts]
27 ref_lengths = [len(t) for t in reference_texts]
28
29 # KS test for distribution difference
30 from scipy.stats import ks_2samp
31 stat, p_value = ks_2samp(recent_lengths, ref_lengths)
32
33 if p_value < 0.05:
34 logging.warning(f"Data drift detected: KS stat={stat:.3f}")
35 return True
36 return False
37
38 def check_model_degradation(self, labeled_sample):
39 """Check if model performance has degraded"""
40 current_metrics = self.evaluate(labeled_sample)
41
42 for metric, baseline_value in self.baseline_metrics.items():
43 current_value = current_metrics.get(metric, 0)
44 degradation = baseline_value - current_value
45
46 if degradation > self.alert_threshold:
47 logging.error(
48 f"Performance degradation: {metric} dropped by {degradation:.2%}"
49 )
50 return True
51
52 return FalseCreate a complete production-ready NLP classification system with proper evaluation and monitoring.
1import numpy as np
2from collections import Counter, defaultdict
3import re
4from datetime import datetime
5import json
6
7class ProductionNLPSystem:
8 """Complete production-ready NLP classification system"""
9
10 def __init__(self, name="nlp_classifier"):
11 self.name = name
12 self.model = None
13 self.preprocessor = None
14 self.metrics_log = []
15 self.prediction_log = []
16 self.version = "1.0.0"
17
18 # ========== Preprocessing ==========
19
20 def preprocess(self, text):
21 """Standard text preprocessing"""
22 if not text or not isinstance(text, str):
23 return ""
24
25 # Lowercase
26 text = text.lower()
27
28 # Remove URLs
29 text = re.sub(r'https?://\S+', '', text)
30
31 # Remove special characters but keep basic punctuation
32 text = re.sub(r'[^\w\s.!?]', '', text)
33
34 # Normalize whitespace
35 text = ' '.join(text.split())
36
37 return text
38
39 # ========== Feature Extraction ==========
40
41 def extract_features(self, text):
42 """Extract features for classification"""
43 text = self.preprocess(text)
44 words = text.split()
45
46 features = {}
47
48 # Word n-grams
49 for word in words:
50 features[f'word_{word}'] = features.get(f'word_{word}', 0) + 1
51
52 # Character n-grams
53 for i in range(len(text) - 2):
54 ngram = text[i:i+3]
55 features[f'char_{ngram}'] = features.get(f'char_{ngram}', 0) + 1
56
57 # Meta features
58 features['length'] = len(text)
59 features['word_count'] = len(words)
60 features['avg_word_len'] = np.mean([len(w) for w in words]) if words else 0
61
62 return features
63
64 # ========== Training ==========
65
66 def train(self, train_data, val_data=None):
67 """Train the classifier"""
68 print(f"Training {self.name} v{self.version}")
69 print(f"Training samples: {len(train_data)}")
70
71 # Aggregate features per class
72 class_features = defaultdict(Counter)
73 class_counts = Counter()
74
75 for item in train_data:
76 text, label = item['text'], item['label']
77 features = self.extract_features(text)
78 class_features[label].update(features)
79 class_counts[label] += 1
80
81 self.model = {
82 'class_features': dict(class_features),
83 'class_counts': dict(class_counts),
84 'classes': list(class_counts.keys())
85 }
86
87 # Validate if validation data provided
88 if val_data:
89 val_metrics = self.evaluate(val_data)
90 print(f"Validation metrics: {val_metrics}")
91
92 return self
93
94 # ========== Prediction ==========
95
96 def predict(self, text, return_probs=False):
97 """Make prediction with confidence"""
98 start_time = datetime.now()
99
100 features = self.extract_features(text)
101
102 # Score each class
103 scores = {}
104 for cls in self.model['classes']:
105 score = np.log(self.model['class_counts'][cls] + 1)
106 for feat, count in features.items():
107 feat_count = self.model['class_features'][cls].get(feat, 0)
108 score += count * np.log(feat_count + 1)
109 scores[cls] = score
110
111 # Softmax for probabilities
112 max_score = max(scores.values())
113 exp_scores = {k: np.exp(v - max_score) for k, v in scores.items()}
114 total = sum(exp_scores.values())
115 probs = {k: v / total for k, v in exp_scores.items()}
116
117 prediction = max(probs, key=probs.get)
118 confidence = probs[prediction]
119
120 # Log prediction
121 latency_ms = (datetime.now() - start_time).total_seconds() * 1000
122 self._log_prediction(text, prediction, confidence, latency_ms)
123
124 if return_probs:
125 return prediction, confidence, probs
126 return prediction, confidence
127
128 def _log_prediction(self, text, prediction, confidence, latency_ms):
129 """Log prediction for monitoring"""
130 self.prediction_log.append({
131 'timestamp': datetime.now().isoformat(),
132 'text_length': len(text),
133 'prediction': prediction,
134 'confidence': confidence,
135 'latency_ms': latency_ms
136 })
137
138 # ========== Evaluation ==========
139
140 def evaluate(self, test_data):
141 """Comprehensive evaluation"""
142 y_true = [d['label'] for d in test_data]
143 y_pred = [self.predict(d['text'])[0] for d in test_data]
144
145 # Basic metrics
146 correct = sum(1 for t, p in zip(y_true, y_pred) if t == p)
147 accuracy = correct / len(y_true)
148
149 # Per-class metrics
150 classes = self.model['classes']
151 class_metrics = {}
152
153 for cls in classes:
154 tp = sum(1 for t, p in zip(y_true, y_pred) if t == cls and p == cls)
155 fp = sum(1 for t, p in zip(y_true, y_pred) if t != cls and p == cls)
156 fn = sum(1 for t, p in zip(y_true, y_pred) if t == cls and p != cls)
157
158 precision = tp / (tp + fp) if (tp + fp) > 0 else 0
159 recall = tp / (tp + fn) if (tp + fn) > 0 else 0
160 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
161
162 class_metrics[cls] = {
163 'precision': precision,
164 'recall': recall,
165 'f1': f1,
166 'support': sum(1 for t in y_true if t == cls)
167 }
168
169 metrics = {
170 'accuracy': accuracy,
171 'class_metrics': class_metrics,
172 'n_samples': len(test_data)
173 }
174
175 # Log metrics
176 self.metrics_log.append({
177 'timestamp': datetime.now().isoformat(),
178 'metrics': metrics
179 })
180
181 return metrics
182
183 # ========== Monitoring ==========
184
185 def get_prediction_stats(self, last_n=100):
186 """Get statistics on recent predictions"""
187 recent = self.prediction_log[-last_n:]
188 if not recent:
189 return {}
190
191 return {
192 'count': len(recent),
193 'avg_confidence': np.mean([p['confidence'] for p in recent]),
194 'avg_latency_ms': np.mean([p['latency_ms'] for p in recent]),
195 'prediction_distribution': Counter([p['prediction'] for p in recent])
196 }
197
198 def export_model(self, filepath):
199 """Export model for deployment"""
200 export_data = {
201 'name': self.name,
202 'version': self.version,
203 'model': self.model,
204 'exported_at': datetime.now().isoformat()
205 }
206 with open(filepath, 'w') as f:
207 json.dump(export_data, f)
208 print(f"Model exported to {filepath}")
209
210# ========== Demo ==========
211
212# Training data
213train_data = [
214 {'text': 'I love this amazing product!', 'label': 'positive'},
215 {'text': 'This is wonderful and fantastic', 'label': 'positive'},
216 {'text': 'Great experience, highly recommend', 'label': 'positive'},
217 {'text': 'Absolutely brilliant service', 'label': 'positive'},
218 {'text': 'Perfect solution for my needs', 'label': 'positive'},
219 {'text': 'I hate this terrible product', 'label': 'negative'},
220 {'text': 'This is awful and disappointing', 'label': 'negative'},
221 {'text': 'Horrible experience, waste of money', 'label': 'negative'},
222 {'text': 'Completely useless and frustrating', 'label': 'negative'},
223 {'text': 'Worst purchase I ever made', 'label': 'negative'},
224]
225
226# Test data
227test_data = [
228 {'text': 'Amazing quality and service!', 'label': 'positive'},
229 {'text': 'Really disappointed with this', 'label': 'negative'},
230 {'text': 'Excellent product, love it', 'label': 'positive'},
231 {'text': 'Terrible, do not buy', 'label': 'negative'},
232]
233
234# Build system
235system = ProductionNLPSystem(name="sentiment_classifier")
236system.train(train_data)
237
238print("\nPRODUCTION NLP SYSTEM")
239print("=" * 60)
240
241# Evaluate
242metrics = system.evaluate(test_data)
243print(f"\nEvaluation Results:")
244print(f" Accuracy: {metrics['accuracy']:.1%}")
245for cls, m in metrics['class_metrics'].items():
246 print(f" {cls}: P={m['precision']:.2f}, R={m['recall']:.2f}, F1={m['f1']:.2f}")
247
248# Make predictions
249print("\n" + "-" * 60)
250print("Sample Predictions:")
251test_texts = [
252 "This product exceeded my expectations!",
253 "Complete waste of money, very disappointed",
254 "Not bad but could be better",
255]
256
257for text in test_texts:
258 pred, conf, probs = system.predict(text, return_probs=True)
259 print(f"\n Text: '{text}'")
260 print(f" Prediction: {pred} (confidence: {conf:.1%})")
261
262# Check monitoring
263print("\n" + "-" * 60)
264print("System Statistics:")
265stats = system.get_prediction_stats()
266print(f" Predictions logged: {stats['count']}")
267print(f" Avg confidence: {stats['avg_confidence']:.1%}")
268print(f" Avg latency: {stats['avg_latency_ms']:.2f}ms")This demonstrates a complete production NLP system with preprocessing, training, evaluation, prediction logging, and monitoring capabilities.
Congratulations on completing the NLP & Text Analytics course! You're now equipped to build, evaluate, and deploy production-ready NLP systems. Consider exploring the Computer Vision course or diving deeper into specific NLP applications.
Your progress is saved in this browser, on this device only.