How semantic caching reduces redundant LLM calls

Semantic caching for LLM calls stores responses keyed by meaning rather than exact string match, so near-duplicate prompts skip the model entirely. This how-to builds a working cache layer in Python that wraps any OpenAI-compatible chat endpoint and shows how to measure the savings.

Step 1: Choose an embedding model and cache store

The core idea behind semantic caching for LLM calls is to embed the prompt, then compare that embedding against previously seen prompts. You need two pieces: an embedding model and a store that can do similarity search.

For a prototype, an in-memory list with cosine similarity is enough. In production you’ll want a vector index (pgvector, Redis Stack, or a dedicated ANN engine) because linear scans degrade past a few thousand entries.

Pick a small, fast embedding model. text-embedding-3-small gives 1536 dimensions and low latency. Set a similarity threshold up front—0.90 to 0.95 is a sane starting band for natural language paraphrases.

import time
import numpy as np
from openai import OpenAI

EMBED_MODEL = "text-embedding-3-small"
SIMILARITY_THRESHOLD = 0.92

class SemanticCache:
    def __init__(self, ttl_seconds=3600, max_entries=1000):
        self.entries = []  # (embedding, response, timestamp)
        self.ttl = ttl_seconds
        self.max = max_entries
        self.hits = 0
        self.misses = 0

    def _embed(self, text: str) -> np.ndarray:
        client = OpenAI()  # reads OPENAI_API_KEY
        resp = client.embeddings.create(model=EMBED_MODEL, input=text)
        return np.array(resp.data[0].embedding)

Step 2: Implement cosine similarity and threshold lookup

Cosine similarity is the dot product of two normalized vectors. Cache hits are entries above your threshold and within TTL.

    def query(self, prompt: str):
        vec = self._embed(prompt)
        now = time.time()
        best_sim = 0.0
        best_entry = None
        for emb, resp, ts in self.entries:
            if now - ts > self.ttl:
                continue
            norm = np.linalg.norm(vec) * np.linalg.norm(emb)
            sim = np.dot(vec, emb) / norm if norm > 0 else 0.0
            if sim > best_sim:
                best_sim = sim
                best_entry = resp
        if best_entry is not None and best_sim >= SIMILARITY_THRESHOLD:
            self.hits += 1
            return best_entry, best_sim
        self.misses += 1
        return None, best_sim

    def add(self, prompt: str, response: dict):
        vec = self._embed(prompt)
        self.entries.append((vec, response, time.time()))
        if len(self.entries) > self.max:
            self.entries.pop(0)

Keep the similarity loop O(n). If you later move to a real vector DB, this method gets replaced by a single query call.

Step 3: Build the cache wrapper

This wrapper implements semantic caching for LLM calls without changing your call site. It checks the cache, falls back to the model, and stores the result.

def make_cached_client(base_url=None, api_key=None):
    client = OpenAI(base_url=base_url, api_key=api_key)
    cache = SemanticCache()

    def get_completion(prompt: str, model: str = "gpt-4o-mini"):
        cached, score = cache.query(prompt)
        if cached:
            cached["cache_hit"] = True
            cached["similarity"] = round(score, 4)
            return cached
        resp = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        result = {
            "content": resp.choices[0].message.content,
            "model": model,
            "cache_hit": False,
        }
        cache.add(prompt, result)
        return result

    return get_completion, cache

The returned cache object exposes hits and misses so you can compute hit rate later.

Step 4: Point the client at a gateway and forward cache hints

In real deployments you rarely call providers directly. Point your OpenAI client at a gateway such as n4n.ai’s OpenAI-compatible endpoint (a single base URL covering 240+ models with automatic fallback). It forwards provider cache-control hints, so you can combine semantic caching at the app layer with provider-level prompt caching.

get_completion, cache = make_cached_client(
    base_url="https://api.n4n.ai/v1",
    api_key="your-gateway-key"
)

# Optional: pass routing directive via extra headers if your gateway supports it
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1")
client.chat.completions.create(
    model="anthropic/claude-3.5-sonnet",
    messages=[{"role": "user", "content": "Summarize: ..."}],
    extra_headers={"x-routing": "prefer=us-east"}
)

The semantic cache sits in front of this call and intercepts repeats before they ever leave your process.

Step 5: Add TTL, eviction, and invalidation controls

The SemanticCache constructor already takes ttl_seconds and max_entries. For multi-tenant systems, add a namespace parameter so different users don’t cross-contaminate cached answers.

class NamespacedCache:
    def __init__(self):
        self.stores = {}

    def for_namespace(self, ns: str) -> SemanticCache:
        if ns not in self.stores:
            self.stores[ns] = SemanticCache(ttl_seconds=1800, max_entries=500)
        return self.stores[ns]

Invalidation is explicit: delete entries when underlying data changes. Expose a invalidate_prefix method that embeds a topic tag and drops matches above 0.8 similarity.

Step 6: Measure savings and prove it works

To prove semantic caching for LLM calls works, track token flow. Wrap the completion call with a token counter using resp.usage if your gateway returns it. n4n.ai provides per-token usage metering, so you can diff billed tokens with and without the cache.

import json

prompts = [
    "What is the refund policy for annual plans?",
    "How do I get a refund on my yearly subscription?",
    "Explain the difference between TCP and UDP.",
]

fn, cache = make_cached_client(base_url="https://api.n4n.ai/v1")
for p in prompts:
    out = fn(p)
    print(json.dumps({"prompt": p[:30], **out}, indent=2))

print(f"Hit rate: {cache.hits}/{cache.hits + cache.misses}")

The first two prompts are semantic paraphrases. Expect the second to return cache_hit: True with similarity ≥ 0.92. The third is unrelated and misses.

Verify success

Run the script above. Success criteria:

1. The second prompt returns cache_hit: True and similarity above your threshold.
2. Hit rate: prints at least 1/3.
3. Token usage on the hit is zero at the model layer (your embedding call still costs a few tokens, which is the cache overhead).

If you see misses on clear paraphrases, lower SIMILARITY_THRESHOLD to 0.88 and re-test. If you see false hits on unrelated prompts, raise it to 0.95.

Semantic caching for LLM calls is not a silver bullet—it breaks on factual volatility and user-specific data—but for stable knowledge queries it removes the majority of redundant spend with about sixty lines of code.