AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build a persistent semantic cache

Build a Persistent Semantic Cache

An in-memory semantic cache loses all entries on restart. A production semantic cache needs persistence through a key-value store, TTL-based expiration, and similarity-based lookups.

What to build

PersistentSemanticCache(store: dict, embedding_fn, similarity_threshold=0.92, default_ttl=3600):

CacheEntry: query, response, embedding, model, created_at, ttl_seconds, hit_count. Add is_expired property.
async get(query) -> CacheEntry | None — embed query, find best non-expired match above threshold, increment hit_count
async set(query, response, model, ttl=None) -> str — returns cache key
evict_expired() -> int
stats() -> dict — total_entries, expired_entries, total_hits, hit_rate

Cache key: first 16 hex chars of sha256(json(embedding)).

Constraints

embedding_fn is sync: (text) -> list[float]
Implement cosine similarity (no numpy)
store dict simulates Redis (serialize entries as JSON strings)

Api Async / hard / Step 19 of 23

Practice stage

Async and provider control

Hint

Make waiting behavior explicit. Timeouts, retries, and concurrency limits matter more than squeezing everything into one helper.

Next drill

Async provider client with semaphore rate limiting

Success criteria

- Uses async boundaries coherently
- Makes timeout and retry decisions legible
- Would be maintainable under provider instability

Review checklist

- Is timeout behavior explicit?
- Is retryable failure separate from terminal failure?
- Would logs reveal what actually timed out?

from __future__ import annotations
import hashlib, json, math, time
from dataclasses import dataclass, asdict

@dataclass
class CacheEntry:
    query: str
    response: str
    embedding: list[float]
    model: str
    created_at: float
    ttl_seconds: int
    hit_count: int = 0

@property
    def is_expired(self) -> bool:
        raise NotImplementedError

class PersistentSemanticCache:
    def __init__(self, store: dict, embedding_fn, similarity_threshold: float = 0.92, default_ttl: int = 3600):
        raise NotImplementedError

def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        raise NotImplementedError

def _cache_key(self, embedding: list[float]) -> str:
        raise NotImplementedError

async def get(self, query: str) -> CacheEntry | None:
        raise NotImplementedError

async def set(self, query: str, response: str, model: str, ttl: int | None = None) -> str:
        raise NotImplementedError

def evict_expired(self) -> int:
        raise NotImplementedError

def stats(self) -> dict:
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.