Caching LLM outputs at your service boundary is the fastest win for latency and spend. This guide builds a complete fastapi cache llm responses redis layer that sits between your API clients and any OpenAI-compatible model endpoint, using async Redis and deterministic keys.
Step 1: Scaffold the FastAPI app and Redis client
Install the minimal dependency set. You need an async web framework, an async Redis driver, and an HTTP client that won’t block the event loop.
pip install fastapi uvicorn redis httpx pydanticUse redis.asyncio, not the sync redis client. A blocking get inside a request handler will stall every other request on the same worker. The snippet below initializes the client and validates connectivity at startup.
import os
import redis.asyncio as aioredis
from contextlib import asynccontextmanager
from fastapi import FastAPI
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
@asynccontextmanager
async def lifespan(app: FastAPI):
await redis_client.ping()
yield
await redis_client.aclose()
app = FastAPI(lifespan=lifespan)decode_responses=True returns strings instead of bytes, which keeps JSON handling clean. If you run Redis in cluster mode, swap from_url for RedisCluster from redis.asyncio.cluster. The async client manages its own connection pool; do not recreate the client per request. One module-level instance is correct for a single process. If you deploy multiple workers, each gets its own pool, which is fine because Redis is the shared store.
Step 2: Define the request shape and a deterministic cache key
Your cache key must be a pure function of the inputs that affect the model output. Any non-determinism—timestamps, client IP, trace IDs—leaks into the key and destroys hit rate.
import hashlib
import json
from pydantic import BaseModel
class ChatMessage(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
model: str
messages: list[ChatMessage]
temperature: float = 0.7
max_tokens: int = 512
stream: bool = False
def make_cache_key(req: ChatRequest) -> str:
payload = {
"model": req.model,
"messages": [m.model_dump() for m in req.messages],
"temperature": req.temperature,
"max_tokens": req.max_tokens,
}
raw = json.dumps(payload, sort_keys=True)
digest = hashlib.sha256(raw.encode()).hexdigest()
model_tag = req.model.replace("/", ":")
return f"llm:{model_tag}:{digest}"sort_keys=True ensures that {"a":1,"b":2} and {"b":2,"a":1} hash identically. Exclude stream because a streaming response is a different wire format and should bypass the fastapi cache llm responses redis path entirely. If you later add seed or top_p, include them in the payload dict. Treat the key builder as a contract: any field that changes the bytes sent to the model must be present, any field that doesn’t must be absent.
Step 3: Implement the cached completion endpoint
The endpoint checks Redis before calling the model. On a miss, it proxies to the upstream, stores the normalized JSON, and returns it. Only successful responses get cached.
import httpx
from fastapi import HTTPException
LLM_ENDPOINT = os.getenv("LLM_ENDPOINT", "https://api.openai.com/v1/chat/completions")
LLM_API_KEY = os.getenv("LLM_API_KEY", "")
async def call_llm(req: ChatRequest) -> dict:
headers = {
"Authorization": f"Bearer {LLM_API_KEY}",
"Content-Type": "application/json",
}
payload = req.model_dump(exclude={"stream"})
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(LLM_ENDPOINT, headers=headers, json=payload)
if r.status_code != 200:
raise HTTPException(status_code=r.status_code, detail=r.text)
return r.json()
@app.post("/v1/chat/completions")
async def chat(req: ChatRequest):
if req.stream:
raise HTTPException(status_code=400, detail="streaming bypasses cache")
key = make_cache_key(req)
cached = await redis_client.get(key)
if cached:
return json.loads(cached)
resp = await call_llm(req)
await redis_client.set(key, json.dumps(resp), ex=3600)
return respIf you route through n4n.ai, it forwards provider cache-control hints, so you can set the ex parameter to match the upstream cache lifetime instead of hard-coding 3600 seconds. That keeps your fastapi cache llm responses redis layer coherent with the model provider’s own caching window. The endpoint above is intentionally thin. Put auth, rate limiting, and request logging in separate dependencies so the cache logic stays readable.
Step 4: Choose a TTL and namespace strategy
A blanket one-hour TTL is fine for a prototype, but production traffic needs coarser control. Namespace by model so you can invalidate a single model when you rotate prompts or fine-tunes.
# key already prefixed with model_tag from Step 2
await redis_client.set(key, json.dumps(resp), ex=3600)To drop all cached outputs for a model:
redis-cli --scan --pattern 'llm:gpt-4o:*' | xargs -L 100 redis-cli delAvoid KEYS in production; --scan is non-blocking. For multi-tenant systems, embed a tenant hash in the prefix so one customer’s cache never serves another. If a model version is deprecated, bump the model string in your client calls and the old keys naturally expire. Do not attempt to parse provider version from the response and rewrite keys at runtime—that creates a race between reads and writes.
Step 5: Prevent cache stampedes
A hot key with no value will trigger a thundering herd of identical LLM calls. Use a Redis lock with SET NX before computing.
import asyncio
async def chat_locked(req: ChatRequest):
key = make_cache_key(req)
cached = await redis_client.get(key)
if cached:
return json.loads(cached)
lock_key = key + ":lock"
if not await redis_client.set(lock_key, "1", nx=True, ex=30):
# someone else is computing; wait and retry
for _ in range(10):
await asyncio.sleep(0.2)
cached = await redis_client.get(key)
if cached:
return json.loads(cached)
raise HTTPException(status_code=504, detail="cache fill timeout")
try:
resp = await call_llm(req)
await redis_client.set(key, json.dumps(resp), ex=3600)
finally:
await redis_client.delete(lock_key)
return respThis pattern caps concurrent misses at one per key. Tune the retry loop to your p99 model latency. If your upstream supports request deduplication, the lock is still useful because it saves the round trip. Set the lock TTL longer than the expected compute time; a dead lock expiring early is safer than a permanent one blocking fills.
Step 6: Serialization and error boundaries
LLM responses contain nested dicts, usage objects, and sometimes None. json.dumps handles that, but watch for non-serializable types if you augment the response with local metadata. Store only the upstream payload.
Never cache error bodies. The call_llm helper above raises on non-200, so the set line is unreachable on failure. If you add retries or fallback, ensure the error path returns without writing to Redis. A 429 from the provider is not a successful completion; caching it would poison the key with a useless body.
If you use Pydantic v2, model_dump() is safe. On v1 use dict(). Keep the request model strict: extra="forbid" prevents clients from sneaking cache-busting fields. Log cache hits versus misses with a structured logger so you can alert on hit-rate drops.
Step 7: Verify the setup end to end
Start Redis and the app:
redis-server --daemonize yes
uvicorn main:app --port 8000Send the same request twice and measure time:
curl -s -X POST localhost:8000/v1/chat/completions \
-H 'Content-Type: application/json' \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"Say pong"}]}' \
-o /dev/null -w "first: %{time_total}s\n"
curl -s -X POST localhost:8000/v1/chat/completions \
-H 'Content-Type: application/json' \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"Say pong"}]}' \
-o /dev/null -w "second: %{time_total}s\n"The second call should return in single-digit milliseconds versus hundreds of milliseconds for the first. Confirm the key exists:
redis-cli --scan --pattern 'llm:gpt-4o-mini:*'For automated verification, use fakeredis in a pytest fixture and assert that a repeated call hits redis_client.get before httpx.AsyncClient.post. That catches regressions where the key function drifts from the request schema. A minimal test:
import fakeredis, pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_cache_hit(monkeypatch):
monkeypatch.setattr("main.redis_client", fakeredis.aioredis.FakeRedis())
payload = {"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}]}
async with AsyncClient(app=main.app, base_url="http://test") as ac:
await ac.post("/v1/chat/completions", json=payload)
r2 = await ac.post("/v1/chat/completions", json=payload)
assert r2.json() # second response served from fakeredisThis proves the fastapi cache llm responses redis flow without external services.
Step 8: Operational notes
Monitor cache hit rate with a Redis INFO stats poll or a custom counter incremented in the endpoint. A hit rate below 20% on immutable prompts means your key is non-deterministic—audit make_cache_key. Set maxmemory-policy allkeys-lru so Redis evicts old LLM responses under pressure instead of refusing writes.
The fastapi cache llm responses redis pattern here assumes stateless model calls. If you add conversation state outside the message list, include that state in the hash. Keep the cache layer dumb: it stores bytes, not semantics. When you outgrow a single Redis instance, move to a cluster and keep key sizes small—SHA-256 digests are fixed length and cluster-friendly.
That is a shippable starting point. Extend with per-tenant quotas, streaming aggregation, or provider-specific TTLs as your traffic demands.