Coverage for astrocyte/pipeline/utility.py: 99%

1"""Utility scoring — per-memory usage tracking and composite scoring. 

2 

3Combines recency, frequency, relevance, and freshness into a 0-1 score 

4that drives TTL decisions, recall ranking boosts, and bank health metrics. 

5 

6Sync, self-contained — Rust migration candidate. 

7Inspired by ByteRover's lifecycle metadata and Hindsight's consolidation quality. 

8""" 

9 

10from __future__ import annotations 

11 

12import math 

13import time 

14from collections import OrderedDict 

15from dataclasses import dataclass 

16 

17 

18@dataclass 

19class UtilityScore: 

20 """Composite utility score with individual components.""" 

21 

22 recency: float # 0-1: how recently was this memory recalled 

23 frequency: float # 0-1: how often is it recalled 

24 relevance: float # 0-1: average relevance score when recalled 

25 freshness: float # 0-1: how new is the memory 

26 composite: float # 0-1: weighted combination 

27 

28 

29@dataclass 

30class _MemoryStats: 

31 recall_count: int = 0 

32 last_recalled_at: float = 0.0 # monotonic time 

33 total_relevance: float = 0.0 

34 created_at: float = 0.0 # monotonic time 

35 

36 

37def compute_utility( 

38 recall_count: int, 

39 last_recalled_seconds_ago: float, 

40 avg_relevance: float, 

41 created_seconds_ago: float, 

42 *, 

43 recency_half_life_days: float = 7.0, 

44 max_frequency: int = 100, 

45 weight_recency: float = 0.3, 

46 weight_frequency: float = 0.2, 

47 weight_relevance: float = 0.3, 

48 weight_freshness: float = 0.2, 

49) -> UtilityScore: 

50 """Compute composite utility score for a memory. 

51 

52 All inputs are non-negative. Returns UtilityScore with components in [0, 1]. 

53 Sync, pure computation — Rust migration candidate. 

54 """ 

55 half_life_seconds = recency_half_life_days * 86400.0 

56 

57 # Recency: exponential decay from last recall (1.0 = just recalled, decays to 0) 

58 if last_recalled_seconds_ago <= 0: 

59 recency = 1.0 

60 else: 

61 recency = math.exp(-0.693 * last_recalled_seconds_ago / max(half_life_seconds, 1.0)) 

62 

63 # Frequency: normalized recall count (capped at max_frequency) 

64 frequency = min(recall_count / max(max_frequency, 1), 1.0) 

65 

66 # Relevance: average score when recalled (already 0-1) 

67 relevance = max(0.0, min(1.0, avg_relevance)) 

68 

69 # Freshness: how new the memory is (exponential decay from creation) 

70 if created_seconds_ago <= 0: 

71 freshness = 1.0 

72 else: 

73 freshness = math.exp(-0.693 * created_seconds_ago / max(half_life_seconds * 4, 1.0)) 

74 

75 # Composite: weighted sum 

76 composite = ( 

77 weight_recency * recency 

78 + weight_frequency * frequency 

79 + weight_relevance * relevance 

80 + weight_freshness * freshness 

81 ) 

82 # Normalize to [0, 1] 

83 total_weight = weight_recency + weight_frequency + weight_relevance + weight_freshness 

84 if total_weight > 0: 

85 composite /= total_weight 

86 

87 return UtilityScore( 

88 recency=recency, 

89 frequency=frequency, 

90 relevance=relevance, 

91 freshness=freshness, 

92 composite=composite, 

93 ) 

94 

95 

96class UtilityTracker: 

97 """Per-memory usage tracking for utility scoring. 

98 

99 Maintains recall counts, timestamps, and relevance scores in memory. 

100 LRU eviction when over capacity. 

101 

102 Sync, self-contained — Rust migration candidate. 

103 """ 

104 

105 def __init__( 

106 self, 

107 max_entries: int = 10000, 

108 recency_half_life_days: float = 7.0, 

109 ) -> None: 

110 self.max_entries = max_entries 

111 self.recency_half_life_days = recency_half_life_days 

112 self._stats: OrderedDict[str, _MemoryStats] = OrderedDict() # memory_id → stats (LRU order) 

113 

114 def record_recall(self, memory_id: str, relevance_score: float) -> None: 

115 """Record that a memory was recalled with a given relevance score.""" 

116 now = time.monotonic() 

117 

118 if memory_id not in self._stats: 

119 self._stats[memory_id] = _MemoryStats(created_at=now) 

120 

121 stats = self._stats[memory_id] 

122 stats.recall_count += 1 

123 stats.last_recalled_at = now 

124 stats.total_relevance += relevance_score 

125 

126 # Move to end (most recently used) — O(1) with OrderedDict 

127 self._stats.move_to_end(memory_id) 

128 

129 # Evict LRU if over capacity 

130 while len(self._stats) > self.max_entries: 

131 self._stats.popitem(last=False) 

132 

133 def record_creation(self, memory_id: str) -> None: 

134 """Record that a new memory was created.""" 

135 now = time.monotonic() 

136 self._stats[memory_id] = _MemoryStats(created_at=now) 

137 

138 # Move to end (most recently used) — O(1) with OrderedDict 

139 self._stats.move_to_end(memory_id) 

140 

141 # Evict LRU if over capacity 

142 while len(self._stats) > self.max_entries: 

143 self._stats.popitem(last=False) 

144 

145 def get_utility(self, memory_id: str) -> UtilityScore | None: 

146 """Compute current utility score for a memory. Returns None if not tracked.""" 

147 stats = self._stats.get(memory_id) 

148 if stats is None: 

149 return None 

150 

151 now = time.monotonic() 

152 avg_relevance = stats.total_relevance / max(stats.recall_count, 1) 

153 last_recalled_ago = now - stats.last_recalled_at if stats.last_recalled_at > 0 else now - stats.created_at 

154 created_ago = now - stats.created_at 

155 

156 return compute_utility( 

157 recall_count=stats.recall_count, 

158 last_recalled_seconds_ago=last_recalled_ago, 

159 avg_relevance=avg_relevance, 

160 created_seconds_ago=created_ago, 

161 recency_half_life_days=self.recency_half_life_days, 

162 ) 

163 

164 def get_recall_count(self, memory_id: str) -> int: 

165 """Get the recall count for a memory.""" 

166 stats = self._stats.get(memory_id) 

167 return stats.recall_count if stats else 0 

168 

169 def clear(self) -> None: 

170 """Clear all tracked stats.""" 

171 self._stats.clear()