1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

import numpy as np
import pandas as pd
from typing import Dict, List, Tuple
from dataclasses import dataclass
 
@dataclass
class UserTier:
    name: str
    min_interactions: int
    max_interactions: int
 
# Standard user tier definitions
USER_TIERS = [
    UserTier("ice_cold", 0, 0),
    UserTier("cold", 1, 5),
    UserTier("cool", 6, 20),
    UserTier("warm", 21, 100),
    UserTier("hot", 101, float('inf')),
]
 
class ColdStartAnalyzer:
    """
    Analyze cold start severity and track metrics by user/item activity tiers.
    """
    
    def __init__(
        self,
        user_interaction_counts: Dict[str, int],
        item_interaction_counts: Dict[str, int],
    ):
        """
        Args:
            user_interaction_counts: {user_id: total_interactions}
            item_interaction_counts: {item_id: total_interactions}
        """
        self.user_counts = user_interaction_counts
        self.item_counts = item_interaction_counts
        
    def get_user_tier(self, user_id: str) -> str:
        """Get activity tier for a user."""
        count = self.user_counts.get(user_id, 0)
        for tier in USER_TIERS:
            if tier.min_interactions <= count <= tier.max_interactions:
                return tier.name
        return "unknown"
    
    def get_item_tier(self, item_id: str) -> str:
        """Get popularity tier for an item."""
        count = self.item_counts.get(item_id, 0)
        
        # Item tiers often defined by percentiles
        if count == 0:
            return "new_item"
        elif count <= 10:
            return "long_tail"
        elif count <= 100:
            return "mid_popularity"
        else:
            return "popular"
    
    def compute_tier_distribution(self) -> pd.DataFrame:
        """
        Compute distribution of users/items across tiers.
        
        Returns DataFrame with tier counts and percentages.
        """
        user_tiers = [self.get_user_tier(uid) for uid in self.user_counts]
        item_tiers = [self.get_item_tier(iid) for iid in self.item_counts]
        
        user_dist = pd.Series(user_tiers).value_counts(normalize=True)
        item_dist = pd.Series(item_tiers).value_counts(normalize=True)
        
        return pd.DataFrame({
            'user_distribution': user_dist,
            'item_distribution': item_dist
        })
    
    def cold_start_exposure(
        self,
        recommendations: List[Tuple[str, str]],  # (user_id, item_id)
    ) -> Dict[str, float]:
        """
        Analyze how often cold users see cold items.
        
        The nightmare scenario: ice_cold user + new_item = random walk.
        
        Returns frequency of each (user_tier, item_tier) combination.
        """
        combinations = {}
        total = len(recommendations)
        
        for user_id, item_id in recommendations:
            user_tier = self.get_user_tier(user_id)
            item_tier = self.get_item_tier(item_id)
            key = f"{user_tier}_user_x_{item_tier}_item"
            combinations[key] = combinations.get(key, 0) + 1
        
        return {k: v / total for k, v in combinations.items()}
    
    def metrics_by_tier(
        self,
        interactions: pd.DataFrame,  # user_id, item_id, clicked, converted
    ) -> pd.DataFrame:
        """
        Compute recommendation metrics segmented by user tier.
        
        Critical for understanding cold start impact on business metrics.
        """
        # Add tier to each interaction
        interactions = interactions.copy()
        interactions['user_tier'] = interactions['user_id'].apply(self.get_user_tier)
        
        # Aggregate metrics by tier
        tier_metrics = interactions.groupby('user_tier').agg({
            'clicked': 'mean',  # CTR
            'converted': 'mean',  # Conversion rate
            'user_id': 'nunique',  # Number of users
        }).rename(columns={
            'clicked': 'ctr',
            'converted': 'conversion_rate',
            'user_id': 'n_users'
        })
        
        return tier_metrics
    
    def estimate_revenue_at_risk(
        self,
        tier_metrics: pd.DataFrame,
        avg_order_value: float,
        target_conversion_rate: float = None  # Use warm tier if not specified
    ) -> Dict[str, float]:
        """
        Estimate revenue impact of cold start.
        
        Compares conversion rates of cold tiers vs warm tier to estimate
        revenue lost due to suboptimal cold recommendations.
        """
        if target_conversion_rate is None:
            target_conversion_rate = tier_metrics.loc['warm', 'conversion_rate']
        
        revenue_at_risk = {}
        
        for tier in ['ice_cold', 'cold', 'cool']:
            if tier not in tier_metrics.index:
                continue
                
            current_rate = tier_metrics.loc[tier, 'conversion_rate']
            n_users = tier_metrics.loc[tier, 'n_users']
            
            # Gap between cold and warm performance
            rate_gap = target_conversion_rate - current_rate
            
            # Estimated lost conversions
            lost_conversions = n_users * rate_gap
            
            # Revenue impact
            revenue_at_risk[tier] = lost_conversions * avg_order_value
        
        revenue_at_risk['total'] = sum(revenue_at_risk.values())
        
        return revenue_at_risk

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217

import numpy as np
from typing import List, Dict, Optional
from dataclasses import dataclass
 
@dataclass
class UserProfile:
    user_id: str
    demographics: Dict[str, str]  # e.g., {"age_group": "25-34", "country": "US"}
    onboarding_preferences: List[str]  # Item IDs selected during onboarding
    interaction_history: List[str]  # Item IDs (chronological)
 
 
class UserColdStartSolver:
    """
    Comprehensive approach to new user recommendations.
    
    Combines multiple signals and strategies, with graceful
    transitions as user warms up.
    """
    
    def __init__(
        self,
        item_popularity: Dict[str, float],
        segment_preferences: Dict[str, Dict[str, float]],  # segment -> item -> score
        item_embeddings: np.ndarray,  # Shape: (n_items, embedding_dim)
        item_ids: List[str],
    ):
        self.item_popularity = item_popularity
        self.segment_preferences = segment_preferences
        self.item_embeddings = item_embeddings
        self.item_ids = item_ids
        self.item_to_idx = {item_id: i for i, item_id in enumerate(item_ids)}
        
    def _get_user_tier(self, user: UserProfile) -> str:
        """Determine user's warmth tier."""
        n_interactions = len(user.interaction_history)
        
        if n_interactions == 0:
            if user.onboarding_preferences:
                return "has_onboarding"
            return "ice_cold"
        elif n_interactions <= 5:
            return "cold"
        elif n_interactions <= 20:
            return "cool"
        else:
            return "warm"
    
    def recommend(
        self, 
        user: UserProfile, 
        n_recommendations: int = 10,
        exploration_rate: float = 0.2
    ) -> List[str]:
        """
        Generate recommendations adapting to user's warmth level.
        
        Strategy blending weights depend on available data.
        """
        tier = self._get_user_tier(user)
        
        if tier == "ice_cold":
            return self._recommend_ice_cold(user, n_recommendations, exploration_rate)
        elif tier == "has_onboarding":
            return self._recommend_with_onboarding(user, n_recommendations)
        elif tier == "cold":
            return self._recommend_cold(user, n_recommendations, exploration_rate)
        else:
            # Defer to main recommendation system for cool/warm users
            return self._recommend_warm(user, n_recommendations)
    
    def _recommend_ice_cold(
        self,
        user: UserProfile,
        n_recommendations: int,
        exploration_rate: float
    ) -> List[str]:
        """
        Complete cold start: no history, no onboarding.
        
        Strategy: Blend segment popularity with exploration.
        """
        # Try to use demographic segment
        segment_key = self._get_segment_key(user.demographics)
        
        if segment_key in self.segment_preferences:
            segment_prefs = self.segment_preferences[segment_key]
            base_scores = np.array([
                segment_prefs.get(item_id, 0) 
                for item_id in self.item_ids
            ])
        else:
            # Fall back to global popularity
            base_scores = np.array([
                self.item_popularity.get(item_id, 0)
                for item_id in self.item_ids
            ])
        
        # Add exploration: epsilon chance of random selection
        n_exploit = int(n_recommendations * (1 - exploration_rate))
        n_explore = n_recommendations - n_exploit
        
        # Top items by score
        top_indices = np.argsort(base_scores)[-n_exploit:][::-1]
        exploit_items = [self.item_ids[i] for i in top_indices]
        
        # Random exploration (potentially excluding already selected)
        explore_pool = list(set(self.item_ids) - set(exploit_items))
        explore_items = list(np.random.choice(
            explore_pool, 
            size=min(n_explore, len(explore_pool)),
            replace=False
        ))
        
        return exploit_items + explore_items
    
    def _recommend_with_onboarding(
        self,
        user: UserProfile,
        n_recommendations: int
    ) -> List[str]:
        """
        User completed onboarding: leverage stated preferences.
        
        Strategy: Content-based similarity to onboarding items.
        """
        # Compute average embedding of onboarding items
        onboarding_indices = [
            self.item_to_idx[item_id] 
            for item_id in user.onboarding_preferences
            if item_id in self.item_to_idx
        ]
        
        if not onboarding_indices:
            return self._recommend_ice_cold(user, n_recommendations, 0.2)
        
        user_profile_embedding = self.item_embeddings[onboarding_indices].mean(axis=0)
        
        # Score all items by similarity to user profile
        similarities = self.item_embeddings @ user_profile_embedding
        
        # Exclude onboarding items from recommendations
        for idx in onboarding_indices:
            similarities[idx] = -np.inf
        
        top_indices = np.argsort(similarities)[-n_recommendations:][::-1]
        return [self.item_ids[i] for i in top_indices]
    
    def _recommend_cold(
        self,
        user: UserProfile,
        n_recommendations: int,
        exploration_rate: float
    ) -> List[str]:
        """
        User has few interactions (1-5): blend content + limited CF.
        
        Strategy: Heavy weight on content similarity, light weight
        on collaborative signal (if any), continued exploration.
        """
        # Content-based: similar to interacted items
        history_indices = [
            self.item_to_idx[item_id]
            for item_id in user.interaction_history
            if item_id in self.item_to_idx
        ]
        
        if history_indices:
            user_embedding = self.item_embeddings[history_indices].mean(axis=0)
            content_scores = self.item_embeddings @ user_embedding
        else:
            content_scores = np.zeros(len(self.item_ids))
        
        # Popularity prior
        pop_scores = np.array([
            self.item_popularity.get(item_id, 0)
            for item_id in self.item_ids
        ])
        
        # Blend: 60% content, 40% popularity for cold users
        combined_scores = 0.6 * content_scores + 0.4 * pop_scores
        
        # Exclude already interacted items
        for idx in history_indices:
            combined_scores[idx] = -np.inf
        
        # Top recommendations with exploration
        n_exploit = int(n_recommendations * (1 - exploration_rate))
        top_indices = np.argsort(combined_scores)[-n_exploit:][::-1]
        exploit_items = [self.item_ids[i] for i in top_indices]
        
        # Exploration from mid-ranked items
        n_explore = n_recommendations - n_exploit
        mid_range = np.argsort(combined_scores)[-100:-n_exploit]
        explore_items = list(np.random.choice(
            [self.item_ids[i] for i in mid_range],
            size=min(n_explore, len(mid_range)),
            replace=False
        ))
        
        return exploit_items + explore_items
    
    def _recommend_warm(
        self,
        user: UserProfile,
        n_recommendations: int
    ) -> List[str]:
        """Placeholder for full recommendation system."""
        # In production, this would call the main CF-based recommender
        raise NotImplementedError("Defer to main recommendation system")
    
    def _get_segment_key(self, demographics: Dict[str, str]) -> str:
        """Create segment key from demographics."""
        parts = []
        for key in sorted(demographics.keys()):
            parts.append(f"{key}={demographics[key]}")
        return "|".join(parts)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer
from sklearn.neighbors import NearestNeighbors
 
@dataclass
class Item:
    item_id: str
    title: str
    description: str
    categories: List[str]
    creator_id: Optional[str]
    metadata: Dict[str, any]
 
 
class ItemColdStartSolver:
    """
    Handle new item recommendations using content features.
    
    Core idea: Map new items to the same representation space as
    existing items, then leverage learned user preferences.
    """
    
    def __init__(
        self,
        text_encoder: SentenceTransformer = None,
        existing_items: List[Item] = None,
        existing_item_embeddings: np.ndarray = None,
        user_item_preferences: Dict[str, Dict[str, float]] = None  # user -> item -> score
    ):
        self.text_encoder = text_encoder or SentenceTransformer('all-MiniLM-L6-v2')
        self.existing_items = {item.item_id: item for item in (existing_items or [])}
        self.existing_embeddings = existing_item_embeddings  # Shape: (n_items, dim)
        self.user_preferences = user_item_preferences or {}
        
        # Build nearest neighbor index for existing items
        if existing_item_embeddings is not None:
            self.nn_index = NearestNeighbors(n_neighbors=100, metric='cosine')
            self.nn_index.fit(existing_item_embeddings)
        
        # Creator -> items mapping for creator transfer
        self.creator_items: Dict[str, List[str]] = {}
        for item in (existing_items or []):
            if item.creator_id:
                if item.creator_id not in self.creator_items:
                    self.creator_items[item.creator_id] = []
                self.creator_items[item.creator_id].append(item.item_id)
    
    def encode_new_item(self, item: Item) -> np.ndarray:
        """
        Encode new item into embedding space using content features.
        
        Combines text embedding with structured features.
        """
        # Text embedding from title + description
        text_input = f"{item.title}. {item.description}"
        text_embedding = self.text_encoder.encode(text_input)
        
        # Could also incorporate category embeddings, but text usually dominates
        return text_embedding
    
    def find_similar_existing_items(
        self, 
        new_item: Item, 
        k: int = 20
    ) -> List[Tuple[str, float]]:
        """
        Find existing items most similar to new item.
        
        Returns: List of (item_id, similarity_score) pairs.
        """
        new_embedding = self.encode_new_item(new_item)
        new_embedding = new_embedding.reshape(1, -1)
        
        distances, indices = self.nn_index.kneighbors(new_embedding, n_neighbors=k)
        
        # Convert distances to similarities (cosine distance -> similarity)
        similarities = 1 - distances[0]
        
        existing_item_ids = list(self.existing_items.keys())
        return [
            (existing_item_ids[idx], sim) 
            for idx, sim in zip(indices[0], similarities)
        ]
    
    def predict_user_preference(
        self, 
        user_id: str, 
        new_item: Item
    ) -> float:
        """
        Predict user's preference for new item based on content similarity
        to items they've liked before.
        
        Uses weighted average of preferences for similar items.
        """
        if user_id not in self.user_preferences:
            return 0.0  # No preference data
        
        user_prefs = self.user_preferences[user_id]
        similar_items = self.find_similar_existing_items(new_item, k=30)
        
        # Weighted average of user's preferences for similar items
        weighted_sum = 0.0
        weight_total = 0.0
        
        for similar_item_id, similarity in similar_items:
            if similar_item_id in user_prefs:
                weighted_sum += similarity * user_prefs[similar_item_id]
                weight_total += similarity
        
        if weight_total == 0:
            return 0.0
        
        return weighted_sum / weight_total
    
    def find_target_users(
        self, 
        new_item: Item, 
        n_users: int = 100
    ) -> List[Tuple[str, float]]:
        """
        Find users most likely to enjoy new item.
        
        Strategy: Find similar items, then find users who loved those items.
        """
        similar_items = self.find_similar_existing_items(new_item, k=50)
        similar_item_ids = {item_id for item_id, _ in similar_items}
        
        user_scores = {}
        
        for user_id, prefs in self.user_preferences.items():
            # How much does this user like items similar to new item?
            overlap_score = 0.0
            for item_id, pref_score in prefs.items():
                if item_id in similar_item_ids:
                    # Could weight by similarity here
                    overlap_score += pref_score
            
            if overlap_score > 0:
                user_scores[user_id] = overlap_score
        
        # Return top users sorted by score
        sorted_users = sorted(user_scores.items(), key=lambda x: -x[1])
        return sorted_users[:n_users]
    
    def creator_transfer(
        self, 
        new_item: Item, 
        n_users: int = 100
    ) -> List[Tuple[str, float]]:
        """
        Transfer preferences from creator's previous items.
        
        Users who liked this creator's previous work → potential audience.
        """
        if not new_item.creator_id:
            return []
        
        creator_previous_items = self.creator_items.get(new_item.creator_id, [])
        
        if not creator_previous_items:
            return []
        
        # Find users who liked previous items from this creator
        user_scores = {}
        
        for user_id, prefs in self.user_preferences.items():
            creator_pref_sum = 0.0
            n_rated = 0
            
            for item_id in creator_previous_items:
                if item_id in prefs:
                    creator_pref_sum += prefs[item_id]
                    n_rated += 1
            
            if n_rated > 0:
                avg_creator_pref = creator_pref_sum / n_rated
                # Weight by number of rated items (more data = more confidence)
                user_scores[user_id] = avg_creator_pref * np.sqrt(n_rated)
        
        sorted_users = sorted(user_scores.items(), key=lambda x: -x[1])
        return sorted_users[:n_users]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from typing import List, Tuple, Dict
from copy import deepcopy
 
class UserPersonalizationNetwork(nn.Module):
    """
    Base network for user preference prediction.
    
    Architecture: User embedding + Item embedding -> Score
    """
    
    def __init__(self, n_items: int, embedding_dim: int = 64):
        super().__init__()
        self.item_embeddings = nn.Embedding(n_items, embedding_dim)
        self.user_transform = nn.Linear(embedding_dim, embedding_dim)
        self.output = nn.Linear(embedding_dim, 1)
        
    def forward(
        self, 
        user_embedding: torch.Tensor, 
        item_ids: torch.Tensor
    ) -> torch.Tensor:
        """
        Predict preference score.
        
        Args:
            user_embedding: (batch, embedding_dim) 
            item_ids: (batch,) item indices
        
        Returns:
            (batch, 1) prediction scores
        """
        item_emb = self.item_embeddings(item_ids)
        user_transformed = self.user_transform(user_embedding)
        
        # Simple dot product with non-linearity
        interaction = user_transformed * item_emb
        return self.output(torch.relu(interaction))
 
 
class MAMLRecommender:
    """
    MAML-style meta-learning for cold start recommendations.
    
    Core idea: Learn a model initialization that quickly adapts
    to new users with just a few interactions.
    
    Training:
    1. Sample a batch of users (tasks)
    2. For each user:
       a. Take a few gradient steps on their support set
       b. Evaluate on their query set
    3. Update meta-parameters to minimize query set loss
    
    At test time:
    1. New user provides a few interactions (support set)
    2. Take a few gradient steps to personalize
    3. Use adapted model for recommendations
    """
    
    def __init__(
        self,
        n_items: int,
        embedding_dim: int = 64,
        inner_lr: float = 0.01,
        outer_lr: float = 0.001,
        n_inner_steps: int = 3
    ):
        self.n_items = n_items
        self.embedding_dim = embedding_dim
        self.inner_lr = inner_lr
        self.outer_lr = outer_lr
        self.n_inner_steps = n_inner_steps
        
        self.model = UserPersonalizationNetwork(n_items, embedding_dim)
        self.meta_optimizer = optim.Adam(self.model.parameters(), lr=outer_lr)
        
        # Meta-learned user initialization (what new users start with)
        self.user_init = nn.Parameter(torch.randn(embedding_dim) * 0.01)
        
    def inner_loop(
        self,
        support_items: torch.Tensor,  # Items user interacted with
        support_labels: torch.Tensor,  # Preference labels
        user_embedding: torch.Tensor
    ) -> torch.Tensor:
        """
        Adapt user embedding based on their support interactions.
        
        Few gradient steps to personalize.
        """
        user_emb = user_embedding.clone().requires_grad_(True)
        
        for _ in range(self.n_inner_steps):
            predictions = self.model(user_emb.unsqueeze(0).expand(len(support_items), -1), support_items)
            loss = nn.functional.mse_loss(predictions.squeeze(), support_labels)
            
            # Gradient step on user embedding only
            grad = torch.autograd.grad(loss, user_emb, create_graph=True)[0]
            user_emb = user_emb - self.inner_lr * grad
        
        return user_emb
    
    def meta_train_step(
        self,
        user_tasks: List[Dict]  # Each dict: {support_items, support_labels, query_items, query_labels}
    ) -> float:
        """
        One step of meta-training.
        
        For each user, adapt to their support set, evaluate on query set.
        Update meta-parameters to minimize query loss.
        """
        self.meta_optimizer.zero_grad()
        total_query_loss = 0.0
        
        for task in user_tasks:
            # Start from meta-learned initialization
            user_emb = self.user_init.clone()
            
            # Adapt to this user's support set
            adapted_user_emb = self.inner_loop(
                task['support_items'],
                task['support_labels'],
                user_emb
            )
            
            # Evaluate on query set with adapted embedding
            query_predictions = self.model(
                adapted_user_emb.unsqueeze(0).expand(len(task['query_items']), -1),
                task['query_items']
            )
            query_loss = nn.functional.mse_loss(
                query_predictions.squeeze(), 
                task['query_labels']
            )
            
            total_query_loss = total_query_loss + query_loss
        
        # Meta-update: improve initialization based on query performance
        avg_loss = total_query_loss / len(user_tasks)
        avg_loss.backward()
        self.meta_optimizer.step()
        
        return avg_loss.item()
    
    def adapt_to_new_user(
        self,
        support_items: List[int],
        support_labels: List[float]
    ) -> torch.Tensor:
        """
        Adapt model to a new user based on their few interactions.
        
        Returns: Personalized user embedding
        """
        support_items_t = torch.tensor(support_items)
        support_labels_t = torch.tensor(support_labels, dtype=torch.float32)
        
        with torch.no_grad():
            user_emb = self.user_init.clone()
        
        # Use inner loop to adapt (without graph for inference)
        user_emb = user_emb.requires_grad_(True)
        
        for _ in range(self.n_inner_steps):
            predictions = self.model(
                user_emb.unsqueeze(0).expand(len(support_items_t), -1), 
                support_items_t
            )
            loss = nn.functional.mse_loss(predictions.squeeze(), support_labels_t)
            
            grad = torch.autograd.grad(loss, user_emb)[0]
            user_emb = user_emb.detach() - self.inner_lr * grad
            user_emb = user_emb.requires_grad_(True)
        
        return user_emb.detach()
    
    def recommend(
        self,
        user_embedding: torch.Tensor,
        n_recommendations: int = 10,
        exclude_items: List[int] = None
    ) -> List[int]:
        """
        Generate top-N recommendations for adapted user.
        """
        with torch.no_grad():
            all_items = torch.arange(self.n_items)
            user_expanded = user_embedding.unsqueeze(0).expand(self.n_items, -1)
            
            scores = self.model(user_expanded, all_items).squeeze()
            
            # Exclude already interacted items
            if exclude_items:
                for item_id in exclude_items:
                    scores[item_id] = -float('inf')
            
            top_k = torch.topk(scores, n_recommendations).indices.tolist()
        
        return top_k