Model routing, caching and graceful fallbacks

Read a sample · Chapter 01 of 06

A cheaper call is useful only when the resulting answer still meets the task.

Imagine a fictional learning service that handles short lookups and longer explanations. We want to avoid paying for an expensive route when a simpler one is adequate, while preserving an explicit standard for an acceptable answer. The policy needs evidence about task types and outcomes. A short prompt is not necessarily easy, and a long prompt is not necessarily difficult. This lab therefore uses a trusted task label instead of pretending that prompt length measures reasoning difficulty.

The experiment is deliberately synthetic. FAST costs one invented unit and takes ten simulated milliseconds. STRONG costs four units and takes thirty. These names are fixture labels, not commercial products or claims about model sizes. A scripted function supplies answers and failures. The simulator does not sleep, call a network, measure execution latency or estimate tokens. Its clock advances by the assigned duration of each attempted model call.

Scroll sideways to see every column.

Create router_lab.py and append its sections in order. The first section below defines immutable request and model records and an integer clock. Request.validate rejects empty identifiers and unsupported task labels before a lookup or call. The dataclass is an internal contract, not an HTTP request parser: the caller must construct a Request after authentication, authorisation and input-size checks. Neither a type annotation nor a scope string performs those checks.

"""Serial, synthetic routing lab. No provider calls or real billing."""
from collections import OrderedDict
from dataclasses import dataclass

def natural(value, name, minimum=0):
    if type(value) is not int or value < minimum:
        raise ValueError(name)

@dataclass(frozen=True)
class Request:
    scope: str
    prompt: str
    context_version: str = 'docs-v1'
    task: str = 'simple'
    cacheable: bool = True

    def validate(self):
        for value in (self.scope, self.prompt, self.context_version):
            if type(value) is not str or not value.strip():
                raise ValueError('empty or non-string request field')
        if self.task not in ('simple', 'complex'):
            raise ValueError('task')
        if type(self.cacheable) is not bool:
            raise ValueError('cacheable')

@dataclass(frozen=True)
class Model:
    name: str
    units: int
    duration: int

FAST = Model('fast-v1', 1, 10)
STRONG = Model('strong-v1', 4, 30)

class Clock:
    def __init__(self):
        self.now = 0

    def advance(self, milliseconds):
        natural(milliseconds, 'milliseconds')
        self.now += milliseconds

Treat scope as an opaque identifier for a permitted answer-sharing group, including the relevant permission generation. It could represent one user within a tenant and a current authorisation version. A client must not choose another person's scope. Changing permissions requires a new scope or explicit invalidation before looking up an answer. The lab checks equality of supplied scopes; it does not prove that they were issued correctly.

context_version identifies the exact information and instruction context affecting the answer. In this fixture it is a manually supplied string. A real adapter must derive and maintain a version for the actual system instructions, retrieved documents, tool schema, output format and generation settings. Omitting one changing input can make two different tasks look equivalent. Keep raw user text exact here: even trailing whitespace forms a different request.

The three policies are strong, fast and route. The first two always choose their named fixture and never fall back. The route policy uses the task label, then permits one upward fallback for a simple task after a transient failure. It never downgrades a complex task to FAST. Eligibility is an application decision that should be established before availability or price chooses among candidates.

Read a sample · Chapter 02 of 06

Freshness, equivalence and eviction solve different problems.

A complete-response cache returns stored answer text without generating a new answer. Provider prompt caching reuses processing of an eligible repeated prompt prefix while generation continues; it does not mean that your application has stored and replayed a final answer. OpenAI's prompt-caching guide describes prefix reuse. Its eligibility and retention are provider-specific. This lab implements only the application response cache and makes no provider token-discount claim.

RFC 9111 describes HTTP cache freshness, reuse and restrictions around shared and authenticated responses. It is useful background for asking whether a response may be reused, but the following class is not an HTTP cache and does not implement that standard. It has no headers, validators or revalidation protocol. Do not place it in front of private HTTP responses and assume its small key replaces the required access controls.

class Cache:
    def __init__(self, clock, capacity=4, ttl=100):
        natural(capacity, 'capacity', 1)
        natural(ttl, 'ttl', 1)
        self.clock, self.capacity, self.ttl = clock, capacity, ttl
        self.items = OrderedDict()

    def purge(self):
        for key, (expires, _) in list(self.items.items()):
            if self.clock.now >= expires:
                del self.items[key]

    def get(self, key):
        self.purge()
        if key not in self.items:
            return None
        self.items.move_to_end(key)
        return self.items[key][1]

    def put(self, key, text):
        self.purge()
        self.items[key] = (self.clock.now + self.ttl, text)
        self.items.move_to_end(key)
        while len(self.items) > self.capacity:
            self.items.popitem(last=False)

The cache stores an expiry timestamp and text for each key. put runs only after the first model attempt has completed successfully, so expiry is completion time plus TTL. get removes entries when current time is greater than or equal to expiry. An entry is usable just before that boundary and unusable at it. A hit changes recency but does not extend the original expiry. This is a fixed lifetime, not a sliding one.

For example, a FAST answer started at time zero finishes at ten. With TTL 100 it expires at 110, not at 100. A hit at 109 succeeds, while a read at 110 misses. If repeated reads restarted the lifetime, a frequently requested answer could remain indefinitely despite the intended freshness interval. These endpoint choices are observable requirements, so the tests cover both completion time and exact expiry.

OrderedDict supplies the recency order. A successful read moves its key to the newest end. Insertion does the same, and overflowing capacity removes the oldest end. Purging expired entries happens before both operations so expired entries do not displace live ones. This simple purge scans the whole small cache. Capacity bounds the entry count only; there is no bound on the size of a prompt or answer in bytes.

The simulated clock is advanced only forwards through Clock.advance. Production elapsed durations should use a monotonic clock, whose differences are unaffected by wall-clock corrections, as described in Python's time documentation. Do not persist a process-local monotonic timestamp and assume it remains meaningful after a restart. This lab stores nothing on disk and does not implement a production clock adapter.

TTL limits reuse but does not establish factual freshness. A changed document can make an answer obsolete one millisecond after insertion. Update context_version when the source changes, or disable caching for the request. The cacheable field is a trusted host decision. False bypasses both reads and writes; it should be used for requests whose answers must not be retained or repeated under the chosen policy.

Read a sample · Chapter 03 of 06

A transient failure, a refusal and a programming error should not share a retry path.

Append the remaining router_lab.py source below. The key contains the full immutable Request plus the primary Model, policy and policy revision. The model record includes its versioned name. A policy revision can intentionally separate results after the routing or answer contract changes. The per-request spending budget is not part of the key: a valid cached answer can be served with zero new units even when that budget is zero.

@dataclass(frozen=True)
class Result:
    status: str
    text: str
    model: str
    hit: bool
    attempts: int
    units: int
    elapsed: int

class Router:
    def __init__(self, provider, clock, cache=None,
                 policy='route', revision='policy-v1'):
        if policy not in ('route', 'fast', 'strong'):
            raise ValueError('policy')
        if type(revision) is not str or not revision.strip():
            raise ValueError('revision')
        if cache is not None and cache.clock is not clock:
            raise ValueError('cache must share the clock')
        self.provider, self.clock, self.cache = provider, clock, cache
        self.policy, self.revision = policy, revision

    def run(self, request, budget=5):
        request.validate()
        natural(budget, 'budget')
        primary = FAST
        if self.policy == 'strong' or (
                self.policy == 'route' and request.task == 'complex'):
            primary = STRONG

        key = (request, primary, self.policy, self.revision)
        usable = self.cache is not None and request.cacheable
        if usable:
            text = self.cache.get(key)
            if text is not None:
                return Result('ok', text, primary.name, True, 0, 0, 0)

        started, spent, attempts = self.clock.now, 0, 0
        models = [primary]
        if self.policy == 'route' and primary == FAST:
            models.append(STRONG)
        for model in models:
            if spent + model.units > budget:
                return Result('budget', '', '', False, attempts,
                              spent, self.clock.now - started)
            spent += model.units
            attempts += 1
            self.clock.advance(model.duration)
            status, text = self.provider(model, request)
            if status not in ('ok', 'transient', 'refused'):
                raise ValueError('unknown provider status')
            if type(text) is not str or (status == 'ok' and not text):
                raise ValueError('invalid provider text')

            if status == 'ok':
                if usable and attempts == 1:
                    self.cache.put(key, text)
                return Result('ok', text, model.name, False, attempts,
                              spent, self.clock.now - started)
            if status == 'refused':
                return Result('refused', '', model.name, False,
                              attempts, spent, self.clock.now - started)
        return Result('unavailable', '', '', False, attempts,
                      spent, self.clock.now - started)

def fixture(model, request):
    """Invented outputs and failures, not a model quality assessment."""
    if request.prompt == 'restricted':
        return 'refused', ''
    if request.prompt == 'outage' and model == FAST:
        return 'transient', ''
    if request.task == 'complex':
        return 'ok', 'reasoned' if model == STRONG else 'incomplete'
    return 'ok', 'answer'

run validates the request before consulting the cache. On a miss it selects a finite list of eligible models: one normally, two only for a simple request in route mode. Before each attempt it compares spent plus that model's units with the budget. The default budget of five allows a one-unit FAST failure followed by a four-unit STRONG attempt. Budget four allows the first attempt but stops before the second; the returned result still reports the one unit already spent.

Every attempted call is charged its fixed synthetic units and duration, including a transient failure or refusal. This is an accounting assumption for the fixture, not a statement about provider billing. Real prices can depend on input, output and cached tokens and on the service contract. Reserving an upper bound before starting a live call and reconciling its actual usage is a separate adapter responsibility. This program never charges money.

The provider returns one of three statuses. ok requires a non-empty string. transient permits the next eligible attempt if there is one. refused stops immediately, never trying to find a provider that will answer the same refused request. Unrecognised statuses, invalid output and unexpected provider exceptions raise errors rather than being disguised as temporary outages. A real application needs a visible outer error handler and accounting for such adapter faults; that handler is outside this lab.

Only a successful first attempt enters the cache. A successful fallback returns its actual model name but is not cached under the failed primary route. That keeps the next request free to try the recovered primary. It also means repeated outages can repeatedly pay for a failed first call. There is no circuit breaker, failure-memory window, shared retry budget or backoff here. Those are additional policies that must be designed and tested, not inferred from this two-attempt loop.

Google's discussion of cascading failures explains why retries need limits and why retries at multiple layers can multiply load. This fixture has one synchronous routing layer with no hidden retries. In a live stack, inspect the client SDK, gateway and server together. Passing two attempts to an adapter that silently retries three times does not yield a two-call system. Fallback also needs enough remaining time and spare downstream capacity.

Scroll sideways to see every column.

Read a sample · Chapter 04 of 06

Keep cost, answer acceptance and refusals on declared denominators.

Save demo.py below beside router_lab.py and run python demo.py. The six requests are two identical simple questions for reader A, one complex question for A, the same simple question for reader B, a temporary FAST outage and one deliberately refused request. The scripted answer for a complex task is reasoned only on STRONG. These labels define the fixture; they are not a learned quality judge or a benchmark dataset.

from router_lab import Cache, Clock, Request, Router, fixture

TRACE = [
    (Request('reader-a', 'sum'), 'answer'),
    (Request('reader-a', 'sum'), 'answer'),
    (Request('reader-a', 'why', task='complex'), 'reasoned'),
    (Request('reader-b', 'sum'), 'answer'),
    (Request('reader-a', 'outage'), 'answer'),
    (Request('reader-a', 'restricted'), None),
]

def evaluate(policy, cached):
    clock = Clock()
    cache = Cache(clock) if cached else None
    router = Router(fixture, clock, cache, policy=policy)
    results = [router.run(request) for request, _ in TRACE]
    accepted = sum(r.status == 'ok' and r.text == expected
                   for r, (_, expected) in zip(results, TRACE)
                   if expected is not None)
    refusals = sum(r.status == 'refused' for r in results)
    print(f'{policy} cache={int(cached)}: '
          f'units={sum(r.units for r in results)} '
          f'ms={sum(r.elapsed for r in results)} '
          f'accepted={accepted}/5 refused={refusals}/1 '
          f'hits={sum(r.hit for r in results)}/6 '
          f'attempts={sum(r.attempts for r in results)}')
    return results

if __name__ == '__main__':
    for policy, cached in [('strong', False), ('fast', False),
                           ('route', False), ('route', True)]:
        evaluate(policy, cached)

strong cache=0: units=24 ms=180 accepted=5/5 refused=1/1 hits=0/6 attempts=6
fast cache=0: units=6 ms=60 accepted=3/5 refused=1/1 hits=0/6 attempts=6
route cache=0: units=13 ms=110 accepted=5/5 refused=1/1 hits=0/6 attempts=7
route cache=1: units=12 ms=100 accepted=5/5 refused=1/1 hits=1/6 attempts=6

The all-STRONG baseline attempts six calls, spending 24 units and advancing time 180 milliseconds. Five answerable requests meet their exact expected strings and the restricted request is refused. All-FAST costs six units and sixty milliseconds, but only three of five answerable requests pass: its complex answer is incomplete and its outage has no fallback. A lower bill alone hides those two failures.

Routing without caching uses FAST for the ordinary simple requests, STRONG for the complex one and an extra STRONG attempt for the outage. This costs 13 units across seven attempts and takes 110 simulated milliseconds. It passes all five answerable cases and preserves the refusal. The extra attempt is counted even though there are only six user requests. Confusing those two denominators would hide work amplification.

Enabling the cache saves the immediate second request for reader A. The reader-B request must miss despite identical prompt text because its scope differs. The complex answer uses a different route and key. The fallback is intentionally not stored, and refusals are not stored. The final row therefore has one hit out of six submitted requests, six attempted calls, twelve units and one hundred simulated milliseconds.

Against all-STRONG, the last row saves twelve of twenty-four invented units, or fifty percent, on this selected trace. Against the same routing policy without a cache, caching alone saves one of thirteen units, about 7.7 percent. Those are different comparisons. Neither number predicts production savings. The fixture intentionally includes one repeat and favourable task labels; changing the traffic, expiry or quality outcomes changes the result.

A hit is assigned zero simulated time because this lab does not model lookup overhead. The total elapsed value is the sum of sequential invented call durations; it is neither measured wall-clock latency nor throughput under concurrent load. Do not divide the six requests by it to claim real serving capacity. Report the model of time alongside the number.

accepted uses the five answerable requests as its denominator. refused uses the one expected refusal. hits uses all six submitted requests. An exact-string acceptance check is suitable for these literal scripted outputs, not for judging open-ended explanations. A real evaluation needs a representative held-out set, an explicit rubric, error categories and a reviewer or validated checking method. Keep appropriately refused requests visible without falsely labelling them successful answers.

Read a sample · Chapter 05 of 06

A fast demo is incomplete if it reuses the wrong reader's answer or retries a refusal.

Save test_router.py below and run python -m unittest -v test_router.py. Fourteen test methods cover route eligibility, zero-cost hits, request dimensions, version separation, exact expiry, non-sliding lifetime, recency eviction, bypass, fallback, refusal, spending and invalid input. They run without downloading a model or requiring an account. The source here is the same source executed for the published evidence.

import unittest
from dataclasses import replace
from router_lab import Cache, Clock, Request, Router, FAST, STRONG, fixture

class RoutingTests(unittest.TestCase):
    def setUp(self):
        self.clock = Clock()
        self.cache = Cache(self.clock, capacity=2, ttl=100)
        self.calls = []
        def provider(model, request):
            self.calls.append((model.name, request))
            return fixture(model, request)
        self.router = Router(provider, self.clock, self.cache)
        self.request = Request('a', 'sum')

    def test_routing_and_quality(self):
        self.assertEqual(self.router.run(self.request).model, FAST.name)
        complex_request = replace(self.request, task='complex')
        result = self.router.run(complex_request)
        self.assertEqual((result.model, result.text), (STRONG.name, 'reasoned'))

    def test_hit_no_new_call_or_units(self):
        self.router.run(self.request)
        hit = self.router.run(self.request, budget=0)
        self.assertEqual((hit.hit, hit.attempts, hit.units, hit.elapsed),
                         (True, 0, 0, 0))
        self.assertEqual(len(self.calls), 1)

    def test_every_request_dimension_separates(self):
        for change in ({'scope':'b'}, {'prompt':'sum '},
                       {'context_version':'docs-v2'}, {'task':'complex'}):
            with self.subTest(change=change):
                self.setUp()
                self.router.run(self.request)
                changed = replace(self.request, **change)
                self.assertFalse(self.router.run(changed).hit)
                self.assertEqual(len(self.calls), 2)

    def test_policy_revision_and_model_separate(self):
        self.router.run(self.request)
        for policy, revision in [('route','v2'), ('strong','policy-v1')]:
            router = Router(fixture, self.clock, self.cache, policy, revision)
            self.assertFalse(router.run(self.request).hit)

    def test_expiry_starts_at_completion_and_is_exclusive(self):
        self.router.run(self.request)
        self.clock.advance(99)
        self.assertTrue(self.router.run(self.request).hit)
        self.clock.advance(1)
        self.assertFalse(self.router.run(self.request).hit)
        self.assertEqual(len(self.calls), 2)

    def test_hit_does_not_extend_ttl(self):
        self.cache.put('a', 'A')
        self.clock.advance(90)
        self.assertEqual(self.cache.get('a'), 'A')
        self.clock.advance(10)
        self.assertIsNone(self.cache.get('a'))

    def test_lru_touch_and_expired_entries(self):
        self.cache.put('a','A')
        self.cache.put('b','B')
        self.cache.get('a')
        self.cache.put('c','C')
        self.assertEqual(list(self.cache.items), ['a','c'])
        self.clock.advance(100)
        self.cache.put('d','D')
        self.assertEqual(list(self.cache.items), ['d'])

    def test_non_cacheable_bypasses_reads_and_writes(self):
        request = replace(self.request, cacheable=False)
        self.router.run(request)
        self.assertFalse(self.router.run(request).hit)
        self.assertEqual(len(self.calls), 2)
        self.assertFalse(self.cache.items)

    def test_transient_fallback_is_not_cached(self):
        request = replace(self.request, prompt='outage')
        result = self.router.run(request)
        self.assertEqual((result.status, result.model, result.attempts,
                          result.units, result.elapsed),
                         ('ok', STRONG.name, 2, 5, 40))
        self.assertFalse(self.router.run(request).hit)
        self.assertEqual(len(self.calls), 4)

    def test_refusal_is_terminal_and_not_cached(self):
        request = replace(self.request, prompt='restricted')
        self.assertEqual(self.router.run(request).status, 'refused')
        self.assertEqual(len(self.calls), 1)
        self.assertEqual(self.router.run(request).status, 'refused')
        self.assertEqual(len(self.calls), 2)

    def test_budget_checked_before_each_attempt(self):
        self.assertEqual(self.router.run(self.request, 0).status, 'budget')
        self.assertFalse(self.calls)
        request = replace(self.request, prompt='outage')
        result = self.router.run(request, 4)
        self.assertEqual((result.status, result.units, result.attempts),
                         ('budget', 1, 1))
        self.assertEqual(len(self.calls), 1)

    def test_exhaustion_and_no_quality_downgrade(self):
        def failed(model, request):
            self.calls.append(model)
            return 'transient', ''
        router = Router(failed, self.clock)
        result = router.run(self.request)
        self.assertEqual((result.status, len(self.calls)), ('unavailable', 2))
        self.calls.clear()
        result = router.run(replace(self.request, task='complex'))
        self.assertEqual((result.status, self.calls), ('unavailable', [STRONG]))

    def test_invalid_input_and_programming_errors(self):
        for change in ({'scope':''}, {'prompt':7}, {'task':'unknown'},
                       {'context_version':' '}, {'cacheable':1}):
            with self.assertRaises(ValueError):
                self.router.run(replace(self.request, **change))
        for budget in (-1, True, 1.5):
            with self.assertRaises(ValueError):
                self.router.run(self.request, budget)
        self.assertFalse(self.calls)
        for status, text in [('wrong',''), ('ok',''), ('ok',None)]:
            router = Router(lambda m,r: (status,text), self.clock)
            with self.assertRaises(ValueError):
                router.run(self.request)

    def test_cache_configuration_rejected(self):
        for capacity, ttl in [(0,1),(1,0),(True,1),(1,1.5)]:
            with self.assertRaises(ValueError):
                Cache(self.clock, capacity, ttl)
        with self.assertRaises(ValueError):
            Router(fixture, Clock(), self.cache)

if __name__ == '__main__':
    unittest.main()

The scope test makes a request for one reader, then changes one request dimension at a time. Every such change must cause a new call. It is an equality-contract check rather than an authentication test: a malicious caller who can forge the trusted Request could still provide somebody else's scope. Put that distinction in a review note instead of calling a passing unit test a security audit.

The expiry test deliberately asks at completion plus TTL minus one, then exactly at expiry. The first must hit and the second must miss. The separate non-sliding test reads an item shortly before expiry and confirms that the read did not extend it. The recency test reads A after writing A and B, inserts C, then checks that B was evicted. Inspecting only final cache size would miss the wrong eviction choice.

The budget test checks both stopping before any call and stopping before the second call after spending on a failure. The refusal test requires only one provider call for each restricted request. The exhaustion test confirms that a complex request never falls down to FAST, even when STRONG fails. These observations establish what the code does for selected cases; they do not establish a reliable model classifier or sufficient capacity during a live outage.

Author review also ran four syntactically valid faulty copies in isolated temporary directories. Omitting scope from the key, accepting an entry exactly at expiry, allowing refusal to reach fallback and ignoring already-spent units each caused its targeted test to fail. Original file hashes stayed unchanged. This makes those checks demonstrably sensitive to four plausible defects, while leaving other defects and production behaviours outside the tested set.

Keep errors distinguishable in reports. An unavailable response is not a refusal; a budget stop is not a model-quality failure; a malformed adapter response is not a transient backend outage. The tests raise invalid-output errors directly. A production adapter must define a narrow mapping from real provider statuses and exceptions into the contract, and must preserve observability when it rejects unexpected responses.

Read a sample · Chapter 06 of 06

A live adapter needs measurements and controls that the scripted lab intentionally leaves open.

The reusable output is a decision contract and an executable example. Before connecting a provider, record each candidate model's exact revision, allowed data boundary, supported output contract and observed quality on the relevant tasks. Establish which routes are eligible first. A provider that cannot meet the required schema or data-handling rules is not an acceptable fallback merely because it is available.

Scroll sideways to see every column.

This cache is single-process and single-threaded. It has no locks, single-flight request coalescing, distributed coordination, encrypted storage or durable invalidation. Two concurrent misses could both call a provider in a future adaptation unless coordinated. Large request text could exhaust memory despite the entry-count limit. Keep this teaching cache out of an untrusted service boundary until those concerns have explicit implementations and tests.

A live call must respect the request's remaining end-to-end time. A model duration field does not enforce a timeout; our field simply advances an invented clock. Cancellation may also need to reach a provider to stop downstream work. Handle in-progress streams and partial output deliberately: the lab only caches complete successful strings and never demonstrates resuming or replacing a partly delivered answer.

Google's overload chapter discusses accepting work that available capacity can serve and rejecting excess work deliberately. Carry that principle into routing: moving every failed request to a second backend can overload it. A bounded per-request fallback is only one limit. The application may need admission control, a service-wide retry budget and an explicitly designed degradation path with observable recovery.

A useful degradation might be a clearly labelled unavailable result with a later retry option, or an approved static explanation for a narrow task. Do not silently substitute a low-quality answer and count it as successful. If stale information is allowed for a particular product feature, define its age, scope and label explicitly. This lab serves no expired entry and has no stale-on-error mode.

For the next experiment, use a replayable, representative request set that the service is permitted to process. Run the same set across the candidate policies, controlling model versions and cache warm-up. Record per-request status, actual route, attempts, cache state, usage and timing, plus the quality judgement. Separate cold and warm cache results; show failures and refusals alongside accepted answers. Obtain owner-agreed thresholds before deciding whether to promote a policy.

The published evidence covers the exact local lab, fourteen tests, four deterministic demo lines and four detected defects. It does not cover actual model quality, provider billing, authentication, concurrent races, network timeout enforcement, sustained load or production deployment. A handover should name those remaining checks and the person or team responsible for acceptance instead of turning the fixture's fifty-percent result into a marketing claim.