Designing a Real-Time Recommendation System

Recommendation systems are deceptively complex. A simple "users who bought X also bought Y" is relatively straightforward. A real-time personalized recommendation system that feels fresh, does not repeat items the user has seen, and stays fast at scale requires thoughtful architecture.

The Three Components

Every recommendation system has three jobs:

1. Candidate generation - narrow from millions of items to hundreds
2. Scoring and ranking - score candidates for this user in this context
3. Serving - return results fast, handle cold starts, enforce business rules

Candidate Generation

You cannot score every item for every user. Candidate generation creates a manageable shortlist.

Collaborative filtering (item-to-item):

pythonCopy
import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity

def build_item_similarity_matrix(interaction_matrix: csr_matrix) -> np.ndarray:
    """
    interaction_matrix: users × items, value = interaction count
    Returns: items × items similarity matrix
    """
    # Normalize by user activity to reduce popularity bias
    item_user_matrix = interaction_matrix.T.tocsr()
    # Compute cosine similarity between item vectors
    similarity = cosine_similarity(item_user_matrix, dense_output=False)
    return similarity

def get_cf_candidates(user_id: int, interaction_history: list[int],
                      item_sim: np.ndarray, k: int = 200) -> list[int]:
    """Get k candidates based on items the user has interacted with."""
    candidate_scores = np.zeros(item_sim.shape[0])

    for item_id in interaction_history[-10:]:  # use recent history
        similar_items = item_sim[item_id].toarray().flatten()
        candidate_scores += similar_items

    # Zero out already-seen items
    candidate_scores[interaction_history] = 0

    top_k = np.argsort(candidate_scores)[-k:][::-1]
    return top_k.tolist()

Embedding-based retrieval (two-tower model):

A two-tower model learns separate embeddings for users and items such that users are close to items they like in embedding space:

pythonCopy
import torch
import torch.nn as nn

class UserTower(nn.Module):
    def __init__(self, n_users, embedding_dim=128, hidden_dim=256):
        super().__init__()
        self.user_embedding = nn.Embedding(n_users, 64)
        self.layers = nn.Sequential(
            nn.Linear(64 + n_user_features, hidden_dim),
            nn.ReLU(),
            nn.LayerNorm(hidden_dim),
            nn.Linear(hidden_dim, embedding_dim)
        )

    def forward(self, user_id, user_features):
        user_emb = self.user_embedding(user_id)
        x = torch.cat([user_emb, user_features], dim=-1)
        return nn.functional.normalize(self.layers(x), dim=-1)

class ItemTower(nn.Module):
    def __init__(self, n_items, embedding_dim=128, hidden_dim=256):
        super().__init__()
        self.item_embedding = nn.Embedding(n_items, 64)
        self.layers = nn.Sequential(
            nn.Linear(64 + n_item_features, hidden_dim),
            nn.ReLU(),
            nn.LayerNorm(hidden_dim),
            nn.Linear(hidden_dim, embedding_dim)
        )

    def forward(self, item_id, item_features):
        item_emb = self.item_embedding(item_id)
        x = torch.cat([item_emb, item_features], dim=-1)
        return nn.functional.normalize(self.layers(x), dim=-1)

# At serving time: embed user, then ANN search for closest items
# Pre-compute and index all item embeddings with FAISS
import faiss

index = faiss.IndexFlatIP(128)  # inner product = cosine sim for normalized vectors
index.add(item_embeddings)

def get_embedding_candidates(user_embedding: np.ndarray, k: int = 200):
    distances, indices = index.search(user_embedding.reshape(1, -1), k)
    return indices[0].tolist()

Ranking the Candidates

With ~500 candidates, you can afford a more expensive model:

pythonCopy
import lightgbm as lgb

def extract_ranking_features(user: dict, item: dict, context: dict) -> np.ndarray:
    return np.array([
        # Relevance signals
        cf_score(user['id'], item['id']),
        embedding_similarity(user['embedding'], item['embedding']),

        # Item popularity (log-scaled to reduce dominance)
        np.log1p(item['view_count_7d']),
        np.log1p(item['click_count_7d']),
        item['click_through_rate'],

        # Personalization
        category_match_score(user['category_history'], item['category']),
        recency_of_last_interaction(user['id'], item['id']),

        # Context
        context['hour_of_day'],
        context['day_of_week'],
        context['device_type_encoded'],

        # Item quality
        item['avg_rating'],
        np.log1p(item['rating_count']),

        # Freshness
        days_since_published(item['published_at']),
    ])

Handling Real-Time Context

What makes recommendations "real-time" is incorporating session context - what the user is doing right now:

pythonCopy
class SessionContextEncoder:
    def __init__(self, embedding_dim=32):
        self.embedding_dim = embedding_dim

    def encode_session(self, recent_views: list[dict]) -> np.ndarray:
        """
        Encode the last N items viewed in this session into a context vector.
        Use exponential decay  -  more recent items matter more.
        """
        if not recent_views:
            return np.zeros(self.embedding_dim)

        context = np.zeros(self.embedding_dim)
        decay = 0.9
        for i, view in enumerate(reversed(recent_views[-5:])):
            weight = decay ** i
            context += weight * get_item_embedding(view['item_id'])

        return context / (np.linalg.norm(context) + 1e-8)

Cold Start

Every recommendation system has a cold start problem:

New user cold start: No history to personalize on.

• Show globally popular items
• Use onboarding preferences (explicitly ask what they like)
• Use implicit signals: location, device, referral source

New item cold start: No interactions to learn from.

• Use content-based features exclusively (category, description, price point)
• Boost new items in exploration slots
• Use similar items' interaction patterns

pythonCopy
def get_recommendations(user: dict, context: dict, k: int = 20) -> list[dict]:
    is_new_user = len(user.get('interaction_history', [])) < 10
    is_new_item_request = context.get('include_new_items', False)

    if is_new_user:
        # Cold start path: content-based + popularity
        candidates = get_popular_candidates(k * 5)
        candidates += get_content_based_candidates(user.get('preferences', {}), k * 3)
    else:
        # Warm path: collaborative + embedding
        candidates = get_cf_candidates(user['id'], user['interaction_history'], k * 3)
        candidates += get_embedding_candidates(user['embedding'], k * 3)

    # Always include some new items for exploration
    candidates += get_new_item_candidates(k)

    # Deduplicate and rank
    unique_candidates = list(set(candidates) - set(user.get('shown_items', [])))
    features = [extract_ranking_features(user, items[c], context)
                for c in unique_candidates]
    scores = ranker.predict(features)

    top_k_indices = np.argsort(scores)[-k:][::-1]
    return [items[unique_candidates[i]] for i in top_k_indices]

The Feedback Loop

Recommendations without a feedback loop do not improve. Log and train on:

Event	Signal strength	How to use
Impression	Weak negative (not clicked)	Negative training example
Click	Moderate positive	Positive training example
Dwell time > 30s	Stronger positive	Weighted positive
Purchase / conversion	Strong positive	High-weight positive
Explicit dislike / hide	Negative	Filter from future recs

pythonCopy
def create_training_examples_from_logs(event_log_df: pd.DataFrame) -> pd.DataFrame:
    """
    Convert event logs to (user, item, label) training set.
    """
    # Merge all events for each (user, item, session) triplet
    grouped = event_log_df.groupby(['user_id', 'item_id', 'session_id']).agg({
        'impressed': 'max',
        'clicked': 'max',
        'dwell_seconds': 'sum',
        'purchased': 'max',
        'hidden': 'max'
    }).reset_index()

    # Build graded relevance label
    def get_label(row):
        if row['hidden'] == 1:
            return -1  # explicit negative
        if row['purchased'] == 1:
            return 3
        if row['dwell_seconds'] > 30:
            return 2
        if row['clicked'] == 1:
            return 1
        if row['impressed'] == 1:
            return 0  # shown but not clicked
        return 0

    grouped['label'] = grouped.apply(get_label, axis=1)
    return grouped

Serving Architecture

User request (< 100ms budget)
    │
    ├─ Check Redis cache for this user's candidates (5ms)
    │
    ├─ If cache miss:
    │   ├─ Fetch user embedding from feature store (10ms)
    │   ├─ ANN search against item index (20ms)
    │   └─ Write to Redis cache with 5-minute TTL
    │
    ├─ Filter already-seen items (5ms)
    ├─ Extract ranking features (10ms)
    ├─ Score with ranker model (20ms)
    ├─ Apply business rules: dedup, diversity, freshness boost (5ms)
    └─ Return top K results + log impression (5ms)

Total: ~80ms with warm cache, ~120ms on cache miss - within typical latency budgets.

The item index (FAISS/pgvector) is updated on a schedule: new items added hourly, full re-index nightly. User embeddings are refreshed after significant activity events (purchases, long sessions).

Common Mistakes

Serving fresh embeddings for all users on every request. Computing user embeddings in real time on every inference request introduces a dependency on the full feature pipeline at serving time, dramatically increasing latency and infrastructure complexity. For most users, the embedding changes slowly enough that pre-computing it every few minutes and serving from a cache provides nearly identical relevance at a fraction of the serving cost.

Ignoring feedback loop effects in real-time systems. A real-time system that immediately incorporates user clicks into its ranking creates a feedback loop: what is recommended gets clicked, which reinforces recommending it further. Without explicit diversity and exploration mechanisms (epsilon-greedy, Thompson sampling, or scheduled content injection), the system converges on a small set of items and the long tail starves of impressions.

Treating offline NDCG as a proxy for real-time business metrics without an A/B test. Offline metrics measure historical relevance; real-time systems affect what users do next, which changes the data distribution the next model version trains on. The only reliable way to know whether an improved offline metric translates to improved real-time business performance is an A/B test. Never deploy a real-time recommendation change to 100% of traffic without a holdback group.

What to Practice Next

• Sketch a complete data flow diagram for a real-time recommendation API: include user event ingestion, feature freshness guarantees, candidate retrieval, ranking, and the feedback path back to the training pipeline.
• Design a cache invalidation policy for pre-computed user embeddings: specify under what conditions the cache should be invalidated (time-based, event-triggered, or drift-triggered) and how stale embeddings are handled during the refresh window.
• Define the A/B test you would run before promoting a new real-time recommendation model to 100% of traffic: primary metric, guardrail metrics, minimum detectable effect, and required sample size.