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

1"""Cache-backed memory backend — persists entries via CacheBackendProtocol.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import TYPE_CHECKING, Any, cast 

7 

8from lexigram.ai.memory.exceptions import MemoryStoreError 

9from lexigram.contracts.ai.memory import ( 

10 MemoryEntry, 

11 MemoryQuery, 

12 MemorySearchResult, 

13) 

14from lexigram.contracts.core import HealthCheckResult, HealthStatus 

15from lexigram.serialization.backends.json import dumps_str, loads 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.infra.cache import CacheBackendProtocol 

19 

20_KEY_PREFIX = "mem:" 

21_INDEX_KEY = "mem:index" 

22 

23 

24class CacheMemoryBackend: 

25 """MemoryStoreProtocol backed by a CacheBackendProtocol (e.g. Redis).""" 

26 

27 def __init__(self, cache: CacheBackendProtocol, ttl: int = 86400 * 30) -> None: 

28 self._cache = cache 

29 self._ttl = ttl 

30 

31 def _key(self, owner_id: str, entry_id: str) -> str: 

32 return f"{_KEY_PREFIX}{owner_id}:{entry_id}" 

33 

34 def _index_key(self, owner_id: str) -> str: 

35 return f"{_INDEX_KEY}:{owner_id}" 

36 

37 def _to_json(self, entry: MemoryEntry) -> str: 

38 payload = { 

39 "id": entry.id, 

40 "owner_id": entry.owner_id, 

41 "content": entry.content, 

42 "role": entry.role, 

43 "timestamp": entry.timestamp.isoformat(), 

44 "importance": entry.importance, 

45 "metadata": entry.metadata, 

46 "embedding": entry.embedding, 

47 } 

48 return dumps_str(payload) 

49 

50 def _from_json(self, raw: str) -> MemoryEntry: 

51 data = cast("dict[str, Any]", loads(raw)) 

52 return MemoryEntry( 

53 id=str(data["id"]), 

54 owner_id=str(data["owner_id"]), 

55 content=str(data["content"]), 

56 role=str(data["role"]), 

57 timestamp=datetime.fromisoformat(str(data["timestamp"])), 

58 importance=float(data.get("importance", 0.5)), 

59 metadata=cast("dict[str, Any]", data.get("metadata", {})), 

60 embedding=cast("list[float] | None", data.get("embedding")), 

61 ) 

62 

63 def _as_json_text(self, value: Any | None) -> str | None: 

64 if value is None: 

65 return None 

66 if isinstance(value, str): 

67 return value 

68 if isinstance(value, bytes): 

69 return value.decode("utf-8") 

70 raise MemoryStoreError( 

71 "Cache value was not JSON-serializable text", store="cache" 

72 ) 

73 

74 async def _cache_get_text(self, key: str) -> str | None: 

75 get_result = await self._cache.get(key) 

76 if get_result.is_err(): 

77 raise MemoryStoreError( 

78 f"Cache get failed for key {key}", 

79 store="cache", 

80 ) from get_result.unwrap_err() 

81 return self._as_json_text(get_result.unwrap_or(None)) 

82 

83 async def _cache_set_text( 

84 self, key: str, value: str, ttl: int | None = None 

85 ) -> None: 

86 set_result = await self._cache.set(key, value, ttl=ttl) 

87 if set_result.is_err(): 

88 raise MemoryStoreError( 

89 f"Cache set failed for key {key}", 

90 store="cache", 

91 ) from set_result.unwrap_err() 

92 

93 async def _cache_delete(self, key: str) -> None: 

94 delete_result = await self._cache.delete(key) 

95 if delete_result.is_err(): 

96 raise MemoryStoreError( 

97 f"Cache delete failed for key {key}", 

98 store="cache", 

99 ) from delete_result.unwrap_err() 

100 

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

102 await self._cache_set_text( 

103 self._key(entry.owner_id, entry.id), self._to_json(entry), ttl=self._ttl 

104 ) 

105 index_raw = await self._cache_get_text(self._index_key(entry.owner_id)) 

106 ids: list[str] = cast("list[str]", loads(index_raw)) if index_raw else [] 

107 if entry.id not in ids: 

108 ids.append(entry.id) 

109 await self._cache_set_text(self._index_key(entry.owner_id), dumps_str(ids)) 

110 

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

112 index_raw = await self._cache_get_text(self._index_key(query.owner_id)) 

113 if index_raw is None: 

114 return [] 

115 

116 ids = cast("list[str]", loads(index_raw)) 

117 entries: list[MemoryEntry] = [] 

118 for entry_id in ids: 

119 raw = await self._cache_get_text(self._key(query.owner_id, entry_id)) 

120 if raw is not None: 

121 entries.append(self._from_json(raw)) 

122 

123 import math 

124 

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

126 for entry in entries: 

127 age_seconds = (datetime.now(UTC) - entry.timestamp).total_seconds() 

128 recency = math.exp(-age_seconds / 86400.0) 

129 score = ( 

130 query.recency_weight * recency 

131 + query.importance_weight * entry.importance 

132 + query.relevance_weight * 0.5 

133 ) 

134 if score < query.min_relevance: 

135 continue 

136 if query.time_range: 

137 start, end = query.time_range 

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

139 continue 

140 scored.append((score, entry)) 

141 

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

143 return [ 

144 MemorySearchResult(entry=entry, score=score, source="cache") 

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

146 ] 

147 

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

149 index_raw = await self._cache_get_text(self._index_key(owner_id)) 

150 if index_raw is None: 

151 return [] 

152 

153 ids = cast("list[str]", loads(index_raw)) 

154 entries: list[MemoryEntry] = [] 

155 for entry_id in ids: 

156 raw = await self._cache_get_text(self._key(owner_id, entry_id)) 

157 if raw is not None: 

158 entries.append(self._from_json(raw)) 

159 return sorted(entries, key=lambda entry: entry.timestamp, reverse=True)[:n] 

160 

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

162 await self._cache_delete(self._key(owner_id, entry_id)) 

163 index_raw = await self._cache_get_text(self._index_key(owner_id)) 

164 if index_raw is None: 

165 return 

166 remaining_ids = [ 

167 item for item in cast("list[str]", loads(index_raw)) if item != entry_id 

168 ] 

169 await self._cache_set_text(self._index_key(owner_id), dumps_str(remaining_ids)) 

170 

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

172 index_raw = await self._cache_get_text(self._index_key(owner_id)) 

173 if index_raw is not None: 

174 for entry_id in cast("list[str]", loads(index_raw)): 

175 await self._cache_delete(self._key(owner_id, entry_id)) 

176 await self._cache_delete(self._index_key(owner_id)) 

177 

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

179 health_result = await self._cache.health_check(timeout=timeout) 

180 status = HealthStatus.HEALTHY 

181 if not health_result.is_healthy(): 

182 status = HealthStatus.DEGRADED 

183 return HealthCheckResult( 

184 component="memory.cache", 

185 status=status, 

186 message=health_result.message, 

187 error=health_result.error, 

188 details={"timeout": timeout}, 

189 ) 

190 

191 

192__all__ = ["CacheMemoryBackend"]