Coverage for agentos/cache/response_cache.py: 44%

1""" 

2Response Cache with TTL — Cached LLM responses with configurable expiry. 

3 

4Supports in-memory LRU cache with TTL, disk persistence, and cache key 

5strategies (exact match, semantic similarity, template-based). 

6""" 

7 

8from __future__ import annotations 

9 

10import hashlib 

11import json 

12import time 

13from collections import OrderedDict 

14from dataclasses import dataclass, field 

15from enum import Enum 

16from typing import Any 

17 

18 

19class CacheKeyStrategy(Enum): 

20 """Strategy for generating cache lookup keys.""" 

21 

22 EXACT = "exact" 

23 """Hash of the full prompt/message.""" 

24 

25 NORMALIZED = "normalized" 

26 """Hash after whitespace/lowercase normalization.""" 

27 

28 TEMPLATE = "template" 

29 """Hash of template name + variables (ignores phrasing variations).""" 

30 

31 

32@dataclass 

33class CacheEntry: 

34 """A single cache entry.""" 

35 

36 key: str 

37 value: Any 

38 created_at: float = field(default_factory=time.time) 

39 ttl_seconds: float = 3600.0 

40 """Time-to-live in seconds. None means no expiry.""" 

41 

42 hit_count: int = 0 

43 last_accessed: float = 0.0 

44 metadata: dict[str, Any] = field(default_factory=dict) 

45 

46 @property 

47 def is_expired(self) -> bool: 

48 if self.ttl_seconds <= 0: 

49 return False 

50 return (time.time() - self.created_at) > self.ttl_seconds 

51 

52 @property 

53 def age_seconds(self) -> float: 

54 return time.time() - self.created_at 

55 

56 

57@dataclass 

58class CacheStats: 

59 """Cache performance statistics.""" 

60 

61 hits: int = 0 

62 misses: int = 0 

63 evictions: int = 0 

64 expirations: int = 0 

65 size: int = 0 

66 max_size: int = 0 

67 

68 @property 

69 def hit_rate(self) -> float: 

70 total = self.hits + self.misses 

71 return self.hits / total if total > 0 else 0.0 

72 

73 @property 

74 def utilization(self) -> float: 

75 return self.size / self.max_size if self.max_size > 0 else 0.0 

76 

77 

78class ResponseCache: 

79 """ 

80 Response cache with TTL and LRU eviction. 

81 

82 Supports: 

83 - In-memory LRU cache with configurable TTL 

84 - Multiple cache key strategies (exact, normalized, template) 

85 - Statistics tracking (hit rate, evictions, expirations) 

86 - Optional disk persistence (planned) 

87 

88 Example:: 

89 

90 cache = ResponseCache(max_entries=1000, default_ttl=3600) 

91 cache.put("What is 2+2?", "4") 

92 result = cache.get("What is 2+2?") # "4" (cache hit) 

93 """ 

94 

95 def __init__( 

96 self, 

97 max_entries: int = 1000, 

98 default_ttl: float = 3600.0, 

99 key_strategy: CacheKeyStrategy = CacheKeyStrategy.EXACT, 

100 ): 

101 self._max_entries = max_entries 

102 self._default_ttl = default_ttl 

103 self._key_strategy = key_strategy 

104 self._store: OrderedDict[str, CacheEntry] = OrderedDict() 

105 self._stats = CacheStats(max_size=max_entries) 

106 

107 def get(self, prompt: str, **context: Any) -> Any | None: 

108 """ 

109 Retrieve cached response for a prompt. 

110 

111 Args: 

112 prompt: The prompt/message text. 

113 **context: Additional context for template-based keys. 

114 

115 Returns: 

116 Cached value if found and not expired, else None. 

117 """ 

118 key = self._make_key(prompt, context) 

119 entry = self._store.get(key) 

120 

121 if entry is None: 

122 self._stats.misses += 1 

123 return None 

124 

125 if entry.is_expired: 

126 self._evict(key) 

127 self._stats.expirations += 1 

128 self._stats.misses += 1 

129 return None 

130 

131 # Move to end for LRU 

132 self._store.move_to_end(key) 

133 entry.hit_count += 1 

134 entry.last_accessed = time.time() 

135 self._stats.hits += 1 

136 return entry.value 

137 

138 def put( 

139 self, 

140 prompt: str, 

141 value: Any, 

142 ttl: float | None = None, 

143 **context: Any, 

144 ) -> str: 

145 """ 

146 Cache a response. 

147 

148 Args: 

149 prompt: The prompt/message text. 

150 value: The response to cache. 

151 ttl: Custom TTL in seconds (default: self._default_ttl). 

152 **context: Additional context for template-based keys. 

153 

154 Returns: 

155 The cache key string. 

156 """ 

157 key = self._make_key(prompt, context) 

158 effective_ttl = ttl if ttl is not None else self._default_ttl 

159 

160 if key in self._store: 

161 self._store.move_to_end(key) 

162 

163 self._store[key] = CacheEntry( 

164 key=key, 

165 value=value, 

166 ttl_seconds=effective_ttl, 

167 last_accessed=time.time(), 

168 ) 

169 

170 self._stats.size = len(self._store) 

171 

172 # Evict oldest if over capacity 

173 while len(self._store) > self._max_entries: 

174 oldest_key, _ = self._store.popitem(last=False) 

175 self._stats.evictions += 1 

176 

177 return key 

178 

179 def invalidate(self, prompt: str, **context: Any) -> bool: 

180 """Remove a specific cache entry. Returns True if found and removed.""" 

181 key = self._make_key(prompt, context) 

182 if key in self._store: 

183 del self._store[key] 

184 self._stats.size = len(self._store) 

185 return True 

186 return False 

187 

188 def clear(self) -> None: 

189 """Clear all cached entries.""" 

190 self._store.clear() 

191 self._stats.size = 0 

192 

193 def clear_expired(self) -> int: 

194 """Remove all expired entries. Returns count removed.""" 

195 expired = [k for k, e in self._store.items() if e.is_expired] 

196 for k in expired: 

197 del self._store[k] 

198 self._stats.expirations += len(expired) 

199 self._stats.size = len(self._store) 

200 return len(expired) 

201 

202 def get_stats(self) -> CacheStats: 

203 """Return current cache statistics snapshot.""" 

204 self._stats.size = len(self._store) 

205 return self._stats 

206 

207 def get_entry(self, prompt: str, **context: Any) -> CacheEntry | None: 

208 """Get the full cache entry (including metadata) without updating LRU.""" 

209 key = self._make_key(prompt, context) 

210 return self._store.get(key) 

211 

212 def _evict(self, key: str) -> None: 

213 """Evict a specific entry.""" 

214 if key in self._store: 

215 del self._store[key] 

216 self._stats.evictions += 1 

217 self._stats.size = len(self._store) 

218 

219 def _make_key(self, prompt: str, context: dict[str, Any]) -> str: 

220 """Generate a cache key based on the configured strategy.""" 

221 if self._key_strategy == CacheKeyStrategy.NORMALIZED: 

222 prompt = " ".join(prompt.lower().split()) 

223 

224 if self._key_strategy == CacheKeyStrategy.TEMPLATE: 

225 key_data = json.dumps({"template": prompt, "vars": context}, sort_keys=True) 

226 return hashlib.sha256(key_data.encode()).hexdigest()[:32] 

227 

228 if context: 

229 prompt = prompt + json.dumps(context, sort_keys=True) 

230 

231 return hashlib.sha256(prompt.encode()).hexdigest()[:32] 

232 

233 @property 

234 def size(self) -> int: 

235 return len(self._store) 

236 

237 @property 

238 def is_full(self) -> bool: 

239 return len(self._store) >= self._max_entries 

240 

241 def __contains__(self, prompt: str) -> bool: 

242 key = self._make_key(prompt, {}) 

243 entry = self._store.get(key) 

244 return entry is not None and not entry.is_expired 

245 

246 def __len__(self) -> int: 

247 return len(self._store)