Semantic Caching for AI Agents: Reduce LLM Costs by 40–60% with Embedding-Based Response Caching

The bottom line: Semantic caching — returning cached LLM responses for semantically similar queries — is the single highest-leverage cost optimization for production agent systems. With embedding models at ~$0.02M input tokens and vector databases handling 10K+ queries/sec at sub-10ms latency, a well-tuned cache can serve 40–60% of agent requests from cache, cutting latency from seconds to milliseconds and reducing LLM spend by thousands of dollars per month 1. This guide walks through architecture, code, and production deployment.

---

Why Semantic Caching Matters for Agents

Standard LLM caching (exact-match on input text) catches almost nothing in agent systems. Agents frame the same intent in endlessly varied language:

• “What’s the weather in Tokyo?”
• “How’s the weather looking in Tokyo today?”
• “Tell me the current weather conditions for Tokyo, Japan”

An exact-match cache sees three different keys and makes three LLM calls. A semantic cache sees one intent and returns one cached response.

In production agent logs, 60–75% of user queries are semantically near-duplicates of previously seen questions 2. The distribution follows a power law: a small set of query intents accounts for the majority of volume. This is especially true for:

• Customer support agents — users ask the same 20 questions in 200 different phrasings
• Documentation RAG agents — users search for the same concepts with different wording
• Code review agents — the same code smells trigger the same analysis patterns
• Monitoring agents — recurring alert patterns generate semantically identical queries

The cost math is compelling. If your agent spends $5,000/month on LLM inference and a semantic cache serves 50% of requests from cache, you save $2,500/month — minus ~$50 in vector database hosting and ~$10 in embedding API costs.

Architecture Overview

A semantic cache sits between your application and the LLM, intercepting every agent request:

Application
    │
    ▼
┌──────────────────────┐
│  Semantic Cache      │
│                      │
│  Embed query → search│ ←── Vector DB (Chroma/Qdrant)
│  Similarity ≥ τ?     │
│  ├─ Yes → return     │
│  └─ No → call LLM    │────→ LLM API
│         store result │────→ Vector DB (write)
└──────────────────────┘

The critical parameter is the similarity threshold τ (tau). Set it too high and you miss valid cache hits (low recall). Set it too low and you return incorrect responses (low precision). Finding the right threshold for your use case is the central engineering challenge.

Step 1: Choose Your Embedding Model

The embedding model determines semantic quality. For caching, you want fast, cheap embeddings that cluster similar intents close in vector space.

Model	Dimensions	Cost per 1K inputs	Speed	Quality
text-embedding-3-small	512–1536	$0.02	Fast	Good
text-embedding-3-large	256–3072	$0.13	Slower	Best
BAAI/bge-small-en-v1.5	384	Free (local)	Fastest	Adequate
intfloat/e5-mistral-7b-instruct	4096	Free (local GPU)	Slow	Best (local)

For production caching, text-embedding-3-small (OpenAI) at 512 dimensions offers the best cost/quality tradeoff — $0.02 per 1K inputs, sub-50ms embedding time, and excellent semantic discrimination 3. For fully offline setups, BAAI/bge-small-en-v1.5 runs on CPU at 15ms per embedding with decent quality.

import openai

client = openai.Client(api_key="sk-...")

def embed(text: str, dimensions: int = 512) -> list[float]:
    """Generate embedding vector for a query string."""
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
        dimensions=dimensions,
    )
    return resp.data[0].embedding

Step 2: Set Up the Vector Store

ChromaDB is the easiest starting point — runs in-process, needs no separate server, and supports cosine similarity search out of the box. For production at scale, Qdrant or Pinecone offer better performance.

ChromaDB (development and low-traffic production)

import chromadb
from chromadb.utils import embedding_functions

# Using Chroma's built-in OpenAI embedding function
chroma_client = chromadb.PersistentClient(path="./cache_store")
collection = chroma_client.get_or_create_collection(
    name="semantic_cache",
    metadata={"hnsw:space": "cosine"},  # cosine similarity
)

Qdrant (high-traffic production)

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance

qdrant = QdrantClient("localhost", port=6333)
qdrant.recreate_collection(
    collection_name="semantic_cache",
    vectors_config=VectorParams(
        size=512,              # Must match embedding dimension
        distance=Distance.COSINE,
    ),
)

Step 3: Core Cache Logic

The cache has two operations — lookup and store — wrapped around a similarity search.

import numpy as np
from typing import Optional, Any

class SemanticCache:
    def __init__(
        self,
        threshold: float = 0.92,
        max_results: int = 3,
    ):
        self.threshold = threshold
        self.max_results = max_results
        # Initialize Chroma client
        self.chroma = chromadb.PersistentClient(path="./cache_store")
        self.collection = self.chroma.get_or_create_collection(
            name="semantic_cache",
            metadata={"hnsw:space": "cosine"},
        )

    def lookup(self, query: str) -> Optional[dict]:
        """Check if a semantically similar query exists in cache."""
        query_embedding = embed(query)

        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=self.max_results,
            include=["distances", "metadatas"],
        )

        if not results["distances"][0]:
            return None

        # Chroma returns cosine distance (0 = identical, 2 = opposite)
        # Convert to similarity: 1 - distance/2
        best_distance = results["distances"][0][0]
        best_similarity = 1 - best_distance / 2

        if best_similarity >= self.threshold:
            return {
                "response": results["metadatas"][0][0]["response"],
                "similarity": best_similarity,
                "cached_query": results["metadatas"][0][0]["query"],
            }
        return None

    def store(self, query: str, response: str) -> None:
        """Store a query-response pair in the cache."""
        query_embedding = embed(query)
        doc_id = str(hash(query))

        self.collection.upsert(
            ids=[doc_id],
            embeddings=[query_embedding],
            metadatas=[{
                "query": query,
                "response": response,
                "timestamp": str(datetime.now()),
            }],
        )

Step 4: Integrate with Your Agent Loop

The cache wraps the LLM call in your agent’s execution loop:

def agent_with_cache(user_query: str) -> str:
    """Agent loop with semantic caching."""
    # 1. Check cache first
    cached = cache.lookup(user_query)
    if cached:
        metrics.record_hit(cached["similarity"])
        logger.info(f"Cache hit (similarity={cached['similarity']:.3f})")
        return cached["response"]

    # 2. Cache miss — call LLM
    metrics.record_miss()
    response = llm_call(agent_prompt + user_query)

    # 3. Store result asynchronously
    executor.submit(cache.store, user_query, response)

    return response

For agent systems with multiple tool calls per turn, cache at the individual tool call level rather than the entire response. A weather agent might cache the get_forecast tool result while always computing the send_email tool fresh:

def execute_tool(tool_name: str, args: dict) -> str:
    """Execute a tool with caching for idempotent tools."""
    if tool_name in CACHEABLE_TOOLS:
        cache_key = f"{tool_name}:{json.dumps(args, sort_keys=True)}"
        cached = cache.lookup(cache_key)
        if cached:
            return cached["response"]

    result = call_downstream_api(tool_name, args)
    
    if tool_name in CACHEABLE_TOOLS:
        executor.submit(cache.store, cache_key, result)
    
    return result

Step 5: Production Hardening

TTL and Cache Invalidation

Stale responses are the biggest risk with semantic caching. Implement time-to-live (TTL) per cache entry and force-refresh for time-sensitive queries:

from datetime import datetime, timedelta

def store_with_ttl(
    self, query: str, response: str,
    ttl_minutes: int = 60
) -> None:
    """Store with expiration timestamp."""
    expires_at = datetime.now() + timedelta(minutes=ttl_minutes)
    # Include in metadata for filtering
    ...

def is_expired(self, metadata: dict) -> bool:
    """Check if cache entry has expired."""
    expires = datetime.fromisoformat(metadata["expires_at"])
    return datetime.now() > expires

Set TTL aggressively for volatile data (stock prices: 30 seconds) and leniently for stable data (company policies: 24 hours).

Query Normalization

Improve cache hit rates by normalizing queries before embedding:

import re

def normalize_query(query: str) -> str:
    """Normalize query to improve cache hit rate."""
    # Lowercase
    query = query.lower()
    # Strip punctuation
    query = re.sub(r'[^\w\s]', '', query)
    # Collapse whitespace
    query = re.sub(r'\s+', ' ', query).strip()
    # Remove stopwords (configurable list)
    stopwords = {"a", "an", "the", "is", "are", "was", "were"}
    tokens = [t for t in query.split() if t not in stopwords]
    return " ".join(tokens)

Multi-Tier Cache Architecture

For high-traffic systems, layer multiple caches:

class MultiTierCache:
    """L1: exact-match (μs), L2: semantic (ms), L3: LLM (seconds)."""
    
    def lookup(self, query: str) -> str:
        # L1: Exact match (in-memory dict, ~1μs)
        normalized = normalize_query(query)
        if normalized in self.exact_cache:
            metrics.record_hit("l1")
            return self.exact_cache[normalized]
        
        # L2: Semantic match (vector DB, ~10ms)
        semantic = self.semantic_cache.lookup(query)
        if semantic:
            metrics.record_hit("l2")
            return semantic["response"]
        
        # L3: LLM call (~1-5s)
        metrics.record_miss()
        response = call_llm(query)
        
        # Populate both caches
        self.exact_cache[normalized] = response
        executor.submit(self.semantic_cache.store, query, response)
        return response

Metrics and Monitoring

Every cache operation should emit metrics:

from prometheus_client import Counter, Histogram, Gauge

cache_hits = Counter("cache_hits_total", "Semantic cache hits", ["tier"])
cache_misses = Counter("cache_misses_total", "Semantic cache misses")
cache_latency = Histogram(
    "cache_lookup_duration_seconds",
    "Cache lookup latency",
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1],
)
cache_hit_ratio = Gauge("cache_hit_ratio", "Current cache hit ratio")
similarity_distribution = Histogram(
    "cache_similarity_score",
    "Similarity scores for cache hits",
    buckets=[0.7, 0.8, 0.85, 0.9, 0.95, 0.99],
)

Track these on a dashboard and alert when hit ratio drops below your baseline (e.g., <30% suggests threshold drift or query pattern shift).

Threshold Tuning: The Critical Parameter

The similarity threshold τ determines cache behavior. Tune it on your real data:

def tune_threshold(
    historical_queries: list[str],
    min_similarity: float = 0.75,
    max_similarity: float = 0.99,
    step: float = 0.01,
) -> dict:
    """Find optimal similarity threshold for your data."""
    # Build cache from first 80% of queries
    train = historical_queries[:int(len(historical_queries) * 0.8)]
    test = historical_queries[int(len(historical_queries) * 0.8):]
    
    # Pre-compute all embeddings
    train_embeddings = [embed(q) for q in train]
    test_embeddings = [embed(q) for q in test]
    
    results = []
    threshold = min_similarity
    while threshold <= max_similarity:
        hits = 0
        for i, test_emb in enumerate(test_embeddings):
            # Find nearest neighbor in train
            similarities = cosine_similarity(test_emb, train_embeddings)
            max_sim = max(similarities)
            if max_sim >= threshold:
                hits += 1
        
        hit_rate = hits / len(test)
        results.append({
            "threshold": threshold,
            "hit_rate": hit_rate,
            "estimated_savings": hit_rate * monthly_llm_cost,
        })
        threshold += step
    
    return results

In practice, most teams converge on τ = 0.88–0.95 depending on query diversity and the cost of stale responses 4. Customer support agents often use lower thresholds (τ ≈ 0.85) because similar questions deserve similar answers. Code generation agents use higher thresholds (τ ≈ 0.95) because incorrect caching can produce broken code.

Performance Benchmarks

These numbers come from a production deployment serving a documentation RAG agent handling 50K queries/day 5:

Metric	Without cache	With semantic cache	Improvement
p50 latency	2.8s	85ms	97% reduction
p95 latency	6.1s	210ms	96.6% reduction
LLM calls/day	50,000	19,500	61% reduction
Monthly LLM cost	$4,200	$1,638	$2,562/month saved
Vector DB cost	$0	$49/month	negligible

The 61% cache hit rate was achieved with τ = 0.91 on a corpus of 120K historical query-response pairs. The embedding step added 45ms median overhead to cache misses — far outweighed by the 2.8s saved on cache hits.

Anti-Patterns to Avoid

1. Caching non-deterministic outputs. Agent responses that depend on real-time data (stock prices, weather, server status) should have aggressive TTLs or bypass the cache entirely. Tag tools as cacheable: false in your tool registry.

2. Using the same threshold across all tools. A get_company_policy tool can use τ = 0.85. A generate_code_snippet tool needs τ = 0.95+. Configure thresholds per tool or per domain.

3. Ignoring embedding drift. Embedding models get deprecated and replaced. When you upgrade your embedding model, the old vector embeddings become incompatible with new queries. Maintain a version tag in your metadata and rebuild the index when the model version changes.

4. Caching PII-containing queries. User-specific queries that contain names, emails, or account IDs should be excluded from caching. Add a PII detector before the cache lookup:

def contains_pii(text: str) -> bool:
    """Basic PII detection for cache exclusion."""
    import re
    patterns = [
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # email
        r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',  # phone
        r'\b\d{16}\b',  # credit card
    ]
    return any(re.search(p, text) for p in patterns)

When Not to Use Semantic Caching

Semantic caching is not a universal solution. Skip it when:

• Your queries are all unique. If every user request is genuinely different (e.g., creative writing generation), you’ll see <10% hit rates.
• Latency isn’t a bottleneck. If your agent is already sub-500ms, the 45ms embedding overhead on cache misses is a net negative.
• Response accuracy is critical and context-dependent. Medical diagnosis or legal analysis agents should rarely serve cached responses.
• Your traffic is <1K queries/day. The engineering complexity and vector DB hosting cost aren’t worth the savings.

---

The 80/20 Takeaway

Start simple: ChromaDB + text-embedding-3-small at 512 dimensions with τ = 0.91. Instrument with Prometheus metrics. Run for one week, analyze the hit ratio distribution, then tune. The setup takes an afternoon and the savings start accruing immediately.

Most teams are surprised by how much semantic overlap exists in their agent traffic. The first week of metrics will tell you whether caching is worth the investment — and for the majority, it is.

---

📖 Related Reads

• ToolBrain — tool reviews, LLM comparisons, and AI workflow guides
• CodeIntel Log — code quality, debugging, and software engineering benchmarks

Cross-links automatically generated from NiteAgent.

← Back to all posts