""" FrequencyTracker Tracks how often a memory is recalled and calculates frequency/recency scores. """ from datetime import UTC, datetime from typing import TYPE_CHECKING if TYPE_CHECKING: from app.models.memory import UserMemory class FrequencyTracker: """Track and score memory recall frequency""" # Score weights MAX_FREQUENCY = 10 # Cap frequency count for scoring RECENCY_DECAY_DAYS = 30 # After 30 days, recency score drops significantly def increment(self, memory: "UserMemory") -> "UserMemory": """Increment recall count and update last recalled timestamp""" memory.frequency_count = (memory.frequency_count or 0) + 1 memory.last_recalled_at = datetime.now(UTC) return memory def get_frequency_score(self, memory: "UserMemory") -> float: """Calculate normalized frequency score (0.0 - 1.0) Uses logarithmic scaling to prevent high-frequency memories from dominating completely. """ count = memory.frequency_count or 0 if count == 0: return 0.0 # Logarithmic scaling: more recalls have diminishing returns # log(1+x) / log(1+MAX) gives 0-1 range import math score = math.log(1 + count) / math.log(1 + self.MAX_FREQUENCY) return min(1.0, max(0.0, score)) def get_recency_score(self, memory: "UserMemory") -> float: """Calculate recency score (0.0 - 1.0) Memory recalled recently scores higher. Uses exponential decay. """ last_recalled = memory.last_recalled_at if last_recalled is None: return 0.0 now = datetime.now(UTC) if isinstance(last_recalled, datetime): if last_recalled.tzinfo is None: last_recalled = last_recalled.replace(tzinfo=UTC) days_since = (now - last_recalled).total_seconds() / 86400 else: days_since = self.RECENCY_DECAY_DAYS # Exponential decay: half-life of RECENCY_DECAY_DAYS import math decay = math.exp(-days_since / self.RECENCY_DECAY_DAYS) return min(1.0, max(0.0, decay)) def get_time_decay(self, memory: "UserMemory") -> float: """Calculate time-based decay factor for forgetting curve""" last_accessed = getattr(memory, "last_accessed_at", None) if last_accessed is None: last_accessed = memory.last_recalled_at if last_accessed is None: return 1.0 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 import math return math.exp(-days_since / self.RECENCY_DECAY_DAYS)