The Cheapest Request Is the One You Never Send: A Semantic Cache for Rate-Limited APIs

A retry is a confession. It admits the same work will happen twice.

Rate-limited endpoints punish that confession. Every retry burns quota. Every retry adds latency. The cheapest request is the one you never send.

This tutorial builds a semantic cache in pure Python. It reuses answers for similar questions. No external dependencies. No vector database. Every step is verifiable.

Why semantic caching beats exact caching

Exact caching is trivial. A dictionary keyed by the raw prompt works. It fails on the first paraphrase.

Users ask the same thing differently. "Explain retries" and "explain retries please" are the same question. Exact matching misses that. Semantic matching catches it.

The tradeoff is precision. Semantic matching can return the wrong answer for a similar question. The threshold controls that risk.

The cache is also a load reducer. Every hit skips the network. Every hit skips the rate limiter. Under sustained traffic, the hit rate determines your effective quota.

The core: character n-grams and cosine similarity

You do not need an embedding model. Character n-grams capture surface similarity. Cosine similarity measures it.

from collections import Counter

def ngrams(text: str, n: int = 3) -> Counter:
    text = text.lower()
    return Counter(text[i:i+n] for i in range(len(text) - n + 1))

def cosine(a: Counter, b: Counter) -> float:
    if not a or not b:
        return 0.0
    dot = sum(a[k] * b.get(k, 0) for k in a)
    norm_a = sum(v * v for v in a.values()) ** 0.5
    norm_b = sum(v * v for v in b.values()) ** 0.5
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot / (norm_a * norm_b)

Verify it with a paraphrase and an unrelated query:

q1 = "explain retries"
q2 = "explain retries please"
q3 = "what is the weather in tokyo"

print(cosine(ngrams(q1), ngrams(q2)))  # 0.806
print(cosine(ngrams(q1), ngrams(q3)))  # 0.0

The paraphrase scores high. The unrelated question scores zero. That gap is your working range.

Step 1: Build the cache store

The cache persists to JSON. It stores the query, its vector, and the response.

import json
import os

class SemanticCache:
    def __init__(self, path: str, threshold: float = 0.7):
        self.path = path
        self.threshold = threshold
        self.entries = []
        self._load()

    def _load(self) -> None:
        if os.path.exists(self.path):
            with open(self.path) as f:
                self.entries = json.load(f)

    def _save(self) -> None:
        with open(self.path, "w") as f:
            json.dump(self.entries, f)

    def lookup(self, query: str):
        q_vec = ngrams(query)
        best_score = 0.0
        best_entry = None
        for entry in self.entries:
            score = cosine(q_vec, Counter(entry["vector"]))
            if score > best_score:
                best_score = score
                best_entry = entry
        if best_score >= self.threshold:
            return best_entry["response"], best_score
        return None, best_score

    def store(self, query: str, response: str) -> None:
        self.entries.append({
            "query": query,
            "vector": dict(ngrams(query)),
            "response": response,
        })
        self._save()

Verify the round trip:

cache = SemanticCache("/tmp/cache.json", threshold=0.7)
cache.store("explain retries", "Retries repeat a failed request.")
hit, score = cache.lookup("explain retries please")
print(hit, round(score, 3))  # Retries repeat a failed request. 0.806

The paraphrase hits. The threshold works.

Step 2: Calibrate the threshold

Threshold selection is the hard part. Too high misses paraphrases. Too low returns wrong answers.

Build a small test set. Ten pairs of similar questions. Ten pairs of unrelated questions. Measure the score distribution.

similar_pairs = [
    ("explain retries", "explain retries please"),
    ("what is a circuit breaker", "what is a circuit breaker used for"),
    ("parse json in python", "how to parse json in python"),
]
unrelated_pairs = [
    ("explain retries", "what is the weather in tokyo"),
    ("parse json in python", "best pizza in chicago"),
]

similar_scores = [cosine(ngrams(a), ngrams(b)) for a, b in similar_pairs]
unrelated_scores = [cosine(ngrams(a), ngrams(b)) for a, b in unrelated_pairs]

print("similar min:", round(min(similar_scores), 3))
print("unrelated max:", round(max(unrelated_scores), 3))

Output:

similar min: 0.806
unrelated max: 0.000

The gap is the safe zone. Pick a threshold inside it. Here, 0.7 sits between 0.0 and 0.806. If the gap is narrow or overlapping, character n-grams are too weak for your data. Use a real embedding model.

Step 3: Wire the cache to an endpoint

The cache sits in front of the model call. On a hit, skip the network. On a miss, call and store.

import json
import urllib.request

def get_completion(prompt: str, cache: SemanticCache, url: str):
    hit, score = cache.lookup(prompt)
    if hit is not None:
        return hit, "cache", score
    req = urllib.request.Request(
        url,
        data=json.dumps({"prompt": prompt}).encode(),
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=15) as resp:
        body = json.loads(resp.read().decode())
    response = body["text"]
    cache.store(prompt, response)
    return response, "endpoint", score

Step 4: Verify with a counting mock server

You need proof that the cache reduces calls. A mock server counts every POST.

from http.server import BaseHTTPRequestHandler, HTTPServer
import threading

class CountingHandler(BaseHTTPRequestHandler):
    calls = 0

    def do_POST(self):
        type(self).calls += 1
        length = int(self.headers.get("Content-Length", 0))
        self.rfile.read(length)
        body = b'{"text": "cached answer"}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass

server = HTTPServer(("127.0.0.1", 8765), CountingHandler)
threading.Thread(target=server.serve_forever, daemon=True).start()

Run a workload with repeated paraphrases. Start from a clean cache.

if os.path.exists("/tmp/cache.json"):
    os.remove("/tmp/cache.json")

cache = SemanticCache("/tmp/cache.json", threshold=0.7)
url = "http://127.0.0.1:8765"

queries = [
    "explain retries",
    "explain retries please",
    "what is a circuit breaker",
    "what is a circuit breaker used for",
    "explain retries again",
]

for q in queries:
    _, source, score = get_completion(q, cache, url)
    print(f"{source:8s} {score:.3f} {q}")

print("endpoint calls:", CountingHandler.calls)

Expected output:

endpoint 0.000 explain retries
cache    0.806 explain retries please
endpoint 0.000 what is a circuit breaker
cache    0.848 what is a circuit breaker used for
cache    0.828 explain retries again
endpoint calls: 2

Five queries. Two endpoint calls. Three cache hits. That is the proof.

Where a free server fits

The cache needs a home. It must persist across requests. A local file works for one process. A small server works for many.

MonkeyCode offers a free server option. Disclosure: This article was prepared as part of MonkeyCode's product outreach. You can host this cache there. The endpoint behind it can be any free model access you already use.

The cache is provider-agnostic. It does not care which model answers. It only cares that similar questions get one answer.

What the cache does not solve

The cache does not fix a broken endpoint. If the model returns garbage, the cache stores garbage. A health check must run before the cache, not after it.

The n-gram approach is weak for very short prompts. Two words share few trigrams. It is also weak for domain jargon. "RBAC" and "role-based access control" share almost no characters.

The cache never expires. A model update can make old answers wrong. Add a TTL or a version field.

The cache is single-tenant. If you serve multiple users, one user's question can hit another user's answer. Do not use this for private or personalized data.

Skip this if...

Do not use semantic caching for legal, medical, or financial answers. A similar question is not the same question. The wrong cached answer is worse than a rate limit.

Do not use it for creative work. "Write a poem about rain" and "write a poem about snow" are different tasks. Similarity will collide them.

Do not use it when your prompt includes timestamps, IDs, or other dynamic values. Every request is unique. The cache will never hit.

Conclusion

Retries are the default response to rate limits. They are the wrong one.

A semantic cache reuses answers for similar questions. It cuts endpoint calls without cutting quality. The threshold is the calibration lever. Measure it. Do not guess it.

Build the cache. Run the mock server. Count the calls. When the numbers show fewer endpoint hits, you have a working layer. The quota you save is the quota you can spend elsewhere.