Semantic Cache Design

import hashlib
import time

class SemanticCache:
    def __init__(self, vector_db, embedding_fn, similarity_threshold=0.92):
        self.vector_db = vector_db
        self.embed = embedding_fn
        self.threshold = similarity_threshold
        self.stats = {"hits": 0, "misses": 0}

    def get(self, query, filters=None):
        """Look up a cached response for a similar query."""
        query_emb = self.embed(query)
        results = self.vector_db.search(
            query_emb,
            top_k=1,
            filters=filters
        )

        if results and results[0]["similarity"] >= self.threshold:
            entry = results[0]
            if not self.is_expired(entry):
                self.stats["hits"] += 1
                log_cache_hit(query, entry["similarity"], entry["cached_at"])
                return {
                    "response": entry["response"],
                    "cached": True,
                    "similarity": entry["similarity"],
                    "age_hours": (time.time() - entry["cached_at"]) / 3600
                }

        self.stats["misses"] += 1
        return None

    def set(self, query, response, filters=None, ttl_hours=24):
        """Cache a query-response pair."""
        query_emb = self.embed(query)
        cache_key = hashlib.sha256(query.encode()).hexdigest()

        self.vector_db.insert(
            text=query,
            embedding=query_emb,
            metadata={
                "response": response,
                "cached_at": time.time(),
                "ttl_hours": ttl_hours,
                "cache_key": cache_key,
                **(filters or {})
            }
        )

    def is_expired(self, entry):
        """Check if a cache entry has expired."""
        age_hours = (time.time() - entry["cached_at"]) / 3600
        return age_hours > entry.get("ttl_hours", 24)

    def hit_rate(self):
        """Calculate cache hit rate."""
        total = self.stats["hits"] + self.stats["misses"]
        return self.stats["hits"] / total if total > 0 else 0

class CachePolicy:
    def __init__(self):
        self.policies = {
            "runbook_lookup": {"ttl_hours": 168, "invalidate_on": ["runbook_update"]},
            "alert_classification": {"ttl_hours": 0.5, "invalidate_on": []},
            "incident_summary": {"ttl_hours": 24, "invalidate_on": ["incident_resolved"]},
            "status_page": {"ttl_hours": 1, "invalidate_on": ["status_change"]},
        }

    def get_ttl(self, query_type):
        """Get TTL for a query type."""
        return self.policies.get(query_type, {}).get("ttl_hours", 24)

    def invalidate(self, query_type, event_type):
        """Invalidate cache entries based on events."""
        if event_type in self.policies.get(query_type, {}).get("invalidate_on", []):
            self.vector_db.delete_by_metadata({"query_type": query_type})
            log_invalidation(query_type, event_type)

# Example: invalidate runbook cache when runbook is updated
def on_runbook_updated(runbook_path):
    cache.invalidate("runbook_lookup", "runbook_update")

def warm_cache(queries, cache, rag_pipeline):
    """Pre-warm cache with common queries."""
    for query in queries:
        # Check if already cached
        existing = cache.get(query)
        if existing:
            continue

        # Generate response
        response = rag_pipeline.answer(query)

        # Cache it
        cache.set(query, response, ttl_hours=168)

    print(f"Cache warmed with {len(queries)} queries")

# Common queries from search logs
common_queries = [
    "How do I failover Redis?",
    "What is the pod scaling procedure?",
    "How to check disk usage on a node?",
    "What causes OOMKill errors?",
    "How to roll back a deployment?",
]