Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-memory/src/lexigram/ai/memory/backends/in_memory.py: 42%

1"""In-memory persistence backend — stores MemoryEntry objects in a dict.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6import math 

7 

8from lexigram.contracts.ai.memory import ( 

9 MemoryEntry, 

10 MemoryQuery, 

11 MemorySearchResult, 

12) 

13from lexigram.contracts.core import HealthCheckResult, HealthStatus 

14 

15_EPOCH = datetime(1970, 1, 1, tzinfo=UTC) 

16 

17 

18def _seconds_since(dt: datetime) -> float: 

19 """Return elapsed seconds since *dt*.""" 

20 return (datetime.now(UTC) - dt).total_seconds() 

21 

22 

23def _score(entry: MemoryEntry, query: MemoryQuery) -> float: 

24 """Compute a weighted relevance score for *entry* against *query*.""" 

25 age_s = _seconds_since(entry.timestamp) 

26 recency = math.exp(-age_s / 86400.0) 

27 relevance = 0.5 

28 return ( 

29 query.recency_weight * recency 

30 + query.importance_weight * entry.importance 

31 + query.relevance_weight * relevance 

32 ) 

33 

34 

35class InMemoryMemoryBackend: 

36 """MemoryStoreProtocol backed by an in-process dictionary. 

37 

38 Entries are partitioned by ``owner_id`` so that every operation only 

39 ever touches the owning partition. 

40 """ 

41 

42 def __init__(self) -> None: 

43 self._store: dict[str, dict[str, MemoryEntry]] = {} 

44 

45 async def store(self, entry: MemoryEntry) -> None: 

46 self._store.setdefault(entry.owner_id, {})[entry.id] = entry 

47 

48 async def retrieve(self, query: MemoryQuery) -> list[MemorySearchResult]: 

49 scored: list[tuple[float, MemoryEntry]] = [] 

50 for entry in self._store.get(query.owner_id, {}).values(): 

51 score = _score(entry, query) 

52 if score < query.min_relevance: 

53 continue 

54 if query.time_range: 

55 start, end = query.time_range 

56 if not (start <= entry.timestamp <= end): 

57 continue 

58 if query.filters and not all( 

59 entry.metadata.get(key) == value for key, value in query.filters.items() 

60 ): 

61 continue 

62 scored.append((score, entry)) 

63 

64 scored.sort(key=lambda item: item[0], reverse=True) 

65 return [ 

66 MemorySearchResult(entry=entry, score=score, source="in_memory") 

67 for score, entry in scored[: query.top_k] 

68 ] 

69 

70 async def get_recent(self, n: int, owner_id: str) -> list[MemoryEntry]: 

71 sorted_entries = sorted( 

72 self._store.get(owner_id, {}).values(), 

73 key=lambda entry: entry.timestamp, 

74 reverse=True, 

75 ) 

76 return sorted_entries[:n] 

77 

78 async def delete(self, entry_id: str, owner_id: str) -> None: 

79 self._store.get(owner_id, {}).pop(entry_id, None) 

80 

81 async def clear(self, owner_id: str) -> None: 

82 self._store.pop(owner_id, None) 

83 

84 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

85 return HealthCheckResult( 

86 component="memory.in_memory", 

87 status=HealthStatus.HEALTHY, 

88 details={ 

89 "entries": sum(len(v) for v in self._store.values()), 

90 "timeout": timeout, 

91 }, 

92 ) 

93 

94 def __len__(self) -> int: 

95 return sum(len(v) for v in self._store.values()) 

96 

97 

98__all__ = ["InMemoryMemoryBackend"]