Real-Time AI Personalization

What is Real-Time Personalization?

Personalization adapts content, recommendations, and experiences to individual users based on their behavior, preferences, and context - all in milliseconds.

User Profiling

python
from dataclasses import dataclass, field
from typing import Dict, List
import numpy as np
@dataclass
class UserProfile:
    user_id: str
    explicit_preferences: Dict[str, float] = field(default_factory=dict)
    implicit_signals: List[dict] = field(default_factory=list)
    embedding: np.ndarray = None
    
    def update_from_event(self, event: dict):
        if event['type'] == 'click':
            category = event['content_category']
            self.explicit_preferences[category] = (
                self.explicit_preferences.get(category, 0) * 0.9 + 0.1
            )
        elif event['type'] == 'skip':
            category = event['content_category']
            self.explicit_preferences[category] = (
                self.explicit_preferences.get(category, 0) * 0.9
            )

Content Ranking with Multi-Armed Bandit

python
import numpy as np
from collections import defaultdict
class ThompsonSamplingBandit:
    def __init__(self, n_arms: int):
        self.alpha = np.ones(n_arms)  # Successes
        self.beta = np.ones(n_arms)   # Failures
    
    def select_arm(self) -> int:
        samples = np.random.beta(self.alpha, self.beta)
        return np.argmax(samples)
    
    def update(self, arm: int, reward: float):
        if reward > 0:
            self.alpha[arm] += reward
        else:
            self.beta[arm] += 1
def personalized_feed(user_id: str, candidate_items: list, user_profile: UserProfile) -> list:
    bandit = get_user_bandit(user_id, len(candidate_items))
    
    # Score items
    scores = []
    for i, item in enumerate(candidate_items):
        arm = bandit.select_arm()
        preference_score = user_profile.explicit_preferences.get(item['category'], 0.5)
        scores.append(preference_score + np.random.beta(2, 2))  # Thompson sampling
    
    return [x for _, x in sorted(zip(scores, candidate_items), reverse=True)]

Real-Time Feature Store

python
import redis
import json
class FeatureStore:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379)
    
    def get_user_features(self, user_id: str) -> dict:
        key = f"user_features:{user_id}"
        data = self.redis.get(key)
        return json.loads(data) if data else {}
    
    def update_user_features(self, user_id: str, features: dict, ttl: int = 86400):
        key = f"user_features:{user_id}"
        self.redis.setex(key, ttl, json.dumps(features))

A/B Testing Framework

Track personalization variants and their impact:

python
def get_experiment_variant(user_id: str, experiment_name: str) -> str:
    hash_val = hash(f"{user_id}:{experiment_name}") % 100
    if hash_val < 50:
        return "control"
    else:
        return "treatment"