Quickstart

This page walks through both entry points the package exposes: CacheManager (semantic LLM caching) and MultiLevelCache (generic tiered caching).

1. Semantic LLM caching with CacheManager

CacheManager is async-first. The canonical pattern is: look up the query, call your LLM only on a miss, store the result, and optionally report feedback later.

import asyncio
from zerda.cache import CacheManager

async def call_llm(prompt: str) -> str:
    # Replace with your real LLM call.
    return f"(pretend LLM answer to: {prompt})"

async def main():
    # CacheManager.create() builds every subsystem (storage, embedder, security,
    # tenant manager, ...) from a CacheConfig and warms up the embedding provider.
    manager = await CacheManager.create()

    query = "What is retrieval-augmented generation?"

    result = await manager.get(query, tenant_id="acme-corp")

    if result.hit:
        print("cache hit:", result.decision, "similarity:", result.similarity)
        answer = result.response
    else:
        answer = await call_llm(query)
        await manager.put(
            query,
            answer,
            tenant_id="acme-corp",
            response_tokens=len(answer.split()),
        )

    print(answer)

    # Later, once you know whether the cached/generated answer was actually good:
    if result.hit and result.entry is not None:
        await manager.feedback(result.entry.entry_id, positive=True, tenant_id="acme-corp")

    manager.close()

asyncio.run(main())

[!TIP] The default config uses SQLite and a mock embedder With no arguments, CacheManager.create() builds a CacheConfig() whose storage_backend is StorageBackend.SQLITE (writing to ./zerda_cache.db) and whose embedding.provider is EmbeddingProvider.MOCK (deterministic fake embeddings — no network calls, no API key required). This makes the snippet above runnable as-is. See Configuration for how to point it at Redis and a real embedding provider.

Using it from synchronous code

Every async method has a _sync counterpart that runs the coroutine to completion, whether or not an event loop is already running in the current thread (see CacheManager._run_sync):

from zerda.cache import CacheManager, CacheConfig

manager = CacheManager(CacheConfig())     # sync construction (no warmup)
result = manager.get_sync("hello", tenant_id="default")
if not result.hit:
    manager.put_sync("hello", "hi there!", tenant_id="default")
manager.close()

Note the difference between the two construction paths:

• CacheManager(config) — synchronous constructor; does not warm up the embedding model.
• await CacheManager.create(config) — async factory; calls CacheManager(config) and then performs a warm-up embedding call, logging (not raising) on failure.

Multi-tenancy

from zerda.cache import CacheManager, TenantRegistration

manager = CacheManager(CacheConfig())
manager.register_tenant(TenantRegistration(
    tenant_id="acme-corp",
    display_name="Acme Corp",
    memory_quota_mb=1024,
    max_entries=50_000,
    request_quota_rpm=500,
    allow_shared_read=True,
))

Every get/put call takes a tenant_id; unseen tenant IDs are auto-registered under the default quota policy the first time they're used (see TenantManager._get_state).

2. Generic object caching with MultiLevelCache

MultiLevelCache has no notion of LLMs, tenants, or embeddings — it's a three-level (memory / memory / disk) cache with configurable eviction, usable as a drop-in accelerator for any expensive, repeatable computation.

from zerda.cache import MultiLevelCache, CacheStrategy

cache = MultiLevelCache(
    l1_max_items=50,
    l2_max_items=500,
    strategy=CacheStrategy.LRU,
)

def expensive_computation(x: int) -> int:
    return x * x  # stand-in for something costly

def get_or_compute(x: int) -> int:
    key = f"square:{x}"
    cached = cache.get(key)
    if cached is not None:
        return cached
    value = expensive_computation(x)
    cache.set(key, value, ttl=300)   # expires in 5 minutes
    return value

print(get_or_compute(7))   # computes and caches
print(get_or_compute(7))   # served from L1
print(cache.get_stats()["overall_hit_rate"])

MultiLevelCache is also a context manager, and exposes aget/aset/adelete for async call sites (each simply runs the sync method in a thread via asyncio.to_thread):

with MultiLevelCache(persist_l3=False) as cache:
    cache.set("k", {"nested": "value"})
    assert "k" in cache          # __contains__ → exists()
    assert len(cache) == 1       # __len__ → total items across L1+L2+L3
# __exit__ clears L1/L2 (and, since persist_l3=False, also wipes L3 files from disk)

Continue to Architecture to see how the pieces fit together, or jump straight to the API Reference.