Real-Time Recommendation System
Embeddings + ANN + Redis | Sub-Millisecond Serving
What You'll Learn
- 1Split a recommender into offline embedding work and an online path with a measurable cost model
- 2Size an HNSW index in memory and choose M and ef_search from a recall target
- 3Apply Little's Law to derive the concurrency a target QPS actually needs
- 4Handle cold-start users and power A/B tests to detect lifts that are real
Project Overview
Problem Statement
E-commerce and content platforms need to deliver personalized recommendations in real-time. Traditional collaborative filtering is too slow for online serving. This system uses pre-computed embeddings with approximate nearest neighbor (ANN) search in Redis to serve recommendations in under 1ms.
Objectives
- Build user and item embedding models from interaction data
- Implement ANN search using HNSW indexes in Redis
- Create a real-time feature store for user context
- Deploy with A/B testing infrastructure
- Achieve sub-millisecond recommendation latency at 10K QPS
| Component | Technology |
|---|---|
| Embedding Model | Sentence Transformers + Custom |
| Vector Store | Redis + HNSW |
| Feature Store | Redis + Feature Store |
| API Framework | FastAPI + gRPC |
| A/B Testing | Optimizely / LaunchDarkly |
| Monitoring | Prometheus + Grafana |
| Experiment Tracking | MLflow |
Design Rationale
The system separates what can be pre-computed from what must happen per request. Item embeddings are refreshed by a batch job, user context lives in Redis with a 30 day TTL, and the online path does exactly three things: fetch a vector, run ANN search, then rank. That split is what makes sub-millisecond median latency possible, and the cost model shows why:
At a 90 percent cache hit rate with ms for a Redis get, ms for a Redis Stack HNSW query over 10 million items, and ms, expected latency is 0.9 x 0.3 plus 0.1 x 0.9 plus 0.4, which is 0.76 ms. Every 10 points of hit rate is worth about 0.06 ms, so cache tuning only pays after the index itself is healthy.
Throughput follows Little's Law, which ties concurrency to arrival rate and latency:
Serving 10,000 queries per second with a mean response time of 5 ms requires only requests in flight. Two FastAPI replicas with a pool of 32 connections each cover it, which is why the real bottleneck is usually a slow embedding lookup rather than process count.
HNSW trades memory for recall. A 384 dimension float32 vector costs 1,536 bytes, and the graph adds roughly bytes of links per node:
With that is 16.64 GB for 10 million items; raising M to 32 adds about 128 bytes per node and pushes the index past 18 GB for a couple of points of recall. Measure the trade instead of guessing:
A weekly brute-force sample of 1,000 queries gives ground truth cheaply, and it is the only way to know whether ef_search still meets the 0.95 target.
Architecture Diagram
Step-by-Step Implementation
Step 1: Environment Setup
mkdir rec-system && cd rec-system
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn redis[async] sentence-transformers
pip install numpy faiss-cpu torch pydantic
pip install mlflow kafka-python asyncpg
pip install prometheus-client grpcio
Step 2: Embedding Model Training
Train a custom embedding model using user interaction data with contrastive learning.
# src/embeddings/train_embeddings.py
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from sentence_transformers import SentenceTransformer, InputExample, losses
from sentence_transformers.evaluation import EmbeddingSimilarityEvaluator
import numpy as np
from typing import List, Tuple, Dict
import mlflow
class InteractionDataset(Dataset):
def __init__(self, interactions: List[Dict]):
self.pairs = []
for inter in interactions:
# Positive: items the user interacted with
# Negative: random items the user did not interact with
self.pairs.append(InputExample(
texts=[inter["user_profile"], inter["item_description"]],
label=float(inter["engagement_score"])
))
def __len__(self):
return len(self.pairs)
def __getitem__(self, idx):
return self.pairs[idx]
class EmbeddingTrainer:
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
self.model = SentenceTransformer(model_name)
def train(
self,
train_data: List[Dict],
eval_data: List[Dict],
epochs: int = 10,
batch_size: int = 64,
learning_rate: float = 2e-5,
):
train_dataset = InteractionDataset(train_data)
train_dataloader = DataLoader(train_dataset, shuffle=True, batch_size=batch_size)
# Cosine similarity loss
train_loss = losses.CosineSimilarityLoss(self.model)
# Evaluator
eval_examples = [InputExample(
texts=[e["user_profile"], e["item_description"]],
label=float(e["engagement_score"])
) for e in eval_data]
evaluator = EmbeddingSimilarityEvaluator.from_input_examples(
eval_examples, name="eval"
)
mlflow.set_experiment("embedding-training")
with mlflow.start_run():
mlflow.log_params({
"model_name": "all-MiniLM-L6-v2",
"epochs": epochs,
"batch_size": batch_size,
"learning_rate": learning_rate,
})
self.model.fit(
train_objectives=[(train_dataloader, train_loss)],
evaluator=evaluator,
epochs=epochs,
warmup_steps=100,
output_path="./trained_embeddings",
show_progress_bar=True,
)
def encode(self, texts: List[str], batch_size: int = 256) -> np.ndarray:
return self.model.encode(texts, batch_size=batch_size, show_progress_bar=True)
Step 3: Redis ANN Index Setup
Build and manage HNSW indexes in Redis for fast approximate nearest neighbor search.
# src/storage/redis_index.py
import redis.asyncio as redis
import numpy as np
from typing import List, Tuple, Optional
import json
import logging
logger = logging.getLogger(__name__)
class RedisVectorIndex:
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url, decode_responses=False)
async def create_index(
self, index_name: str, dimension: int = 384,
ef_construction: int = 200, m: int = 16
):
try:
await self.redis.execute_command(
"FT.CREATE", index_name,
"ON", "HASH",
"PREFIX", "1", f"{index_name}:",
"SCHEMA",
"vector", "VECTOR", "HNSW", "6",
"TYPE", "FLOAT32",
"DIM", str(dimension),
"DISTANCE_METRIC", "COSINE",
"EF_CONSTRUCTION", str(ef_construction),
"M", str(m),
)
logger.info(f"Created index {index_name} with dim={dimension}")
except redis.ResponseError as e:
if "Index already exists" not in str(e):
raise
async def upsert_embeddings(
self, index_name: str,
ids: List[str], embeddings: np.ndarray,
metadata: Optional[List[dict]] = None
):
pipe = self.redis.pipeline()
for i, (item_id, embedding) in enumerate(zip(ids, embeddings)):
key = f"{index_name}:{item_id}"
data = {"vector": embedding.astype(np.float32).tobytes()}
if metadata and i < len(metadata):
data["meta"] = json.dumps(metadata[i])
pipe.hset(key, mapping=data)
await pipe.execute()
logger.info(f"Upserted {len(ids)} embeddings to {index_name}")
async def search(
self, index_name: str, query_vector: np.ndarray,
top_k: int = 10,
) -> List[Tuple[str, float]]:
query = f"*=>[KNN {top_k} @vector $query_vec AS score]"
result = await self.redis.execute_command(
"FT.SEARCH", index_name, query,
"PARAMS", "2", "query_vec", query_vector.astype(np.float32).tobytes(),
"DIALECT", "2",
"SORTBY", "score", "ASC",
)
matches = []
for i in range(2, len(result), 2):
key = result[i].decode().split(":")[-1]
score = float(result[i+1][result[i+1].index(b"score") + 6])
matches.append((key, score))
return matches
Step 4: Real-Time Serving API
Build a high-performance FastAPI service for real-time recommendation serving.
# src/api/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
from contextlib import asynccontextmanager
import numpy as np
import time
from src.storage.redis_index import RedisVectorIndex
from src.embeddings.train_embeddings import EmbeddingTrainer
from src.features.user_features import UserFeatureStore
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = RedisVectorIndex()
app.state.embedder = EmbeddingTrainer("./trained_embeddings")
app.state.features = UserFeatureStore()
await app.state.redis.create_index("items", dimension=384)
yield
await app.state.redis.redis.close()
app = FastAPI(title="Rec System API", version="1.0.0", lifespan=lifespan)
class RecRequest(BaseModel):
user_id: str
context: Optional[dict] = None
num_results: int = Field(10, ge=1, le=100)
exclude_items: Optional[List[str]] = None
class RecItem(BaseModel):
item_id: str
score: float
metadata: Optional[dict] = None
class RecResponse(BaseModel):
recommendations: List[RecItem]
latency_ms: float
model_version: str
@app.post("/api/v1/recommend", response_model=RecResponse)
async def recommend(request: RecRequest):
start = time.time()
# Get user embedding from feature store
user_embedding = await app.state.features.get_user_embedding(
request.user_id, request.context
)
# ANN search in Redis
results = await app.state.redis.search(
"items", user_embedding, top_k=request.num_results * 2
)
# Filter excluded items
if request.exclude_items:
results = [r for r in results if r[0] not in request.exclude_items]
# Take top_k results
results = results[:request.num_results]
latency = (time.time() - start) * 1000
return RecResponse(
recommendations=[RecItem(item_id=r[0], score=r[1]) for r in results],
latency_ms=latency,
model_version="v1.2.0",
)
Step 5: User Feature Store
Build a real-time feature store for computing and caching user embeddings from recent interactions.
# src/features/user_features.py
import redis.asyncio as redis
import numpy as np
from typing import Optional, Dict, List
import json
from datetime import datetime, timedelta
class UserFeatureStore:
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url, decode_responses=True)
self.TTL = timedelta(days=30)
async def get_user_embedding(
self, user_id: str, context: Optional[Dict] = None
) -> np.ndarray:
# Try cache first
cached = await self.redis.get(f"user_emb:{user_id}")
if cached:
return np.frombuffer(cached, dtype=np.float32)
# Compute from recent interactions
interactions = await self.get_recent_interactions(user_id)
if not interactions:
return np.random.randn(384).astype(np.float32) # Cold start
# Weighted average of item embeddings
weights = np.array([i["weight"] for i in interactions])
weights = weights / weights.sum()
embeddings = np.array([np.frombuffer(i["embedding"], dtype=np.float32) for i in interactions])
user_emb = np.average(embeddings, axis=0, weights=weights)
# Cache result
await self.redis.setex(
f"user_emb:{user_id}",
int(self.TTL.total_seconds()),
user_emb.astype(np.float32).tobytes(),
)
return user_emb
async def get_recent_interactions(
self, user_id: str, limit: int = 50
) -> List[Dict]:
return await self.redis.lrange(f"user_interactions:{user_id}", 0, limit - 1)
async def record_interaction(
self, user_id: str, item_id: str,
interaction_type: str, weight: float
):
entry = json.dumps({"item_id": item_id, "type": interaction_type, "weight": weight})
await self.redis.lpush(f"user_interactions:{user_id}", entry)
await self.redis.ltrim(f"user_interactions:{user_id}", 0, 999)
# Invalidate cached embedding
await self.redis.delete(f"user_emb:{user_id}")
Step 6: Docker Deployment
# docker-compose.yml
version: "3.8"
services:
rec-api:
build: .
ports:
- "8000:8000"
environment:
- REDIS_URL=redis://redis:6379
- MODEL_PATH=/models/embeddings
depends_on:
- redis
deploy:
replicas: 3
redis:
image: redis/redis-stack-server:latest
ports:
- "6379:6379"
volumes:
- redis-data:/data
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
volumes:
redis-data:
âšī¸
Pre-compute embeddings in batch and upsert them to Redis. Computing embeddings at request time adds unnecessary latency. Use a background job to refresh embeddings periodically.
đĄ
For cold-start users, combine content-based features (demographics, device info) with popularity-based fallbacks until enough interaction data is collected.
Performance Metrics
| Metric | Target | Description |
|---|---|---|
| ANN Latency | < 1ms | Redis HNSW search |
| End-to-End Latency | < 10ms | Full pipeline |
| Throughput | > 10K QPS | Per API instance |
| Recall@10 | > 0.95 | vs exact NN search |
| NDCG@10 | > 0.80 | Ranking quality |
| Cache Hit Rate | > 90% | User embedding cache |
What Breaks in Production
Recommendation services fail in predictable ways. Cold start dominates the first weeks: a brand new user has no interactions, and returning a random 384 dimension vector produces unrelated results. Fall back to popularity weighted by device and region until roughly five interactions exist, then switch to the learned embedding.
The second failure is index skew. HNSW recall decays as the index grows if ef_search stays fixed, so a setting tuned for one million items at recall@10 of 0.97 can sit near 0.90 at ten million. Track the gap directly:
Raise ef_search whenever the gap exceeds 0.02 for two consecutive weekly checks, because a silent recall slide shows up first as a drop in click-through rate that nobody links back to the index.
Third, invalidation races poison the cache. record_interaction deletes the cached user embedding, but a concurrent reader can repopulate it from pre-write data. Delete only after the write commits and version the cache key with the interaction count so stale values cannot win.
Finally, under-powered experiments. Detecting a 0.3 point lift against a 30 percent click-through baseline at 95 percent confidence and 80 percent power needs about 366,000 impressions per arm, so a two day test on 5 percent of traffic will read noise as a win and ship a regression.
Interview Talking Points
- ANN vs Exact NN: HNSW provides 1000x speedup over brute-force with minimal recall loss (typically 95%+).
- Two-Tower Architecture: Separate user and item towers enable pre-computing item embeddings and real-time user embedding computation.
- Cold Start Strategy: Content-based features plus popularity fallback for new users, transitioning to collaborative filtering as data accumulates.
- Feature Store Design: Redis provides sub-millisecond reads for real-time features while maintaining high throughput.
- A/B Testing: Infrastructure for comparing model versions and ranking strategies with statistical significance.
- Scaling: Horizontal scaling through Redis clustering and API replicas behind a load balancer.
â ī¸
Monitor ANN index quality over time. As the index grows, HNSW parameters may need tuning to maintain recall targets.
âšī¸
This project demonstrates a complete real-time recommendation pipeline. For production deployment, add model retraining automation and drift detection.
Common Mistakes
Encoding item embeddings inside the request handler
Fix: Pre-compute item vectors in a batch job and keep the online path to fetch, search, and rank.
Tuning HNSW once and never rechecking recall
Fix: Sample brute-force ground truth weekly and raise ef_search when recall@10 falls short of target.
Calling a short A/B test a win
Fix: Size the test first: a 0.3 point lift on a 30 percent CTR needs about 366,000 impressions per arm.
Best Practices
- Return the model version in every response so experiments can be attributed to a model
- Keep a popularity fallback path running in production every day so it works when you need it
- Separate TTLs by data type: 30 days for user embeddings, one hour for session features
Key Takeaways
- Expected latency is h times T_cache plus (1 minus h) times T_ann plus T_rank; cache hits buy about 0.06 ms per 10 points
- Little's Law says 10K QPS at 5 ms mean latency needs only 50 requests in flight
- A 384 dimension float32 vector with M = 16 costs 1,664 bytes including graph links, so 10M items need 16.64 GB
- ANN recall must be measured against exact search on a sample, never assumed from the parameter values
Frequently Asked Questions
Why Redis instead of a dedicated vector database?
Redis already holds the session and feature data this service needs, so one system serves the cache, the feature store, and the HNSW index with a single network hop. A dedicated engine earns its keep past roughly 100 million vectors or when you need filtered search across many namespaces, neither of which a 10 million item catalog requires.
How many interactions before a learned user embedding beats popularity?
Usually five to ten weighted events. Below that the average of item vectors has high variance and popularity with contextual filters wins; above it the learned vector starts picking up niche intent that popularity ranking cannot express. Measure it as a switch point in an offline replay rather than a fixed rule.
What should happen when an item is deleted or goes out of stock?
Remove it from the index immediately and let the feature store mark it unavailable, because filtering after retrieval silently shrinks your top-k and can leave the API returning fewer than requested results. Fetch extra candidates so post-filters still fill the list.
Interview Questions
Q1: Walk me through the latency budget of a recommendation request.
Feature fetch about 0.3 ms on a cache hit, HNSW search about 0.9 ms for 10 million items, ranking about 0.4 ms, giving 0.76 ms expected at a 90 percent hit rate. Network and serialization add roughly 1 ms, so p50 stays under 2 ms.
Q2: How do you measure ANN quality in production?
Take a weekly sample of 1,000 live queries, run them against exact brute-force search on the same data, and compute recall@10 = intersection divided by 10. Alert when the gap to the 0.95 target exceeds 0.02 twice in a row.
Q3: How do you handle new users and new items?
New users fall back to popularity filtered by device and region until five interactions accumulate. New items enter the index on the next batch upsert and get a content-based vector first, then a collaborative one once enough interactions exist.