I'm Aman Kumar, and I run an OpenAI-compatible gateway, so I see a lot of traffic patterns. One thing stands out: a surprising share of requests in real apps are exact repeats. The same system prompt, the same user question, the same parameters, sent again because a page re-rendered, a test suite re-ran, a cron job retried, or two users asked the same FAQ.
Every one of those repeats costs latency, and depending on how you pay, money or quota. A small cache on your side fixes most of it. This post walks through the one I actually use: what goes into the cache key, where to store it, how to handle streaming, and the cases where you should not cache at all.
Provider prompt caching is not the same thing
Some providers offer "prompt caching", where a long, repeated prefix (like a big system prompt) is billed at a discount. That's useful, but it's a different feature. The model still runs, you still wait for generation, and you still make a request.
A client-side response cache skips the request entirely. If the input is identical and you're fine with an identical answer, you return the stored answer in a millisecond. The two stack well together: provider caching helps with requests that differ only at the end, your own cache helps with requests that don't differ at all.
What the cache key must include
This is where most homemade caches go wrong. If the key misses anything that changes the output, you'll serve wrong answers. My key is a hash of:
- the base URL (so staging and production, or two providers, never share entries)
- the model ID, exactly as sent
- the full
messagesarray, including the system prompt -
temperature,top_p,max_tokens,stop, andseedif you use them -
tools,tool_choiceandresponse_format - a version string you bump whenever you change prompts or post-processing
The API key should not be in the cache key in plain form, and it shouldn't be stored anywhere in the cache. If you need per-tenant isolation, add a tenant ID instead.
Serialize with sorted keys so that {"a":1,"b":2} and {"b":2,"a":1} hash the same:
import hashlib, json
CACHE_VERSION = "v3"
def cache_key(base_url: str, payload: dict) -> str:
material = {
"v": CACHE_VERSION,
"base_url": base_url.rstrip("/"),
**payload,
}
blob = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
return hashlib.sha256(blob.encode("utf-8")).hexdigest()
A minimal SQLite-backed cache
For a single process or a small service, SQLite is enough. It survives restarts and needs no extra infrastructure.
import sqlite3, time
from openai import OpenAI
BASE_URL = "https://apiclaw.biz/v1"
client = OpenAI(base_url=BASE_URL, api_key="sk-your-key")
db = sqlite3.connect("llm_cache.sqlite3", check_same_thread=False)
db.execute("""CREATE TABLE IF NOT EXISTS cache (
k TEXT PRIMARY KEY, body TEXT NOT NULL, created REAL NOT NULL)""")
TTL_SECONDS = 7 * 24 * 3600
def cached_chat(**payload) -> dict:
payload.pop("stream", None) # this path is non-streaming only
k = cache_key(BASE_URL, payload)
row = db.execute("SELECT body, created FROM cache WHERE k=?", (k,)).fetchone()
if row and time.time() - row[1] < TTL_SECONDS:
return json.loads(row[0])
resp = client.chat.completions.create(**payload)
data = resp.model_dump()
choice = data["choices"][0]
if choice.get("finish_reason") in ("stop", "tool_calls"):
db.execute("INSERT OR REPLACE INTO cache VALUES (?,?,?)",
(k, json.dumps(data), time.time()))
db.commit()
return data
result = cached_chat(
model="deepseek-v4-flash",
temperature=0,
messages=[
{"role": "system", "content": "Answer in one short paragraph."},
{"role": "user", "content": "What does HTTP 429 mean?"},
],
)
print(result["choices"][0]["message"]["content"])
Two details matter here.
First, I only store responses whose finish_reason is stop or tool_calls. If the answer was cut off by length, or something went wrong mid-way, I don't want that frozen in the cache for a week.
Second, errors are never cached. A 429 or a 5xx raises an exception before the insert, so the next call tries again, which is what you want.
Streaming responses
Streaming and caching look incompatible, but they're not. On a miss, you stream to the user as normal and collect the chunks. When the stream ends cleanly, you join the text and store it. On a hit, you can either return the full text at once or replay it in small pieces if your UI expects a stream.
def cached_stream(**payload):
payload.pop("stream", None)
k = cache_key(BASE_URL, payload)
row = db.execute("SELECT body FROM cache WHERE k=?", (k,)).fetchone()
if row:
text = json.loads(row[0])["choices"][0]["message"]["content"]
for i in range(0, len(text), 40):
yield text[i:i+40]
return
parts, finish = [], None
for chunk in client.chat.completions.create(stream=True, **payload):
if not chunk.choices:
continue
delta = chunk.choices[0].delta.content or ""
finish = chunk.choices[0].finish_reason or finish
parts.append(delta)
yield delta
if finish == "stop":
body = {"choices": [{"message": {"role": "assistant",
"content": "".join(parts)}, "finish_reason": "stop"}]}
db.execute("INSERT OR REPLACE INTO cache VALUES (?,?,?)",
(k, json.dumps(body), time.time()))
db.commit()
If the user disconnects halfway, the generator is closed, finish never becomes stop, and nothing partial is stored.
Temperature and "should this even be cached?"
A cache returns the same answer every time. That's perfect when you'd want the same answer anyway, and wrong when variety is the point.
Good candidates:
- classification, tagging, extraction and routing at
temperature=0 - summaries of documents that don't change (key on a content hash, not the file name)
- FAQ-style answers and help-center bots
- embeddings-like "describe this item" jobs in batch pipelines
- test suites and CI runs, where repeat calls are pure waste
Bad candidates:
- creative generation where users hit "regenerate" expecting something new
- anything that depends on the current time, live data or tool results fetched at request time
- multi-turn chats, where the full history rarely repeats exactly (the cache won't hurt, it just won't hit)
My rule: cache by default when temperature is 0 and the inputs are fully contained in the request. Opt in explicitly for everything else.
Moving to Redis when you scale out
Once you have several workers, a per-process SQLite file stops being shared. Swap the storage for Redis with SET key value EX ttl; the key function stays the same. Keep values compact (store only the fields you read back), and set a memory limit with an LRU eviction policy so a burst of unique prompts can't fill the box.
One more thing worth adding at that point is request coalescing: if ten workers miss on the same key at the same moment, only one should call the API while the others wait for its result. A short Redis lock on the key (SET lock:<key> 1 NX EX 30) is enough for most apps.
Measuring whether it's worth it
Log three numbers per feature: hits, misses, and the latency of each. In my experience, internal tools and batch jobs often see high hit rates, while open-ended chat sees very few. If a feature's hit rate stays near zero after a week, remove the cache there; it's just extra code.
Also watch for stale answers. If you change a prompt and forget to bump CACHE_VERSION, users will get old-style responses for up to the TTL. I bump the version in the same commit as any prompt edit, and I keep TTLs to days, not months.
Where this fits with a flat-rate plan
On per-token billing, a cache saves money directly. On a flat monthly plan with a daily request allowance, it saves allowance instead, which matters if you run agents or batch jobs that would otherwise hit the limit before the day ends. Either way, the request you don't send is also the fastest one.
Disclosure: I build APIClaw, an OpenAI-compatible gateway with flat monthly pricing, which is the base URL in the examples. Everything above works the same with any OpenAI-compatible endpoint; change BASE_URL, the key and the model ID to whatever your provider lists.
Top comments (0)