Personalized GenAI Systems

# Personalized GenAI Pipeline Components

1. User Profiling Layer
   ├── Behavioral Analytics (interaction patterns)
   ├── Preference Modeling (explicit & implicit)
   ├── Context Extraction (temporal, spatial, social)
   └── Privacy-Preserving Profiling

2. Personalization Engine
   ├── User Embedding Generation
   ├── Dynamic Prompt Construction
   ├── Context-Aware Retrieval
   └── Preference-Based Filtering

3. Model Adaptation Layer
   ├── Fine-tuning Pipeline (LoRA, QLoRA)
   ├── Prompt Template Library
   ├── Mixture of Experts Routing
   └── Real-time Adaptation

4. Privacy Protection Layer
   ├── Differential Privacy (training & inference)
   ├── Federated Learning Framework
   ├── Secure Multi-party Computation
   └── Data Anonymization

5. Evaluation & Monitoring
   ├── Personalization Quality Metrics
   ├── Privacy Leakage Detection
   ├── User Satisfaction Tracking
   └── A/B Testing Framework

# Key Technical Requirements:
- User embedding dimension: 256-1024
- Adaptation latency: < 100ms
- Privacy budget (ε): < 1.0
- Memory per user: < 1MB
- Personalization accuracy: > 85%

import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
import numpy as np

class UserAdaptiveModel:
    def __init__(self, base_model_name="meta-llama/Llama-2-7b-chat-hf"):
        self.base_model_name = base_model_name
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # Load base model
        self.base_model = AutoModelForCausalLM.from_pretrained(
            base_model_name,
            torch_dtype=torch.float16,
            device_map="auto",
            load_in_4bit=True  # QLoRA for memory efficiency
        )
        self.tokenizer = AutoTokenizer.from_pretrained(base_model_name)
        
        # User-specific LoRA adapters
        self.user_adapters = {}
        
        # User profile embeddings
        self.user_embeddings = {}
        self.embedding_dim = 512
        
    def create_user_adapter(self, user_id: str, user_data: dict):
        """Create personalized LoRA adapter for a user"""
        
        # LoRA configuration for personalization
        lora_config = LoraConfig(
            r=16,  # Rank of adaptation
            lora_alpha=32,  # Scaling parameter
            target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
            lora_dropout=0.1,
            bias="none",
            task_type="CAUSAL_LM"
        )
        
        # Clone base model for user adaptation
        user_model = prepare_model_for_kbit_training(self.base_model)
        user_model = get_peft_model(user_model, lora_config)
        
        # Generate user embedding from profile data
        user_embedding = self._create_user_embedding(user_data)
        self.user_embeddings[user_id] = user_embedding
        
        # Store user adapter
        self.user_adapters[user_id] = user_model
        
        return user_model
    
    def _create_user_embedding(self, user_data: dict) -> torch.Tensor:
        """Create user embedding from profile data"""
        
        # Extract features from user profile
        features = []
        
        # Demographic features
        age = user_data.get('age', 30) / 100.0  # Normalize
        features.append(age)
        
        # Preference features (encoded as embeddings)
        preferences = user_data.get('preferences', {})
        style_pref = self._encode_categorical(preferences.get('style', 'neutral'))
        tone_pref = self._encode_categorical(preferences.get('tone', 'professional'))
        complexity_pref = preferences.get('complexity', 0.5)  # 0-1 scale
        
        features.extend(style_pref)
        features.extend(tone_pref)
        features.append(complexity_pref)
        
        # Behavioral features
        interaction_history = user_data.get('interaction_history', [])
        avg_session_length = np.mean([len(session) for session in interaction_history])
        features.append(avg_session_length / 1000.0)  # Normalize
        
        # Pad or truncate to fixed dimension
        features = features[:self.embedding_dim]
        features.extend([0.0] * (self.embedding_dim - len(features)))
        
        return torch.tensor(features, dtype=torch.float16)
    
    def _encode_categorical(self, category: str, categories=['neutral', 'casual', 'formal', 'technical', 'creative']) -> list:
        """One-hot encode categorical features"""
        encoding = [0.0] * len(categories)
        if category in categories:
            encoding[categories.index(category)] = 1.0
        return encoding
    
    def personalized_generate(self, 
                            user_id: str, 
                            prompt: str, 
                            max_length: int = 512,
                            temperature: float = 0.7) -> str:
        """Generate personalized response for a user"""
        
        # Get user adapter and embedding
        if user_id not in self.user_adapters:
            raise ValueError(f"No adapter found for user {user_id}")
        
        user_model = self.user_adapters[user_id]
        user_embedding = self.user_embeddings[user_id]
        
        # Construct personalized prompt
        personalized_prompt = self._construct_personalized_prompt(
            prompt, user_id, user_embedding
        )
        
        # Tokenize input
        inputs = self.tokenizer(
            personalized_prompt, 
            return_tensors="pt", 
            truncation=True, 
            max_length=max_length
        ).to(self.device)
        
        # Generate with user-specific adapter
        with torch.no_grad():
            outputs = user_model.generate(
                **inputs,
                max_new_tokens=max_length,
                temperature=temperature,
                do_sample=True,
                pad_token_id=self.tokenizer.eos_token_id
            )
        
        # Decode response
        response = self.tokenizer.decode(
            outputs[0][inputs['input_ids'].shape[1]:], 
            skip_special_tokens=True
        )
        
        return response.strip()
    
    def _construct_personalized_prompt(self, 
                                     base_prompt: str, 
                                     user_id: str, 
                                     user_embedding: torch.Tensor) -> str:
        """Construct prompt with user context"""
        
        # Extract user preferences from embedding (simplified)
        style_score = user_embedding[1:6].argmax().item()
        styles = ['neutral', 'casual', 'formal', 'technical', 'creative']
        user_style = styles[style_score]
        
        complexity_score = user_embedding[-2].item()
        complexity_level = "simple" if complexity_score < 0.3 else "detailed" if complexity_score > 0.7 else "moderate"
        
        # Construct personalized system prompt
        system_prompt = f"""You are an AI assistant adapting to user preferences.
User Style: {user_style}
Complexity Level: {complexity_level}
Respond in a way that matches these preferences."""
        
        return f"{system_prompt}

User: {base_prompt}
Assistant:"
    
    def update_user_adapter(self, 
                          user_id: str, 
                          feedback_data: list, 
                          learning_rate: float = 1e-4):
        """Update user adapter based on feedback"""
        
        if user_id not in self.user_adapters:
            raise ValueError(f"No adapter found for user {user_id}")
        
        user_model = self.user_adapters[user_id]
        optimizer = torch.optim.AdamW(user_model.parameters(), lr=learning_rate)
        
        # Prepare training data from feedback
        for feedback in feedback_data:
            prompt = feedback['prompt']
            preferred_response = feedback['preferred_response']
            
            # Tokenize
            inputs = self.tokenizer(
                f"{prompt} {preferred_response}",
                return_tensors="pt",
                truncation=True,
                max_length=512
            ).to(self.device)
            
            # Forward pass
            outputs = user_model(**inputs, labels=inputs['input_ids'])
            loss = outputs.loss
            
            # Backward pass
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        
        print(f"Updated adapter for user {user_id} with {len(feedback_data)} feedback samples")

class PersonalizedPromptEngine:
    """Dynamic prompt construction based on user context"""
    
    def __init__(self):
        self.prompt_templates = self._load_prompt_templates()
        self.user_contexts = {}
        
    def _load_prompt_templates(self) -> dict:
        """Load domain-specific prompt templates"""
        return {
            'writing_assistant': {
                'casual': "Help me write this in a friendly, conversational tone: {content}",
                'formal': "Please assist me in writing this in a professional manner: {content}",
                'creative': "Help me write this with creativity and flair: {content}"
            },
            'code_assistant': {
                'beginner': "Explain this code concept in simple terms with examples: {content}",
                'intermediate': "Help me understand and improve this code: {content}",
                'expert': "Provide advanced insights and optimizations for: {content}"
            },
            'tutor': {
                'visual': "Explain this concept using analogies and examples: {content}",
                'analytical': "Break down this topic step-by-step with logical reasoning: {content}",
                'practical': "Show me how to apply this concept with real-world examples: {content}"
            }
        }
    
    def update_user_context(self, user_id: str, interaction_data: dict):
        """Update user context from interactions"""
        if user_id not in self.user_contexts:
            self.user_contexts[user_id] = {
                'preferences': {},
                'history': [],
                'learning_style': 'balanced'
            }
        
        context = self.user_contexts[user_id]
        context['history'].append(interaction_data)
        
        # Infer preferences from interaction patterns
        self._infer_preferences(user_id, interaction_data)
    
    def _infer_preferences(self, user_id: str, interaction_data: dict):
        """Infer user preferences from interaction data"""
        context = self.user_contexts[user_id]
        
        # Analyze response length preference
        if 'response_length' in interaction_data:
            length_pref = interaction_data['response_length']
            context['preferences']['response_length'] = length_pref
        
        # Analyze style preference from feedback
        if 'style_feedback' in interaction_data:
            style_feedback = interaction_data['style_feedback']
            context['preferences']['style'] = style_feedback
    
    def generate_personalized_prompt(self, 
                                   user_id: str, 
                                   base_content: str, 
                                   domain: str = 'general') -> str:
        """Generate personalized prompt for user"""
        
        if user_id not in self.user_contexts:
            return base_content  # Return base content if no context
        
        context = self.user_contexts[user_id]
        preferences = context['preferences']
        
        # Select appropriate template
        if domain in self.prompt_templates:
            templates = self.prompt_templates[domain]
            
            # Choose template based on user preferences
            style = preferences.get('style', 'balanced')
            if style in templates:
                template = templates[style]
            else:
                template = list(templates.values())[0]  # Default
            
            personalized_prompt = template.format(content=base_content)
        else:
            # Generic personalization
            style_instruction = self._get_style_instruction(preferences)
            personalized_prompt = f"{style_instruction}

{base_content}"
        
        return personalized_prompt
    
    def _get_style_instruction(self, preferences: dict) -> str:
        """Generate style instruction from preferences"""
        instructions = []
        
        if preferences.get('response_length') == 'concise':
            instructions.append("Please be concise and to the point.")
        elif preferences.get('response_length') == 'detailed':
            instructions.append("Please provide detailed explanations.")
        
        if preferences.get('style') == 'casual':
            instructions.append("Use a friendly, conversational tone.")
        elif preferences.get('style') == 'formal':
            instructions.append("Use a professional, formal tone.")
        
        return " ".join(instructions) if instructions else "Please respond helpfully."
    
    def get_user_insights(self, user_id: str) -> dict:
        """Get insights about user preferences and patterns"""
        if user_id not in self.user_contexts:
            return {}
        
        context = self.user_contexts[user_id]
        history = context['history']
        
        if not history:
            return {}
        
        # Analyze interaction patterns
        avg_session_length = np.mean([len(h.get('session', [])) for h in history])
        common_domains = {}
        
        for interaction in history:
            domain = interaction.get('domain', 'general')
            common_domains[domain] = common_domains.get(domain, 0) + 1
        
        most_common_domain = max(common_domains, key=common_domains.get)
        
        return {
            'total_interactions': len(history),
            'avg_session_length': avg_session_length,
            'most_common_domain': most_common_domain,
            'preferences': context['preferences'],
            'engagement_level': self._calculate_engagement(history)
        }
    
    def _calculate_engagement(self, history: list) -> str:
        """Calculate user engagement level"""
        if len(history) < 5:
            return 'new_user'
        elif len(history) < 20:
            return 'casual'
        else:
            return 'active'

import numpy as np
import torch
import torch.nn as nn
from typing import Dict, List, Tuple
import hashlib

class DifferentiallyPrivatePersonalization:
    def __init__(self, epsilon=1.0, delta=1e-5):
        self.epsilon = epsilon  # Privacy budget
        self.delta = delta      # Failure probability
        self.user_budgets = {}  # Track per-user privacy budget
        self.global_sensitivity = 1.0  # Sensitivity of queries
        
    def add_noise(self, data: torch.Tensor, sensitivity: float = None) -> torch.Tensor:
        """Add Gaussian noise for differential privacy"""
        if sensitivity is None:
            sensitivity = self.global_sensitivity
            
        # Calculate noise scale for Gaussian mechanism
        sigma = np.sqrt(2 * np.log(1.25 / self.delta)) * sensitivity / self.epsilon
        
        # Add Gaussian noise
        noise = torch.normal(0, sigma, size=data.shape)
        return data + noise
    
    def private_user_embedding(self, user_id: str, user_data: Dict) -> torch.Tensor:
        """Create differentially private user embedding"""
        
        # Check privacy budget
        if user_id in self.user_budgets:
            if self.user_budgets[user_id] <= 0:
                raise ValueError(f"Privacy budget exhausted for user {user_id}")
        else:
            self.user_budgets[user_id] = self.epsilon
        
        # Extract features (same as before)
        features = self._extract_features(user_data)
        embedding = torch.tensor(features, dtype=torch.float32)
        
        # Add differential privacy noise
        private_embedding = self.add_noise(embedding)
        
        # Update privacy budget
        query_cost = 0.1  # Cost of this query
        self.user_budgets[user_id] -= query_cost
        
        return private_embedding
    
    def _extract_features(self, user_data: Dict) -> List[float]:
        """Extract features from user data"""
        features = []
        
        # Numerical features with bounded sensitivity
        age = min(max(user_data.get('age', 30), 18), 100) / 100.0
        features.append(age)
        
        # Categorical features (one-hot)
        preferences = user_data.get('preferences', {})
        style_categories = ['casual', 'formal', 'technical', 'creative']
        style = preferences.get('style', 'casual')
        
        for category in style_categories:
            features.append(1.0 if style == category else 0.0)
        
        return features
    
    def private_gradient_update(self, 
                              gradients: List[torch.Tensor], 
                              clip_norm: float = 1.0) -> torch.Tensor:
        """Compute differentially private gradient update"""
        
        # Clip gradients to bound sensitivity
        clipped_gradients = []
        for grad in gradients:
            grad_norm = torch.norm(grad)
            if grad_norm > clip_norm:
                grad = grad * clip_norm / grad_norm
            clipped_gradients.append(grad)
        
        # Average gradients
        avg_gradient = torch.mean(torch.stack(clipped_gradients), dim=0)
        
        # Add noise for privacy
        private_gradient = self.add_noise(avg_gradient, sensitivity=clip_norm)
        
        return private_gradient

class FederatedPersonalization:
    """Federated learning for privacy-preserving personalization"""
    
    def __init__(self, global_model_config: Dict):
        self.global_model = self._initialize_global_model(global_model_config)
        self.client_models = {}
        self.aggregation_weights = {}
        
    def _initialize_global_model(self, config: Dict) -> nn.Module:
        """Initialize global model"""
        # Simplified model for demonstration
        return nn.Sequential(
            nn.Linear(config['input_dim'], config['hidden_dim']),
            nn.ReLU(),
            nn.Linear(config['hidden_dim'], config['output_dim'])
        )
    
    def add_client(self, client_id: str, local_data_size: int):
        """Add a new client to the federation"""
        # Clone global model for client
        client_model = type(self.global_model)()
        client_model.load_state_dict(self.global_model.state_dict())
        
        self.client_models[client_id] = client_model
        self.aggregation_weights[client_id] = local_data_size
    
    def local_training(self, 
                      client_id: str, 
                      local_data: List[Dict], 
                      epochs: int = 1,
                      learning_rate: float = 0.01) -> Dict:
        """Perform local training on client data"""
        
        if client_id not in self.client_models:
            raise ValueError(f"Client {client_id} not registered")
        
        client_model = self.client_models[client_id]
        optimizer = torch.optim.SGD(client_model.parameters(), lr=learning_rate)
        criterion = nn.MSELoss()
        
        # Local training loop
        for epoch in range(epochs):
            total_loss = 0
            for batch in local_data:
                inputs = torch.tensor(batch['features'], dtype=torch.float32)
                targets = torch.tensor(batch['targets'], dtype=torch.float32)
                
                optimizer.zero_grad()
                outputs = client_model(inputs)
                loss = criterion(outputs, targets)
                loss.backward()
                optimizer.step()
                
                total_loss += loss.item()
        
        # Return model updates (only send gradients, not raw data)
        updates = {}
        for name, param in client_model.named_parameters():
            updates[name] = param.data - self.global_model.state_dict()[name]
        
        return {
            'updates': updates,
            'loss': total_loss / len(local_data),
            'data_size': len(local_data)
        }
    
    def federated_averaging(self, client_updates: Dict[str, Dict]):
        """Perform federated averaging to update global model"""
        
        # Calculate weighted average of updates
        total_weight = sum(self.aggregation_weights[client_id] 
                          for client_id in client_updates.keys())
        
        aggregated_updates = {}
        
        # Initialize aggregated updates
        for name, param in self.global_model.named_parameters():
            aggregated_updates[name] = torch.zeros_like(param.data)
        
        # Weighted aggregation
        for client_id, update_data in client_updates.items():
            weight = self.aggregation_weights[client_id] / total_weight
            updates = update_data['updates']
            
            for name in aggregated_updates:
                aggregated_updates[name] += weight * updates[name]
        
        # Update global model
        with torch.no_grad():
            for name, param in self.global_model.named_parameters():
                param.data += aggregated_updates[name]
        
        # Update client models with new global model
        for client_model in self.client_models.values():
            client_model.load_state_dict(self.global_model.state_dict())

class SecureAggregation:
    """Secure multi-party computation for privacy-preserving aggregation"""
    
    def __init__(self, num_parties: int):
        self.num_parties = num_parties
        self.shares = {}
        
    def secret_share(self, value: torch.Tensor, party_id: str) -> List[torch.Tensor]:
        """Create secret shares of a value"""
        # Simplified secret sharing (Shamir's Secret Sharing in practice)
        shares = []
        
        # Generate random shares that sum to the original value
        for i in range(self.num_parties - 1):
            share = torch.rand_like(value)
            shares.append(share)
        
        # Last share ensures sum equals original value
        last_share = value - sum(shares)
        shares.append(last_share)
        
        return shares
    
    def aggregate_shares(self, all_shares: Dict[str, torch.Tensor]) -> torch.Tensor:
        """Aggregate secret shares from all parties"""
        
        if len(all_shares) != self.num_parties:
            raise ValueError(f"Expected {self.num_parties} shares, got {len(all_shares)}")
        
        # Sum all shares to recover the aggregated value
        aggregated = torch.zeros_like(list(all_shares.values())[0])
        for share in all_shares.values():
            aggregated += share
        
        return aggregated
    
    def secure_federated_learning(self, 
                                client_updates: Dict[str, torch.Tensor]) -> torch.Tensor:
        """Perform secure aggregation for federated learning"""
        
        # Each client creates secret shares of their update
        all_shares = {party: [] for party in range(self.num_parties)}
        
        for client_id, update in client_updates.items():
            shares = self.secret_share(update, client_id)
            for party_id, share in enumerate(shares):
                all_shares[party_id].append(share)
        
        # Each party aggregates their shares
        party_results = {}
        for party_id, shares in all_shares.items():
            party_results[party_id] = sum(shares)
        
        # Combine party results to get final aggregation
        final_result = self.aggregate_shares(party_results)
        
        return final_result

class PrivacyBudgetManager:
    """Manage privacy budgets across users and queries"""
    
    def __init__(self, total_epsilon: float = 1.0):
        self.total_epsilon = total_epsilon
        self.user_budgets = {}
        self.query_costs = {
            'embedding_generation': 0.1,
            'preference_update': 0.05,
            'model_training': 0.2,
            'inference': 0.01
        }
        
    def allocate_budget(self, user_id: str, initial_budget: float = None):
        """Allocate privacy budget to a user"""
        if initial_budget is None:
            initial_budget = self.total_epsilon
            
        self.user_budgets[user_id] = initial_budget
        
    def check_budget(self, user_id: str, query_type: str) -> bool:
        """Check if user has sufficient budget for a query"""
        if user_id not in self.user_budgets:
            return False
            
        cost = self.query_costs.get(query_type, 0.1)
        return self.user_budgets[user_id] >= cost
    
    def consume_budget(self, user_id: str, query_type: str) -> bool:
        """Consume privacy budget for a query"""
        if not self.check_budget(user_id, query_type):
            return False
            
        cost = self.query_costs.get(query_type, 0.1)
        self.user_budgets[user_id] -= cost
        return True
    
    def get_remaining_budget(self, user_id: str) -> float:
        """Get remaining privacy budget for a user"""
        return self.user_budgets.get(user_id, 0.0)
    
    def reset_budget(self, user_id: str):
        """Reset user's privacy budget (e.g., monthly reset)"""
        self.user_budgets[user_id] = self.total_epsilon

import asyncio
import torch
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import redis
import json
import hashlib
import logging

class PersonalizedGenAIService:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # Initialize components
        self.user_adaptive_model = UserAdaptiveModel(config.get('base_model'))
        self.prompt_engine = PersonalizedPromptEngine()
        self.privacy_manager = DifferentiallyPrivatePersonalization(
            epsilon=config.get('privacy_epsilon', 1.0)
        )
        self.budget_manager = PrivacyBudgetManager()
        
        # User data storage (encrypted)
        self.redis_client = redis.Redis(
            host=config.get('redis_host', 'localhost'),
            port=config.get('redis_port', 6379),
            decode_responses=False
        )
        
        # Performance tracking
        self.metrics = {
            'total_requests': 0,
            'personalized_requests': 0,
            'privacy_budget_exhausted': 0,
            'adaptation_time': 0
        }
        
    async def initialize_user(self, user_id: str, user_profile: Dict[str, Any]) -> Dict[str, Any]:
        """Initialize a new user with privacy-preserving profile"""
        try:
            # Check if user already exists
            if await self._user_exists(user_id):
                return {'status': 'user_already_exists', 'user_id': user_id}
            
            # Allocate privacy budget
            self.budget_manager.allocate_budget(user_id)
            
            # Create privacy-preserving user embedding
            if self.budget_manager.check_budget(user_id, 'embedding_generation'):
                user_embedding = self.privacy_manager.private_user_embedding(
                    user_id, user_profile
                )
                self.budget_manager.consume_budget(user_id, 'embedding_generation')
            else:
                # Use default embedding if budget insufficient
                user_embedding = torch.zeros(512)
            
            # Create user adapter
            adapter = self.user_adaptive_model.create_user_adapter(
                user_id, user_profile
            )
            
            # Store encrypted user data
            await self._store_user_data(user_id, {
                'profile': user_profile,
                'embedding': user_embedding.tolist(),
                'created_at': datetime.now().isoformat(),
                'privacy_budget': self.budget_manager.get_remaining_budget(user_id)
            })
            
            return {
                'status': 'success',
                'user_id': user_id,
                'privacy_budget_remaining': self.budget_manager.get_remaining_budget(user_id)
            }
            
        except Exception as e:
            logging.error(f"Failed to initialize user {user_id}: {e}")
            return {'status': 'error', 'message': str(e)}
    
    async def generate_personalized_response(self,
                                           user_id: str,
                                           prompt: str,
                                           domain: str = 'general',
                                           privacy_level: str = 'medium') -> Dict[str, Any]:
        """Generate personalized response with privacy protection"""
        try:
            start_time = asyncio.get_event_loop().time()
            self.metrics['total_requests'] += 1
            
            # Check if user exists
            if not await self._user_exists(user_id):
                # Use non-personalized response for new users
                return await self._generate_generic_response(prompt)
            
            # Check privacy budget
            if not self.budget_manager.check_budget(user_id, 'inference'):
                logging.warning(f"Privacy budget exhausted for user {user_id}")
                self.metrics['privacy_budget_exhausted'] += 1
                return await self._generate_generic_response(prompt)
            
            # Load user context
            user_data = await self._load_user_data(user_id)
            
            # Update user context with current interaction
            interaction_data = {
                'prompt': prompt,
                'domain': domain,
                'timestamp': datetime.now().isoformat(),
                'privacy_level': privacy_level
            }
            
            self.prompt_engine.update_user_context(user_id, interaction_data)
            
            # Generate personalized prompt
            personalized_prompt = self.prompt_engine.generate_personalized_prompt(
                user_id, prompt, domain
            )
            
            # Generate response using user adapter
            if privacy_level == 'high':
                # Use more privacy-preserving generation
                response = await self._generate_private_response(
                    user_id, personalized_prompt
                )
            else:
                # Standard personalized generation
                response = self.user_adaptive_model.personalized_generate(
                    user_id, personalized_prompt
                )
            
            # Consume privacy budget
            self.budget_manager.consume_budget(user_id, 'inference')
            
            # Update metrics
            end_time = asyncio.get_event_loop().time()
            self.metrics['personalized_requests'] += 1
            self.metrics['adaptation_time'] += (end_time - start_time)
            
            # Store interaction for learning
            await self._store_interaction(user_id, {
                'prompt': prompt,
                'response': response,
                'personalized_prompt': personalized_prompt,
                'domain': domain,
                'timestamp': datetime.now().isoformat()
            })
            
            return {
                'response': response,
                'personalized': True,
                'privacy_budget_remaining': self.budget_manager.get_remaining_budget(user_id),
                'generation_time_ms': int((end_time - start_time) * 1000)
            }
            
        except Exception as e:
            logging.error(f"Failed to generate personalized response: {e}")
            return await self._generate_generic_response(prompt)
    
    async def update_user_preferences(self,
                                    user_id: str,
                                    feedback: Dict[str, Any]) -> Dict[str, Any]:
        """Update user preferences based on feedback"""
        try:
            # Check privacy budget
            if not self.budget_manager.check_budget(user_id, 'preference_update'):
                return {
                    'status': 'error',
                    'message': 'Insufficient privacy budget for preference update'
                }
            
            # Load current user data
            user_data = await self._load_user_data(user_id)
            
            # Update preferences based on feedback
            if 'rating' in feedback:
                # Explicit feedback
                rating = feedback['rating']
                response_id = feedback.get('response_id')
                
                # Update user model based on rating
                if rating >= 4:  # Positive feedback
                    await self._reinforce_preferences(user_id, feedback)
                elif rating <= 2:  # Negative feedback
                    await self._adjust_preferences(user_id, feedback)
            
            if 'style_preference' in feedback:
                # Style preference update
                new_style = feedback['style_preference']
                user_data['profile']['preferences']['style'] = new_style
                
                # Update user embedding
                updated_embedding = self.privacy_manager.private_user_embedding(
                    user_id, user_data['profile']
                )
                user_data['embedding'] = updated_embedding.tolist()
            
            # Consume privacy budget
            self.budget_manager.consume_budget(user_id, 'preference_update')
            
            # Store updated user data
            await self._store_user_data(user_id, user_data)
            
            return {
                'status': 'success',
                'privacy_budget_remaining': self.budget_manager.get_remaining_budget(user_id)
            }
            
        except Exception as e:
            logging.error(f"Failed to update preferences for user {user_id}: {e}")
            return {'status': 'error', 'message': str(e)}
    
    async def _generate_private_response(self, user_id: str, prompt: str) -> str:
        """Generate response with enhanced privacy protection"""
        
        # Use differential privacy for response generation
        base_response = self.user_adaptive_model.personalized_generate(
            user_id, prompt, temperature=0.8
        )
        
        # Apply privacy-preserving post-processing
        # 1. Remove potential personally identifiable information
        sanitized_response = self._sanitize_response(base_response)
        
        # 2. Add calibrated noise to response features (simplified)
        # In practice, this would involve more sophisticated techniques
        
        return sanitized_response
    
    def _sanitize_response(self, response: str) -> str:
        """Remove potentially sensitive information from response"""
        import re
        
        # Remove potential email addresses
        response = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]', response)
        
        # Remove potential phone numbers
        response = re.sub(r'\b\d{3}-\d{3}-\d{4}\b', '[PHONE]', response)
        
        # Remove potential names (simplified)
        response = re.sub(r'\bMr\.|Mrs\.|Ms\.|Dr\.', '[TITLE]', response)
        
        return response
    
    async def _generate_generic_response(self, prompt: str) -> Dict[str, Any]:
        """Generate non-personalized response"""
        # Use base model without personalization
        base_model = self.user_adaptive_model.base_model
        inputs = self.user_adaptive_model.tokenizer(prompt, return_tensors="pt")
        
        with torch.no_grad():
            outputs = base_model.generate(**inputs, max_new_tokens=512)
        
        response = self.user_adaptive_model.tokenizer.decode(
            outputs[0], skip_special_tokens=True
        )
        
        return {
            'response': response,
            'personalized': False,
            'message': 'Generic response - no personalization applied'
        }
    
    async def _user_exists(self, user_id: str) -> bool:
        """Check if user exists in the system"""
        user_key = self._get_user_key(user_id)
        return bool(await asyncio.get_event_loop().run_in_executor(
            None, self.redis_client.exists, user_key
        ))
    
    async def _store_user_data(self, user_id: str, user_data: Dict[str, Any]):
        """Store encrypted user data"""
        user_key = self._get_user_key(user_id)
        encrypted_data = self._encrypt_user_data(json.dumps(user_data))
        
        await asyncio.get_event_loop().run_in_executor(
            None, self.redis_client.set, user_key, encrypted_data
        )
        
        # Set expiration for data retention compliance
        retention_days = self.config.get('data_retention_days', 365)
        await asyncio.get_event_loop().run_in_executor(
            None, self.redis_client.expire, user_key, retention_days * 24 * 3600
        )
    
    async def _load_user_data(self, user_id: str) -> Dict[str, Any]:
        """Load and decrypt user data"""
        user_key = self._get_user_key(user_id)
        encrypted_data = await asyncio.get_event_loop().run_in_executor(
            None, self.redis_client.get, user_key
        )
        
        if encrypted_data:
            decrypted_data = self._decrypt_user_data(encrypted_data)
            return json.loads(decrypted_data)
        else:
            raise ValueError(f"No data found for user {user_id}")
    
    def _get_user_key(self, user_id: str) -> str:
        """Generate Redis key for user data"""
        # Hash user ID for additional privacy
        hashed_id = hashlib.sha256(user_id.encode()).hexdigest()
        return f"user_data:{hashed_id}"
    
    def _encrypt_user_data(self, data: str) -> bytes:
        """Encrypt user data (simplified - use proper encryption in production)"""
        # This is a placeholder - implement proper encryption
        return data.encode('utf-8')
    
    def _decrypt_user_data(self, encrypted_data: bytes) -> str:
        """Decrypt user data (simplified - use proper decryption in production)"""
        # This is a placeholder - implement proper decryption
        return encrypted_data.decode('utf-8')
    
    async def get_user_insights(self, user_id: str) -> Dict[str, Any]:
        """Get privacy-preserving insights about user"""
        if not await self._user_exists(user_id):
            return {'error': 'User not found'}
        
        # Get anonymized insights
        insights = self.prompt_engine.get_user_insights(user_id)
        
        # Remove sensitive information
        safe_insights = {
            'total_interactions': insights.get('total_interactions', 0),
            'engagement_level': insights.get('engagement_level', 'unknown'),
            'most_common_domain': insights.get('most_common_domain', 'general'),
            'privacy_budget_remaining': self.budget_manager.get_remaining_budget(user_id)
        }
        
        return safe_insights
    
    def get_service_metrics(self) -> Dict[str, Any]:
        """Get service-level metrics"""
        total_requests = self.metrics['total_requests']
        personalized_requests = self.metrics['personalized_requests']
        
        return {
            'total_requests': total_requests,
            'personalized_requests': personalized_requests,
            'personalization_rate': personalized_requests / max(total_requests, 1),
            'privacy_budget_exhausted': self.metrics['privacy_budget_exhausted'],
            'avg_adaptation_time_ms': int(
                (self.metrics['adaptation_time'] / max(personalized_requests, 1)) * 1000
            ),
            'active_users': len(self.user_adaptive_model.user_adapters)
        }

# Personalized GenAI Service Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: personalized-genai-service
  labels:
    app: personalized-genai
spec:
  replicas: 3
  selector:
    matchLabels:
      app: personalized-genai
  template:
    metadata:
      labels:
        app: personalized-genai
    spec:
      containers:
      - name: genai-service
        image: ml-platform/personalized-genai:v2.0.0
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "8Gi"
            cpu: "2"
            nvidia.com/gpu: "1"
          limits:
            memory: "16Gi"
            cpu: "4"
            nvidia.com/gpu: "1"
        env:
        - name: PRIVACY_EPSILON
          value: "1.0"
        - name: DATA_RETENTION_DAYS
          value: "365"
        - name: ENCRYPTION_KEY
          valueFrom:
            secretKeyRef:
              name: encryption-secret
              key: key
        - name: REDIS_HOST
          value: "redis-cluster"
        - name: MODEL_CACHE_DIR
          value: "/models"
        volumeMounts:
        - name: model-storage
          mountPath: /models
        - name: encryption-keys
          mountPath: /keys
          readOnly: true
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 120
          periodSeconds: 60
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 30
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-storage-pvc
      - name: encryption-keys
        secret:
          secretName: encryption-keys
      nodeSelector:
        privacy-compliant: "true"
        gpu-type: "a100"

---
# Privacy-compliant Redis with encryption
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: redis-encrypted
spec:
  serviceName: "redis"
  replicas: 3
  selector:
    matchLabels:
      app: redis-encrypted
  template:
    metadata:
      labels:
        app: redis-encrypted
    spec:
      containers:
      - name: redis
        image: redis:7-alpine
        ports:
        - containerPort: 6379
        command:
        - redis-server
        - --requirepass
        - $(REDIS_PASSWORD)
        - --save
        - "900 1"
        - --save
        - "300 10"
        env:
        - name: REDIS_PASSWORD
          valueFrom:
            secretKeyRef:
              name: redis-secret
              key: password
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        volumeMounts:
        - name: redis-storage
          mountPath: /data
  volumeClaimTemplates:
  - metadata:
      name: redis-storage
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 20Gi

---
# Privacy Budget Monitor
apiVersion: batch/v1
kind: CronJob
metadata:
  name: privacy-budget-reset
spec:
  schedule: "0 0 1 * *"  # Monthly reset
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: budget-reset
            image: ml-platform/privacy-tools:v1.0.0
            command:
            - python
            - reset_privacy_budgets.py
            env:
            - name: REDIS_HOST
              value: "redis-cluster"
          restartPolicy: OnFailure

---
# Data Retention Job
apiVersion: batch/v1
kind: CronJob
metadata:
  name: data-retention-cleanup
spec:
  schedule: "0 2 * * *"  # Daily cleanup
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: data-cleanup
            image: ml-platform/privacy-tools:v1.0.0
            command:
            - python
            - cleanup_expired_data.py
            env:
            - name: REDIS_HOST
              value: "redis-cluster"
            - name: RETENTION_DAYS
              value: "365"
          restartPolicy: OnFailure