LLM Production Patterns

Real-World Scenario

A SaaS product ships AI document analysis. Week one: the bill is triple the estimate, and a 20-minute provider incident turns into an hour of timeouts because every request retried in lockstep. Nothing here needs a smarter model. Each fix is ordinary engineering around the API call.

This lesson runs every pattern against FlakyProvider, a stand-in for a hosted LLM API that fails on a script you choose. No API key and no billing account are needed, and the failure modes are the real HTTP ones (429 rate limited, 529 overloaded, 400 bad request). The wrappers work unchanged around a real SDK client, because they only catch its exception classes and call its complete method.

The Failure: One 429 Takes Down the Request

The stand-in provider raises the exception you script for it, then answers normally.

class RateLimitError(Exception): status = 429
class OverloadedError(Exception): status = 529
class BadRequestError(Exception): status = 400

class FlakyProvider:
    """Stands in for a hosted LLM API: fails on a fixed script, then succeeds."""
    def __init__(self, script=()):
        self.script = list(script)
        self.calls = 0

    def complete(self, prompt: str, max_tokens: int = 256) -> str:
        self.calls += 1
        outcome = self.script.pop(0) if self.script else "ok"
        if outcome != "ok":
            raise outcome(f"HTTP {outcome.status}")
        return f"answer to {prompt!r}"

if __name__ == "__main__":
    api = FlakyProvider(script=[RateLimitError, "ok"])
    print(api.complete("What are your opening hours?"))

$ python provider.py
Traceback (most recent call last):
  File "/home/user/llm-demo/production/provider.py", line 20, in <module>
    print(api.complete("What are your opening hours?"))
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/llm-demo/production/provider.py", line 15, in complete
    raise outcome(f"HTTP {outcome.status}")
RateLimitError: HTTP 429

The provider would have answered on the very next call, but the user got a stack trace instead. Rate limits are routine under real traffic, and a bare API call turns every one of them into a failed request. Save this file as provider.py. Every example below imports it.

Retry With Exponential Backoff and Jitter

Only transient errors are worth another attempt, and each wait should be longer than the last.

import random, time
from provider import FlakyProvider, RateLimitError, OverloadedError, BadRequestError

RETRYABLE = (RateLimitError, OverloadedError)   # 429, 529: try again later

def call_with_retry(api, prompt, max_retries=3, base=0.2, cap=2.0):
    for attempt in range(max_retries + 1):
        try:
            return api.complete(prompt)
        except RETRYABLE as e:
            if attempt == max_retries:
                raise
            wait = min(cap, base * 2 ** attempt) * random.uniform(0.5, 1.0)  # jitter
            print(f"  attempt {attempt + 1}: {type(e).__name__} -> sleep {wait:.2f}s")
            time.sleep(wait)

random.seed(7)
api = FlakyProvider(script=[RateLimitError, OverloadedError, "ok"])
print(call_with_retry(api, "hours?"), f"after {api.calls} calls")

for script in ([BadRequestError], [OverloadedError] * 5):
    api = FlakyProvider(script)
    try:
        call_with_retry(api, "hours?")
    except Exception as e:
        print(f"gave up: {type(e).__name__} after {api.calls} call(s)")

$ python retry.py
  attempt 1: RateLimitError -> sleep 0.13s
  attempt 2: OverloadedError -> sleep 0.23s
answer to 'hours?' after 3 calls
gave up: BadRequestError after 1 call(s)
  attempt 1: OverloadedError -> sleep 0.17s
  attempt 2: OverloadedError -> sleep 0.21s
  attempt 3: OverloadedError -> sleep 0.61s
gave up: OverloadedError after 4 call(s)

There are three outcomes here. First, two transient failures are absorbed and the caller gets an answer on the third call. Second, BadRequestError is not in RETRYABLE, so it propagates after one call: a malformed request fails identically every time, and retrying it only burns time. Third, a persistent 529 is given up on after max_retries + 1 = 4 calls instead of looping forever.

The sleeps are not a clean 0.2, 0.4, 0.8. The jitter multiplier (0.5 to 1.0) spreads them out so that a thousand clients rate-limited at the same instant don’t all retry at the same instant and trip the limit again. The base delays are tiny here so the demo runs fast. Against a real API you would start around 1 second and cap near 60.

Caching Identical Requests

A support bot answers the same questions all day. Paying the provider again for an identical request is waste.

import hashlib, json, sqlite3, time
from provider import FlakyProvider

db = sqlite3.connect("llm_cache.db")
db.execute("CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, response TEXT)")
api = FlakyProvider()

def cached_complete(prompt: str, max_tokens: int = 256) -> str:
    request = {"model": "small-v1", "prompt": prompt, "max_tokens": max_tokens}
    key = hashlib.sha256(json.dumps(request, sort_keys=True).encode()).hexdigest()
    row = db.execute("SELECT response FROM cache WHERE key = ?", (key,)).fetchone()
    if row:
        return row[0]
    time.sleep(0.8)                      # a real API round trip costs time and money
    text = api.complete(prompt, max_tokens)
    db.execute("INSERT INTO cache VALUES (?, ?)", (key, text)); db.commit()
    return text

for prompt, mt in [("What is 2 + 2?", 256), ("What is 2 + 2?", 256),
                   ("What is 2+2?", 256), ("What is 2 + 2?", 64)]:
    t = time.perf_counter()
    cached_complete(prompt, mt)
    print(f"{prompt!r:18} max_tokens={mt:<3} {time.perf_counter() - t:.3f}s  api.calls={api.calls}")

$ python cache.py
'What is 2 + 2?'   max_tokens=256 0.803s  api.calls=1
'What is 2 + 2?'   max_tokens=256 0.000s  api.calls=1
'What is 2+2?'     max_tokens=256 0.803s  api.calls=2
'What is 2 + 2?'   max_tokens=64  0.803s  api.calls=3
$ python cache.py
'What is 2 + 2?'   max_tokens=256 0.000s  api.calls=0
'What is 2 + 2?'   max_tokens=256 0.000s  api.calls=0
'What is 2+2?'     max_tokens=256 0.000s  api.calls=0
'What is 2 + 2?'   max_tokens=64  0.000s  api.calls=0

The repeat request drops from 0.803s to effectively zero and never reaches the API. Two details are easy to miss:

• The key is exact. "What is 2+2?" without spaces is a miss, and so is the same prompt with a different max_tokens. Everything that can change the answer (model, system prompt, parameters) belongs in the key, and nothing fuzzier than that. Normalizing whitespace before hashing is fine. Treating two different questions as one is a wrong-answer bug.
• The second run made zero API calls. The cache is on disk, so it survives restarts, which is the point. But nothing ever expires. Answers generated before you changed the model or the system prompt will be served forever. Practice task 1 adds a TTL.

Cache only what should be deterministic: temperature-0 classification, extraction, FAQ answers. A cached creative reply, or a cached answer containing one user’s data served to another, is a bug.

Where the Money Goes

Cost is input tokens times the input price plus output tokens times the output price, and output is usually priced several times higher. The prices below are placeholders with a realistic ratio. Check your provider’s pricing page for real ones.

# Illustrative prices in USD per million tokens -- check your provider's pricing page.
PRICES = {"small": (1.00, 5.00), "large": (5.00, 25.00)}   # (input, output)

def cost(model: str, input_tokens: int, output_tokens: int) -> float:
    p_in, p_out = PRICES[model]
    return (input_tokens * p_in + output_tokens * p_out) / 1_000_000

if __name__ == "__main__":
    REQUESTS = 100_000          # one month of a document-analysis feature
    SYSTEM, QUESTION = 2_000, 300

    setups = {
        "large, full system prompt, 900-token answers": ("large", SYSTEM + QUESTION, 900),
        "large, system prompt trimmed to 400 tokens":   ("large", 400 + QUESTION, 900),
        "  + max_tokens and 'answer briefly' -> 250":   ("large", 400 + QUESTION, 250),
        "  + small model for the same traffic":         ("small", 400 + QUESTION, 250),
    }
    baseline = None
    for name, (model, tin, tout) in setups.items():
        monthly = cost(model, tin, tout) * REQUESTS
        baseline = baseline or monthly
        print(f"{name:46} ${monthly:>9,.2f}/month  ({monthly / baseline:6.1%} of baseline)")

$ python cost.py
large, full system prompt, 900-token answers   $ 3,400.00/month  (100.0% of baseline)
large, system prompt trimmed to 400 tokens     $ 2,600.00/month  ( 76.5% of baseline)
  + max_tokens and 'answer briefly' -> 250     $   975.00/month  ( 28.7% of baseline)
  + small model for the same traffic           $   195.00/month  (  5.7% of baseline)

Trimming a 2,000-token system prompt to 400 tokens saved 23.5%. That is real money, but it’s the smallest lever here. Cutting answers from 900 to 250 tokens saved far more, because every output token costs five times an input token at these prices. One caveat: max_tokens is only a ceiling that truncates. It’s the “answer briefly” instruction that actually makes answers shorter, and the cap stops a runaway one. The single biggest step is moving traffic to a smaller model, a 5x price drop on both sides. Whether that’s acceptable is a quality question, which the next section answers per task rather than globally.

Save this as cost.py. The __main__ guard lets the next example import cost without rerunning this table.

Routing Each Task to the Cheapest Model That Handles It

Classification and extraction rarely need the largest model. Multi-step reasoning often does.

from collections import Counter
from cost import cost

ROUTES = {                      # task type -> (model tier, max_tokens)
    "classify":  ("small", 16),
    "extract":   ("small", 200),
    "summarize": ("small", 300),
    "reason":    ("large", 1200),
}

# A day of traffic: (task type, input tokens, output tokens actually generated)
traffic = [("classify", 350, 3)] * 6000 + [("extract", 900, 120)] * 2500 \
        + [("summarize", 3000, 250)] * 1200 + [("reason", 1500, 700)] * 300

spend, calls = Counter(), Counter()
for task, tin, tout in traffic:
    model, max_tokens = ROUTES[task]
    spend[model] += cost(model, tin, min(tout, max_tokens))
    calls[model] += 1

everything_large = sum(cost("large", tin, tout) for _, tin, tout in traffic)
for model in spend:
    print(f"{model:5}  {calls[model]:>5} calls  ${spend[model]:7.2f}")
print(f"routed total ${sum(spend.values()):.2f}  vs  all-large ${everything_large:.2f}")

$ python router.py
small   9700 calls  $  11.04
large    300 calls  $   7.50
routed total $18.54  vs  all-large $62.70

The 300 reasoning calls (3% of traffic) cost two-thirds as much as the other 9,700 combined. That’s typical: a small share of hard requests dominates spend, so it’s the easy majority that you move to the small model. Routing cuts the day from $62.70 to $18.54.

The max_tokens values are per-route ceilings. No request in this traffic hits one, which is how it should be. A cap you hit regularly is truncating real answers.

The routing table encodes a claim, “the small model is good enough for extraction”, that you have to verify with an evaluation set rather than assume. See the evaluation lesson. Practice task 3 prices in what happens when that claim is only partly true.

A Circuit Breaker for Provider Outages

During a real outage, retries make things worse: every request hammers a failing service several times, and every user waits through all the backoff delays. A circuit breaker stops calling the provider for a while.

import time
from provider import FlakyProvider, OverloadedError

FALLBACK = "[fallback] busy, try again shortly"

class CircuitBreaker:
    def __init__(self, api, threshold=3, cooldown=1.0):
        self.api, self.threshold, self.cooldown = api, threshold, cooldown
        self.failures, self.opened_at = 0, None

    def complete(self, prompt):
        if self.opened_at and time.monotonic() - self.opened_at < self.cooldown:
            return FALLBACK                           # open: don't touch the API at all
        try:
            text = self.api.complete(prompt)          # closed, or the half-open trial call
            self.failures, self.opened_at = 0, None
            return text
        except OverloadedError:
            self.failures += 1
            if self.failures >= self.threshold:
                self.opened_at = time.monotonic()
            return FALLBACK

api = FlakyProvider(script=[OverloadedError] * 3)    # a provider outage, then recovery
breaker = CircuitBreaker(api)
for i in range(1, 8):
    if i == 6: time.sleep(1.1)                        # the cooldown passes
    reply = breaker.complete("hours?")
    print(f"request {i}: api.calls={api.calls}  {reply}")

$ python breaker.py
request 1: api.calls=1  [fallback] busy, try again shortly
request 2: api.calls=2  [fallback] busy, try again shortly
request 3: api.calls=3  [fallback] busy, try again shortly
request 4: api.calls=3  [fallback] busy, try again shortly
request 5: api.calls=3  [fallback] busy, try again shortly
request 6: api.calls=4  answer to 'hours?'
request 7: api.calls=5  answer to 'hours?'

Watch api.calls. The first three requests reach the provider and fail. The third failure hits threshold and opens the circuit. Requests 4 and 5 get the fallback instantly, and the call count stays at 3: the struggling provider sees no traffic from us at all. Once the cooldown passes, request 6 is let through as a trial (the “half-open” state). It succeeds, the failure count resets, and traffic flows normally again.

In production the breaker wraps the retry function, not the other way round: retries absorb a single blip, and the breaker absorbs a sustained outage. The fallback can be a smaller backup model, a cached answer, or an honest “try again” message. The provider abstraction lesson builds the backup-model version with a real local model.

Practice

1. The SQLite cache never expires. Add a created timestamp column and a ttl parameter so an entry older than ttl seconds counts as a miss. Test it with ttl=0.5 and pauses of 0, 0.1 and 0.6 seconds between identical requests.

   Solution

   import hashlib, json, sqlite3, time
   from provider import FlakyProvider
   
   db = sqlite3.connect(":memory:")
   db.execute("CREATE TABLE cache (key TEXT PRIMARY KEY, response TEXT, created REAL)")
   api = FlakyProvider()
   
   def cached_complete(prompt, max_tokens=256, ttl=0.5):
       key = hashlib.sha256(json.dumps([prompt, max_tokens]).encode()).hexdigest()
       row = db.execute("SELECT response, created FROM cache WHERE key = ?", (key,)).fetchone()
       if row and time.time() - row[1] < ttl:
           return row[0]
       text = api.complete(prompt, max_tokens)
       db.execute("INSERT OR REPLACE INTO cache VALUES (?, ?, ?)", (key, text, time.time()))
       db.commit()
       return text
   
   for pause in (0, 0.1, 0.6):
       time.sleep(pause)
       cached_complete("What is 2 + 2?")
       print(f"after {pause}s pause: api.calls={api.calls}")

   $ python sol_ttl.py
   after 0s pause: api.calls=1
   after 0.1s pause: api.calls=1
   after 0.6s pause: api.calls=2

   The 0.1s repeat is a hit. After 0.6s the entry is older than the TTL, so the provider is called again. INSERT OR REPLACE overwrites the stale row instead of failing on the primary key.

2. Real 429 responses often include a Retry-After header. Make the retry loop wait for a retry_after attribute on the exception when one is present, and fall back to computed backoff otherwise.

   Solution

   import time
   from provider import FlakyProvider, RateLimitError
   
   class RateLimited(RateLimitError):
       retry_after = 0.3                      # what the server's Retry-After header said
   
   def call(api, prompt, max_retries=3):
       for attempt in range(max_retries + 1):
           try:
               return api.complete(prompt)
           except RateLimitError as e:
               if attempt == max_retries:
                   raise
               wait = getattr(e, "retry_after", None) or 0.2 * 2 ** attempt
               print(f"  attempt {attempt + 1}: waiting {wait:.2f}s (server-specified)")
               time.sleep(wait)
   
   print(call(FlakyProvider([RateLimited, RateLimited]), "hours?"))

   $ python sol_retry_after.py
     attempt 1: waiting 0.30s (server-specified)
     attempt 2: waiting 0.30s (server-specified)
   answer to 'hours?'

   The server knows when its rate-limit window resets, and your client doesn’t. Honouring its value avoids retrying too early, which wastes an attempt, and waiting too long, which wastes the user’s time.

3. Routing assumes the small model handles extraction. Suppose some fraction of its extractions fail validation and are retried on the large model. Price 2,500 extractions (900 in, 120 out) at failure rates of 0%, 10% and 50%, against sending everything to the large model.

   Solution

   from cost import cost
   
   # 2,500 extractions/day; small model's output fails validation on some fraction,
   # and each failure is retried on the large model.
   for fail_rate in (0.0, 0.10, 0.50):
       failed = int(2500 * fail_rate)
       routed = 2500 * cost("small", 900, 120) + failed * cost("large", 900, 120)
       all_large = 2500 * cost("large", 900, 120)
       print(f"fail rate {fail_rate:4.0%}: routed ${routed:6.2f}  vs  all-large ${all_large:6.2f}")

   $ python sol_escalate.py
   fail rate   0%: routed $  3.75  vs  all-large $ 18.75
   fail rate  10%: routed $  5.62  vs  all-large $ 18.75
   fail rate  50%: routed $ 13.12  vs  all-large $ 18.75

   Even at a 50% failure rate, try-small-then-escalate is cheaper, because the small call costs a fifth of the large one. Break-even is an 80% failure rate. Latency is a different story: a failed request now takes two round trips, so the escalation rate matters for p95 latency long before it matters for cost.

4. The breaker only counts OverloadedError. Script the provider with [RateLimitError, OverloadedError, RateLimitError] (import RateLimitError too) and the original breaker crashes on request 1. Make it treat rate limits as failures too.

   Solution

   Change the import to from provider import FlakyProvider, OverloadedError, RateLimitError, the handler to except (OverloadedError, RateLimitError):, and the script to [RateLimitError, OverloadedError, RateLimitError].

   $ python sol_breaker.py
   request 1: api.calls=1  [fallback] busy, try again shortly
   request 2: api.calls=2  [fallback] busy, try again shortly
   request 3: api.calls=3  [fallback] busy, try again shortly
   request 4: api.calls=3  [fallback] busy, try again shortly
   request 5: api.calls=3  [fallback] busy, try again shortly
   request 6: api.calls=4  answer to 'hours?'
   request 7: api.calls=5  answer to 'hours?'

   The output matches the all-529 run, because the breaker now treats both as “the provider is struggling”. Keep BadRequestError out of that tuple: a malformed request says nothing about the provider’s health, and letting it open the circuit would take the feature down for every user because of one bad input.