Recommendation Systems with GenAI

Recommendation Approaches

Collaborative Filtering

Collaborative filtering leverages user behavior patterns to find similar users or items, assuming users who agreed in the past will agree in the future.

Content-Based Filtering

Content-based methods recommend items similar to those a user liked before, using item features and user profiles.

Hybrid Approaches

Modern systems combine both approaches using deep learning to capture complex patterns and avoid limitations of individual methods.

Building a Recommendation System

import numpy as np
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer

class HybridRecommender:
    def __init__(self, alpha=0.7):
        self.alpha = alpha
        self.embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
        self.user_profiles = {}
        self.item_embeddings = None
        self.items = []
    
    def fit(self, interactions_df, items_df):
        self.items = items_df["description"].tolist()
        self.item_ids = items_df["id"].tolist()
        
        self.item_embeddings = self.embedding_model.encode(self.items)
        
        self.user_interactions = {}
        for user_id, group in interactions_df.groupby("user_id"):
            self.user_interactions[user_id] = {
                "liked": group[group["rating"] >= 4]["item_id"].tolist(),
                "disliked": group[group["rating"] <= 2]["item_id"].tolist()
            }
    
    def get_user_profile(self, user_id):
        if user_id not in self.user_interactions:
            return np.mean(self.item_embeddings, axis=0)
        
        liked_items = self.user_interactions[user_id]["liked"]
        if not liked_items:
            return np.mean(self.item_embeddings, axis=0)
        
        liked_indices = [
            self.item_ids.index(item_id) 
            for item_id in liked_items 
            if item_id in self.item_ids
        ]
        
        if not liked_indices:
            return np.mean(self.item_embeddings, axis=0)
        
        return np.mean(self.item_embeddings[liked_indices], axis=0)
    
    def recommend(self, user_id, k=5, exclude_interacted=True):
        user_profile = self.get_user_profile(user_id)
        
        similarities = cosine_similarity(
            [user_profile], 
            self.item_embeddings
        )[0]
        
        if exclude_interacted and user_id in self.user_interactions:
            interacted = set(self.user_interactions[user_id]["liked"])
            interacted_idx = [
                self.item_ids.index(item_id)
                for item_id in interacted
                if item_id in self.item_ids
            ]
            similarities[interacted_idx] = -1
        
        top_k_idx = np.argsort(similarities)[-k:][::-1]
        
        return [
            {
                "item_id": self.item_ids[i],
                "score": similarities[i],
                "description": self.items[i][:100]
            }
            for i in top_k_idx
        ]

# Usage example
recommender = HybridRecommender(alpha=0.8)
recommender.fit(interactions_df, items_df)
recommendations = recommender.recommend(user_id="user_123", k=5)

GenAI-Powered Recommendations

import openai
from typing import List, Dict

class GenAIRecommender:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
    
    def generate_recommendations(
        self, 
        user_history: List[Dict], 
        candidate_items: List[Dict],
        context: str = ""
    ) -> str:
        history_text = "\n".join([
            f"- {item['title']} (rating: {item['rating']})"
            for item in user_history[-10:]
        ])
        
        candidates_text = "\n".join([
            f"{i+1}. {item['title']}: {item['description'][:100]}"
            for i, item in enumerate(candidate_items)
        ])
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": """You are a recommendation expert. 
                Analyze user preferences and recommend items with explanations."""},
                {"role": "user", "content": f"""User History:
{history_text}

Candidate Items:
{candidates_text}

{context}

Provide top 3 recommendations with brief explanations."""}
            ],
            temperature=0.7
        )
        
        return response.choices[0].message.content
    
    def generate_user_summary(self, user_data: Dict) -> str:
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "Create a concise user preference profile."},
                {"role": "user", "content": f"User data: {user_data}"}
            ]
        )
        return response.choices[0].message.content

recommender = GenAIRecommender(api_key="your-api-key")
summary = recommender.generate_user_summary({
    "favorite_genres": ["sci-fi", "thriller"],
    "recent_ratings": [5, 4, 5, 3, 4],
    "watch_history": ["Inception", "Interstellar", "The Matrix"]
})

Evaluation Metrics

import numpy as np
from typing import List, Set

class RecommenderEvaluator:
    def __init__(self):
        self.metrics = {}
    
    def precision_at_k(
        self, 
        recommended: List[str], 
        relevant: Set[str], 
        k: int
    ) -> float:
        recommended_k = recommended[:k]
        hits = len(set(recommended_k) & relevant)
        return hits / k
    
    def recall_at_k(
        self, 
        recommended: List[str], 
        relevant: Set[str], 
        k: int
    ) -> float:
        recommended_k = recommended[:k]
        hits = len(set(recommended_k) & relevant)
        return hits / len(relevant) if relevant else 0
    
    def ndcg_at_k(
        self, 
        recommended: List[str], 
        relevance_scores: Dict[str, float], 
        k: int
    ) -> float:
        dcg = 0
        for i, item in enumerate(recommended[:k]):
            rel = relevance_scores.get(item, 0)
            dcg += rel / np.log2(i + 2)
        
        ideal_relevance = sorted(relevance_scores.values(), reverse=True)[:k]
        idcg = sum(rel / np.log2(i + 2) for i, rel in enumerate(ideal_relevance))
        
        return dcg / idcg if idcg > 0 else 0
    
    def mean_average_precision(
        self, 
        user_recommendations: Dict[str, List[str]], 
        user_relevant: Dict[str, Set[str]]
    ) -> float:
        ap_scores = []
        for user_id, recommended in user_recommendations.items():
            relevant = user_relevant.get(user_id, set())
            hits = 0
            sum_precisions = 0
            
            for i, item in enumerate(recommended):
                if item in relevant:
                    hits += 1
                    sum_precisions += hits / (i + 1)
            
            ap = sum_precisions / len(relevant) if relevant else 0
            ap_scores.append(ap)
        
        return np.mean(ap_scores)

evaluator = RecommenderEvaluator()
recommended = ["item_1", "item_2", "item_3", "item_4", "item_5"]
relevant = {"item_1", "item_3", "item_5"}
print(f"P@5: {evaluator.precision_at_k(recommended, relevant, 5):.2f}")
print(f"R@5: {evaluator.recall_at_k(recommended, relevant, 5):.2f}")

Cold Start Solutions

• Use content-based features for new users/items
• Implement onboarding to collect preferences
• Leverage demographic information
• Use transfer learning from similar domains
• Apply knowledge-based recommendations

Best Practices

• Balance exploration and exploitation
• Handle implicit and explicit feedback
• Consider diversity in recommendations
• Implement real-time learning capabilities
• Monitor for bias and fairness
• A/B test recommendation strategies