Skip to content
BranchesTags

Repository files navigation

Semacache

A production-grade semantic caching library for LLM applications. Reduce latency and cost by serving semantically similar responses from cache.

from semacache import CacheMetadata, CachePolicy, SemanticCache

async with SemanticCache.in_memory(namespace="demo") as cache:
    result = await cache.get_or_generate(
        query="What is the capital of France?",
        generator=lambda: "Paris",
        metadata=CacheMetadata(agent="bot"),
        policy=CachePolicy.READ,
    )
    print(result.status, result.value)

Features

  • Semantic caching — cache hits on semantically similar (not just identical) queries with configurable similarity threshold
  • Exact caching — fast SHA-256 keyed lookups as the first cache layer
  • Dual-layer architecture — exact + semantic cache for optimal latency
  • Pluggable backends — in-memory (default), Redis, or custom via protocols
  • Flexible deployment — single-process in-memory or distributed with Redis
  • Request coalescing — deduplicate concurrent identical LLM requests
  • Multi-tenant — namespace and tenant isolation built-in
  • Type-safe — generic SemanticCache[T] and CacheResult[T]
  • FastAPI integration — REST API for cache query, stats, invalidation, warmup
  • Extensible — protocols for stores, embeddings, serialization, validation, eviction, coalescing, metrics
  • Configurable failure policies — raise, bypass, return_none on failures
  • Response validation — reject empty or malformed responses before caching
  • Health checks — cache.health() returns per-component status
  • Serialization — pluggable ValueSerializer[T] with JSON support

Installation

pip install semacache

Optional extras:

pip install "semacache[fastapi]"    # FastAPI integration
pip install "semacache[prometheus]" # Prometheus metrics

Requires Python 3.11+.

Quick start

In-memory (no external services)

import asyncio
from semacache import CacheMetadata, CachePolicy, SemanticCache


async def main():
    async with SemanticCache.in_memory(namespace="quickstart", ttl_seconds=3600) as cache:
        meta = CacheMetadata(agent="demo")

        r1 = await cache.get_or_generate(
            query="What is semantic caching?",
            generator=lambda: "Caching based on meaning, not exact keys.",
            metadata=meta,
            policy=CachePolicy.READ,
        )
        print(f"{r1.status}: {r1.value}")  # miss: ...

        r2 = await cache.get_or_generate(
            query="What is semantic caching?",
            generator=lambda: "Caching based on meaning, not exact keys.",
            metadata=meta,
            policy=CachePolicy.READ,
        )
        print(f"{r2.status}: {r2.value}")  # exact_hit: ...

        r3 = await cache.get_or_generate(
            query="Tell me about semantic caching",
            generator=lambda: "Caching based on meaning, not exact keys.",
            metadata=meta,
            policy=CachePolicy.READ,
        )
        print(f"{r3.status}: {r3.value}")  # semantic_hit: ...


asyncio.run(main())

With Redis and Ollama

from semacache import SemanticCache, SemanticCacheConfig

config = SemanticCacheConfig(
    namespace="my-app",
    ttl_seconds=3600,
    exact_cache={"provider": "redis", "url": "redis://localhost:6379/0"},
    semantic_cache={"provider": "redis", "enabled": True, "similarity_threshold": 0.9},
    embeddings={"provider": "ollama", "model": "nomic-embed-text"},
)

cache = SemanticCache(config=config)

From YAML

config = SemanticCacheConfig.from_yaml("semacache.yaml")
# semacache.yaml
namespace: my-app
ttl_seconds: 3600
exact_cache:
  provider: redis
  url: redis://localhost:6379/0
semantic_cache:
  provider: redis
  enabled: true
  similarity_threshold: 0.9

From environment variables

config = SemanticCacheConfig()  # reads SEMACACHE_* env vars
# e.g. SEMACACHE_NAMESPACE=my-app SEMACACHE_TTL_SECONDS=7200

Usage

Decorator

from semacache import semantic_cached

cache = SemanticCache.in_memory(namespace="qa")


@semantic_cached(cache, policy=CachePolicy.READ)
async def ask(question: str) -> str:
    return f"Answer to: {question}"

Low-level API

# Lookup only
lookup = await cache.lookup(query="What is Python?", metadata=meta)
if lookup.hit:
    return lookup.entry.response

# Store only
await cache.store(query="What is Python?", value="A programming language", metadata=meta)

# Get stats
stats = await cache.get_stats()
print(stats.hit_rate)

# Invalidate
result = await cache.invalidate(InvalidateRequest(tenant="acme"))
print(f"Invalidated {result.invalidated_count} entries")

# Clear
count = await cache.clear()

Cache policies

from semacache import CachePolicy

# Normal operation — read from cache, fall back to generator
policy = CachePolicy.READ

# Bypass cache — always call generator, don't store result
policy = CachePolicy.BYPASS

# Refresh — always call generator, update cache
policy = CachePolicy.REFRESH

# Disabled — skip cache entirely
policy = CachePolicy.DISABLED

Cache metadata

from semacache import CacheMetadata

meta = CacheMetadata(
    tenant="acme-corp",           # Multi-tenant isolation
    user_scope="user-abc123",     # Per-user namespace
    model="gpt-4",                # Track which model generated the response
    agent="support-bot",          # Logical agent name
    extra={"custom_field": "x"},  # Arbitrary metadata dict
)

Serialization

from semacache import JSONSerializer, SemanticCache

# Store complex types as JSON
cache: SemanticCache[str] = SemanticCache.in_memory(
    serializer=JSONSerializer(),
)

Health check

health = await cache.health()
print(f"Healthy: {health.healthy}")
print(f"Exact cache: {health.exact_cache}")
print(f"Semantic cache: {health.semantic_cache}")

Request coalescing

Prevents duplicate LLM calls when multiple requests arrive simultaneously for the same cache miss:

config = SemanticCacheConfig(
    coalescing={
        "enabled": True,
        "mode": "memory",           # "redis" for distributed coalescing
        "lock_timeout_seconds": 30,
    },
)

Failure policies

Configure how the cache behaves when downstream components fail:

from semacache import SemanticCacheConfig

config = SemanticCacheConfig(
    failure_policy={
        "on_cache_read_failure": "bypass",      # raise, return_none, bypass
        "on_cache_write_failure": "log",         # raise, log
        "on_embedding_failure": "raise",         # raise, return_none
    },
)

Concurrency control

config = SemanticCacheConfig(
    max_concurrent_llm_requests=5,          # Max concurrent LLM calls
    llm_request_queue_timeout_seconds=60,   # Wait timeout for concurrency slot
    request_timeout_seconds=120,            # Per-request timeout
)

FastAPI integration

from semacache import SemanticCache
from semacache.integrations.fastapi import create_cache_app

cache = SemanticCache.in_memory(namespace="api")
app = create_cache_app(cache, require_api_key=False)

Run with: uvicorn app:app

Available endpoints

Method Path Description
POST /v1/query Query the cache
GET /health/live Liveness probe
GET /health/ready Readiness probe
GET /v1/cache/stats Cache statistics
POST /v1/cache/invalidate Invalidate cache entries
POST /v1/cache/warm Pre-warm cache with entries

API key auth

app = create_cache_app(cache, api_keys=["sk-123"], api_key_header="X-API-Key")

Configuration reference

Config field Default Description
namespace "default" Logical cache namespace
ttl_seconds 3600 Entry TTL in seconds
max_entries 10000 Max cache entries
max_memory_estimate_mb 512 Max estimated memory
eviction_strategy "ttl_lru" ttl, lru, lfu, size, ttl_lru
serve_stale False Serve stale entries when available
deterministic_test_mode False Deterministic embeddings for testing
request_timeout_seconds 120 Per-request timeout
max_concurrent_llm_requests 5 Max concurrent LLM calls
exact_cache.provider "memory" "memory" or "redis"
semantic_cache.provider "memory" "memory" or "redis"
semantic_cache.enabled True Enable semantic cache
semantic_cache.similarity_threshold 0.85 Min similarity for semantic hit
semantic_cache.similarity_strategy "cosine" "cosine" or "euclidean"
embeddings.provider "ollama" "ollama" or "deterministic"
embeddings.model "nomic-embed-text" Embedding model name
coalescing.enabled True Enable request coalescing
coalescing.mode "memory" "memory" or "redis"
failure_policy.on_cache_read_failure "bypass" raise, return_none, bypass
failure_policy.on_embedding_failure "raise" raise, return_none
validation.reject_empty True Reject empty responses
metrics.enabled False Enable metrics recording

API reference

SemanticCache

Method Returns Description
in_memory(namespace, ttl_seconds) SemanticCache Create cache with in-memory stores
lookup(query, metadata) CacheLookupResult Look up a query without generating
store(query, value, metadata) bool Store a value in cache
get_or_generate(query, generator, metadata, policy) CacheResult[T] Primary cache-through API
get_stats() CacheStats Cache statistics
invalidate(request) InvalidateResult Invalidate matching entries
clear() int Clear all entries
warm(query, value, metadata) bool Pre-warm a cache entry
health() HealthStatus Component health status
close() None Clean up resources

CacheResult[T]

Attribute Type Description
.value T The cached/generated value
.status CacheStatus Shortcut for cache_status
.cache_status CacheStatus Exact/semantic hit, miss...
.similarity_score float or None Similarity score if hit
.execution_info CacheExecutionInfo or None Timing and metadata

CacheMetadata

Attribute Type Description
tenant str Tenant identifier
user_scope str or None User-scoped namespace
model str or None LLM model name
agent str Logical agent name
extra dict[str, Any] Arbitrary metadata

Exception hierarchy

SemacacheError
├── CacheConfigurationError
├── CacheStoreError
├── EmbeddingError
├── CacheMissError
├── SerializationError
├── HealthCheckError
├── CacheClosedError
├── InvalidKeyError
└── ValidationError

Examples

See the examples/ directory:

  • basic_memory.py — Minimal in-memory example (no deps required)
  • redis_cache.py — Redis + Ollama backend
  • decorator_usage.py — Using @semantic_cached
  • custom_generator.py — Custom LLM generator function
  • custom_embedding_provider.py — Pluggable embedder
  • fastapi_integration.py — FastAPI app (run with uvicorn)
  • distributed_coalescing.py — Distributed request coalescing with Redis

Versioning

This library follows Semantic Versioning 2.0.0 (MAJOR.MINOR.PATCH).

  • Patch — bug fixes, performance improvements, docs
  • Minor — backward-compatible new features
  • Major — breaking API changes (announced at least one minor release in advance with deprecation warnings)

Migration from v1

See MIGRATION.md for a complete guide with before/after examples for every breaking change.

Extension

See EXTENSION.md for implementing custom:

  • Embedding providers
  • Cache stores (exact and semantic)
  • Serializers
  • Response validators
  • Eviction strategies
  • Request coalescers
  • Metrics recorders

Architecture

See ARCHITECTURE.md for:

  • Directory structure
  • Request flow sequence diagram
  • Store adapter class diagram
  • Extension point class diagram

Contributing

See CONTRIBUTING.md.

License

MIT

About

A library to employ semantic caching for LLM lookups

Topics

cachingllmollamaredissemantic-cachesemantic-search

Resources

Readme

Contributing

Contributing
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages