AI Personalization Engines: Building Customer Journeys That Convert

Personalization isn't about using someone's first name in an email anymore. It's about delivering the right content, offer, and experience to each customer at every touchpoint — powered by ML models that understand individual behavior patterns.

The Business Case for Personalization

The numbers are compelling:

Netflix: 80% of content watched comes from recommendations (saves $1B/year in reduced churn)

Amazon: 35% of revenue attributed to recommendation engine

Starbucks: 16% revenue increase from AI-personalized offers

Sephora: 15-25% higher conversion from personalized emails vs. batch sends

Building a Recommendation Engine

Collaborative Filtering

python
import numpy as np
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity
from scipy.sparse import csr_matrix
from scipy.sparse.linalg import svds
class CollaborativeFilteringRecommender:
    """
    Matrix factorization-based collaborative filtering.
    "Users who did X also did Y" approach.
    Works well for content/product recommendations.
    """
    
    def __init__(self, n_factors: int = 50):
        self.n_factors = n_factors
        self.user_factors = None
        self.item_factors = None
        self.user_bias = None
        self.item_bias = None
        self.global_mean = None
        self.user_index = {}  # user_id → matrix row
        self.item_index = {}  # item_id → matrix col
    
    def fit(self, interactions_df: pd.DataFrame):
        """
        Train on interaction matrix.
        
        interactions_df needs: user_id, item_id, rating/interaction_strength
        """
        # Create user and item indices
        users = interactions_df['user_id'].unique()
        items = interactions_df['item_id'].unique()
        
        self.user_index = {u: i for i, u in enumerate(users)}
        self.item_index = {item: i for i, item in enumerate(items)}
        
        # Build interaction matrix
        rows = [self.user_index[u] for u in interactions_df['user_id']]
        cols = [self.item_index[i] for i in interactions_df['item_id']]
        data = interactions_df['rating'].values
        
        matrix = csr_matrix(
            (data, (rows, cols)),
            shape=(len(users), len(items))
        )
        
        self.global_mean = np.mean(data)
        
        # SVD decomposition
        U, sigma, Vt = svds(matrix.toarray(), k=self.n_factors)
        
        # Reconstruct latent factors
        sigma_diag = np.diag(sigma)
        self.user_factors = U.dot(sigma_diag)
        self.item_factors = Vt.T
        
        print(f"Trained on {len(users)} users, {len(items)} items")
        return self
    
    def recommend(self, user_id: str, n_recommendations: int = 10,
                  exclude_seen: bool = True) -> list[dict]:
        """Get personalized recommendations for a user."""
        
        if user_id not in self.user_index:
            # Cold start: return popular items
            return self._get_popular_items(n_recommendations)
        
        user_row = self.user_index[user_id]
        user_vector = self.user_factors[user_row]
        
        # Calculate scores for all items
        scores = user_vector.dot(self.item_factors.T)
        
        # Create recommendations list
        item_scores = list(zip(self.item_index.keys(), scores))
        item_scores.sort(key=lambda x: x[1], reverse=True)
        
        return [
            {'item_id': item_id, 'score': float(score), 'rank': i+1}
            for i, (item_id, score) in enumerate(item_scores[:n_recommendations])
        ]
    
    def get_similar_items(self, item_id: str, n: int = 5) -> list[dict]:
        """Find items similar to a given item (for 'You might also like')."""
        
        if item_id not in self.item_index:
            return []
        
        item_idx = self.item_index[item_id]
        item_vector = self.item_factors[item_idx]
        
        # Cosine similarity
        similarities = cosine_similarity(
            item_vector.reshape(1, -1),
            self.item_factors
        )[0]
        
        # Sort and exclude the item itself
        similar_indices = similarities.argsort()[::-1][1:n+1]
        
        index_to_item = {v: k for k, v in self.item_index.items()}
        
        return [
            {
                'item_id': index_to_item[idx],
                'similarity_score': float(similarities[idx])
            }
            for idx in similar_indices
        ]
    
    def _get_popular_items(self, n: int) -> list[dict]:
        """Fallback for cold start users."""
        # Would normally query a database
        return [{'item_id': f'popular_{i}', 'score': 1.0 - i*0.1} for i in range(n)]
Content-based filtering for item similarity
class ContentBasedRecommender:
    """
    Recommends based on item features.
    "If you liked this type of content, here's similar content."
    Works well for new items (solves cold start for items).
    """
    
    def __init__(self):
        self.client = OpenAI()
        self.item_embeddings = {}
    
    def embed_item(self, item: dict) -> np.ndarray:
        """Create embedding from item features."""
        
        # Combine item metadata into text
        item_text = f"""
Title: {item.get('title', '')}
Category: {item.get('category', '')}
Description: {item.get('description', '')[:500]}
Tags: {', '.join(item.get('tags', []))}
"""
        
        response = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=item_text
        )
        
        return np.array(response.data[0].embedding)
    
    def find_similar(self, item_id: str, n: int = 5) -> list[dict]:
        """Find similar items using embedding similarity."""
        
        if item_id not in self.item_embeddings:
            return []
        
        query_embedding = self.item_embeddings[item_id]
        
        similarities = {}
        for other_id, other_embedding in self.item_embeddings.items():
            if other_id == item_id:
                continue
            sim = cosine_similarity(
                query_embedding.reshape(1, -1),
                other_embedding.reshape(1, -1)
            )[0][0]
            similarities[other_id] = sim
        
        sorted_items = sorted(similarities.items(), key=lambda x: x[1], reverse=True)
        
        return [
            {'item_id': item_id, 'similarity': float(score)}
            for item_id, score in sorted_items[:n]
        ]

Dynamic Email Personalization

python
import anthropic
def generate_personalized_email(user_profile: dict, email_purpose: str) -> dict:
    """
    Generate a personalized email based on user behavior and context.
    """
    client = anthropic.Anthropic()
    
    message = client.messages.create(
        model="claude-haiku-4-5",  # Fast for email generation
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Write a personalized {email_purpose} email for this customer:
Profile:
Name: {user_profile.get('first_name')}
Customer since: {user_profile.get('customer_since')}
Products used: {', '.join(user_profile.get('products', []))}
Last activity: {user_profile.get('last_activity_days')} days ago
Plan tier: {user_profile.get('plan')}
Usage level: {user_profile.get('usage_level', 'moderate')}
Recent support tickets: {user_profile.get('recent_tickets', 0)}
Email purpose: {email_purpose}
Generate:
Subject line (compelling, personal, under 50 chars)
Preview text (under 100 chars)
Email body (150-200 words, warm tone, specific to their usage)
CTA button text
Return as JSON."""
        }]
    )
    
    try:
        return json.loads(message.content[0].text)
    except:
        return {"body": message.content[0].text}

The Ethics of Personalization

Personalization requires handling personal data responsibly:

Be transparent: Tell users what you're personalizing and why

Give control: Allow users to see and edit their preference profile

Avoid filter bubbles: Some randomness and diversity is good

Don't be creepy: "We noticed you've been looking at X" can feel invasive

Privacy compliance: GDPR/CCPA require explicit consent for certain types of personalization

The best personalization makes customers feel understood, not surveilled.