File size: 3,404 Bytes
07ff2cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """
TTL Cache module for Agent Tools.
Provides an in-memory cache with time-to-live eviction to avoid
redundant API calls and rate limiting.
"""
import json
import threading
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class CacheEntry:
"""A single cache entry with its stored value and creation timestamp."""
value: str
timestamp: float
class TTLCache:
"""In-memory cache with time-to-live eviction.
Stores string results keyed by function name + parameters.
Thread-safe via threading.Lock on all read/write operations.
"""
def __init__(self, default_ttl: int = 300, max_age: int = 900):
"""Initialize the cache.
Args:
default_ttl: Time-to-live in seconds (default 300 = 5 minutes).
Entries older than this are considered expired on get().
max_age: Maximum entry age before forced eviction (default 900 = 15 minutes).
Entries older than this are removed during eviction sweeps.
"""
self.default_ttl = default_ttl
self.max_age = max_age
self._store: dict[str, CacheEntry] = {}
self._lock = threading.Lock()
def make_key(self, func_name: str, **kwargs) -> str:
"""Generate a deterministic cache key from function name and parameters.
Uses sorted JSON serialization to ensure the same parameters always
produce the same key regardless of argument order.
Args:
func_name: The name of the tool function.
**kwargs: The parameters passed to the function.
Returns:
A string key in the format "func_name:{sorted_params_json}".
"""
sorted_params_json = json.dumps(kwargs, sort_keys=True)
return f"{func_name}:{sorted_params_json}"
def get(self, key: str) -> Optional[str]:
"""Return cached value if it exists and is within TTL, else None.
Always triggers _evict_stale() to clean up old entries.
Args:
key: The cache key to look up.
Returns:
The cached string value if valid, or None if expired/missing.
"""
with self._lock:
self._evict_stale()
entry = self._store.get(key)
if entry is None:
return None
if time.time() - entry.timestamp > self.default_ttl:
return None
return entry.value
def set(self, key: str, value: str) -> None:
"""Store a value in the cache with the current timestamp.
Args:
key: The cache key.
value: The string value to cache.
"""
with self._lock:
self._store[key] = CacheEntry(value=value, timestamp=time.time())
def _evict_stale(self) -> None:
"""Remove entries older than max_age (15 minutes by default).
This method is called internally and assumes the lock is already held.
"""
current_time = time.time()
stale_keys = [
key
for key, entry in self._store.items()
if current_time - entry.timestamp > self.max_age
]
for key in stale_keys:
del self._store[key]
def clear(self) -> None:
"""Remove all entries from the cache. Useful for testing."""
with self._lock:
self._store.clear()
|