Coverage for agentos/llm/smart_cache.py: 37%
142 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +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, Optional
19__all__ = [
20 "CacheConfig",
21 "CacheStats",
22 "SmartCache",
23 "CacheEntry",
24]
27# ── 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 """
39 max_entries: int = 1000
40 ttl_seconds: int = 3600 # 1 hour default
41 enable_fuzzy: bool = False
42 fuzzy_threshold: float = 0.85
45@dataclass
46class CacheStats:
47 """Cache performance and cost savings statistics.
49 Attributes:
50 hits: Number of cache hits (exact).
51 fuzzy_hits: Number of fuzzy match hits.
52 misses: Number of cache misses.
53 total_cost_saved_usd: Estimated total API cost saved via caching.
54 entries: Current number of cached entries.
55 evictions: Total evicted entries (LRU + TTL).
56 """
57 hits: int = 0
58 fuzzy_hits: int = 0
59 misses: int = 0
60 total_cost_saved_usd: float = 0.0
61 entries: int = 0
62 evictions: int = 0
64 @property
65 def hit_rate(self) -> float:
66 total = self.hits + self.fuzzy_hits + self.misses
67 if total == 0:
68 return 0.0
69 return (self.hits + self.fuzzy_hits) / total
71 def summary(self) -> dict:
72 return {
73 "hits": self.hits,
74 "fuzzy_hits": self.fuzzy_hits,
75 "misses": self.misses,
76 "hit_rate": round(self.hit_rate, 4),
77 "total_cost_saved_usd": round(self.total_cost_saved_usd, 6),
78 "entries": self.entries,
79 "evictions": self.evictions,
80 }
83# ── Cache Entry ───────────────────────────────────────────────────
85@dataclass
86class CacheEntry:
87 """A single cached LLM response.
89 Attributes:
90 key: Cache key (usually hash of the prompt/model).
91 prompt: Original prompt text.
92 response: Cached LLM response.
93 model: Model name used.
94 cost_usd: Estimated cost of the original API call (savings).
95 tokens: Token count of the response.
96 created_at: Unix timestamp when cached.
97 ttl: Time-to-live in seconds.
98 hit_count: Number of times this entry was used.
99 """
100 key: str
101 prompt: str
102 response: Any
103 model: str = ""
104 cost_usd: float = 0.0
105 tokens: int = 0
106 created_at: float = field(default_factory=time.time)
107 ttl: int = 3600
108 hit_count: int = 0
110 @property
111 def expired(self) -> bool:
112 """Check if this entry has exceeded its TTL."""
113 if self.ttl <= 0:
114 return False
115 return time.time() - self.created_at > self.ttl
117 @property
118 def age_seconds(self) -> float:
119 return time.time() - self.created_at
122# ── Smart Cache ──────────────────────────────────────────────────
124class SmartCache:
125 """LLM response cache with exact + fuzzy matching and cost tracking.
127 Usage:
128 cache = SmartCache(config=CacheConfig(max_entries=500, ttl_seconds=7200))
129 cache.set(prompt="What is quantum computing?", response="...", model="gpt-4o", cost_usd=0.005)
130 cached = cache.get(prompt="What is quantum computing?", model="gpt-4o")
131 if cached:
132 print(f"Cache hit! Saved ${cached.cost_usd}")
134 The cache uses an LRU eviction policy with TTL-based expiration.
135 Keys are derived from a hash of (prompt + model) for exact matching.
136 """
138 def __init__(self, config: Optional[CacheConfig] = None):
139 self._config = config or CacheConfig()
140 self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
141 self.stats = CacheStats()
143 # ── Core API ──────────────────────────────────────────────────
145 def get(
146 self,
147 prompt: str,
148 model: str = "",
149 use_fuzzy: bool = False,
150 ) -> Optional[CacheEntry]:
151 """Look up a cached response for the given prompt.
153 Args:
154 prompt: The prompt text to look up.
155 model: Model name for key disambiguation.
156 use_fuzzy: Enable fuzzy/semantic matching (requires embeddings).
158 Returns:
159 CacheEntry if found, None otherwise.
160 """
161 # 1. Exact match
162 key = self._make_key(prompt, model)
163 if key in self._cache:
164 entry = self._cache[key]
165 # Check TTL
166 if entry.expired:
167 del self._cache[key]
168 self.stats.evictions += 1
169 self.stats.entries = len(self._cache)
170 else:
171 # LRU: move to end
172 self._cache.move_to_end(key)
173 entry.hit_count += 1
174 self.stats.hits += 1
175 return entry
177 # 2. Fuzzy match (optional)
178 if use_fuzzy and self._config.enable_fuzzy:
179 entry = self._fuzzy_lookup(prompt, model)
180 if entry:
181 self.stats.fuzzy_hits += 1
182 return entry
184 # 3. Miss
185 self.stats.misses += 1
186 return None
188 def set(
189 self,
190 prompt: str,
191 response: Any,
192 model: str = "",
193 cost_usd: float = 0.0,
194 tokens: int = 0,
195 ) -> str:
196 """Cache a response.
198 Args:
199 prompt: The original prompt.
200 response: The LLM response to cache.
201 model: Model name used.
202 cost_usd: Cost of the original call (for savings tracking).
203 tokens: Token count of the response.
205 Returns:
206 The cache key.
207 """
208 key = self._make_key(prompt, model)
210 # Update existing entry if present
211 if key in self._cache:
212 entry = self._cache[key]
213 entry.response = response
214 entry.cost_usd = cost_usd
215 entry.tokens = tokens
216 entry.created_at = time.time()
217 self._cache.move_to_end(key)
218 return key
220 # Evict if over capacity
221 while len(self._cache) >= self._config.max_entries:
222 self._evict_one()
224 entry = CacheEntry(
225 key=key,
226 prompt=prompt,
227 response=response,
228 model=model,
229 cost_usd=cost_usd,
230 tokens=tokens,
231 ttl=self._config.ttl_seconds,
232 )
233 self._cache[key] = entry
234 self.stats.entries = len(self._cache)
235 return key
237 def clear(self):
238 """Clear all cached entries."""
239 self._cache.clear()
240 self.stats.entries = 0
242 def gc(self) -> int:
243 """Garbage collect expired entries. Returns number of entries evicted."""
244 count = 0
245 expired_keys = [k for k, v in self._cache.items() if v.expired]
246 for k in expired_keys:
247 del self._cache[k]
248 count += 1
249 self.stats.evictions += count
250 self.stats.entries = len(self._cache)
251 return count
253 # ── Info ─────────────────────────────────────────────────────
255 @property
256 def size(self) -> int:
257 return len(self._cache)
259 @property
260 def config(self) -> CacheConfig:
261 return self._config
263 def contains(self, prompt: str, model: str = "") -> bool:
264 """Check if a prompt is cached (exact match, ignores TTL)."""
265 return self._make_key(prompt, model) in self._cache
267 # ── Internal ─────────────────────────────────────────────────
269 def _make_key(self, prompt: str, model: str = "") -> str:
270 """Generate a deterministic cache key from prompt + model."""
271 content = f"{model or 'default'}:{prompt}"
272 return hashlib.sha256(content.encode()).hexdigest()[:32]
274 def _fuzzy_lookup(self, prompt: str, model: str = "") -> Optional[CacheEntry]:
275 """Semantic similarity lookup using simple keyword overlap.
277 For production, replace with embedding-based similarity (cosine).
278 """
279 prompt_words = set(prompt.lower().split())
280 if not prompt_words:
281 return None
283 best_score = 0.0
284 best_entry: Optional[CacheEntry] = None
285 best_key = ""
287 for key, entry in self._cache.items():
288 if entry.expired:
289 continue
290 if model and entry.model and entry.model != model:
291 continue
293 entry_words = set(entry.prompt.lower().split())
294 if not entry_words:
295 continue
297 # Jaccard similarity
298 intersection = prompt_words & entry_words
299 union = prompt_words | entry_words
300 score = len(intersection) / len(union) if union else 0.0
302 if score > best_score and score >= self._config.fuzzy_threshold:
303 best_score = score
304 best_entry = entry
305 best_key = key
307 if best_entry:
308 self._cache.move_to_end(best_key)
309 best_entry.hit_count += 1
310 return best_entry
312 return None
314 def _evict_one(self):
315 """Evict the oldest entry (LRU front)."""
316 if self._cache:
317 self._cache.popitem(last=False)
318 self.stats.evictions += 1