Caching with Redis

import hashlib, json, time
import fakeredis                     # in-memory stand-in; in production: redis.Redis(host=..., port=6379)
 
r = fakeredis.FakeRedis(decode_responses=True)
CALLS = {"llm": 0}
 
def call_llm(prompt: str) -> str:
    CALLS["llm"] += 1
    time.sleep(0.2)                  # pretend the model takes a while
    return f"answer to: {prompt}"
 
def cached_answer(prompt: str, model: str, ttl_s: int = 3600) -> tuple[str, str]:
    key = "llm:" + hashlib.sha256(json.dumps([model, prompt]).encode()).hexdigest()[:16]
    hit = r.get(key)                                   # 1. look in the cache
    if hit is not None:
        return hit, "HIT"
    answer = call_llm(prompt)                          # 2. miss: do the slow, costly work
    r.set(key, answer, ex=ttl_s)                       # 3. store it with an expiry (TTL)
    return answer, "MISS"
 
for prompt in ["What is RAG?", "What is RAG?", "What is an agent?", "What is RAG?"]:
    start = time.perf_counter()
    answer, status = cached_answer(prompt, "small-model")
    ms = round((time.perf_counter() - start) * 1000, -2)          # rounded to 100 ms for a stable demo
    print(f"{status:4} ~{ms:3.0f} ms  {answer}")
print("model calls:", CALLS["llm"], "| keys:", len(r.keys("llm:*")), "| ttl of one key:", r.ttl(r.keys("llm:*")[0]), "s")
 
# Other Redis jobs in AI apps: rate-limit counters, session state, queues, distributed locks.
r.incr("ratelimit:alice:2026-11-14T10:15")
r.expire("ratelimit:alice:2026-11-14T10:15", 60)
print("alice requests this minute:", r.get("ratelimit:alice:2026-11-14T10:15"))

MISS ~200 ms  answer to: What is RAG?
HIT  ~  0 ms  answer to: What is RAG?
MISS ~200 ms  answer to: What is an agent?
HIT  ~  0 ms  answer to: What is RAG?
model calls: 2 | keys: 2 | ttl of one key: 3600 s
alice requests this minute: 1