AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build a production RAG pipeline with caching and fallbacks

Build a Production RAG Pipeline with Caching and Fallbacks

A RAG pipeline that works in a demo rarely survives its first month in production. The embedding API goes down. The vector database returns stale results. The LLM provider has a service disruption. Users send the same question 50 times in an hour. Without caching and fallback strategies, every one of these situations becomes a user-facing failure.

What you are building

Create a ProductionRAGPipeline class that adds operational resilience to a basic RAG pipeline:

Query result caching — cache query -> (chunks, answer) with configurable TTL using a dict with timestamps. Cache miss triggers full retrieval + generation.
Retrieval fallback — if the primary retriever raises an exception, fall back to a secondary retriever (injectable). Track which path was used.
Generation fallback — if the primary LLM fails, fall back to a secondary LLM. If both fail, return a graceful degraded message.
Per-step timing — track latency for retrieval, generation, and total in milliseconds.
Circuit breaker for the primary retriever — after 3 consecutive failures, skip the primary and use the fallback directly until a success resets the count.
Returns a PipelineResponse with: answer, chunks, from_cache bool, fallback_used string, latency breakdown, and error context.

Why this matters

Caching eliminates redundant work for repeated queries (common in help-center bots where 20–40% of queries repeat within an hour). The fallback chain ensures a degraded-but-working answer instead of an error page. The circuit breaker prevents timeout latency from cascading across all requests when a service is degraded.

Constraints

Standard library only (use time.monotonic). Injectable callables for testability.
Retrieve function: Callable[[str], list[dict]]. LLM function: Callable[[str, list[dict]], str].

Retrieval / hard / Step 13 of 15

Practice stage

Retrieval quality and ranking

Hint

Do not trust a single score blindly. Combine ranking logic with metadata and think about why a candidate deserves to rise.

Next drill

Implement BM25 keyword search from scratch

Success criteria

- Ranking rule is explainable
- Supports future iteration without rewriting everything
- Feels tied to product trust rather than pure math trivia

Review checklist

- Did I account for both relevance and metadata?
- Would I be able to explain the ranking rule in an interview?
- Does the code make future tuning easy?

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Callable, Optional

@dataclass
class PipelineResponse:
    answer: str
    chunks: list[dict]
    from_cache: bool
    fallback_used: str | None  # None, "retrieval", "generation", or "both"
    latency_ms: dict[str, float]
    error_context: str | None = None

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3):
        self.failure_threshold = failure_threshold
        self.consecutive_failures = 0
        self.is_open = False

def record_success(self) -> None:
        # TODO: reset consecutive_failures, set is_open = False
        raise NotImplementedError

def record_failure(self) -> None:
        # TODO: increment consecutive_failures, set is_open = True if threshold reached
        raise NotImplementedError

class ProductionRAGPipeline:
    def __init__(
        self,
        primary_retrieve: Callable[[str], list[dict]],
        fallback_retrieve: Callable[[str], list[dict]],
        primary_llm: Callable[[str, list[dict]], str],
        fallback_llm: Callable[[str, list[dict]], str],
        cache_ttl_seconds: int = 3600,
    ):
        self.primary_retrieve = primary_retrieve
        self.fallback_retrieve = fallback_retrieve
        self.primary_llm = primary_llm
        self.fallback_llm = fallback_llm
        self.cache_ttl = cache_ttl_seconds
        self._cache: dict[str, tuple[float, PipelineResponse]] = {}
        self._circuit_breaker = CircuitBreaker()

def _cache_key(self, query: str) -> str:
        raise NotImplementedError

def _get_cached(self, key: str) -> Optional[PipelineResponse]:
        raise NotImplementedError

def _set_cached(self, key: str, response: PipelineResponse) -> None:
        raise NotImplementedError

def query(self, query: str) -> PipelineResponse:
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.