LRU cache implementations with TTL and background refresh for prompt and data caching. LangSmith provides both synchronous and asynchronous cache implementations optimized for caching prompts and other data with automatic refresh capabilities.
Thread-safe LRU cache with background thread refresh for synchronous operations.
class Cache:
"""
Thread-safe LRU cache with background refresh.
Provides an in-memory cache with automatic expiration and background
refresh capabilities. Used with synchronous Client.
"""
def __init__(
self,
*,
max_size: int = 100,
ttl_seconds: Optional[float] = 3600.0,
refresh_interval_seconds: float = 60.0,
fetch_func: Optional[Callable[[str], Any]] = None,
):
"""
Create a cache instance.
Parameters:
- max_size: Maximum entries in cache (LRU eviction when exceeded)
- ttl_seconds: Time before entry is stale (None for infinite TTL)
- refresh_interval_seconds: How often to check for stale entries
- fetch_func: Callback to fetch fresh data for a cache key
"""
def get(self, key: str) -> Optional[Any]:
"""
Get value from cache.
Parameters:
- key: Cache key
Returns:
Cached value or None if not found or expired
"""
def set(self, key: str, value: Any) -> None:
"""
Set value in cache.
Parameters:
- key: Cache key
- value: Value to cache
"""
def invalidate(self, key: str) -> None:
"""
Remove specific entry from cache.
Parameters:
- key: Cache key to invalidate
"""
def clear(self) -> None:
"""
Clear all entries from cache.
"""
def shutdown(self) -> None:
"""
Stop background refresh thread and cleanup.
"""
def dump(self, path: Union[str, Path]) -> None:
"""
Dump cache to JSON file.
Parameters:
- path: File path to write cache data
"""
def load(self, path: Union[str, Path]) -> int:
"""
Load cache from JSON file.
Parameters:
- path: File path to read cache data from
Returns:
Number of entries loaded
"""
@property
def metrics(self) -> CacheMetrics:
"""
Get cache performance metrics.
Returns:
CacheMetrics object with hits, misses, evictions, etc.
"""Thread-safe LRU cache with asyncio task refresh for asynchronous operations.
class AsyncCache:
"""
Thread-safe LRU cache with asyncio refresh.
Provides an async-compatible in-memory cache with automatic expiration
and background refresh. Used with AsyncClient.
"""
def __init__(
self,
*,
max_size: int = 100,
ttl_seconds: Optional[float] = 3600.0,
refresh_interval_seconds: float = 60.0,
fetch_func: Optional[Callable[[str], Awaitable[Any]]] = None,
):
"""
Create an async cache instance.
Parameters:
- max_size: Maximum entries in cache (LRU eviction when exceeded)
- ttl_seconds: Time before entry is stale (None for infinite TTL)
- refresh_interval_seconds: How often to check for stale entries
- fetch_func: Async callback to fetch fresh data for a cache key
"""
def get(self, key: str) -> Optional[Any]:
"""
Get value from cache (synchronous).
Parameters:
- key: Cache key
Returns:
Cached value or None if not found or expired
"""
def set(self, key: str, value: Any) -> None:
"""
Set value in cache (synchronous).
Parameters:
- key: Cache key
- value: Value to cache
"""
def invalidate(self, key: str) -> None:
"""
Remove specific entry from cache (synchronous).
Parameters:
- key: Cache key to invalidate
"""
def clear(self) -> None:
"""
Clear all entries from cache (synchronous).
"""
async def start(self) -> None:
"""
Start async background refresh loop.
Must be called before using the cache with automatic refresh.
"""
async def stop(self) -> None:
"""
Stop async background refresh loop and cleanup.
"""
def dump(self, path: Union[str, Path]) -> None:
"""
Dump cache to JSON file (synchronous).
Parameters:
- path: File path to write cache data
"""
def load(self, path: Union[str, Path]) -> int:
"""
Load cache from JSON file (synchronous).
Parameters:
- path: File path to read cache data from
Returns:
Number of entries loaded
"""
@property
def metrics(self) -> CacheMetrics:
"""
Get cache performance metrics.
Returns:
CacheMetrics object with hits, misses, evictions, etc.
"""class CacheMetrics:
"""Cache performance metrics."""
hits: int
"""Number of cache hits"""
misses: int
"""Number of cache misses"""
evictions: int
"""Number of entries evicted due to size limit"""
expirations: int
"""Number of entries expired due to TTL"""
@property
def hit_rate(self) -> float:
"""
Calculate cache hit rate.
Returns:
Hit rate as a float between 0 and 1
"""from langsmith.cache import Cache
# Create cache
cache = Cache(max_size=100, ttl_seconds=3600)
# Set values
cache.set("key1", "value1")
cache.set("key2", {"data": "value2"})
# Get values
value = cache.get("key1") # "value1"
value = cache.get("key3") # None (not found)
# Remove entry
cache.invalidate("key1")
# Clear all
cache.clear()
# Cleanup when done
cache.shutdown()from langsmith import Client
from langsmith.cache import Cache
# Create cache for prompts
prompt_cache = Cache(
max_size=50,
ttl_seconds=1800 # 30 minutes
)
# Use cache with client
client = Client(cache=prompt_cache)
# Pull prompt (cached)
prompt = client.pull_prompt("my-prompt")
# Second call uses cache
prompt = client.pull_prompt("my-prompt") # No API call
# Or use True for default cache
client = Client(cache=True)from langsmith.cache import Cache
def fetch_fresh_data(key: str):
"""Fetch fresh data for a key."""
# Your logic to fetch fresh data
return fetch_from_api(key)
# Cache with auto-refresh
cache = Cache(
max_size=100,
ttl_seconds=300, # 5 minutes
refresh_interval_seconds=60, # Check every minute
fetch_func=fetch_fresh_data
)
# Set initial value
cache.set("my-key", "initial-value")
# Value will be automatically refreshed in background
# when it becomes stalefrom langsmith.cache import Cache
cache = Cache(max_size=100)
# Populate cache
cache.set("key1", "value1")
cache.set("key2", "value2")
# Save to disk
cache.dump("./cache.json")
# Later, load from disk
cache2 = Cache(max_size=100)
loaded = cache2.load("./cache.json")
print(f"Loaded {loaded} entries")
value = cache2.get("key1") # "value1"from langsmith.cache import Cache
cache = Cache(max_size=10)
# Use the cache
for i in range(20):
cache.set(f"key{i}", f"value{i}")
for i in range(15):
cache.get(f"key{i}")
# Check metrics
metrics = cache.metrics
print(f"Hit rate: {metrics.hit_rate:.2%}")
print(f"Hits: {metrics.hits}")
print(f"Misses: {metrics.misses}")
print(f"Evictions: {metrics.evictions}")from langsmith.cache import AsyncCache
async def main():
# Create async cache
cache = AsyncCache(max_size=100, ttl_seconds=3600)
# Start background refresh
await cache.start()
try:
# Set values (synchronous)
cache.set("key1", "value1")
cache.set("key2", {"data": "value2"})
# Get values (synchronous)
value = cache.get("key1") # "value1"
# Use with async operations
await do_async_work(cache)
finally:
# Stop refresh and cleanup
await cache.stop()from langsmith import AsyncClient
from langsmith.cache import AsyncCache
async def main():
# Create cache
prompt_cache = AsyncCache(
max_size=50,
ttl_seconds=1800
)
# Start cache
await prompt_cache.start()
try:
# Use with async client
async with AsyncClient(cache=prompt_cache) as client:
# Pull prompt (cached)
prompt = await client.pull_prompt("my-prompt")
# Second call uses cache
prompt = await client.pull_prompt("my-prompt")
finally:
await prompt_cache.stop()from langsmith.cache import AsyncCache
async def fetch_fresh_data(key: str):
"""Async fetch fresh data for a key."""
# Your async logic to fetch fresh data
return await fetch_from_api_async(key)
async def main():
# Cache with async auto-refresh
cache = AsyncCache(
max_size=100,
ttl_seconds=300,
refresh_interval_seconds=60,
fetch_func=fetch_fresh_data
)
await cache.start()
try:
# Set initial value
cache.set("my-key", "initial-value")
# Value will be automatically refreshed in background
await asyncio.sleep(400)
# Get refreshed value
value = cache.get("my-key")
finally:
await cache.stop()from langsmith.cache import Cache
# Using cache with context manager pattern
class CacheManager:
def __init__(self):
self.cache = Cache(max_size=100)
def __enter__(self):
return self.cache
def __exit__(self, exc_type, exc_val, exc_tb):
self.cache.shutdown()
# Usage
with CacheManager() as cache:
cache.set("key", "value")
value = cache.get("key")
# Cache is automatically shut downfrom langsmith.cache import Cache
# Short-lived cache (5 minutes)
short_cache = Cache(
max_size=100,
ttl_seconds=300
)
# Long-lived cache (24 hours)
long_cache = Cache(
max_size=1000,
ttl_seconds=86400
)
# Infinite cache (no expiration)
infinite_cache = Cache(
max_size=100,
ttl_seconds=None
)from langsmith.cache import Cache
# L1 cache: Small, fast, short TTL
l1_cache = Cache(
max_size=10,
ttl_seconds=60
)
# L2 cache: Larger, longer TTL
l2_cache = Cache(
max_size=100,
ttl_seconds=3600
)
def get_with_fallback(key: str):
"""Get from L1, fallback to L2, fallback to fetch."""
# Try L1
value = l1_cache.get(key)
if value is not None:
return value
# Try L2
value = l2_cache.get(key)
if value is not None:
# Promote to L1
l1_cache.set(key, value)
return value
# Fetch fresh
value = fetch_from_api(key)
l1_cache.set(key, value)
l2_cache.set(key, value)
return valuefrom langsmith.cache import Cache
cache = Cache(max_size=100, ttl_seconds=3600)
# Warm cache with frequently accessed data
def warm_cache():
"""Pre-populate cache with common data."""
common_keys = ["prompt1", "prompt2", "prompt3"]
for key in common_keys:
data = fetch_from_api(key)
cache.set(key, data)
# Warm cache at startup
warm_cache()from langsmith.cache import Cache
cache = Cache(max_size=100, ttl_seconds=3600)
# Invalidate on update
def update_prompt(prompt_name: str, new_content: str):
"""Update a prompt and invalidate cache."""
# Update in database
update_in_db(prompt_name, new_content)
# Invalidate cache
cache.invalidate(prompt_name)
# Invalidate by pattern (custom implementation)
def invalidate_pattern(pattern: str):
"""Invalidate all keys matching pattern."""
# Note: Cache doesn't have pattern matching built-in
# This is a custom implementation
all_keys = get_all_cache_keys() # Custom function
for key in all_keys:
if pattern in key:
cache.invalidate(key)
# Selective clear (by tags/metadata)
def clear_by_category(category: str):
"""Clear cache entries for a category."""
# Custom implementation using metadata
# stored alongside cache entries
pass