Shipping AI to Production

Observability, evals, retries, caching, routing, cost control and the architecture of a real streaming AI product.

• Advanced
• 30 min read
• 3 objectives

Before this lessonLesson 16: Multi-Agent Systems, MCP and AI Safety

A demo that works on your laptop and a product that works for ten thousand users at 3am are different animals. This lesson is about the second one: seeing what your AI system is doing (observability), measuring whether it is good (evals), surviving failures (retries and fallbacks), controlling the bill (caching and routing) and the overall architecture of a streaming AI product.

Observability: you cannot debug what you cannot see

An LLM app fails in strange ways: a slightly worse retrieval, a prompt that drifted, a tool that returned garbage. Without logs you are guessing. Trace every request: each model call and tool call becomes a span with inputs, outputs, latency, token counts and cost, nested under one trace per user request. Tools: LangSmith, Langfuse, Arize Phoenix, Helicone, OpenTelemetry (GenAI conventions). Here is the idea from scratch:

import time, functools

TRACE = []                       # in production: send these to a tracing backend
DEPTH = 0

def traced(name):
    def wrap(fn):
        @functools.wraps(fn)
        def inner(*args, **kwargs):
            global DEPTH
            start = time.perf_counter()
            DEPTH += 1
            try:
                return fn(*args, **kwargs)
            finally:
                DEPTH -= 1
                TRACE.append((DEPTH, name, round((time.perf_counter() - start) * 1000)))
        return inner
    return wrap

@traced("retrieve")
def retrieve(q):
    time.sleep(0.02)
    return ["chunk A", "chunk B"]

@traced("llm_call")
def llm_call(prompt):
    time.sleep(0.05)
    return "Laptops can be returned within 30 days."

@traced("handle_request")
def handle(question):
    chunks = retrieve(question)
    return llm_call(f"{chunks} {question}")

handle("Can I return a laptop?")
for depth, name, ms in sorted(TRACE, key=lambda t: t[0]):
    print(f"{'  ' * depth}{name}")
print("(each span also records its own duration, inputs, outputs, tokens and cost)")

handle_request
  retrieve
  llm_call
(each span also records its own duration, inputs, outputs, tokens and cost)

With spans you can answer the questions that matter: which step is slow? which step is expensive? what exactly did the model see when it gave that terrible answer? Always log the retrieved chunks and the final prompt.

Evals: your regression test suite for AI

Traditional software has unit tests. AI systems need evals: a set of realistic inputs with a way of scoring the outputs, run automatically whenever you change a prompt, model, retrieval setting or tool. Three layers:

• Deterministic checks: valid JSON? contains the required field? correct tool called? cost under budget? Fast and free, run on everything.
• Reference-based checks: compare against a known-good answer (exact match, fuzzy match, or LLM judge).
• LLM-as-judge and human review: rubric-based scoring for quality, tone and faithfulness, calibrated against human labels on a sample.

Grow the set from real failures: every production bug becomes a new test case. Track scores over time, and gate deployments on them ("do not ship if faithfulness drops below 0.9"). Add online signals too: thumbs up/down, retries, escalations to humans, session abandonment.

Retries, backoff and fallbacks

Provider APIs throw 429 (rate limited) and 5xx (server errors) regularly. Retry with exponential backoff plus jitter (wait 1s, 2s, 4s, each with random spread so a thousand clients do not retry in lockstep), cap the attempts, and only retry idempotent-safe failures. Add fallbacks: if the primary model is down, use a secondary model or a cached/degraded answer.

import random

def call_with_retry(fn, max_attempts=5, base_delay=1.0, seed=3):
    rng = random.Random(seed)
    for attempt in range(1, max_attempts + 1):
        try:
            return fn(attempt)
        except ConnectionError as err:
            if attempt == max_attempts:
                raise
            delay = base_delay * 2 ** (attempt - 1)          # 1, 2, 4, 8 ...
            delay += rng.uniform(0, delay * 0.25)            # jitter
            print(f"attempt {attempt} failed ({err}); wait {delay:.2f}s")

def flaky_api(attempt):
    if attempt < 3:
        raise ConnectionError("429 rate limited")
    return "answer"

print("result:", call_with_retry(flaky_api))

attempt 1 failed (429 rate limited); wait 1.06s
attempt 2 failed (429 rate limited); wait 2.27s
result: answer

Controlling cost

• Prompt caching. Providers discount a repeated prefix heavily. Keep the system prompt, tools and stable documents at the front, unchanged.
• Response and semantic caching. Identical (or near-identical, by embedding) questions get the stored answer for free.
• Model routing. Cheap model for easy requests, strong model for hard ones, decided by a classifier or a rule.
• Trim context. Fewer, better chunks. Summarise long histories.
• Cap output length and ask for concise answers.
• Batch APIs for non-urgent work: often around half price.
• Set budgets and alerts per user, per feature and per day so a bug cannot burn the month's budget overnight.

A working semantic cache and a router, so the savings are concrete:

import math, re
from collections import Counter

def vec(text):
    return Counter(re.findall(r"[a-z]+", text.lower()))

def cosine(a, b):
    d = sum(a[k] * b[k] for k in a)
    return d / (math.sqrt(sum(v*v for v in a.values())) * math.sqrt(sum(v*v for v in b.values())) or 1)

class SemanticCache:
    def __init__(self, threshold=0.7):
        self.items, self.threshold, self.hits, self.misses = [], threshold, 0, 0

    def get(self, question):
        q = vec(question)
        for stored_q, answer in self.items:
            if cosine(q, vec(stored_q)) >= self.threshold:
                self.hits += 1
                return answer
        self.misses += 1
        return None

    def put(self, question, answer):
        self.items.append((question, answer))

def route(question):
    """Cheap heuristic router: long or analytical questions go to the big model."""
    hard = len(question.split()) > 12 or any(w in question.lower() for w in ("compare", "why", "analyse", "design"))
    return "big-model" if hard else "small-model"

cache = SemanticCache()
cost = {"small-model": 0.0005, "big-model": 0.0100}          # dollars per call (illustrative)
spent = 0.0

questions = ["How do I reset my password", "how do i reset my password please", "What are your opening hours",
             "Compare the warranty and return policies for laptops", "how do I reset my password"]
for q in questions:
    cached = cache.get(q)
    if cached:
        print(f"CACHE HIT   | {q}")
        continue
    model = route(q)
    spent += cost[model]
    cache.put(q, "answer...")
    print(f"{model:<11} | {q}")

print(f"\nspent ${spent:.4f} instead of ${len(questions) * cost['big-model']:.4f} "
      f"(hits: {cache.hits}, misses: {cache.misses})")

small-model | How do I reset my password
CACHE HIT   | how do i reset my password please
small-model | What are your opening hours
big-model   | Compare the warranty and return policies for laptops
CACHE HIT   | how do I reset my password

spent $0.0110 instead of $0.0500 (hits: 2, misses: 3)

The architecture of a real streaming AI product

Putting the whole course together, here is what a production RAG chatbot looks like, with every concept from this course in its place:

flowchart TB B[Browser React] -->|POST /chat| API[API server FastAPI] API --> G[Auth, rate limit, budget] G --> RW[Rewrite query] RW --> RET[Hybrid retrieve] RET --> RR[Rerank top 40 to top 5] RR --> PR[Build prompt] PR --> CACHE{Semantic cache hit?} CACHE -->|Yes| B CACHE -->|No| LLM[LLM streaming call] LLM --> TOOLS[Tool loop] TOOLS --> LLM LLM -->|SSE tokens| B LLM --> FG[Faithfulness / guardrails] API --> TRACE[Tracing] TRACE --> EVAL[Eval dashboards]

Fine-tuning: when is it the right answer?

• Try first: better prompts, few-shot examples, RAG, and a stronger model. These solve most problems.
• Fine-tune for behaviour: a consistent tone, a strict output format, domain shorthand, or squeezing a small model to do one task cheaply and fast (distillation).
• Do not fine-tune for facts. Knowledge that changes belongs in retrieval. Fine-tuning teaches style far better than it teaches facts, and facts go stale.
• Embedding fine-tuning can lift retrieval on specialised jargon when you have labelled query-document pairs.
• Methods: full fine-tuning is heavy; parameter-efficient methods like LoRA adapt a model cheaply.

Go-live checklist

• An eval set with a baseline score, run in CI.
• Tracing on every request, including retrieved chunks and prompts.
• Timeouts, retries with backoff, fallbacks, and graceful error messages.
• Streaming with a stop button; proxy buffering off.
• Per-user rate limits, budgets and cost alerts.
• Permission-scoped retrieval; secrets out of prompts; PII policy.
• Prompt-injection tests and human approval for risky actions.
• A feedback button, and a habit of turning bad answers into eval cases.

# Write your solution here