« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Consolidation strategies — algorithms for pruning and merging memory entries."""
3from __future__ import annotations
5from datetime import UTC, datetime
6import math
8from lexigram.contracts.ai.memory import MemoryEntry
9from lexigram.logging import (
10 get_logger,
11)
13logger = get_logger(__name__)
16class RecencyDecayStrategy:
17 """Prunes entries whose recency score falls below a threshold.
19 Uses an exponential decay model with a configurable half-life.
20 """
22 def __init__(self, half_life_hours: float = 24.0, threshold: float = 0.05) -> None:
23 """Initialise the recency decay strategy.
25 Args:
26 half_life_hours: Time (h) for importance to halve.
27 threshold: Entries with recency below this are pruned.
28 """
29 self._half_life_s = half_life_hours * 3600.0
30 self._threshold = threshold
32 def should_prune(self, entry: MemoryEntry) -> bool:
33 """Return True if *entry* should be pruned.
35 Args:
36 entry: Entry to evaluate.
38 Returns:
39 True if the entry's recency score is below threshold.
40 """
41 age_s = (datetime.now(UTC) - entry.timestamp).total_seconds()
42 recency = math.exp(-math.log(2) * age_s / self._half_life_s)
43 return recency < self._threshold
45 def filter(
46 self, entries: list[MemoryEntry]
47 ) -> tuple[list[MemoryEntry], list[MemoryEntry]]:
48 """Split entries into (kept, pruned).
50 Args:
51 entries: Entries to evaluate.
53 Returns:
54 Tuple of (kept, pruned) entry lists.
55 """
56 kept: list[MemoryEntry] = []
57 pruned: list[MemoryEntry] = []
58 for e in entries:
59 (pruned if self.should_prune(e) else kept).append(e)
60 return kept, pruned
63class AccessFrequencyStrategy:
64 """Preserves high-importance entries; prunes low-importance stale ones."""
66 def __init__(self, importance_threshold: float = 0.1) -> None:
67 """Initialise the importance threshold strategy.
69 Args:
70 importance_threshold: Entries below this value are candidates for pruning.
71 """
72 self._threshold = importance_threshold
74 def should_prune(self, entry: MemoryEntry) -> bool:
75 """Return True if *entry* importance is below threshold.
77 Args:
78 entry: Entry to evaluate.
80 Returns:
81 True if importance is below threshold.
82 """
83 return entry.importance < self._threshold
85 def filter(
86 self, entries: list[MemoryEntry]
87 ) -> tuple[list[MemoryEntry], list[MemoryEntry]]:
88 """Split entries into (kept, pruned).
90 Args:
91 entries: Entries to evaluate.
93 Returns:
94 Tuple of (kept, pruned) entry lists.
95 """
96 kept: list[MemoryEntry] = []
97 pruned: list[MemoryEntry] = []
98 for e in entries:
99 (pruned if self.should_prune(e) else kept).append(e)
100 return kept, pruned
103class DeduplicationStrategy:
104 """Removes near-duplicate entries based on content similarity.
106 Two entries are duplicates if their lowercased content shares more than
107 *similarity_threshold* characters of the shorter one (Jaccard-like).
108 """
110 def __init__(self, similarity_threshold: float = 0.85) -> None:
111 """Initialise the deduplication strategy.
113 Args:
114 similarity_threshold: Jaccard-like overlap above which entries
115 are considered duplicates.
116 """
117 self._threshold = similarity_threshold
119 def _tokens(self, text: str) -> set[str]:
120 return set(text.lower().split())
122 def deduplicate(
123 self, entries: list[MemoryEntry]
124 ) -> tuple[list[MemoryEntry], list[MemoryEntry]]:
125 """Return (unique, duplicates).
127 Args:
128 entries: Entries to deduplicate.
130 Returns:
131 Tuple of (unique, duplicate) entry lists.
132 """
133 unique: list[MemoryEntry] = []
134 dupes: list[MemoryEntry] = []
135 seen_tokens: list[set[str]] = []
137 for entry in entries:
138 tokens = self._tokens(entry.content)
139 is_dup = False
140 for seen in seen_tokens:
141 if not tokens or not seen:
142 continue
143 intersection = len(tokens & seen)
144 union = len(tokens | seen)
145 if union > 0 and intersection / union >= self._threshold:
146 is_dup = True
147 break
148 if is_dup:
149 dupes.append(entry)
150 else:
151 unique.append(entry)
152 seen_tokens.append(tokens)
154 return unique, dupes
157class TimeDecayStrategy:
158 """Exponential decay of memory importance based on entry age.
160 The *effective importance* of an entry degrades over time using the formula::
162 effective = entry.importance * exp(-0.693 * age_hours / half_life_hours)
164 This mirrors radio-isotope half-life: after ``half_life_hours`` the importance
165 is exactly half the original value. After two half-lives it is one quarter,
166 and so on.
168 Unlike :class:`RecencyDecayStrategy` (which *prunes* entries), this strategy
169 *reweights* them — entries are not removed but their importance is updated so
170 that downstream retrieval ranks them lower.
172 Args:
173 half_life_hours: Time after which importance halves. Default is one week
174 (168 h).
175 min_importance: Floor value — entries will never be weighted below this.
177 Example::
179 strategy = TimeDecayStrategy(half_life_hours=72.0)
180 updated_entries = strategy.apply(all_entries)
181 """
183 def __init__(
184 self,
185 half_life_hours: float = 168.0,
186 *,
187 min_importance: float = 0.01,
188 ) -> None:
189 """Initialise the time-decay strategy.
191 Args:
192 half_life_hours: Hours until importance halves (default 168 = 1 week).
193 min_importance: Minimum importance floor after decay.
194 """
195 if half_life_hours <= 0:
196 raise ValueError("half_life_hours must be positive")
197 self._half_life_hours = half_life_hours
198 self._min_importance = min_importance
200 def compute_importance(self, entry: MemoryEntry) -> float:
201 """Return the time-decayed importance for *entry*.
203 Args:
204 entry: Memory entry to evaluate.
206 Returns:
207 Decayed importance value in the range ``[min_importance, entry.importance]``.
208 """
209 age_hours = (datetime.now(UTC) - entry.timestamp).total_seconds() / 3600.0
210 decayed = entry.importance * math.exp(
211 -math.log(2) * age_hours / self._half_life_hours
212 )
213 return max(decayed, self._min_importance)
215 def apply(self, entries: list[MemoryEntry]) -> list[MemoryEntry]:
216 """Return entries with importance reweighted by age.
218 The original entries are not mutated — new :class:`MemoryEntry` objects
219 are returned with updated ``importance`` values.
221 Args:
222 entries: Memory entries to reweight.
224 Returns:
225 New list of reweighted entries sorted by descending effective importance.
226 """
227 reweighted: list[MemoryEntry] = []
228 for entry in entries:
229 new_importance = self.compute_importance(entry)
230 reweighted.append(
231 MemoryEntry(
232 id=entry.id,
233 owner_id=entry.owner_id,
234 content=entry.content,
235 role=entry.role,
236 importance=new_importance,
237 timestamp=entry.timestamp,
238 metadata=entry.metadata,
239 embedding=entry.embedding,
240 )
241 )
242 reweighted.sort(key=lambda e: e.importance, reverse=True)
243 return reweighted
246__all__ = [
247 "AccessFrequencyStrategy",
248 "DeduplicationStrategy",
249 "RecencyDecayStrategy",
250 "TimeDecayStrategy",
251]