Agent Caching Strategies

Why This Matters

LLM API calls are expensive and slow. A single GPT-4 call costs 0.06 and takes 500ms-2s. For a production agent handling 10K requests/day, that's 600/day in API costs and 5-20K seconds of LLM compute. Caching transforms this: an 87% cache hit rate means only 1,300 requests hit the LLM, saving ~$500/day and serving cached responses in <5ms instead of 500ms. Caching isn't just an optimization—it's the difference between a profitable and unprofitable AI product.

Real-World Analogy

Caching an agent is like a librarian's memory. When a patron asks "What's the capital of France?", the librarian doesn't go to the shelf—they already know (L1: in-memory). When asked about a less common fact, they remember where they saw it recently (L2: Redis). For older questions, they check the card catalog (L3: database). Only when nothing is cached do they walk to the stacks to find the book (LLM API). The librarian who remembers 87% of questions answers the room in seconds, not minutes.

Multi-Layer Cache System

import asyncio
import hashlib
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Optional
from collections import OrderedDict

logger = logging.getLogger(__name__)


@dataclass
class CacheEntry:
    key: str
    value: Any
    timestamp: float
    ttl: float
    access_count: int = 0
    size_bytes: int = 0


class LRUCache:
    def __init__(self, max_size: int = 1000, default_ttl: float = 300.0) -> None:
        self._max_size = max_size
        self._default_ttl = default_ttl
        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> Optional[Any]:
        if key in self._cache:
            entry = self._cache[key]
            if time.time() - entry.timestamp < entry.ttl:
                self._cache.move_to_end(key)
                entry.access_count += 1
                self._hits += 1
                return entry.value
            del self._cache[key]
        self._misses += 1
        return None

    def set(self, key: str, value: Any, ttl: Optional[float] = None) -> None:
        if key in self._cache:
            del self._cache[key]
        entry = CacheEntry(
            key=key, value=value, timestamp=time.time(),
            ttl=ttl or self._default_ttl,
            size_bytes=len(str(value).encode()),
        )
        self._cache[key] = entry
        self._cache.move_to_end(key)
        while len(self._cache) > self._max_size:
            self._cache.popitem(last=False)

    def delete(self, key: str) -> None:
        self._cache.pop(key, None)

    def clear(self) -> None:
        self._cache.clear()
        self._hits = 0
        self._misses = 0

    def stats(self) -> dict[str, Any]:
        total = self._hits + self._misses
        return {
            "size": len(self._cache), "max_size": self._max_size,
            "hits": self._hits, "misses": self._misses,
            "hit_rate": self._hits / total if total else 0.0,
            "total_bytes": sum(e.size_bytes for e in self._cache.values()),
        }


class MultiLayerCache:
    def __init__(
        self,
        l1_max_size: int = 1000,
        l1_ttl: float = 300,
        l2_ttl: float = 3600,
        l3_ttl: float = 86400,
    ) -> None:
        self._l1 = LRUCache(max_size=l1_max_size, default_ttl=l1_ttl)
        self._l2: dict[str, CacheEntry] = {}
        self._l3: dict[str, CacheEntry] = {}
        self._l2_ttl = l2_ttl
        self._l3_ttl = l3_ttl

    def _make_key(self, query: str, model: str, **kwargs: Any) -> str:
        key_data = f"{query}:{model}:{sorted(kwargs.items())}"
        return hashlib.md5(key_data.encode()).hexdigest()

    async def get(self, query: str, model: str, **kwargs: Any) -> Optional[Any]:
        key = self._make_key(query, model, **kwargs)

        result = self._l1.get(key)
        if result is not None:
            return result

        result = self._get_l2(key)
        if result is not None:
            self._l1.set(key, result)
            return result

        result = self._get_l3(key)
        if result is not None:
            self._l1.set(key, result)
            self._set_l2(key, result)
            return result

        return None

    async def set(self, query: str, model: str, value: Any, **kwargs: Any) -> None:
        key = self._make_key(query, model, **kwargs)
        self._l1.set(key, value)
        self._set_l2(key, value)
        self._set_l3(key, value)

    def _get_l2(self, key: str) -> Optional[Any]:
        if key in self._l2:
            entry = self._l2[key]
            if time.time() - entry.timestamp < entry.ttl:
                return entry.value
            del self._l2[key]
        return None

    def _set_l2(self, key: str, value: Any) -> None:
        self._l2[key] = CacheEntry(key=key, value=value, timestamp=time.time(), ttl=self._l2_ttl)

    def _get_l3(self, key: str) -> Optional[Any]:
        if key in self._l3:
            entry = self._l3[key]
            if time.time() - entry.timestamp < entry.ttl:
                return entry.value
            del self._l3[key]
        return None

    def _set_l3(self, key: str, value: Any) -> None:
        self._l3[key] = CacheEntry(key=key, value=value, timestamp=time.time(), ttl=self._l3_ttl)

    def stats(self) -> dict[str, Any]:
        return {
            "l1": self._l1.stats(),
            "l2": {"size": len(self._l2)},
            "l3": {"size": len(self._l3)},
        }

Semantic Cache

import hashlib
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Optional

import numpy as np

logger = logging.getLogger(__name__)


@dataclass
class SemanticCacheEntry:
    query: str
    embedding: np.ndarray
    response: Any
    timestamp: float
    ttl: float = 3600.0
    hit_count: int = 0


class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.92, max_size: int = 10_000) -> None:
        self._threshold = similarity_threshold
        self._max_size = max_size
        self._entries: list[SemanticCacheEntry] = []
        self._hits = 0
        self._misses = 0

    def _embed(self, text: str) -> np.ndarray:
        words = text.lower().split()
        embedding = np.zeros(384)
        for i, word in enumerate(words[:384]):
            hash_val = int(hashlib.md5(word.encode()).hexdigest(), 16) % 1000
            embedding[i] = hash_val / 1000.0
        norm = np.linalg.norm(embedding)
        return embedding / norm if norm > 0 else embedding

    @staticmethod
    def _cosine(a: np.ndarray, b: np.ndarray) -> float:
        dot = np.dot(a, b)
        norm_a, norm_b = np.linalg.norm(a), np.linalg.norm(b)
        return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

    def get(self, query: str) -> Optional[Any]:
        query_emb = self._embed(query)
        best_entry = None
        best_sim = 0.0
        now = time.time()

        for entry in self._entries:
            if now - entry.timestamp > entry.ttl:
                continue
            sim = self._cosine(query_emb, entry.embedding)
            if sim > best_sim:
                best_sim = sim
                best_entry = entry

        if best_entry and best_sim >= self._threshold:
            best_entry.hit_count += 1
            self._hits += 1
            return best_entry.response

        self._misses += 1
        return None

    def set(self, query: str, response: Any, ttl: float = 3600.0) -> None:
        embedding = self._embed(query)
        self._entries.append(SemanticCacheEntry(
            query=query, embedding=embedding, response=response,
            timestamp=time.time(), ttl=ttl,
        ))
        if len(self._entries) > self._max_size:
            self._entries.sort(key=lambda e: e.hit_count)
            self._entries = self._entries[1:]

    def clear(self) -> None:
        self._entries.clear()
        self._hits = 0
        self._misses = 0

    def stats(self) -> dict[str, Any]:
        total = self._hits + self._misses
        return {
            "size": len(self._entries), "hits": self._hits, "misses": self._misses,
            "hit_rate": self._hits / total if total else 0.0,
            "avg_hits": sum(e.hit_count for e in self._entries) / len(self._entries) if self._entries else 0,
        }

Prompt Cache

import hashlib
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class PromptCacheStrategy(Enum):
    PREFIX = "prefix"
    FULL = "full"
    PARTIAL = "partial"


@dataclass
class PromptCacheEntry:
    prefix: str
    response: Any
    token_count: int
    timestamp: float
    ttl: float = 86400.0
    hit_count: int = 0


class PromptCache:
    def __init__(
        self,
        strategy: PromptCacheStrategy = PromptCacheStrategy.PREFIX,
        max_prefix_length: int = 500,
    ) -> None:
        self._strategy = strategy
        self._max_prefix = max_prefix_length
        self._cache: dict[str, PromptCacheEntry] = {}
        self._hits = 0
        self._misses = 0
        self._total_saved = 0

    def _extract_prefix(self, prompt: str) -> str:
        if self._strategy == PromptCacheStrategy.PREFIX:
            return prompt[:self._max_prefix]
        elif self._strategy == PromptCacheStrategy.FULL:
            return prompt
        else:
            lines = prompt.split("\n")
            prefix_lines: list[str] = []
            length = 0
            for line in lines:
                if length + len(line) > self._max_prefix:
                    break
                prefix_lines.append(line)
                length += len(line)
            return "\n".join(prefix_lines)

    def _key(self, prefix: str) -> str:
        return hashlib.sha256(prefix.encode()).hexdigest()[:32]

    def get(self, prompt: str) -> Optional[Any]:
        prefix = self._extract_prefix(prompt)
        key = self._key(prefix)
        entry = self._cache.get(key)
        if entry and time.time() - entry.timestamp < entry.ttl:
            entry.hit_count += 1
            self._hits += 1
            self._total_saved += entry.token_count
            return entry.response
        self._misses += 1
        return None

    def set(self, prompt: str, response: Any, token_count: int) -> None:
        prefix = self._extract_prefix(prompt)
        key = self._key(prefix)
        self._cache[key] = PromptCacheEntry(
            prefix=prefix, response=response, token_count=token_count,
            timestamp=time.time(),
        )

    def clear(self) -> None:
        self._cache.clear()
        self._hits = 0
        self._misses = 0
        self._total_saved = 0

    def stats(self) -> dict[str, Any]:
        total = self._hits + self._misses
        return {
            "size": len(self._cache), "hits": self._hits, "misses": self._misses,
            "hit_rate": self._hits / total if total else 0.0,
            "tokens_saved": self._total_saved,
            "avg_per_hit": self._total_saved / self._hits if self._hits else 0,
        }

Mathematical Foundations

Cache Hit Rate:

Semantic Similarity (Cosine):

Cost Savings:

Optimal Cache Size (square-root rule):

Cache Throughput:

Latency Improvement:

Performance Considerations

Cache Layer	Latency	Hit Rate	Cost	Best For
L1 (In-Memory)	<2ms	~60%	Free	Hot data, recent queries
L2 (Redis)	2-10ms	~25%	Low	Shared sessions, distributed
L3 (Database)	10-50ms	~10%	Medium	Persistent, large datasets
Semantic Cache	5-50ms	~40%	Medium	Q&A with paraphrases
Prompt Cache	<2ms	~30%	Free	Repeated system prompts
LLM API	200-2000ms	0%	High	Cache misses only

Security Considerations

Encrypt cached data at rest especially if storing user conversations or PII.
Set cache TTLs based on data sensitivity—shorter for personal data, longer for public facts.
Prevent cache poisoning by validating cache keys and sanitizing inputs before caching.
Access control on shared caches (Redis, database) to prevent unauthorized reads.
Audit cache invalidation events to detect tampering or data manipulation.
Don't cache error responses that might contain sensitive system information.

Common Pitfalls

Pitfall	Consequence	Solution
Stale cache data	Serves outdated information	Implement proper TTL and event-driven invalidation
Cache stampede	Thundering herd on expiry	Use probabilistic early expiration or lock-based refresh
Memory leaks	OOM crashes	Monitor and enforce size limits with LRU eviction
No cache warming	Cold-start latency spikes	Pre-populate for predictable workloads
Ignoring cache coherence	Inconsistent data across instances	Use write-through or pub/sub invalidation
Over-caching	Wasted memory, stale data	Monitor hit rates and adjust TTLs per use case
Single point of failure	Complete cache outage	Use distributed cache with replication
No cache monitoring	Blind to performance	Track hit rates, latency, and eviction rates

Interview Q&A

1. What is the difference between exact match and semantic caching?

Exact match caching uses hash-based lookup—it's fast (O(1)) and reliable but misses semantically similar queries. Semantic caching uses embedding similarity to find related queries—it handles paraphrases and variations but requires vector computation (O(n) scan or approximate nearest neighbor). Use exact match for deterministic API responses where identical inputs must produce identical outputs. Use semantic caching for Q&A systems where users ask the same question in different ways ("What's the capital of France?" ≈ "Which city is France's capital?").

2. How do you implement a multi-layer cache hierarchy?

Multi-layer cache: (1) L1 (in-memory dict or LRU) for fastest access (<2ms), (2) L2 (Redis) for distributed caching across instances (2-10ms), (3) L3 (PostgreSQL or DynamoDB) for persistent cache (10-50ms). On cache miss, check each layer in order. On hit at a lower layer, promote the entry to higher layers. Use shorter TTLs for higher layers (L1: 5min, L2: 1hr, L3: 24hr) to ensure freshness. Implement cache warming by pre-populating L1 from historical access patterns during startup.

3. What is prompt caching and when should you use it?

Prompt caching stores responses for identical or similar prompt prefixes. Use when: (1) system prompts are reused across many conversations, (2) many users ask similar questions to the same knowledge base, (3) conversation context has common prefixes (e.g., "You are a helpful assistant for [company]"). Benefits: reduces token costs (cached prefixes are cheaper in some APIs), decreases latency (fewer tokens to process), improves throughput. Limitations: requires cache management, may serve stale data if prompts change. Use for system prompts and templates, not for dynamic user-specific content.

4. How do you handle cache invalidation?

Invalidation strategies: (1) TTL-based—simple, automatic, but may serve stale data, (2) event-driven—invalidate when underlying data changes (webhook, database trigger), (3) version-based—embed version in cache key, new version = new key, (4) manual—admin cache clearing for emergencies, (5) LRU eviction—natural capacity management. Choose based on data volatility: TTL for semi-static data (product descriptions), event-driven for dynamic data (prices, inventory), version-based for schema changes. The hardest problem in computer science is cache invalidation—start simple (TTL) and add complexity as needed.

5. What metrics should you monitor for cache performance?

Key metrics: (1) hit rate (target >80% for well-configured caches), (2) miss rate and miss reasons (expired vs. never cached), (3) eviction rate (indicates cache is too small or TTLs too short), (4) cache size and memory usage (prevent OOM), (5) latency improvement (cache response time vs. LLM response time), (6) cost savings (total $ saved by cached responses), (7) throughput increase (cached requests served per second). Set up alerts for hit rate drops below 70% and cache size approaching limits. Use dashboards (Grafana) for real-time visibility.

6. How do you implement cache warming?

Cache warming: (1) pre-populate with predicted queries based on historical access patterns, (2) use a background job to seed cache from database during low-traffic periods, (3) implement write-through caching—cache new data as it's created, (4) batch warm during deployment or scheduled maintenance, (5) use predictive analytics to forecast which queries will be popular. Monitor hit rates after warming to measure effectiveness. Start with the top 100-1000 most frequent queries—these typically cover 50-80% of traffic (Zipf's law).

7. What are the tradeoffs of different caching strategies?

Tradeoffs: (1) Exact vs. semantic match—accuracy vs. flexibility (exact misses paraphrases, semantic has false positives), (2) TTL vs. event-based invalidation—simplicity vs. freshness (TTL may serve stale data, event-driven requires infrastructure), (3) Memory vs. distributed cache—speed vs. scalability (in-memory is fast but single-instance, Redis is distributed but adds latency), (4) Aggressive vs. conservative caching—cost vs. staleness (aggressive saves more but may serve outdated data). Choose based on: data volatility (high → short TTL), latency requirements (low → in-memory), consistency needs (strong → event-driven), and budget (limited → aggressive caching).

8. How do you test cache correctness?

Testing approach: (1) unit tests for cache operations—get, set, delete, TTL expiration, LRU eviction, (2) integration tests for cache coherence—verify multi-layer promotion and invalidation propagation, (3) load tests for performance—measure hit rates under realistic traffic patterns, (4) chaos tests for failure scenarios—what happens when Redis goes down? Does the system degrade gracefully to LLM, or does it crash?, (5) A/B tests for optimization—compare hit rates and latency with different TTLs and strategies. Verify: data consistency (same key returns same value), TTL behavior (expired entries are evicted), eviction policy (LRU evicts least recently used), and failure recovery (cache outage doesn't break the agent).

KnowledgeCheck

What is the primary advantage of semantic caching over exact match?
- a) Faster lookup speed
- b) Handles query variations and paraphrases
- c) Lower memory usage
- d) Simpler implementation
What is a typical target cache hit rate for a well-configured system?
- a) 50%
- b) 70%
- c) 80%
- d) 95%
What does TTL stand for in caching?
- a) Total Time Limit
- b) Time To Live
- c) Timeout Threshold Level
- d) Token Total Limit
What is prompt caching used for?
- a) Caching database queries
- b) Storing responses for repeated prompt patterns
- c) Managing memory allocation
- d) Load balancing across servers
What is cache stampede?
- a) Multiple requests hitting an expired cache key simultaneously
- b) Cache memory overflow
- c) Network timeout during cache lookup
- d) Database connection pool exhaustion
What is the benefit of multi-layer caching?
- a) Simplifies code architecture
- b) Balances speed, cost, and persistence across layers
- c) Reduces code complexity
- d) Eliminates need for monitoring

Answers: 1-b, 2-c, 3-b, 4-b, 5-a, 6-b

Agent Caching Strategies: Semantic Caching, Result Caching & Prompt Caching