Coverage for agentos/llm/smart_cache.py: 37%
142 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 21:19 +0800
1"""Smart Cache — LLM response caching with exact and fuzzy matching.
3Reduces API costs by caching LLM responses. Supports:
4 - Exact match: identical prompt → cached response
5 - Fuzzy match: semantically similar prompts → cached response (via embeddings)
6 - TTL-based expiration and LRU eviction
7 - Cost savings tracking
8"""
10from __future__ import annotations
12import hashlib
13import time
14from collections import OrderedDict
15from dataclasses import dataclass, field
16from typing import Any
18__all__ = [
19 "CacheConfig",
20 "CacheStats",
21 "SmartCache",
22 "CacheEntry",
23]
26# ── Config & Stats ─────────────────────────────────────────────────
29@dataclass
30class CacheConfig:
31 """Configuration for SmartCache.
33 Attributes:
34 max_entries: Maximum number of cached entries (LRU eviction).
35 ttl_seconds: Time-to-live in seconds (0 = no expiry).
36 enable_fuzzy: Enable semantic similarity matching.
37 fuzzy_threshold: Similarity threshold for fuzzy matching (0-1).
38 """
40 max_entries: int = 1000
41 ttl_seconds: int = 3600 # 1 hour default
42 enable_fuzzy: bool = False
43 fuzzy_threshold: float = 0.85
46@dataclass
47class CacheStats:
48 """Cache performance and cost savings statistics.
50 Attributes:
51 hits: Number of cache hits (exact).
52 fuzzy_hits: Number of fuzzy match hits.
53 misses: Number of cache misses.
54 total_cost_saved_usd: Estimated total API cost saved via caching.
55 entries: Current number of cached entries.
56 evictions: Total evicted entries (LRU + TTL).
57 """
59 hits: int = 0
60 fuzzy_hits: int = 0
61 misses: int = 0
62 total_cost_saved_usd: float = 0.0
63 entries: int = 0
64 evictions: int = 0
66 @property
67 def hit_rate(self) -> float:
68 total = self.hits + self.fuzzy_hits + self.misses
69 if total == 0:
70 return 0.0
71 return (self.hits + self.fuzzy_hits) / total
73 def summary(self) -> dict:
74 return {
75 "hits": self.hits,
76 "fuzzy_hits": self.fuzzy_hits,
77 "misses": self.misses,
78 "hit_rate": round(self.hit_rate, 4),
79 "total_cost_saved_usd": round(self.total_cost_saved_usd, 6),
80 "entries": self.entries,
81 "evictions": self.evictions,
82 }
85# ── Cache Entry ───────────────────────────────────────────────────
88@dataclass
89class CacheEntry:
90 """A single cached LLM response.
92 Attributes:
93 key: Cache key (usually hash of the prompt/model).
94 prompt: Original prompt text.
95 response: Cached LLM response.
96 model: Model name used.
97 cost_usd: Estimated cost of the original API call (savings).
98 tokens: Token count of the response.
99 created_at: Unix timestamp when cached.
100 ttl: Time-to-live in seconds.
101 hit_count: Number of times this entry was used.
102 """
104 key: str
105 prompt: str
106 response: Any
107 model: str = ""
108 cost_usd: float = 0.0
109 tokens: int = 0
110 created_at: float = field(default_factory=time.time)
111 ttl: int = 3600
112 hit_count: int = 0
114 @property
115 def expired(self) -> bool:
116 """Check if this entry has exceeded its TTL."""
117 if self.ttl <= 0:
118 return False
119 return time.time() - self.created_at > self.ttl
121 @property
122 def age_seconds(self) -> float:
123 return time.time() - self.created_at
126# ── Smart Cache ──────────────────────────────────────────────────
129class SmartCache:
130 """LLM response cache with exact + fuzzy matching and cost tracking.
132 Usage:
133 cache = SmartCache(config=CacheConfig(max_entries=500, ttl_seconds=7200))
134 cache.set(prompt="What is quantum computing?", response="...", model="gpt-4o", cost_usd=0.005)
135 cached = cache.get(prompt="What is quantum computing?", model="gpt-4o")
136 if cached:
137 print(f"Cache hit! Saved ${cached.cost_usd}")
139 The cache uses an LRU eviction policy with TTL-based expiration.
140 Keys are derived from a hash of (prompt + model) for exact matching.
141 """
143 def __init__(self, config: CacheConfig | None = None):
144 self._config = config or CacheConfig()
145 self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
146 self.stats = CacheStats()
148 # ── Core API ──────────────────────────────────────────────────
150 def get(
151 self,
152 prompt: str,
153 model: str = "",
154 use_fuzzy: bool = False,
155 ) -> CacheEntry | None:
156 """Look up a cached response for the given prompt.
158 Args:
159 prompt: The prompt text to look up.
160 model: Model name for key disambiguation.
161 use_fuzzy: Enable fuzzy/semantic matching (requires embeddings).
163 Returns:
164 CacheEntry if found, None otherwise.
165 """
166 # 1. Exact match
167 key = self._make_key(prompt, model)
168 if key in self._cache:
169 entry = self._cache[key]
170 # Check TTL
171 if entry.expired:
172 del self._cache[key]
173 self.stats.evictions += 1
174 self.stats.entries = len(self._cache)
175 else:
176 # LRU: move to end
177 self._cache.move_to_end(key)
178 entry.hit_count += 1
179 self.stats.hits += 1
180 return entry
182 # 2. Fuzzy match (optional)
183 if use_fuzzy and self._config.enable_fuzzy:
184 entry = self._fuzzy_lookup(prompt, model)
185 if entry:
186 self.stats.fuzzy_hits += 1
187 return entry
189 # 3. Miss
190 self.stats.misses += 1
191 return None
193 def set(
194 self,
195 prompt: str,
196 response: Any,
197 model: str = "",
198 cost_usd: float = 0.0,
199 tokens: int = 0,
200 ) -> str:
201 """Cache a response.
203 Args:
204 prompt: The original prompt.
205 response: The LLM response to cache.
206 model: Model name used.
207 cost_usd: Cost of the original call (for savings tracking).
208 tokens: Token count of the response.
210 Returns:
211 The cache key.
212 """
213 key = self._make_key(prompt, model)
215 # Update existing entry if present
216 if key in self._cache:
217 entry = self._cache[key]
218 entry.response = response
219 entry.cost_usd = cost_usd
220 entry.tokens = tokens
221 entry.created_at = time.time()
222 self._cache.move_to_end(key)
223 return key
225 # Evict if over capacity
226 while len(self._cache) >= self._config.max_entries:
227 self._evict_one()
229 entry = CacheEntry(
230 key=key,
231 prompt=prompt,
232 response=response,
233 model=model,
234 cost_usd=cost_usd,
235 tokens=tokens,
236 ttl=self._config.ttl_seconds,
237 )
238 self._cache[key] = entry
239 self.stats.entries = len(self._cache)
240 return key
242 def clear(self):
243 """Clear all cached entries."""
244 self._cache.clear()
245 self.stats.entries = 0
247 def gc(self) -> int:
248 """Garbage collect expired entries. Returns number of entries evicted."""
249 count = 0
250 expired_keys = [k for k, v in self._cache.items() if v.expired]
251 for k in expired_keys:
252 del self._cache[k]
253 count += 1
254 self.stats.evictions += count
255 self.stats.entries = len(self._cache)
256 return count
258 # ── Info ─────────────────────────────────────────────────────
260 @property
261 def size(self) -> int:
262 return len(self._cache)
264 @property
265 def config(self) -> CacheConfig:
266 return self._config
268 def contains(self, prompt: str, model: str = "") -> bool:
269 """Check if a prompt is cached (exact match, ignores TTL)."""
270 return self._make_key(prompt, model) in self._cache
272 # ── Internal ─────────────────────────────────────────────────
274 def _make_key(self, prompt: str, model: str = "") -> str:
275 """Generate a deterministic cache key from prompt + model."""
276 content = f"{model or 'default'}:{prompt}"
277 return hashlib.sha256(content.encode()).hexdigest()[:32]
279 def _fuzzy_lookup(self, prompt: str, model: str = "") -> CacheEntry | None:
280 """Semantic similarity lookup using simple keyword overlap.
282 For production, replace with embedding-based similarity (cosine).
283 """
284 prompt_words = set(prompt.lower().split())
285 if not prompt_words:
286 return None
288 best_score = 0.0
289 best_entry: CacheEntry | None = None
290 best_key = ""
292 for key, entry in self._cache.items():
293 if entry.expired:
294 continue
295 if model and entry.model and entry.model != model:
296 continue
298 entry_words = set(entry.prompt.lower().split())
299 if not entry_words:
300 continue
302 # Jaccard similarity
303 intersection = prompt_words & entry_words
304 union = prompt_words | entry_words
305 score = len(intersection) / len(union) if union else 0.0
307 if score > best_score and score >= self._config.fuzzy_threshold:
308 best_score = score
309 best_entry = entry
310 best_key = key
312 if best_entry:
313 self._cache.move_to_end(best_key)
314 best_entry.hit_count += 1
315 return best_entry
317 return None
319 def _evict_one(self):
320 """Evict the oldest entry (LRU front)."""
321 if self._cache:
322 self._cache.popitem(last=False)
323 self.stats.evictions += 1