Introduction to Recommendation Engines with Redis Caching
A recommendation engine is a system that filters and ranks content or products based on user preferences, behavior patterns, and contextual signals. When you see "Customers who bought this also bought..." on Amazon, or personalized playlists on Spotify, you're experiencing recommendation engines in action. At their core, these systems process large volumes of dataâuser interactions, item metadata, similarity matricesâand must return results with minimal latency to keep users engaged.
Redis enters the picture as a high-performance, in-memory data store that dramatically accelerates the recommendation pipeline. By caching pre-computed recommendations, similarity scores, user profiles, and frequently accessed item features, Redis transforms what could be a slow, database-intensive computation into a sub-millisecond lookup. This tutorial walks you through designing a production-grade recommendation engine backed by Redis caching, from data modeling to deployment considerations.
Why Redis Caching Matters for Recommendation Engines
đ Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →Recommendation engines face three fundamental performance challenges that Redis solves elegantly:
1. Latency Sensitivity
Users expect recommendations to appear instantlyâoften within 100-200 milliseconds. Querying a primary database, joining tables, and computing similarity scores on the fly simply cannot meet this bar at scale. Redis operates entirely in memory, delivering sub-millisecond response times for cached recommendation payloads.
2. High Read-to-Write Ratio
Recommendation data is typically computed periodically (e.g., nightly batch jobs) and read millions of times throughout the day. Redis excels at read-heavy workloads, handling 100,000+ operations per second on modest hardware. This allows a single Redis instance to serve thousands of concurrent recommendation requests.
3. Expensive Computation Offloading
Collaborative filtering, matrix factorization, and deep learning inference are computationally expensive. Recomputing them for every request is wasteful. Redis stores the pre-computed outputâtop-K item lists, user-item affinity scoresâso the heavy lifting happens once, and serving happens millions of times at near-zero cost.
Architecture Overview
A well-designed recommendation engine with Redis caching typically follows a layered architecture:
- Batch / Offline Layer: Computes recommendations periodically using collaborative filtering, content-based models, or hybrid approaches. Results are pushed to Redis.
- Real-time / Online Layer: Handles incoming user requests, fetches cached recommendations from Redis, applies lightweight personalization or re-ranking, and returns the final list.
- Redis Cache Layer: Stores pre-computed user-to-item mappings, item similarity vectors, trending lists, and user session features. Acts as the fast read-path between the online service and the offline computation.
Data Modeling in Redis
Choosing the right Redis data structures is critical. Here are the most effective patterns for recommendation engine workloads:
User Recommendation Lists (Sorted Sets)
Store pre-computed top-N recommendations for each user as a sorted set, where each member is an item ID and the score represents the recommendation confidence or predicted rating. Sorted sets allow you to retrieve the top items in order and paginate efficiently.
ZADD user:recs:user123 0.95 "item:456" 0.89 "item:789" 0.82 "item:101" 0.77 "item:202"
ZADD user:recs:user123 0.74 "item:303" 0.68 "item:404" 0.63 "item:505" 0.58 "item:606"To retrieve the top 5 recommendations for a user with their scores:
ZREVRANGE user:recs:user123 0 4 WITHSCORESItem-to-Item Similarity (Sorted Sets or Hashes)
For item-based collaborative filtering, store the similarity vectors for each item. A sorted set per item keyed on the item ID works beautifully, with similar items as members and similarity coefficients as scores.
ZADD item:sim:item456 0.92 "item:789" 0.87 "item:101" 0.84 "item:303" 0.81 "item:202"
ZADD item:sim:item456 0.76 "item:505" 0.71 "item:606" 0.65 "item:707" 0.60 "item:808"When a user interacts with item:456, you can instantly fetch similar items to generate "because you viewed..." recommendations:
ZREVRANGE item:sim:item456 0 9 WITHSCORESGlobal Trending or Popular Items (Sorted Sets)
Maintain a sorted set for global trending recommendations, scored by a composite metric like a weighted combination of recent views, purchases, and ratings. This serves as a fallback for cold-start users with no history.
ZADD trending:global 15230 "item:456" 12870 "item:101" 11950 "item:789"
ZADD trending:global 10400 "item:303" 9800 "item:202" 8700 "item:606"User Profile Features (Hashes)
Cache lightweight user feature vectorsâcategory preferences, brand affinities, average ratingâas Redis hashes for fast real-time personalization without querying the primary database.
HSET user:profile:user123
favorite_category "electronics"
avg_rating "4.2"
brand_affinity "apple,sony,samsung"
last_active "2025-01-15T14:32:00Z"
total_interactions "47"Session-Based Temporary Recommendations (Lists or Sorted Sets with TTL)
For anonymous users or short-lived browsing sessions, store ephemeral recommendation lists with an expiration time. Redis lists work well for ordered collections, and setting a TTL ensures automatic cleanup.
RPUSH session:recs:abc-xyz-123 "item:456" "item:789" "item:101" "item:303" "item:202"
EXPIRE session:recs:abc-xyz-123 1800Building the Recommendation Pipeline: Practical Code Examples
Let's walk through a complete implementation in Python using the redis-py library. We'll build an item-based collaborative filtering recommender with Redis caching at every stage.
Step 1: Setting Up Redis Connection and Helper Functions
import redis
import json
from typing import List, Dict, Optional
from datetime import timedelta
# Establish connection pool for thread-safety
redis_pool = redis.ConnectionPool(
host='localhost',
port=6379,
db=0,
decode_responses=True,
max_connections=20
)
def get_redis_client():
return redis.Redis(connection_pool=redis_pool)
def cache_user_recommendations(user_id: str, recommendations: List[Dict], ttl_minutes: int = 360):
"""
Store pre-computed recommendations for a user as a sorted set.
Each recommendation is a dict with 'item_id' and 'score' keys.
The sorted set allows O(log N) retrieval of top-K items.
"""
client = get_redis_client()
key = f"user:recs:{user_id}"
# Pipeline for atomic bulk insertion
pipe = client.pipeline()
for rec in recommendations:
pipe.zadd(key, {rec['item_id']: rec['score']})
pipe.expire(key, timedelta(minutes=ttl_minutes))
pipe.execute()
print(f"Cached {len(recommendations)} recommendations for user {user_id}")
def get_cached_recommendations(user_id: str, top_n: int = 10) -> List[tuple]:
"""
Retrieve cached recommendations from Redis.
Returns list of (item_id, score) tuples ordered by score descending.
"""
client = get_redis_client()
key = f"user:recs:{user_id}"
results = client.zrevrange(key, 0, top_n - 1, withscores=True)
return results
Step 2: Computing and Caching Item Similarity Matrix
The item similarity matrix is the backbone of item-based collaborative filtering. We compute cosine similarity between item vectors derived from user interaction data, then cache the top-K similar items for each item in Redis.
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from collections import defaultdict
def build_item_user_matrix(interactions: List[Dict]) -> Dict[str, Dict[str, int]]:
"""
Convert raw interactions into an item-to-user affinity matrix.
interactions: list of {'user_id': str, 'item_id': str, 'rating': int}
Returns: dict of item_id -> {user_id: rating}
"""
item_users = defaultdict(dict)
for interaction in interactions:
item_users[interaction['item_id']][interaction['user_id']] = interaction['rating']
return dict(item_users)
def compute_and_cache_item_similarity(
item_user_matrix: Dict[str, Dict[str, int]],
top_k: int = 50,
ttl_hours: int = 24
):
"""
Compute cosine similarity between all items and cache top-K similar items per item.
"""
client = get_redis_client()
# Get sorted item IDs for consistent vector ordering
all_items = sorted(item_user_matrix.keys())
all_users_set = set()
for users in item_user_matrix.values():
all_users_set.update(users.keys())
all_users = sorted(all_users_set)
# Build sparse vectors
user_index = {u: i for i, u in enumerate(all_users)}
item_vectors = []
for item_id in all_items:
vec = np.zeros(len(all_users))
for user, rating in item_user_matrix[item_id].items():
vec[user_index[user]] = rating
item_vectors.append(vec)
# Compute pairwise cosine similarity
sim_matrix = cosine_similarity(np.array(item_vectors))
# Cache in Redis using pipeline for efficiency
pipe = client.pipeline()
for i, item_id in enumerate(all_items):
key = f"item:sim:{item_id}"
# Get indices of top-K most similar items (excluding self)
sim_scores = sim_matrix[i]
# Sort indices by similarity descending, skip self (index i)
top_indices = np.argsort(sim_scores)[::-1][1:top_k + 1]
mapping = {}
for idx in top_indices:
if sim_scores[idx] > 0: # Only cache positive similarities
mapping[all_items[idx]] = float(sim_scores[idx])
if mapping:
pipe.zadd(key, mapping)
pipe.expire(key, timedelta(hours=ttl_hours))
pipe.execute()
print(f"Cached similarity vectors for {len(all_items)} items")
# Example: Sample interaction data
sample_interactions = [
{'user_id': 'user1', 'item_id': 'itemA', 'rating': 5},
{'user_id': 'user1', 'item_id': 'itemB', 'rating': 4},
{'user_id': 'user1', 'item_id': 'itemC', 'rating': 3},
{'user_id': 'user2', 'item_id': 'itemA', 'rating': 4},
{'user_id': 'user2', 'item_id': 'itemB', 'rating': 5},
{'user_id': 'user3', 'item_id': 'itemB', 'rating': 4},
{'user_id': 'user3', 'item_id': 'itemC', 'rating': 5},
]
item_matrix = build_item_user_matrix(sample_interactions)
compute_and_cache_item_similarity(item_matrix, top_k=10)
Step 3: Real-Time Recommendation Serving
When a user request arrives, we fetch the user's recent items from their session or profile, retrieve similar items from Redis for each, aggregate scores, and return the final ranked list. This entire path hits Redis and never touches the primary database.
def get_recent_user_items(user_id: str, limit: int = 5) -> List[str]:
"""
Retrieve the user's most recently interacted items.
In production, this could come from a Redis list tracking recent views.
"""
client = get_redis_client()
key = f"user:recent:{user_id}"
return client.lrange(key, 0, limit - 1)
def generate_recommendations_from_cache(user_id: str, top_n: int = 10) -> List[Dict]:
"""
Generate real-time recommendations using cached item similarities.
Strategy: For each item the user recently interacted with, fetch similar items
from Redis, aggregate scores, and return top-N.
"""
client = get_redis_client()
# Step 1: Check for pre-computed user recommendations first
precomputed = get_cached_recommendations(user_id, top_n)
if precomputed and len(precomputed) >= top_n:
return [{'item_id': item_id, 'score': score, 'source': 'precomputed'}
for item_id, score in precomputed]
# Step 2: Fallback to item-based aggregation using cached similarities
recent_items = get_recent_user_items(user_id, limit=5)
if not recent_items:
# Cold start: return global trending
trending = client.zrevrange('trending:global', 0, top_n - 1, withscores=True)
return [{'item_id': item_id, 'score': score, 'source': 'trending'}
for item_id, score in trending]
# Aggregate scores across similar items
aggregated_scores = defaultdict(float)
pipe = client.pipeline()
# Fetch similar items for all recent items in one pipeline call
for item_id in recent_items:
pipe.zrevrange(f"item:sim:{item_id}", 0, 19, withscores=True)
results = pipe.execute()
# Weighted aggregation: more recent items get higher weight
for idx, similar_items in enumerate(results):
recency_weight = 1.0 / (idx + 1) # First (most recent) gets weight 1.0
for similar_item, similarity_score in similar_items:
if similar_item not in recent_items: # Don't recommend already-seen items
aggregated_scores[similar_item] += similarity_score * recency_weight
# Sort and return top-N
sorted_recs = sorted(aggregated_scores.items(), key=lambda x: x[1], reverse=True)[:top_n]
return [{'item_id': item_id, 'score': round(score, 3), 'source': 'item-based'}
for item_id, score in sorted_recs]
# Example usage
recommendations = generate_recommendations_from_cache('user123', top_n=5)
for rec in recommendations:
print(f"Item: {rec['item_id']}, Score: {rec['score']}, Source: {rec['source']}")
Step 4: Caching User Sessions and Recent Interactions
Track user behavior in real time using Redis lists with size caps and TTLs. This feeds the recommendation engine with fresh signals.
def record_user_interaction(user_id: str, item_id: str, max_history: int = 50):
"""
Record a user-item interaction in Redis.
Maintains a capped list of recent items per user.
"""
client = get_redis_client()
key = f"user:recent:{user_id}"
pipe = client.pipeline()
# Push to front (left) for reverse chronological order
pipe.lpush(key, item_id)
# Trim to max_history items to prevent unbounded growth
pipe.ltrim(key, 0, max_history - 1)
# Set TTL for inactive users (30 days)
pipe.expire(key, timedelta(days=30))
pipe.execute()
def get_user_session_context(user_id: str) -> Dict:
"""
Retrieve cached user profile features for lightweight personalization.
"""
client = get_redis_client()
key = f"user:profile:{user_id}"
profile = client.hgetall(key)
if not profile:
# Load from primary database and cache (cache-aside pattern)
profile = load_user_profile_from_db(user_id) # hypothetical function
if profile:
client.hset(key, mapping=profile)
client.expire(key, timedelta(hours=6))
return profile
def load_user_profile_from_db(user_id: str) -> Dict:
"""
Placeholder for actual database query.
In production, this queries your user database.
"""
# Simulated database response
return {
'favorite_category': 'electronics',
'avg_rating': '4.2',
'brand_affinity': 'apple,sony',
'total_interactions': '47'
}
Step 5: Periodic Batch Refresh Pipeline
The offline computation runs on a schedule (e.g., nightly) and refreshes the cached recommendations. This function simulates that workflow.
def batch_refresh_all_user_recommendations(user_ids: List[str]):
"""
Nightly batch job: recompute recommendations for all active users
and update Redis cache.
"""
client = get_redis_client()
for user_id in user_ids:
# Get user's recent items from Redis
recent_items = client.lrange(f"user:recent:{user_id}", 0, 49)
if not recent_items:
continue
# Aggregate similar items (same logic as online path, but exhaustive)
pipe = client.pipeline()
for item_id in recent_items[:10]: # Use top 10 recent items
pipe.zrevrange(f"item:sim:{item_id}", 0, 29, withscores=True)
results = pipe.execute()
aggregated = defaultdict(float)
for idx, similar_items in enumerate(results):
weight = 1.0 / (idx + 1)
for sim_item, score in similar_items:
if sim_item not in recent_items:
aggregated[sim_item] += score * weight
# Store top 100 recommendations per user
top_100 = sorted(aggregated.items(), key=lambda x: x[1], reverse=True)[:100]
mapping = {item: score for item, score in top_100}
key = f"user:recs:{user_id}"
pipe = client.pipeline()
# Atomic replace: delete old, insert new
pipe.delete(key)
if mapping:
pipe.zadd(key, mapping)
pipe.expire(key, timedelta(hours=25)) # Slightly longer than refresh interval
pipe.execute()
print(f"Batch refreshed recommendations for {len(user_ids)} users")
Caching Strategies and Patterns
Cache-Aside (Lazy Loading)
The application checks Redis first; on a miss, it computes or queries the primary database, then writes the result back to Redis. This pattern is ideal for recommendation engines because it keeps the cache fresh only for actively requested data and avoids wasting memory on inactive users.
def get_recommendations_cache_aside(user_id: str, top_n: int = 10) -> List[Dict]:
client = get_redis_client()
key = f"user:recs:{user_id}"
# Check cache
cached = client.zrevrange(key, 0, top_n - 1, withscores=True)
if cached:
return [{'item_id': item_id, 'score': score} for item_id, score in cached]
# Cache miss: compute recommendations
recommendations = generate_recommendations_from_cache(user_id, top_n)
# Write-through to cache
pipe = client.pipeline()
mapping = {rec['item_id']: rec['score'] for rec in recommendations}
pipe.zadd(key, mapping)
pipe.expire(key, timedelta(hours=6))
pipe.execute()
return recommendations
Write-Through and Write-Behind
For pre-computed batch recommendations, a write-through approachâwhere the batch job writes directly to Redisâensures the cache is always populated. Write-behind adds an asynchronous queue to decouple computation from cache updates, useful when recommendation computation is slow but cache writes must be fast.
TTL and Eviction Policies
Always set TTLs on recommendation keys to prevent stale data and memory bloat. A reasonable TTL aligns with your batch refresh interval plus a buffer (e.g., 25 hours for daily refreshes). Configure Redis eviction policy to volatile-lru or allkeys-lru so that when memory fills, the least recently used recommendation sets are evicted gracefully.
Handling Cold Start and Edge Cases
New Users (User Cold Start)
For users with no interaction history, fall back to global trending lists cached in Redis. You can also maintain category-specific trending lists for slightly more personalized cold-start recommendations.
def get_trending_by_category(category: str, top_n: int = 10) -> List[str]:
client = get_redis_client()
key = f"trending:category:{category}"
results = client.zrevrange(key, 0, top_n - 1)
# If category trending is empty, fall back to global
if not results:
results = client.zrevrange('trending:global', 0, top_n - 1)
return results
New Items (Item Cold Start)
New items have no similarity scores. Implement a "new items" sorted set that gets boosted in recommendations until enough interaction data accumulates. Use a time-decay score to naturally phase out items as they age.
def add_new_item_with_boost(item_id: str, category: str):
client = get_redis_client()
current_time = time.time()
# Add to new items set with time-based score (newer = higher score)
client.zadd('items:new_arrivals', {item_id: current_time})
# Also add to category trending with initial boost
client.zadd(f"trending:category:{category}", {item_id: 100.0})
# Set TTL to clean up after 30 days
client.expire(f"trending:category:{category}", timedelta(days=30))
Cache Stampede Prevention
When a popular recommendation key expires, multiple concurrent requests might simultaneously hit the database to recompute it. Use Redis locks or probabilistic early recomputation to prevent this stampede.
import hashlib
import time
def get_recommendations_with_lock(user_id: str, top_n: int = 10) -> List[Dict]:
client = get_redis_client()
key = f"user:recs:{user_id}"
lock_key = f"lock:recs:{user_id}"
cached = client.zrevrange(key, 0, top_n - 1, withscores=True)
if cached:
return [{'item_id': item_id, 'score': score} for item_id, score in cached]
# Try to acquire distributed lock with 5-second timeout
lock_value = str(time.time())
acquired = client.set(lock_key, lock_value, nx=True, ex=5)
if acquired:
try:
# Double-check cache after acquiring lock
cached = client.zrevrange(key, 0, top_n - 1, withscores=True)
if cached:
return [{'item_id': item_id, 'score': score} for item_id, score in cached]
# Compute and cache
recommendations = generate_recommendations_from_cache(user_id, top_n)
pipe = client.pipeline()
mapping = {rec['item_id']: rec['score'] for rec in recommendations}
pipe.zadd(key, mapping)
pipe.expire(key, timedelta(hours=6))
pipe.execute()
return recommendations
finally:
# Release lock only if we still own it
if client.get(lock_key) == lock_value:
client.delete(lock_key)
else:
# Wait and retry
time.sleep(0.1)
return get_recommendations_with_lock(user_id, top_n)
Best Practices for Production
- Use Pipelines Aggressively: Group multiple Redis commands into pipelines to reduce network round-trips. In recommendation serving, you often fetch dozens of similarity sets; batching them into one pipeline call can reduce latency from 20ms to 2ms.
- Size Limits Are Essential: Always cap sorted sets and lists (e.g., top 100 recommendations per user, top 50 similar items). Use
ZREMRANGEBYRANKorLTRIMto enforce caps after writes. - Monitor Memory and Hit Rates: Track Redis memory usage, eviction rates, and cache hit ratios. A hit ratio below 90% warrants investigationâperhaps TTLs are too short or the batch refresh pipeline is failing.
- Separate Redis Instances by Workload: Use one Redis instance (or cluster) for recommendation caches and another for session/rate-limiting data. This prevents heavy recommendation cache evictions from impacting unrelated services.
- Compress Large Payloads: If storing full recommendation payloads (item metadata along with IDs), serialize with MessagePack or protobuf and store as binary strings to reduce memory footprint.
- Graceful Degradation: Always wrap Redis calls in try/except blocks with fallbacks. If Redis is unreachable, fall back to a minimal in-memory LRU cache or query the database directly with stricter limits.
- Warm Up After Deployments: After a Redis restart or deployment, pre-populate caches for your most active users before opening traffic. A warm-up script that iterates top users and primes their recommendation keys prevents a thundering herd.
- Use Redis Cluster for Scale: When your item catalog and user base grow beyond a single instance's memory, shard across a Redis Cluster. Hash user IDs to consistent slots so recommendations for a user always reside on the same node.
Performance Tuning and Monitoring
Measure every stage of your recommendation pipeline. Key metrics to track:
- Cache Hit Latency: p50/p99 of Redis
ZREVRANGEcalls for recommendation keys - Cache Miss Latency: Time to compute recommendations on miss, including pipeline calls
- Batch Refresh Duration: Total time for nightly recomputation and Redis writes
- Redis Memory Fragmentation Ratio: Keep near 1.0; high fragmentation indicates inefficient data modeling
- Stale Recommendation Rate: Percentage of users receiving recommendations older than the refresh interval
A practical monitoring snippet using Redis INFO and latency tracking:
def get_cache_health_metrics() -> Dict:
client = get_redis_client()
info = client.info()
metrics = {
'hit_rate_percent': round(
info['keyspace_hits'] / max(1, info['keyspace_hits'] + info['keyspace_misses']) * 100,
2
),
'memory_used_mb': info['used_memory_human'],
'evicted_keys': info['evicted_keys'],
'connected_clients': info['connected_clients'],
'uptime_days': info['uptime_in_days'],
'keyspace_size': client.dbsize(),
}
# Check specific recommendation key counts
rec_key_pattern = 'user:recs:*'
metrics['cached_user_count'] = sum(1 for _ in client.scan_iter(match=rec_key_pattern, count=1000))
return metrics
# Print health dashboard
health = get_cache_health_metrics()
for metric, value in health.items():
print(f"{metric}: {value}")
Putting It All Together: A Complete Recommendation Service
Below is a production-ready recommendation service class that ties together caching, fallback logic, monitoring, and graceful degradation. Use this as a starting template for your own implementation.
import redis
import time
import logging
from typing import List, Dict, Optional
from datetime import timedelta
from collections import defaultdict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RecommendationService:
"""
A production-grade recommendation engine backed by Redis caching.
Supports item-based collaborative filtering with pre-computed and
real-time fallback paths.
"""
def __init__(self, redis_host: str = 'localhost', redis_port: int = 6379, db: int = 0):
self.pool = redis.ConnectionPool(
host=redis_host,
port=redis_port,
db=db,
decode_responses=True,
max_connections=20,
socket_timeout=0.5, # Fail fast if Redis is slow
socket_connect_timeout=0.3
)
self.client = redis.Redis(connection_pool=self.pool)
self.local_cache = {} # In-memory fallback for critical users
def get_client(self) -> redis.Redis:
return redis.Redis(connection_pool=self.pool)
def recommend(self, user_id: str, top_n: int = 10, context: Optional[Dict] = None) -> List[Dict]:
"""
Main recommendation endpoint.
Returns list of {'item_id': str, 'score': float, 'source': str}
"""
start_time = time.perf_counter()
client = self.get_client()
try:
# 1. Try pre-computed recommendations
key = f"user:recs:{user_id}"
cached = client.zrevrange(key, 0, top_n - 1, withscores=True)
if cached and len(cached) >= top_n:
elapsed = (time.perf_counter() - start_time) * 1000
logger.info(f"Cache hit for {user_id} in {elapsed:.2f}ms")
return [{'item_id': item_id, 'score': score, 'source': 'precomputed'}
for item_id, score in cached]
# 2. Cache miss: generate via item-based aggregation
recent_items = client.lrange(f"user:recent:{user_id}", 0, 9)
if not recent_items:
# Cold start fallback
trending = client.zrevrange('trending:global', 0, top_n - 1, withscores=True)
elapsed = (time.perf_counter() - start_time) * 1000
logger.info(f"Cold start fallback for {user_id} in {elapsed:.2f}ms")
return [{'item_id': item_id, 'score': score, 'source': 'trending'}
for item_id, score in trending]
# Aggregate similar items
pipe = client.pipeline()
for item_id in recent_items[:5]:
pipe.zrevrange(f"item:sim:{item_id}", 0, 19, withscores=True)
results = pipe.execute()
aggregated = defaultdict(float)
for idx, similar_items in enumerate(results):
weight = 1.0 / (idx + 1)
for sim_item, score in similar_items:
if sim_item not in recent_items:
aggregated[sim_item] += score * weight
sorted_recs = sorted(aggregated.items(), key=lambda x: x[1], reverse=True)[:top_n]
# Write back to cache for future requests
if sorted_recs:
pipe = client.pipeline()
mapping = {item: score for item, score in sorted_recs}
pipe.zadd(key, mapping)
pipe.expire(key, timedelta(hours=6))
pipe.execute()
elapsed = (time.perf_counter() - start_time) * 1000
logger.info(f"Computed recommendations for {user_id} in {elapsed:.2f}ms")
return [{'item_id': item_id, 'score': round(score, 3), 'source': 'item-based'}
for item_id, score in sorted_recs]
except (redis.ConnectionError, redis.TimeoutError) as e:
# Graceful degradation: use in-memory fallback
logger.warning(f"Redis unavailable for {user_id}: {e}")
elapsed = (time.perf_counter() - start_time) * 1000
logger.info(f"Fallback recommendations for {user_id} in {elapsed:.2f}ms")
# Return from local cache or global defaults
fallback = self.local_cache.get(user_id, [])
if not fallback:
fallback = [
{'item_id': 'default_item_1', 'score': 1.0, 'source': 'fallback'},
{'item_id': 'default_item_2', 'score': 0.9, 'source': 'fallback'},
]
return fallback[:top_n]
def record_interaction(self, user_id: str, item_id: str):
"""Record user interaction for future recommendations."""
try:
client = self.get_client()
key = f"user:recent:{user_id}"
pipe = client.pipeline()
pipe.lpush(key, item_id)
pipe.ltrim(key, 0, 49)
pipe.expire(key, timedelta(days=30))
pipe.execute()
except redis.RedisError as e:
logger.error(f"Failed to record interaction: {e}")
# Instantiate and use
service = RecommendationService()
service.record_interaction('user123', 'item456')
recs = service.recommend('user123', top_n=5)
for r in recs:
print(f" -> {r['item_id']} (score: {r['score']}, source: {r['source']})")
Conclusion
Designing a recommendation engine with Redis caching transforms a computationally intensive, latency-sensitive system into a fast, scalable, and resilient service. By pre-computing similarity matrices and user-specific recommendation lists offline and storing them in Redis sorted sets, you achieve sub-millisecond recommendation responses even under heavy load. The patterns covered in this tutorialâsorted sets for ranked recommendations