FastAPI + Redis: Rate Limiting and Caching for RAG Applications

Redis in a RAG application usually ends up doing three unrelated jobs: rate limiting API keys, caching embeddings to avoid redundant OpenAI calls, and storing conversation/query session state. Mixing these into one ad-hoc Redis client without namespacing is how you end up with a rate limit counter accidentally sharing a key prefix with a cache entry. Here's how to structure it cleanly.

Namespacing: the thing everyone skips

Before writing any Redis logic, decide your key namespace convention. Every key should encode its purpose and be impossible to collide with another:

# app/cache/keys.py
def rate_limit_key(api_key_id: str, window: str) -> str:
    return f"ratelimit:{api_key_id}:{window}"

def embedding_cache_key(text_hash: str) -> str:
    return f"embed:cache:{text_hash}"

def session_key(session_id: str) -> str:
    return f"session:{session_id}"

This looks trivial until you're debugging a production incident and need to SCAN for a specific class of key without accidentally matching three other systems.

Sliding-window rate limiting

A fixed window (reset every 60 seconds on the clock) allows a burst right at the window boundary — 60 requests at 11:59:59 and 60 more at 12:00:01 is 120 requests in two seconds. A sliding window avoids this:

async def check_rate_limit(api_key_id: str, limit: int, window_seconds: int = 60) -> bool:
    key = rate_limit_key(api_key_id, "minute")
    now = time.time()
    pipe = redis.pipeline()
    pipe.zremrangebyscore(key, 0, now - window_seconds)
    pipe.zadd(key, {str(now): now})
    pipe.zcard(key)
    pipe.expire(key, window_seconds)
    _, _, count, _ = await pipe.execute()
    return count <= limit

Using a sorted set keyed by timestamp, then trimming anything outside the window on every check, gives you accurate sliding-window behavior with one round trip via a pipeline.

Caching embeddings without silently going stale

Embedding the same text twice is pure waste — OpenAI's embedding models are deterministic for a given model version, so the same input always produces the same vector.

import hashlib

async def get_embedding_cached(text: str) -> list[float]:
    text_hash = hashlib.sha256(text.encode()).hexdigest()
    key = embedding_cache_key(text_hash)

    cached = await redis.get(key)
    if cached:
        return json.loads(cached)

    embedding = await openai_client.embeddings.create(model="text-embedding-3-small", input=text)
    vector = embedding.data[0].embedding
    await redis.set(key, json.dumps(vector), ex=60 * 60 * 24 * 30)  # 30-day TTL
    return vector

The 30-day TTL isn't arbitrary — it caps how long you'd serve a stale embedding if you ever change embedding models, while still capturing the overwhelming majority of practical cache hits (RAG corpora don't change per-chunk text that often).

Caching RAG query results carefully

Caching the retrieval step (which chunks match a query) is often safe. Caching the full generated answer is riskier — if your underlying documents change, a cached answer can go stale in a way that's hard to detect.

@cache(ttl=300)  # short TTL — only worth it for identical repeated queries
async def cached_retrieval(collection: str, query_embedding: list[float], top_k: int):
    return await vector_store.query(collection, query_embedding, top_k)

Keep generation uncached by default. Cache retrieval results with a short TTL, and only cache full answers if your documents are genuinely static (e.g., a fixed policy handbook that changes quarterly, not daily).

Session state for multi-turn RAG conversations

Follow-up questions in a RAG chat need the prior turn's context to resolve references ("what about the second one?"). Store the last N turns plus the last retrieval's source IDs:

async def append_rag_turn(session_id: str, query: str, answer: str, sources: list[str]):
    key = session_key(session_id)
    turn = {"query": query, "answer": answer, "sources": sources}
    await redis.rpush(key, json.dumps(turn))
    await redis.ltrim(key, -10, -1)  # keep last 10 turns
    await redis.expire(key, 60 * 60 * 24)  # 24h session TTL

What FastAPI AI Kit includes

The kit ships all three Redis roles pre-configured: sliding-window rate limiting on @rate_limit()-decorated endpoints, a @cache() decorator for deterministic function results, and a session store used by both the chat and RAG query endpoints — with sensible key namespacing already in place so these three systems don't collide.