Lesson 8: Capstone — Design a Production AI System

Bring together every pattern from this course into one production-grade AI system: a three-tier model cascade with semantic caching, circuit breakers, and an A/B testing harness — all behind a single FastAPI endpoint and load-tested to 200 concurrent users.

                  ┌────────────────────────┐
  User request →  │   FastAPI /chat        │
                  │   • Variant assignment  │
                  │   • Semantic cache ──── cache hit → return
                  └────────────┬───────────┘
                               │ cache miss
                  ┌────────────▼───────────┐
                  │   Model Cascade        │
                  │   Tier 1: haiku  ──── confidence OK → return
                  │   Tier 2: sonnet ──── confidence OK → return
                  │   Tier 3: opus   ──────────────────────────→ return
                  └────────────┬───────────┘
                               │
                  ┌────────────▼───────────┐
                  │   Circuit Breaker      │
                  │   • OPEN: fallback     │
                  │   • CLOSED: normal     │
                  └────────────────────────┘

Phase 1 — Model Cascade with Confidence Routing

Implement classify_query_complexity(query) using keyword heuristics — no LLM call needed for obvious simple or complex queries.
Implement cascade_query(question) that tries haiku first, escalates to sonnet if confidence < 0.85, then opus if confidence < 0.75.
Log model_used, confidence, and escalated on every response. Verify that >80% of simple queries use haiku.
Add a /cascade/stats endpoint that returns tier usage percentages over the last 1,000 requests.

Phase 2 — Semantic Cache Layer

Set up ChromaDB (dev) or Redis + faiss (production) as the cache backend with TTL of 1 hour.
Wire cache_lookup(query) before the cascade — return cached response immediately on hit.
Wire cache_store(query, response) after a cascade miss — store embedding + response.
Expose X-Cache-Hit: true/false response header. Run 10 identical queries and confirm all but the first are cache hits with <50ms latency.

Phase 3 — Circuit Breaker & Fallback

Implement CircuitBreaker(failure_threshold=3, recovery_timeout=30) wrapping the Anthropic client.
Configure httpx.Timeout(connect=5, read=30) on the Anthropic client — never allow unbounded blocking.
Define a static fallback: "Our AI assistant is temporarily unavailable. We'll be back shortly."
Write a chaos test: mock the Anthropic client to raise anthropic.APIConnectionError 5× in a row. Assert the circuit opens and the fallback is returned without any further API calls.

Phase 4 — A/B Test Harness

Define two prompts: PROMPT_V1 (baseline) and PROMPT_V2 (challenger with chain-of-thought instruction).
Implement assign_variant(user_id, experiment_id) with deterministic MD5 hashing — same user always sees same variant.
Log experiment_id, variant, quality_score (LLM-as-judge, 1–10), and latency_ms for every request in the experiment cohort.
After 200 requests per variant, run analyze_experiment(control_scores, treatment_scores). Record: effect size, p-value, recommended action.

Phase 5 — Load Test to 200 Concurrent Users

Write a Locust file with two task weights: simple queries (weight 3) and complex queries (weight 1), reflecting realistic traffic distribution.
Run the load test at 50, 100, 200 concurrent users. Record p50/p95/p99 latency and error rate at each level.
Confirm p95 latency stays under 5,000ms with cache enabled, and that the circuit breaker does not open during normal load (error rate < 1%).
Identify the concurrency level where p99 exceeds 10,000ms or error rate exceeds 1%. This is your system's capacity ceiling — document it.

# Full integration: FastAPI app with all patterns wired together
from fastapi import FastAPI, Request, Response
from contextlib import asynccontextmanager
import anthropic, chromadb, time, hashlib

app = FastAPI()
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

@app.post("/chat")
async def chat(request: ChatRequest, response: Response) -> ChatResponse:
    start = time.monotonic()

    # Phase 2: Semantic cache
    cached = cache_lookup(request.message)
    if cached:
        response.headers["X-Cache-Hit"] = "true"
        return ChatResponse(answer=cached, latency_ms=(time.monotonic() - start) * 1000)

    # Phase 4: A/B test variant
    variant = assign_variant(request.user_id, "prompt-v2-test")
    prompt = PROMPT_V2 if variant == "treatment" else PROMPT_V1

    # Phase 1 + 3: Cascade with circuit breaker
    def do_cascade():
        return cascade_query(prompt.format(message=request.message))

    result = breaker.call(
        do_cascade,
        fallback=lambda: "Our AI assistant is temporarily unavailable.",
    )

    latency_ms = (time.monotonic() - start) * 1000

    # Store in cache (background)
    cache_store(request.message, result.answer if hasattr(result, "answer") else result)

    # Log for A/B analysis
    log_experiment_event(request.user_id, "prompt-v2-test", variant, result, latency_ms)

    response.headers["X-Cache-Hit"] = "false"
    return ChatResponse(answer=result, latency_ms=latency_ms)

Graduation criteria:Your system must pass all of: p95 latency < 5s at 100 users, cache hit rate > 40% on repeated queries, circuit breaker test passes without calling the API after threshold is hit, and A/B assignment is deterministic (same user always gets the same variant across restarts).

┌────────────────────────┐ User request → │ FastAPI /chat │ │ • Variant assignment │ │ • Semantic cache ──── cache hit → return └────────────┬───────────┘ │ cache miss ┌────────────▼───────────┐ │ Model Cascade │ │ Tier 1: haiku ──── confidence OK → return │ Tier 2: sonnet ──── confidence OK → return │ Tier 3: opus ──────────────────────────→ return └────────────┬───────────┘ │ ┌────────────▼───────────┐ │ Circuit Breaker │ │ • OPEN: fallback │ │ • CLOSED: normal │ └────────────────────────┘

# Full integration: FastAPI app with all patterns wired together from fastapi import FastAPI, Request, Response from contextlib import asynccontextmanager import anthropic, chromadb, time, hashlib app = FastAPI() breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30) @app.post("/chat") async def chat(request: ChatRequest, response: Response) -> ChatResponse: start = time.monotonic() # Phase 2: Semantic cache cached = cache_lookup(request.message) if cached: response.headers["X-Cache-Hit"] = "true" return ChatResponse(answer=cached, latency_ms=(time.monotonic() - start) * 1000) # Phase 4: A/B test variant variant = assign_variant(request.user_id, "prompt-v2-test") prompt = PROMPT_V2 if variant == "treatment" else PROMPT_V1 # Phase 1 + 3: Cascade with circuit breaker def do_cascade(): return cascade_query(prompt.format(message=request.message)) result = breaker.call( do_cascade, fallback=lambda: "Our AI assistant is temporarily unavailable.", ) latency_ms = (time.monotonic() - start) * 1000 # Store in cache (background) cache_store(request.message, result.answer if hasattr(result, "answer") else result) # Log for A/B analysis log_experiment_event(request.user_id, "prompt-v2-test", variant, result, latency_ms) response.headers["X-Cache-Hit"] = "false" return ChatResponse(answer=result, latency_ms=latency_ms)