Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-memory/src/lexigram/ai/memory/pruning/scorer.py: 39%

1"""Scoring strategies for context pruning.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Protocol, runtime_checkable 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.ai.memory import MemoryEntry 

9 

10 

11@runtime_checkable 

12class RelevanceScorerProtocol(Protocol): 

13 """Scores a MemoryEntry for pruning priority (higher = keep).""" 

14 

15 def score(self, entry: MemoryEntry, query: str | None = None) -> float: 

16 """Score a memory entry for retention. 

17 

18 Args: 

19 entry: The memory entry to score. 

20 query: Optional query context for relevance scoring. 

21 

22 Returns: 

23 A score in range [0, 1] or higher, where higher means more important to keep. 

24 """ 

25 ... 

26 

27 

28class RecencyScorerImpl: 

29 """Scores memory entries by recency — more recent entries get higher scores. 

30 

31 Uses the entry's timestamp to produce a normalized score relative to the 

32 entire batch of entries being pruned. 

33 """ 

34 

35 def score(self, entry: MemoryEntry, query: str | None = None) -> float: 

36 """Score entry by recency. 

37 

38 Args: 

39 entry: The memory entry to score. 

40 query: Optional query (unused by RecencyScorerImpl). 

41 

42 Returns: 

43 Score based on entry's timestamp (normalized in batch context). 

44 """ 

45 ts = getattr(entry, "timestamp", None) or getattr(entry, "created_at", None) 

46 return ts.timestamp() if ts is not None else 0.0 

47 

48 

49class HybridScorerImpl: 

50 """Weighted blend of recency and content length as a proxy for relevance. 

51 

52 Content length serves as a simple heuristic for information density: longer 

53 entries are assumed to contain more contextual information. 

54 

55 Attributes: 

56 recency_weight: Weight for the recency component (default 0.6). 

57 relevance_weight: Weight for the content length component (default 0.4). 

58 """ 

59 

60 def __init__( 

61 self, recency_weight: float = 0.6, relevance_weight: float = 0.4 

62 ) -> None: 

63 """Initialize the hybrid scorer. 

64 

65 Args: 

66 recency_weight: Weight for recency in blended score. Default 0.6. 

67 relevance_weight: Weight for content length (relevance proxy). Default 0.4. 

68 """ 

69 self._recency_weight = recency_weight 

70 self._relevance_weight = relevance_weight 

71 

72 def score_batch( 

73 self, 

74 entries: list, 

75 query: str | None = None, 

76 ) -> list[float]: 

77 """Score all entries together, normalizing recency to [0, 1]. 

78 

79 Args: 

80 entries: List of memory entries to score. 

81 query: Optional query context (unused by HybridScorerImpl). 

82 

83 Returns: 

84 List of scores parallel to the input entries list. 

85 """ 

86 # Get timestamps (use 0.0 as fallback for missing timestamps) 

87 timestamps = [] 

88 for entry in entries: 

89 ts = getattr(entry, "timestamp", None) or getattr(entry, "created_at", None) 

90 timestamps.append(ts.timestamp() if ts is not None else 0.0) 

91 

92 min_ts = min(timestamps) if timestamps else 0.0 

93 max_ts = max(timestamps) if timestamps else 0.0 

94 ts_range = max_ts - min_ts or 1.0 # avoid division by zero 

95 

96 scores = [] 

97 for entry, raw_ts in zip(entries, timestamps, strict=True): 

98 recency_score = (raw_ts - min_ts) / ts_range # normalized [0, 1] 

99 length_score = min( 

100 len(str(getattr(entry, "content", "") or "")) / 1000, 1.0 

101 ) 

102 scores.append( 

103 self._recency_weight * recency_score 

104 + self._relevance_weight * length_score 

105 ) 

106 return scores 

107 

108 def score(self, entry: MemoryEntry, query: str | None = None) -> float: 

109 """Score a single entry (recency not normalized — use score_batch for batches). 

110 

111 Args: 

112 entry: The memory entry to score. 

113 query: Optional query (unused by HybridScorerImpl). 

114 

115 Returns: 

116 Weighted score of content length (recency omitted without batch context). 

117 """ 

118 # Length score: normalize to [0, 1], with 1000 chars = max score 

119 # Entries with more content are assumed more important 

120 length_score = min(len(entry.content) / 1000.0, 1.0) 

121 

122 # Return only relevance component (recency requires batch context) 

123 return self._relevance_weight * length_score 

124 

125 

126__all__ = ["HybridScorerImpl", "RecencyScorerImpl", "RelevanceScorerProtocol"]