Spaces:
Runtime error
Runtime error
File size: 11,977 Bytes
eeb0f9c |
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 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 |
"""
Personalization Engine - Progressive learning from user interactions
Continuously improves recommendations based on user data and feedback
"""
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
from collections import Counter
import json
from health_data import HealthContext, UserPreferences
class PersonalizationEngine:
"""
Progressive personalization engine that learns from user interactions
Improves recommendations over time based on accumulated user data
"""
def __init__(self, health_context: HealthContext):
self.health_context = health_context
self.user_id = health_context.user_id
# ===== Pattern Analysis =====
def analyze_user_patterns(self, days: int = 90) -> Dict[str, Any]:
"""Analyze user interaction patterns"""
history = self.health_context.get_health_history(days)
if not history:
return {
'total_interactions': 0,
'interaction_types': {},
'most_common_topics': [],
'engagement_level': 'low'
}
# Count interactions by type
type_counts = Counter(r.record_type for r in history)
# Calculate engagement level
total_interactions = len(history)
days_active = len(set(r.timestamp.date() for r in history))
engagement_score = min(total_interactions / (days / 7), 1.0) # Normalize to 0-1
engagement_level = 'high' if engagement_score > 0.7 else 'medium' if engagement_score > 0.3 else 'low'
return {
'total_interactions': total_interactions,
'days_active': days_active,
'interaction_types': dict(type_counts),
'engagement_score': round(engagement_score, 2),
'engagement_level': engagement_level,
'most_common_topics': [t[0] for t in type_counts.most_common(3)]
}
def extract_preferences(self) -> UserPreferences:
"""Extract and update user preferences from interaction history"""
prefs = self.health_context.get_preferences()
# Analyze exercise preferences from history
exercise_records = self.health_context.get_records_by_type('exercise')
if exercise_records:
# Extract exercise types mentioned
exercise_types = set()
for record in exercise_records[-10:]: # Last 10 exercise records
if 'exercise_type' in record.data:
exercise_types.add(record.data['exercise_type'])
prefs.preferred_exercise_types = list(exercise_types)
# Analyze nutrition preferences
nutrition_records = self.health_context.get_records_by_type('nutrition')
if nutrition_records:
dietary_prefs = set()
for record in nutrition_records[-10:]:
if 'dietary_preference' in record.data:
dietary_prefs.add(record.data['dietary_preference'])
prefs.dietary_preferences = list(dietary_prefs)
# Analyze goals from history
goals = set()
for record in self.health_context.get_health_history(days=180):
if 'goal' in record.data:
goals.add(record.data['goal'])
prefs.goals = list(goals)
self.health_context.update_preferences(
preferred_exercise_types=prefs.preferred_exercise_types,
dietary_preferences=prefs.dietary_preferences,
goals=prefs.goals
)
return prefs
def identify_health_trends(self, days: int = 90) -> Dict[str, Any]:
"""Identify health trends from historical data"""
trends = {
'symptom_frequency': {},
'health_improvements': [],
'health_concerns': [],
'activity_trends': {}
}
# Analyze symptom frequency
symptom_records = self.health_context.get_records_by_type('symptom')
symptom_counts = Counter()
for record in symptom_records:
if 'symptom' in record.data:
symptom_counts[record.data['symptom']] += 1
trends['symptom_frequency'] = dict(symptom_counts.most_common(5))
# Analyze fitness trends
fitness_history = self.health_context.get_fitness_history(days)
if fitness_history:
total_workouts = len(fitness_history)
total_minutes = sum(f.duration_minutes for f in fitness_history)
avg_intensity = sum(1 for f in fitness_history if f.intensity == 'high') / total_workouts if total_workouts > 0 else 0
trends['activity_trends'] = {
'total_workouts': total_workouts,
'total_minutes': total_minutes,
'avg_intensity': round(avg_intensity, 2),
'adherence': self.health_context.get_workout_adherence(days)
}
return trends
def calculate_engagement_score(self, days: int = 30) -> float:
"""Calculate user engagement score (0-1)"""
patterns = self.analyze_user_patterns(days)
return patterns['engagement_score']
# ===== Adaptation Methods =====
def adapt_nutrition_plan(self, current_plan: Dict[str, Any]) -> Dict[str, Any]:
"""Adapt nutrition plan based on user history"""
adapted_plan = current_plan.copy()
# Get user preferences
prefs = self.health_context.get_preferences()
# Apply dietary restrictions
if prefs.dietary_preferences:
adapted_plan['dietary_preferences'] = prefs.dietary_preferences
# Analyze nutrition history for effectiveness
nutrition_records = self.health_context.get_records_by_type('nutrition')
if nutrition_records:
# Check if user is following recommendations
adherence = len(nutrition_records) / max(1, (30 / 7)) # Expected ~1 per week
adapted_plan['adherence_score'] = min(adherence, 1.0)
# Add personalization note
adapted_plan['personalized'] = True
adapted_plan['personalization_date'] = datetime.now().isoformat()
return adapted_plan
def adapt_exercise_plan(self, current_plan: Dict[str, Any]) -> Dict[str, Any]:
"""Adapt exercise plan based on progress"""
adapted_plan = current_plan.copy()
# Get fitness history
fitness_history = self.health_context.get_fitness_history(days=30)
if fitness_history:
# Calculate adherence
adherence = self.health_context.get_workout_adherence(days=30)
# Adjust difficulty based on adherence
if adherence > 0.8:
adapted_plan['difficulty_adjustment'] = 'increase'
adapted_plan['recommendation'] = 'Great adherence! Consider increasing intensity.'
elif adherence < 0.3:
adapted_plan['difficulty_adjustment'] = 'decrease'
adapted_plan['recommendation'] = 'Let\'s make the plan more manageable.'
else:
adapted_plan['difficulty_adjustment'] = 'maintain'
adapted_plan['recommendation'] = 'Keep up the good work!'
# Add exercise preferences
prefs = self.health_context.get_preferences()
if prefs.preferred_exercise_types:
adapted_plan['preferred_exercises'] = prefs.preferred_exercise_types
adapted_plan['personalized'] = True
adapted_plan['personalization_date'] = datetime.now().isoformat()
return adapted_plan
def adapt_communication_style(self) -> str:
"""Adapt communication style based on user interactions"""
patterns = self.analyze_user_patterns(days=30)
# Analyze interaction frequency
if patterns['engagement_level'] == 'high':
return 'detailed' # More detailed responses
elif patterns['engagement_level'] == 'low':
return 'brief' # Shorter, more concise responses
else:
return 'balanced' # Standard responses
def generate_personalized_insights(self) -> List[str]:
"""Generate personalized health insights"""
insights = []
# Analyze trends
trends = self.identify_health_trends(days=90)
# Symptom insights
if trends['symptom_frequency']:
top_symptom = list(trends['symptom_frequency'].keys())[0]
count = trends['symptom_frequency'][top_symptom]
insights.append(f"You've reported '{top_symptom}' {count} times in the last 90 days. Consider consulting a specialist.")
# Activity insights
if 'activity_trends' in trends and trends['activity_trends']:
activity = trends['activity_trends']
if activity['adherence'] > 0.7:
insights.append(f"Excellent fitness adherence! You've completed {activity['total_workouts']} workouts in the last month.")
elif activity['adherence'] < 0.3:
insights.append("Your fitness adherence is low. Let's create a more achievable plan together.")
# Goal progress
prefs = self.health_context.get_preferences()
if prefs.goals:
insights.append(f"Your current goals: {', '.join(prefs.goals)}")
return insights
# ===== Feedback Methods =====
def record_user_feedback(self, feedback_type: str, feedback_data: Dict[str, Any]) -> None:
"""Record user feedback for learning"""
feedback_record = {
'feedback_type': feedback_type, # helpful/not_helpful/confusing/etc
'data': feedback_data,
'timestamp': datetime.now().isoformat()
}
self.health_context.add_health_record(
'feedback',
feedback_record,
agent_name='personalization_engine',
confidence=1.0
)
def update_preferences_from_feedback(self) -> None:
"""Update preferences based on accumulated feedback"""
# Get recent feedback
feedback_records = self.health_context.get_records_by_type('feedback')
if not feedback_records:
return
# Analyze feedback patterns
helpful_count = sum(1 for r in feedback_records[-20:] if r.data.get('feedback_type') == 'helpful')
total_feedback = min(len(feedback_records), 20)
if total_feedback > 0:
helpfulness_score = helpful_count / total_feedback
# Update communication style if needed
if helpfulness_score < 0.3:
self.health_context.update_preferences(
communication_style='brief'
)
elif helpfulness_score > 0.7:
self.health_context.update_preferences(
communication_style='detailed'
)
def get_personalization_summary(self) -> Dict[str, Any]:
"""Get summary of personalization status"""
patterns = self.analyze_user_patterns()
trends = self.identify_health_trends()
prefs = self.extract_preferences()
return {
'engagement': patterns,
'trends': trends,
'preferences': {
'goals': prefs.goals,
'exercise_types': prefs.preferred_exercise_types,
'dietary_preferences': prefs.dietary_preferences,
'communication_style': prefs.communication_style
},
'insights': self.generate_personalized_insights(),
'last_updated': datetime.now().isoformat()
}
|