Real-time Feature Computation at Serving Time

Comprehensive system design for real-time feature computation in machine learning serving systems with low latency and high throughput requirements.

PublishedSeptember 15, 2026
Length7,117 words
Reading time36 min read

AI/ML System Design

Real-time feature computation at serving time represents a critical challenge in production ML systems, requiring careful balance between computational efficiency, data freshness, and system scalability to meet low-latency prediction requirements.

1. Problem Statement & ML Task Definition

Clarifying questions

What is the expected QPS and latency budget for real-time feature computation?
- This determines whether we can compute features in-memory or need to pre-compute and cache.
What is the scale of users/items and feature space?
- This affects embedding size, feature store design, and computational complexity.
What is the data freshness requirement for features?
- Are features updated every hour, minute, or in real-time?
What is the expected feature update frequency and volume?
- This impacts whether to use batch or streaming feature computation.
What is the target ML task (ranking, classification, etc.)?
- This determines whether we're optimizing for feature relevance or real-time signal processing.
What is the acceptable trade-off between model accuracy and serving latency?
- This influences whether to pre-compute features or compute them on the fly.
What is the expected user/item lifecycle and cold start scenario?
- This determines fallback strategies for new users or items.

Business use case and success criteria

The system enables real-time feature computation at serving time to power ML models that require up-to-the-minute user and item context for accurate ranking, personalization, or recommendation. The system must support low-latency inference (e.g., < 100ms p99) for real-time personalization, while maintaining high model accuracy and supporting dynamic user behavior.

Success criteria:

Latency: p99 < 100ms for feature computation
Accuracy: Model AUC-ROC > 0.85 (offline)
QPS: Support 10,000+ requests/sec
Freshness: Features updated within 5 minutes of user activity

ML problem formulation

This is a ranking task (e.g., personalizing content or product recommendations) where the model learns to rank items based on user features, item features, and interaction context. Real-time feature computation is critical for capturing recent behavior (e.g., last viewed item, time since last interaction) that improves ranking accuracy.

ML evaluation metrics

Offline:

AUC-ROC (for binary classification tasks)
NDCG@K (for ranking tasks)
Precision@K (for top-K retrieval)

Online:

CTR (Click-Through Rate) — primary metric for ad or content recommendation systems
Revenue per user (RPU) — business impact of personalization
Engagement time or session duration — indirect signal of relevance

Scale and constraints

Back-of-envelope estimates:

Training data size: 100M user-item interactions
Number of features: ~50K (user, item, context, interaction)
Embedding matrix memory footprint:
- 50K features × 128 dimensions × 4 bytes = ~256MB (per embedding)
Serving QPS math:
- 10,000 QPS × 100ms = 1,000,000 ms (1 second) per request batch

Constraints:

Latency budget: < 100ms p99 for feature computation
QPS: 10,000+ requests/sec (peak)
Freshness: Features updated within 5 minutes of user activity
Data volume: 100M+ training samples, 50K+ features

Non-ML baseline strategy

Rule-based fallback: For new users or items, use default features (e.g., popular items, default user preferences)
Cold start handling: Use item popularity or category-based features for new items, and default user behavior for new users
Static feature cache: Pre-computed features (e.g., item category, user age group) for low-latency fallback

class FeatureComputationBaseline:
    def __init__(self):
        self.default_user_features = {"age_group": "25-34", "location": "US"}
        self.default_item_features = {"category": "electronics", "popularity_score": 0.8}
    
    def compute_features(self, user_id, item_id):
        # Fallback to default features for new users/items
        if not self.is_user_known(user_id):
            return self.default_user_features
        if not self.is_item_known(item_id):
            return self.default_item_features
        # Otherwise compute real-time features
        return self.compute_realtime_features(user_id, item_id)
    
    def is_user_known(self, user_id):
        return user_id in self.known_users
    
    def is_item_known(self, item_id):
        return item_id in self.known_items

2. Data Strategy

Training Data Sources

Explicit feedback: User ratings, clicks, purchases, and direct engagement signals from user actions (e.g., star ratings, "like" buttons)
Implicit signals: Session duration, scroll depth, time spent on content, page views, and bounce rates
Third-party data: Demographic information from partner APIs, contextual signals (weather, location), and external content metadata
Synthetic data: Generated user-item interactions for rare event modeling or to simulate edge cases

Labeling Pipeline Design

Human annotation: For critical signals like user satisfaction or content quality, using internal annotators with quality control (e.g., 3x annotation per sample, inter-annotator agreement > 85%)
Weak supervision: Using heuristics and rule-based labeling for large-scale initial labeling (e.g., "user clicked within 2 seconds" → positive label)
Programmatic labeling: For real-time signals like "user spent > 10s on item" or "user viewed same content twice in 5 minutes" to create high-confidence labels

Class Imbalance Handling Strategy

Oversampling: Using SMOTE or random oversampling for minority classes (e.g., rare user preferences)
Weighted loss: Apply class weights in model training to penalize misclassification of rare classes more heavily (e.g., weight ratio = 1:5 for positive vs negative samples)
Focal Loss: For highly imbalanced datasets, focal loss reduces the focus on easy examples and increases the importance of hard-to-classify samples

Negative Sampling Approach

Hard negative sampling: Sample items that are similar to the user's preferences but not selected (e.g., items from same category with high engagement but low click rate)
Negative item pool: Maintain a dynamic pool of 10,000–50,000 items that are not relevant to the user but have high engagement or are popular in the system
Stratified sampling: Ensure that negative samples reflect user diversity (e.g., age, region, content type) to avoid overfitting to specific user segments

Data Freshness and Update Frequency Requirements

Real-time signals: Updated every 5–10 seconds (e.g., session activity, user scroll behavior)
Batch features: Updated daily or hourly (e.g., user preferences, item popularity scores)
External data: Updated weekly or monthly (e.g., weather, seasonal trends)

Privacy Compliance

GDPR/CCPA compliance: All user data is pseudonymized and anonymized where necessary. Personal identifiers are removed or hashed before training.
Data retention: 24-month retention policy with automatic deletion of data older than 24 months.
User consent: Explicit opt-in for data collection and usage, with granular control over personalization features.

Data Quality Validation and Monitoring

Data validation: Implement schema validation, missing value checks, and consistency checks (e.g., session duration > 0)
Drift detection: Use statistical tests like Kolmogorov-Smirnov or PSI to detect feature drift in real-time
Monitoring dashboard: Track data quality metrics (e.g., % of valid records, missing feature rates) and alert on anomalies

# Example data validation function in Python
def validate_training_data(df):
    assert df['user_id'].notna().all(), "Missing user IDs"
    assert (df['session_duration'] >= 0).all(), "Negative session durations"
    assert df['timestamp'].notna().all(), "Missing timestamps"
    return True

3. Feature Engineering

Feature Taxonomy

Features are categorized into four main types:

User Features: User demographics, behavior patterns, preferences (e.g., age, gender, historical engagement)
Item Features: Content attributes, metadata (e.g., category, popularity, description embeddings)
Context Features: Temporal, geographic, device information (e.g., time of day, location, device type)
Interaction Features: User-item interaction patterns (e.g., click history, time since last interaction)

Embedding Design

Entity Embeddings:

User ID embeddings: 128-dimensional, trained via collaborative filtering
Item ID embeddings: 128-dimensional, trained via matrix factorization

Text Embeddings:

Item descriptions: 256-dimensional using BERT-base embeddings
User preferences: 128-dimensional using TF-IDF weighted average of item descriptions

Graph Embeddings:

User-item interaction graph: 64-dimensional using Node2Vec
Content similarity graph: 32-dimensional using GraphSAGE

Real-time vs Batch Feature Computation

Feature Type	Real-time?	Reasoning
Recent engagement	Yes	Low latency requirement, user behavior changes rapidly
User preference vectors	No	Computationally expensive, can be pre-computed
Item popularity	Yes	Changes frequently, requires real-time aggregation
Historical behavior window	No	Pre-computed with sliding window

Feature Store Design

Online Store:

Low-latency key-value store (e.g., Redis or DynamoDB)
Optimized for read-heavy workloads with TTL support
Serves features within 10ms p99 latency target

Offline Store:

Data warehouse (e.g., Snowflake or BigQuery)
Used for training data generation, feature engineering pipelines
Supports complex aggregations and historical analysis

Feature Freshness and TTL

Feature	Source	Update Frequency	Store Type
User session count	Real-time stream	Every 5 minutes	Online
Item popularity score	Batch job	Hourly	Offline
User preference vector	Batch job	Daily	Offline
Recent item interactions	Real-time stream	Every 1 minute	Online
User age	User profile API	On update	Online

Cross Features and Feature Interactions

Cross features are engineered to capture complex user-item relationships:

user_age * item_category: Age-based category preferences
user_session_count * item_popularity: Engagement with popular items
time_of_day * user_preference_vector: Temporal preference alignment

Feature Importance and Selection Methodology

Method: SHAP-based feature importance analysis on production data

Rank features by contribution to model output variance
Remove features with SHAP importance < 0.01 for model efficiency
Retain top 200 features for serving, with 50% of features being cross-features

Implementation Example: Real-time Feature Computation Engine

import time
from typing import Dict, List, Any
from collections import defaultdict, deque

class RealTimeFeatureEngine:
    def __init__(self):
        self.user_sessions = defaultdict(deque)
        self.item_popularity = {}
        self.user_preferences = {}
        
    def compute_user_features(self, user_id: str, recent_interactions: List[Dict]) -> Dict[str, Any]:
        # Compute real-time user features
        features = {}
        
        # Recent session count (real-time)
        current_time = time.time()
        self.user_sessions[user_id].append(current_time)
        
        # Keep only last 10 sessions
        while len(self.user_sessions[user_id]) > 10:
            self.user_sessions[user_id].popleft()
            
        features['recent_session_count'] = len(self.user_sessions[user_id])
        
        # Time-based engagement (real-time)
        recent_engagement = [t for t in self.user_sessions[user_id] 
                           if current_time - t < 3600]  # Last hour
        features['hourly_engagement_count'] = len(recent_engagement)
        
        # User preference vector (batch computed, cached)
        features['preference_vector'] = self.user_preferences.get(user_id, [0.0] * 128)
        
        return features
    
    def compute_item_features(self, item_id: str) -> Dict[str, Any]:
        # Item features from batch pipeline
        features = {}
        
        # Popularity score (real-time updated)
        features['popularity_score'] = self.item_popularity.get(item_id, 0.0)
        
        # Category features (pre-computed)
        features['category'] = self.get_item_category(item_id)
        
        return features
    
    def get_item_category(self, item_id: str) -> str:
        # Placeholder for actual category lookup
        return "electronics" if item_id.startswith("item_") else "books"

Feature Caching Strategy

Prediction Cache: Store computed features for user-item pairs with TTL of 5 minutes
Embedding Cache: Maintain user and item embedding vectors in memory with LRU eviction
KV Cache: For transformer-based models, cache attention keys/values for recent sequences

Feature Versioning and Rollout Strategy

Versioned Features: Each feature has semantic version (e.g., user_session_count_v2)
Gradual Rollout: 10% of traffic to new features, monitored for drift
Feature Flags: Enable/disable features at serving time for A/B testing

Performance Considerations

Latency Budget: 50ms p99 for feature computation pipeline
Memory Footprint: < 1GB for user session cache, < 2GB for embedding store
Throughput: Support 10,000 QPS with concurrent feature computation
Consistency: Eventual consistency model for real-time features with strong consistency for user preferences

4. Model Architecture

Model selection with justification

For real-time feature computation at serving time, we select a Two-Tower Neural Network architecture with Multi-Task Learning (MTL) capabilities. This choice balances model expressiveness with serving latency constraints.

Justification:

Two-Tower Architecture: Separates user and item representations, enabling efficient real-time embedding computation. Each tower processes user/item features independently before computing dot-product similarity for ranking.
Multi-Task Learning: Allows simultaneous optimization of multiple objectives (e.g., CTR prediction, engagement, revenue) without significant latency overhead.
Scalability: Can be easily distributed across multiple model replicas for horizontal scaling.
Latency: Forward pass through two towers with embedding lookups and dot product is sub-10ms on GPU, meeting our 50ms p99 latency target.

We reject:

Transformer-based models: While powerful, they introduce significant latency overhead due to attention computation and are not well-suited for real-time serving.
Logistic Regression: Insufficient expressiveness for complex user-item interactions.
GBM models: Not designed for real-time serving and require expensive feature engineering.

Architecture diagram

graph TD
    A[User Features] --> B[User Tower]
    C[Item Features] --> D[Item Tower]
    B --> E[User Embedding]
    D --> F[Item Embedding]
    E --> G[Dot Product]
    F --> G
    G --> H[Scoring Layer]
    H --> I[CTR Prediction]
    H --> J[Engagement Score]
    H --> K[Revenue Prediction]

Two-tower / multi-task / ensemble design if applicable

We implement a Two-Tower Multi-Task Neural Network:

User Tower: Processes user features (user ID, demographics, behavior history)
Item Tower: Processes item features (item ID, category, metadata)
Multi-Task Heads: Separate output layers for:
- CTR prediction (binary classification)
- Engagement score (regression)
- Revenue prediction (regression)

Pre-trained model selection and fine-tuning strategy

For foundational features, we leverage:

Pre-trained user embeddings from historical behavioral data (e.g., using contrastive learning)
Item embeddings from item metadata and content features
Fine-tuning: Model is fine-tuned on recent user-item interaction data with a learning rate of 1e-4 for 5 epochs

Model complexity vs latency trade-off analysis

Model	Metric	Latency (p99)	Size
Two-Tower (Small)	CTR AUC	15ms	2MB
Two-Tower (Medium)	CTR AUC	30ms	8MB
Two-Tower (Large)	CTR AUC	50ms	32MB
Transformer	CTR AUC	150ms	128MB

Offline evaluation baseline framing

Baseline: Logistic Regression model achieving AUC-ROC of 0.72 Proposed: Two-Tower MTL model achieving AUC-ROC of 0.82 (14% improvement)

Core model code snippet

import torch
import torch.nn as nn
import torch.nn.functional as F

class TwoTowerMTL(nn.Module):
    def __init__(self, user_vocab_size, item_vocab_size, 
                 user_emb_dim=128, item_emb_dim=128, 
                 hidden_dims=[256, 128], num_tasks=3):
        super().__init__()
        
        # Embedding layers
        self.user_embedding = nn.Embedding(user_vocab_size, user_emb_dim)
        self.item_embedding = nn.Embedding(item_vocab_size, item_emb_dim)
        
        # User tower
        user_layers = [nn.Linear(user_emb_dim, hidden_dims[0])]
        for i in range(len(hidden_dims)-1):
            user_layers.append(nn.ReLU())
            user_layers.append(nn.Linear(hidden_dims[i], hidden_dims[i+1]))
        self.user_tower = nn.Sequential(*user_layers)
        
        # Item tower
        item_layers = [nn.Linear(item_emb_dim, hidden_dims[0])]
        for i in range(len(hidden_dims)-1):
            item_layers.append(nn.ReLU())
            item_layers.append(nn.Linear(hidden_dims[i], hidden_dims[i+1]))
        self.item_tower = nn.Sequential(*item_layers)
        
        # Multi-task heads
        self.ctr_head = nn.Linear(hidden_dims[-1], 1)
        self.engagement_head = nn.Linear(hidden_dims[-1], 1)
        self.revenue_head = nn.Linear(hidden_dims[-1], 1)
        
    def forward(self, user_ids, item_ids):
        # Compute embeddings
        user_emb = self.user_embedding(user_ids).squeeze(1)
        item_emb = self.item_embedding(item_ids).squeeze(1)
        
        # Process through towers
        user_repr = self.user_tower(user_emb)
        item_repr = self.item_tower(item_emb)
        
        # Compute dot product for ranking
        scores = torch.sum(user_repr * item_repr, dim=1, keepdim=True)
        
        # Multi-task predictions
        ctr_pred = self.ctr_head(user_repr).sigmoid()
        engagement_pred = self.engagement_head(user_repr)
        revenue_pred = self.revenue_head(user_repr)
        
        return {
            'scores': scores,
            'ctr': ctr_pred,
            'engagement': engagement_pred,
            'revenue': revenue_pred
        }

# Example usage for inference
model = TwoTowerMTL(user_vocab_size=10000, item_vocab_size=50000)
user_ids = torch.LongTensor([123, 456])
item_ids = torch.LongTensor([789, 101])

# Forward pass
outputs = model(user_ids, item_ids)
print(f"Scores: {outputs['scores']}")
print(f"CTR predictions: {outputs['ctr']}")

5. Serving & Inference Architecture

System Diagram

graph TD
    A[User Request] --> B[API Gateway]
    B --> C[Feature Store]
    C --> D[Real-time Feature Computation Service]
    D --> E[Model Serving Layer]
    E --> F[Scoring/Inference]
    F --> G[Response]
    
    subgraph "Real-time Feature Computation"
        D --> H[Feature Lookup Cache]
        D --> I[Stream Processing Engine]
        I --> J[Event Stream]
        J --> K[Feature Computation]
    end

Online vs Batch Serving Decision

For real-time feature computation, we choose online serving over batch because:

Latency requirements: 200ms p99 target for user-facing applications
User context dependency: Features like recent clicks, session state, and real-time user preferences must be computed immediately
Dynamic feature updates: Real-time signals (like user location, time-based features) require immediate availability
Personalization needs: Individual user preferences and behavior patterns must be reflected instantly

Two-Stage Pipeline

We implement a two-stage pipeline:

Candidate Retrieval: Fast approximate matching using pre-computed embeddings and inverted indexes
Scoring & Re-ranking: Detailed feature computation and model scoring with real-time features

Serving Infrastructure

We use TorchServe for model serving with the following configuration:

Model format: TorchScript (for optimized inference)
Hardware: GPU-accelerated instances (NVIDIA A10 or H100)
Scaling: Auto-scaling based on QPS with warm-up periods
Load balancing: Round-robin with sticky sessions for session-based features

Latency Budget Breakdown

Stage	p50 Latency	p99 Latency	Budget
API Gateway	1ms	5ms	20ms
Feature Computation	15ms	40ms	50ms
Model Inference	25ms	60ms	100ms
Response Assembly	5ms	15ms	20ms
Total	46ms	120ms	190ms

Model Compression Strategy

We implement model compression with the following trade-offs:

Quantization: INT8 quantization reduces model size by 4x with <2% accuracy degradation
Pruning: 50% weight pruning for dense layers with minimal impact on performance
Distillation: Student model trained to mimic teacher model with 20% size reduction

Caching Strategy

We implement multi-layer caching:

Prediction Cache: LRU cache for recent user predictions (500MB)
Embedding Cache: Redis-based cache for frequently accessed embeddings (1GB)
KV Cache: For LLMs, caching attention keys/values to reduce recomputation (2GB)
Feature Cache: TTL-based cache for computed features with 5-minute expiration

Auto-scaling and Capacity Planning

We design auto-scaling based on:

QPS prediction: Historical pattern analysis with 95% confidence intervals
Resource utilization: GPU memory and CPU usage monitoring
Capacity planning: 2x headroom for peak traffic (estimated 50K QPS)
Graceful degradation: Fallback to pre-computed features when system is overloaded

Real-time Feature Computation Pipeline

import asyncio
from typing import Dict, List, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
import redis
import numpy as np

@dataclass
class UserContext:
    user_id: str
    session_id: str
    timestamp: datetime
    location: str
    device_type: str

class RealTimeFeatureComputation:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.feature_cache_ttl = 300  # 5 minutes
    
    async def compute_features(self, user_context: UserContext) -> Dict[str, Any]:
        """Compute real-time features for a user request"""
        # Check cache first
        cache_key = f"features:{user_context.user_id}:{user_context.session_id}"
        cached_features = self.redis.get(cache_key)
        
        if cached_features:
            return self._deserialize_features(cached_features)
        
        # Compute features in parallel
        feature_tasks = [
            self._compute_recent_interactions(user_context),
            self._compute_session_features(user_context),
            self._compute_time_based_features(user_context),
            self._compute_location_features(user_context)
        ]
        
        # Execute all feature computations concurrently
        results = await asyncio.gather(*feature_tasks, return_exceptions=True)
        
        # Aggregate results
        features = {}
        for result in results:
            if isinstance(result, Exception):
                # Log error but continue with other features
                print(f"Feature computation error: {result}")
                continue
            features.update(result)
        
        # Cache computed features
        self.redis.setex(
            cache_key, 
            self.feature_cache_ttl,
            self._serialize_features(features)
        )
        
        return features
    
    async def _compute_recent_interactions(self, user_context: UserContext) -> Dict[str, Any]:
        """Compute features based on recent user interactions"""
        # Simulate fetching from stream processor
        recent_events = await self._fetch_recent_events(
            user_context.user_id,
            time_window=timedelta(minutes=30)
        )
        
        # Compute interaction counts and recency
        interaction_counts = {}
        for event in recent_events:
            item_type = event.get('item_type', 'default')
            interaction_counts[item_type] = interaction_counts.get(item_type, 0) + 1
        
        # Compute recency scores
        recency_scores = []
        for event in recent_events:
            time_diff = (user_context.timestamp - event['timestamp']).seconds
            recency_scores.append(1.0 / (1 + time_diff / 3600))  # Hours-based decay
        
        return {
            'recent_interaction_count': len(recent_events),
            'recent_interaction_types': interaction_counts,
            'recency_weighted_score': sum(recency_scores) if recency_scores else 0.0
        }
    
    async def _fetch_recent_events(self, user_id: str, time_window: timedelta) -> List[Dict]:
        """Simulate fetching recent events from stream processor"""
        # In real implementation, this would query a streaming data store
        # like Kafka or Redis Streams
        return [
            {
                'item_id': f'item_{i}',
                'item_type': 'product',
                'timestamp': datetime.now() - timedelta(minutes=i*5),
                'event_type': 'view'
            } for i in range(5)
        ]

# Example usage
async def main():
    # Initialize feature computation service
    redis_client = redis.Redis(host='localhost', port=6379, db=0)
    feature_service = RealTimeFeatureComputation(redis_client)
    
    # Simulate user request
    user_context = UserContext(
        user_id="user_123",
        session_id="session_456",
        timestamp=datetime.now(),
        location="NYC",
        device_type="mobile"
    )
    
    # Compute features
    features = await feature_service.compute_features(user_context)
    print("Computed features:", features)

# Run example
# asyncio.run(main())

Fault Tolerance and Fallback Strategy

Graceful degradation: When real-time feature computation fails, fall back to:
- Cached features from previous successful computations
- Default/average feature values for new users/items
- Rule-based fallbacks for critical features
Circuit breaker pattern: For streaming feature services, implement circuit breaker to prevent cascading failures
Stale model handling: When serving models are unavailable, use last-known good model version with appropriate warning logs

Performance Monitoring

We instrument the feature computation pipeline with:

Latency metrics: p50/p99/p99.9 latencies for each feature computation
Error rates: Percentage of failed feature computations
Cache hit ratios: Efficiency of feature caching strategy
Throughput metrics: QPS and concurrent request handling capacity

This architecture supports sub-200ms response times while maintaining high availability and scalability for real-time personalization systems.

6. Evaluation Framework

Offline Evaluation Methodology

For offline evaluation, we use temporal cross-validation to ensure our model generalizes to future data. This approach splits the dataset chronologically, with older data used for training and newer data for validation. We use a 3-month rolling window for training and 1-month holdout for validation to simulate real-world temporal dynamics. For model comparison, we use a 5-fold temporal split where each fold uses 20% of the most recent data for validation.

Slice-Based Evaluation

We evaluate model performance across critical user segments:

User demographics: Age groups (18-25, 26-35, 36+), gender distribution
Geographic regions: US, EU, APAC with different engagement patterns
Device types: Mobile vs desktop vs tablet
Engagement levels: New users (0-7 days), active users (>30 days)
Content categories: Different content types (video, article, image)

A/B Test Design

We design a randomized controlled experiment with:

Randomization unit: User ID (to prevent leakage between users)
Sample size calculation: Using power analysis with 80% power, 5% significance level, and expected effect size of 2% CTR improvement. With 1M users, we need ~50k users per variant.
Test duration: 4 weeks to capture seasonal variation and reduce noise
Primary metric: Click-through rate (CTR) for content recommendations

Guardrail Metrics and Blocking Conditions

We implement the following guardrails:

Recall drop below 0.8: Block rollout if recall drops more than 10% from baseline
CTR decrease >5%: Stop serving if CTR drops more than 5% in any segment
Latency increase >20ms: Prevent deployment if p99 latency exceeds 150ms
Error rate >0.5%: Block if serving errors exceed 0.5% of requests

Shadow Mode Rollout Plan

Before full A/B testing, we run shadow mode for 2 weeks where:

5% of live traffic is routed to the new model in shadow mode
Shadow predictions are logged but not served to users
Offline evaluation compares shadow model performance vs baseline
If shadow results show >2% improvement, proceed to full A/B test

Long-term Metric Measurement

We track delayed conversion metrics:

7-day conversion rate: Measure if recommendations lead to content consumption within 7 days
Engagement decay: Monitor if engagement drops more than 15% after 30 days
User retention: Track if users return to platform more than 20% more often with recommendations

# Example A/B test evaluation code
import numpy as np
from scipy import stats

def ab_test_evaluation(control_ctr, test_ctr, control_count, test_count):
    """
    Evaluate A/B test results for CTR improvement
    Returns: p-value, confidence interval, and statistical significance
    """
    # Calculate pooled variance for z-test
    pooled_var = (control_ctr * (1 - control_ctr) / control_count + 
                  test_ctr * (1 - test_ctr) / test_count)
    
    # Z-score calculation
    z_score = (test_ctr - control_ctr) / np.sqrt(pooled_var)
    
    # P-value calculation
    p_value = 2 * (1 - stats.norm.cdf(abs(z_score)))
    
    # Confidence interval calculation
    margin_error = 1.96 * np.sqrt(pooled_var)
    ci_lower = (test_ctr - control_ctr) - margin_error
    ci_upper = (test_ctr - control_ctr) + margin_error
    
    return {
        'p_value': p_value,
        'confidence_interval': (ci_lower, ci_upper),
        'statistically_significant': p_value < 0.05,
        'improvement_percentage': (test_ctr - control_ctr) / control_ctr * 100
    }

# Example usage:
# ab_results = ab_test_evaluation(0.02, 0.023, 100000, 100000)
# print(f"Improvement: {ab_results['improvement_percentage']:.2f}%")

7. Training Pipeline

Training Data Pipeline

The training pipeline follows a multi-stage ETL process:

Data Extraction: Raw data flows from various sources (user events, item metadata, session logs) via Kafka streams
Preprocessing: Feature engineering and cleaning using Spark or Dask for distributed processing
Shuffling & Batching: Data is shuffled and batched for efficient training using TFRecord or Parquet formats
Data Validation: Schema validation and quality checks performed before training

# Example data pipeline component in Python using PySpark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, lit

def preprocess_training_data(raw_df):
    # Handle missing values
    processed_df = raw_df.fillna({
        "user_age": 25,
        "item_category": "unknown"
    })
    
    # Feature engineering
    processed_df = processed_df.withColumn(
        "user_engagement_score",
        when(col("time_spent") > 300, 1.0).otherwise(0.0)
    )
    
    # Remove duplicates and invalid records
    processed_df = processed_df.dropDuplicates(["user_id", "item_id"])
    
    return processed_df

# Example batch processing for training
def create_training_batches(df, batch_size=10000):
    """Create batches for distributed training"""
    from sklearn.model_selection import train_test_split
    
    # Split into train/validation/test
    train_df, temp_df = train_test_split(df, test_size=0.2, random_state=42)
    val_df, test_df = train_test_split(temp_df, test_size=0.5, random_state=42)
    
    # Convert to batches
    train_batches = [train_df.iloc[i:i+batch_size] 
                     for i in range(0, len(train_df), batch_size)]
    
    return train_batches, val_df, test_df

Distributed Training Strategy

For large-scale training, we implement data parallelism with model parallelism for transformer-based models:

Data Parallelism: Each GPU processes a different batch of data, with gradients synchronized via AllReduce
Model Parallelism: For very large models (e.g., >10B parameters), we split model layers across GPUs
Pipeline Parallelism: For very large batch sizes, we pipeline forward/backward passes

Training Infrastructure

Cluster Sizing: For a 50M user, 10M item model with 200D embeddings:
- GPU Cluster: 32x A100 GPUs (40GB VRAM each) for distributed training
- Training Time: ~2-3 days for full dataset with 10 epochs
- Memory Usage: ~50GB per GPU (embeddings + activations)
Storage: SSD-backed storage for fast data access and model checkpointing

Hyperparameter Tuning Approach

We use Bayesian optimization with Optuna for efficient search:

import optuna
from sklearn.ensemble import RandomForestClassifier

def objective(trial):
    # Hyperparameter search space
    n_estimators = trial.suggest_int('n_estimators', 100, 1000)
    max_depth = trial.suggest_int('max_depth', 3, 15)
    learning_rate = trial.suggest_float('learning_rate', 0.01, 0.3)
    
    model = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        learning_rate=learning_rate
    )
    
    # Evaluate model performance (mocked for example)
    score = evaluate_model(model, X_train, y_train, X_val, y_val)
    
    return score

# Run optimization
study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100)

Experiment Tracking and Reproducibility

We use MLflow for experiment tracking:

import mlflow
import mlflow.sklearn

# Example experiment tracking
with mlflow.start_run():
    # Log parameters
    mlflow.log_param("n_estimators", n_estimators)
    mlflow.log_param("max_depth", max_depth)
    
    # Log metrics
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("f1_score", f1_score)
    
    # Log model
    mlflow.sklearn.log_model(model, "model")
    
    # Log features used
    mlflow.log_param("features", feature_names)

Model Versioning and Registry Strategy

Versioning: Semantic versioning (v1.0.0, v1.1.0) with git commit hash as suffix
Registry: MLflow Model Registry for model lifecycle management
Tagging: Version tags include "production", "staging", "candidate"

Continual / Online Learning Design

For real-time adaptation, we implement:

Online Learning: Incremental updates using SGD with learning rate decay
Sliding Window: Recent 30 days of data for model retraining
Concept Drift Detection: Using ADWIN or DDM algorithms to detect shifts in data distribution

# Example online learning update logic
class OnlineModelUpdater:
    def __init__(self, model, learning_rate=0.01):
        self.model = model
        self.learning_rate = learning_rate
        
    def update_model(self, new_data_batch):
        """Update model with new batch of data"""
        for sample in new_data_batch:
            # Compute gradients
            grad = self.compute_gradient(sample)
            
            # Apply gradient update
            self.model.update_parameters(grad, self.learning_rate)
            
            # Decay learning rate
            self.learning_rate *= 0.999
            
    def compute_gradient(self, sample):
        # Simplified gradient computation
        prediction = self.model.predict(sample.features)
        error = prediction - sample.label
        
        # Return gradient for parameter update
        return self.model.compute_gradients(error, sample.features)

The training pipeline is designed for robustness and scalability, supporting both batch and incremental learning paradigms with proper experiment tracking and version control.

8. Monitoring & Observability

Data Drift Detection

Data drift detection is critical for maintaining model performance in production. We implement drift detection using Population Stability Index (PSI) and Kullback-Leibler divergence for continuous features, and Chi-squared tests for categorical features. Drift thresholds are set at 0.1 PSI or 0.05 KL divergence for continuous features, with 5% significance level for categorical features.

import numpy as np
from scipy import stats

def calculate_psi(expected, actual, bins=10):
    """Calculate Population Stability Index"""
    expected = np.array(expected)
    actual = np.array(actual)
    
    # Create bins
    bin_edges = np.histogram_bin_edges(expected, bins=bins)
    
    # Count occurrences in each bin
    expected_counts = np.histogram(expected, bins=bin_edges)[0]
    actual_counts = np.histogram(actual, bins=bin_edges)[0]
    
    # Avoid zero division
    expected_counts = np.where(expected_counts == 0, 1e-6, expected_counts)
    actual_counts = np.where(actual_counts == 0, 1e-6, actual_counts)
    
    psi = np.sum((actual_counts - expected_counts) * 
                 np.log(actual_counts / expected_counts))
    return psi

def calculate_kl_divergence(p, q):
    """Calculate KL divergence between two distributions"""
    return np.sum(p * np.log(p / q))

Prediction Drift and Label Drift Monitoring

Prediction drift is monitored by comparing model predictions over time using statistical tests. Label drift detection uses change point detection algorithms to identify shifts in label distributions, particularly important for ranking models where relevance labels may evolve over time.

Model Performance Degradation Alerting

We implement alerting for model performance degradation using:

AUC-ROC drop below 5% from baseline
NDCG@K decrease by 3% from historical average
CTR drop below 2% from baseline
Prediction confidence variance increase by 15%

Alerts are configured with escalation paths:

P1 (immediate): Slack alerts + on-call notification
P2 (within 24h): Email to ML team + dashboard alert

Retraining Trigger Strategy

Retraining is triggered by:

Scheduled: Monthly retraining with fresh data
Drift-based: When PSI > 0.15 or KL divergence > 0.08
Performance-based: When AUC drops below threshold or prediction drift exceeds limits

Logging Design

Prediction logs include:

User ID, item ID, timestamp, features, prediction scores, actual labels (when available)
Feature logs with feature values and timestamps for drift analysis
Label collection pipeline with human review queue

import json
from datetime import datetime

class PredictionLogger:
    def __init__(self):
        self.logs = []
    
    def log_prediction(self, user_id, item_id, features, prediction_score, 
                       actual_label=None, timestamp=None):
        log_entry = {
            "user_id": user_id,
            "item_id": item_id,
            "features": features,
            "prediction_score": prediction_score,
            "actual_label": actual_label,
            "timestamp": timestamp or datetime.now().isoformat()
        }
        self.logs.append(log_entry)
        
        # Log to monitoring system
        self._send_to_monitoring_system(log_entry)
    
    def _send_to_monitoring_system(self, log_entry):
        # Implementation for sending to monitoring systems like Prometheus or ELK
        pass

Feedback Loop for Continuous Improvement

Feedback collection includes:

User engagement signals (clicks, time spent, conversions)
Manual label corrections from human reviewers
A/B test results for model variants

We implement a feedback loop that:

Collects user interactions with model predictions
Triggers retraining when feedback indicates performance degradation
Updates training data with corrected labels for continuous learning

The system monitors model confidence distributions and flags when confidence drops below thresholds, indicating potential model instability or data drift that requires investigation.

9. Responsible AI & Safety

Bias Analysis

Training data audit: Conduct comprehensive analysis of user demographics, geographic distribution, and content diversity in training data to identify potential representation gaps
Slice-based fairness evaluation: Evaluate model performance across key user segments (age groups, gender, geographic regions) to detect disparate impact
Disparate treatment detection: Monitor for systematic differences in model predictions between user groups that could indicate bias

Fairness Metrics and Constraints

Primary fairness metric: Demographic parity (equal positive prediction rates across protected attributes)
Secondary metrics: Equalized odds (equal true positive and false positive rates) and calibration across slices
Decision rationale: Demographic parity chosen as primary metric to ensure equitable treatment of all user groups, with equalized odds as secondary check for nuanced fairness

Explainability Requirements and Approach

SHAP values: Implement SHAP-based feature importance for top 50% of predictions to provide model transparency
Attention visualization: For transformer-based models, surface attention weights to show which user/item features drove predictions
Stakeholder-specific explainability:
- Business stakeholders: High-level feature impact scores
- Engineering team: Per-feature contribution analysis
- Compliance team: Audit trail of model decisions with confidence intervals

For LLM Systems: Safety Guardrails

Content filtering: Implement real-time content safety checks with toxicity, hate speech, and misinformation detection
Output validation: Validate model outputs against factual constraints and logical consistency
Rate limiting: Apply per-user rate limits to prevent abuse and ensure system stability
Red teaming: Regular adversarial testing with diverse user scenarios to identify edge cases and safety gaps

Model Card Documentation Plan

Content requirements:
- Model architecture and training methodology
- Performance metrics across different user segments
- Known limitations and edge cases
- Data sources and preprocessing steps
- Fairness assessment results and mitigation strategies

Incident Response Plan for Model Failures or Harmful Outputs

Failure detection thresholds:
- Sudden drop in prediction confidence (5% threshold)
- Spike in error rates for specific user segments (10% threshold)
- Outlier prediction patterns exceeding 3σ from baseline
Escalation protocol:
- Tier 1: Automated alert to ML team (within 5 minutes)
- Tier 2: Manual investigation and model rollback if needed (within 30 minutes)
- Tier 3: Business impact assessment and user communication (within 2 hours)
Communication strategy:
- Internal: Real-time alerts to engineering and product teams
- External: User-facing notifications for critical safety issues with clear mitigation steps

10. Scalability & Reliability

Bottleneck Analysis per Pipeline Stage

Back-of-envelope estimates for a real-time recommendation system serving 100K QPS with 100M users and 10M items:

Feature Computation Stage:
- 100K QPS × 50 features per user = 5M feature computations/sec
- Assuming 1ms per feature computation → 5000ms total latency = 5s per second of processing
- For 100K QPS, we need 500x parallelism or 500ms per request budget
Model Inference Stage:
- 100K QPS × 50ms inference time = 5M ms/sec = 5 seconds per second of processing
- Requires 100x parallelism or 50ms budget per request
Embedding Lookup Stage:
- User embedding matrix (100M × 256) = ~100GB
- Item embedding matrix (10M × 256) = ~10GB
- With 100K QPS, we need to support 100K concurrent lookups with <50ms latency

Scale-up Strategy

Horizontal Scaling of Serving Components:

Feature Computation: Distribute across 500+ compute nodes, each handling 200 QPS
Model Inference: Use model parallelism with 100+ GPU instances, batch size of 32
Embedding Stores: Use distributed embedding tables across 100+ nodes with 256GB RAM each

Distributed Feature Computation:

Implement feature computation as a microservice with 500+ instances
Use consistent hashing to distribute user IDs across feature computation nodes
Each node maintains 200K users in-memory for fast lookup

Fault Tolerance

Model Serving Fallbacks:

Implement graceful degradation: if primary model fails, fallback to rule-based scoring with 80% accuracy
Use circuit breaker pattern for feature services (fail fast if >20% error rate)
Implement model versioning with rollback capability

Stale Model Serving on Failure:

class FallbackModel:
    def __init__(self, primary_model, fallback_model):
        self.primary = primary_model
        self.fallback = fallback_model
        self.failure_count = 0
        self.max_failures = 5
        
    def predict(self, input_data):
        try:
            return self.primary.predict(input_data)
        except Exception as e:
            self.failure_count += 1
            if self.failure_count > self.max_failures:
                return self.fallback.predict(input_data)
            return self.primary.predict(input_data)  # Retry primary

Cold Start Handling Strategy

New User Cold Start:

Use content-based features from user profile (interests, demographics)
Apply default item popularity scores for first 100 items
Implement user embedding warm-up using collaborative filtering with minimal data

New Item Cold Start:

Use item metadata features (category, tags, content embeddings)
Apply popularity-based ranking for initial items
Implement item embedding training with minimal feedback

class ColdStartHandler:
    def __init__(self):
        self.popularity_threshold = 1000
        self.default_ranking = self._get_default_items()
        
    def handle_new_user(self, user_profile):
        if not user_profile.get('features'):
            return self.default_ranking
        # Use content-based features for new users
        return self._content_based_ranking(user_profile)
        
    def handle_new_item(self, item_metadata):
        if not item_metadata.get('popularity'):
            return self._default_item_ranking(item_metadata)
        # Use popularity and content features
        return self._rank_by_popularity_and_content(item_metadata)

Multi-Region Serving Strategy

For global scale, implement multi-region serving with:

Primary Region: Handle 80% of traffic with low latency (e.g., US-West)
Secondary Region: Handle 20% of traffic with higher latency (e.g., US-East)
Cross-region replication: Use read replicas for embeddings and feature stores
Traffic steering: Route users based on geographic location and latency measurements

class MultiRegionServing:
    def __init__(self):
        self.regions = {
            'us-west': {'latency_target': 20, 'weight': 0.8},
            'us-east': {'latency_target': 50, 'weight': 0.2}
        }
        
    def route_request(self, user_location):
        # Implement routing logic based on region and latency
        region = self._determine_optimal_region(user_location)
        return self._forward_to_region(region)

Performance Metrics & Monitoring

QPS Target: 100K QPS with <50ms p99 latency
Error Rate: <0.1% error rate with automatic failover
Feature Freshness: <5min TTL for real-time features, <1hr for batch features
Model Accuracy: Maintain >95% of baseline performance during degradation

Capacity Planning

Based on 100K QPS requirement:

Feature Computation: 500 nodes × 200 QPS = 100K QPS capacity
Model Inference: 100 GPU nodes × 500 QPS = 50K QPS capacity (requires scaling)
Embedding Storage: 110GB total (user + item embeddings) with 2x redundancy
Network Bandwidth: 10Gbps per node for inter-node communication

Reliability Patterns

Circuit Breaker: Prevent cascading failures in feature services
Retry Logic: Implement exponential backoff for transient failures
Graceful Degradation: Provide basic functionality when components fail
Health Checks: Monitor service availability and latency metrics

This architecture ensures the system can scale to handle massive traffic while maintaining high reliability and low latency through careful distribution, fault tolerance, and capacity planning.

11. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering / policy alignment

Real-time feature computation vs. pre-computed features: Should we compute features in real-time or pre-compute and store them? This decision impacts latency, system complexity, and data freshness.
Feature freshness policy: How often should features be recomputed? Should we use streaming or batch processing for feature updates?
User privacy vs. feature richness: How much real-time behavioral data should we collect and process? This requires balancing personalization with privacy compliance.
Cross-feature dependency management: How do we handle features that depend on other real-time features, particularly in a distributed system?
Feature versioning strategy: How do we handle feature schema evolution and backward compatibility during real-time computation?

Key ML risks

Risk	Impact	Likelihood	Mitigation
Feature drift in real-time pipeline	High	Medium	Implement continuous drift detection and feature retraining triggers
Latency spikes during real-time computation	High	Medium	Implement feature caching, async computation with fallbacks
Data freshness inconsistency	Medium	High	Enforce strict TTLs, implement feature versioning
System overload from real-time feature computation	High	Medium	Implement rate limiting, circuit breaker patterns
Feature computation errors causing prediction failures	High	Low	Design robust fallbacks and error handling
Privacy compliance violations from real-time data processing	High	Medium	Implement strict data access controls and audit logs

Trade-offs made with justification

Trade-off	Justification
Accuracy vs. latency	Real-time features add latency but improve model accuracy; we accept some accuracy degradation for sub-100ms response times
Freshness vs. cost	Real-time feature computation is more expensive than batch; we prioritize freshness for high-value user segments
Model size vs. feature richness	Larger models can leverage more features but increase serving latency; we use model compression and selective feature inclusion
System complexity vs. flexibility	Real-time computation increases system complexity but enables dynamic feature selection and A/B testing
Consistency vs. availability	Real-time features may be eventually consistent; we prioritize availability for user experience over strict consistency

Alternatives considered and rejected

Pre-computed features: Rejected due to freshness limitations and inability to support dynamic personalization
Hybrid approach with feature caching: Partially adopted but not fully implemented due to added complexity and maintenance overhead
Streaming feature store: Rejected for initial implementation due to high engineering cost and complexity in maintaining consistency
Feature aggregation at batch time: Rejected as it doesn't support real-time personalization requirements

Technical debt acknowledged

Feature computation logic coupling: Real-time feature computation logic is tightly coupled to model serving, making it harder to evolve independently
No feature versioning in production: Current implementation lacks robust feature versioning, which could cause prediction inconsistencies during feature updates
Limited error handling: Basic error handling in real-time feature computation, with no circuit breaker or fallback strategies for downstream systems
No feature lineage tracking: Lack of clear tracking of which features were computed from what sources, making debugging and audit challenging
Inconsistent feature TTL handling: Different feature types have inconsistent TTL policies, leading to potential freshness issues in real-time serving

# Example of feature computation with error handling and fallback logic
import time
from typing import Dict, Any, Optional

class RealTimeFeatureComputation:
    def __init__(self):
        self.feature_cache = {}
        self.cache_ttl = 300  # 5 minutes
    
    def compute_user_features(self, user_id: str, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            # Simulate real-time feature computation
            features = {}
            
            # Feature 1: Recent activity count (real-time)
            recent_activity = self._get_recent_user_activity(user_id, context.get('time_window', 3600))
            features['recent_activity_count'] = len(recent_activity)
            
            # Feature 2: User preference score (real-time)
            user_preference = self._compute_user_preference(user_id, context.get('items'))
            features['user_preference_score'] = user_preference
            
            # Feature 3: Session duration (real-time)
            session_duration = context.get('session_duration', 0)
            features['session_duration'] = session_duration
            
            return features
            
        except Exception as e:
            # Fallback to cached or default values
            print(f"Feature computation failed for user {user_id}: {e}")
            return self._get_fallback_features(user_id)
    
    def _get_recent_user_activity(self, user_id: str, time_window: int) -> list:
        # Simulate real-time data fetch
        return [f"activity_{i}" for i in range(5)]  # Mock data
    
    def _compute_user_preference(self, user_id: str, items: list) -> float:
        # Simulate real-time preference computation
        return 0.85  # Mock value
    
    def _get_fallback_features(self, user_id: str) -> Dict[str, Any]:
        # Fallback to default or cached values
        return {
            'recent_activity_count': 0,
            'user_preference_score': 0.5,
            'session_duration': 0
        }

12. Phased Implementation Roadmap

Phase 1: MVP (Baseline to First ML Model)

Timeline: 4-6 weeks
Deliverables:

Rule-based baseline system with basic feature computation
Simple ML model (logistic regression) trained on static features
Basic online serving infrastructure with 100 QPS capacity
Feature store with minimal feature set (user ID, item ID, timestamp)
A/B testing infrastructure for model comparison

Critical Path Dependencies:

Feature store setup (Week 1-2)
Model training pipeline (Week 2-3)
Online serving stack (Week 3-4)
A/B testing framework (Week 4)

Phase 2: Scale (Feature Store & Distributed Training)

Timeline: 6-8 weeks
Deliverables:

Full feature store with online and offline stores
Distributed training pipeline (data parallelism)
Real-time feature computation engine (streaming + batch hybrid)
Support for 1000 QPS serving capacity
Feature freshness monitoring and TTL policies

Critical Path Dependencies:

Feature store scaling (Week 5-6)
Real-time feature computation engine (Week 6-7)
Training infrastructure (Week 7-8)
Model versioning and registry (Week 8)

Phase 3: Excellence (Advanced Features & Continuous Learning)

Timeline: 8-12 weeks
Deliverables:

Real-time feature computation with contextual embeddings (user/item content, session features)
Multi-task model architecture with ensemble support
Online learning capability for model updates (incremental training)
Advanced feature interaction and cross-feature computation
Production-grade monitoring and alerting (drift detection, performance degradation)

Critical Path Dependencies:

Contextual feature embedding pipelines (Week 9-10)
Multi-task model support (Week 10-11)
Online learning infrastructure (Week 11-12)
Production monitoring stack (Week 12)

Critical Path Dependencies Summary:

Feature Store must be operational before any real-time feature computation
Model Training Pipeline must support distributed training for scale
Online Serving Stack must handle QPS requirements with low latency
A/B Testing Framework must be in place before model launch
Real-time Feature Computation Engine requires integration with streaming data sources

Risk Mitigation:

Feature Store Bottlenecks: Start with minimal feature set and scale gradually
Training Pipeline Delays: Use pre-trained embeddings to reduce training time
Serving Latency Issues: Implement caching and model compression strategies early
Data Freshness: Implement TTL policies and streaming ingestion for critical features

Success Metrics:

QPS capacity achieved (100 → 1000+)
Feature computation latency < 50ms p99
Model accuracy improvement > 10% vs baseline
A/B test statistical significance (p < 0.05)
Production monitoring coverage > 95%