""" ForgettingCurve Calculates memory decay based on Ebbinghaus forgetting curve. decay_score = exp(-days_since_access / half_life) Importance level affects half-life: - high: half_life = 30 * 3 = 90 days (slowest decay) - medium: half_life = 30 * 1 = 30 days - low: half_life = 30 * 0.5 = 15 days (fastest decay) """ import math from datetime import UTC, datetime from typing import TYPE_CHECKING if TYPE_CHECKING: from app.models.memory import UserMemory class ForgettingCurve: """Calculate memory decay based on time and importance.""" BASE_HALF_LIFE_DAYS = 30 # Half-life multipliers by importance level HALF_LIFE_MULTIPLIERS = { "high": 3.0, "medium": 1.0, "low": 0.5, } def calculate_decay(self, memory: "UserMemory") -> float: """Calculate decay score (0.0-1.0). Higher = more remembered. Uses exponential decay: exp(-days_since_access / half_life) """ last_accessed = getattr(memory, "last_accessed_at", None) or getattr( memory, "last_recalled_at", None ) if last_accessed is None: return 1.0 # Never accessed = full retention now = datetime.now(UTC) if isinstance(last_accessed, datetime): if last_accessed.tzinfo is None: last_accessed = last_accessed.replace(tzinfo=UTC) days_since = (now - last_accessed).total_seconds() / 86400 else: days_since = 0 half_life = self.get_half_life(memory) decay = math.exp(-days_since / half_life) return min(1.0, max(0.0, decay)) def get_half_life(self, memory: "UserMemory") -> float: """Get half-life in days based on importance level.""" importance_level = getattr(memory, "importance_level", "medium") or "medium" multiplier = self.HALF_LIFE_MULTIPLIERS.get(importance_level, 1.0) return self.BASE_HALF_LIFE_DAYS * multiplier def should_archive(self, memory: "UserMemory") -> bool: """decay < 0.2 → memory should be archived (cold storage).""" decay = self.calculate_decay(memory) return decay < 0.2 def should_deprioritize(self, memory: "UserMemory") -> bool: """decay < 0.5 → memory should be deprioritized (not in active reminders).""" decay = self.calculate_decay(memory) return decay < 0.5