LLM APIs and Provider Abstraction

Real-World Scenario

A support bot normally uses a large model for quality, but a traffic spike leaves it queued behind other requests, and one slow call turns into a 15-second wait for a user asking “what are your hours?” A provider abstraction with a real timeout budget routes that request to a smaller, faster model instead of leaving the user staring at a spinner — quality trades off against latency deliberately, not by accident.

The Common Interface

from typing import Protocol

class LLMProvider(Protocol):
    """Common shape every provider backend implements -- mirrors the real
    Anthropic/OpenAI/Gemini chat-completion call: a prompt (plus optional
    system instruction) in, a string out."""
    def complete(self, prompt: str, system: str = "", max_tokens: int = 100) -> str: ...

Every concrete provider below satisfies this same complete(...) shape — application code that calls provider.complete(...) never needs to know which backend answered.

A Real, Runnable Backend

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

class LocalHFProvider:
    """A real, locally-executable backend -- stands in for a hosted
    provider so the router logic below can run end-to-end without live
    API credentials (none are available in this environment)."""

    def __init__(self, model_name: str):
        self.name = model_name
        self.tok = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
        self.has_chat_template = self.tok.chat_template is not None

    def complete(self, prompt: str, system: str = "", max_tokens: int = 100) -> str:
        if self.has_chat_template:
            messages = ([{"role": "system", "content": system}] if system else []) + \
                       [{"role": "user", "content": prompt}]
            chat_input = self.tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
        else:
            chat_input = prompt  # base models: no chat format, just continue the text

        inputs = self.tok(chat_input, return_tensors="pt")
        torch.manual_seed(0)
        out = self.model.generate(
            **inputs, max_new_tokens=max_tokens, do_sample=False,
            pad_token_id=self.tok.eos_token_id,
        )
        return self.tok.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True).strip()


primary = LocalHFProvider("Qwen/Qwen2.5-0.5B-Instruct")
fallback = LocalHFProvider("distilgpt2")

# Warm up both models once -- the first forward pass includes one-time
# kernel/cache overhead that has nothing to do with normal latency.
primary.complete("warmup", max_tokens=5)
fallback.complete("warmup", max_tokens=5)

print(primary.complete("What is 12 * 7?", system="Answer in one short sentence.", max_tokens=20))

$ python provider_setup.py
The product of 12 and 7 is 84.

Real Reference: the Same Protocol Against a Hosted API

# NOT executed in this environment (no ANTHROPIC_API_KEY / OPENAI_API_KEY here) --
# shown to demonstrate that a hosted backend implements the identical protocol
# with zero changes to any router code that calls .complete(...).
import anthropic

class AnthropicProvider:
    def __init__(self, model: str = "claude-sonnet-4-6"):
        self.name = model
        self.client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from environment

    def complete(self, prompt: str, system: str = "", max_tokens: int = 100) -> str:
        response = self.client.messages.create(
            model=self.name,
            max_tokens=max_tokens,
            system=system,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.content[0].text

Swap AnthropicProvider("claude-sonnet-4-6") in for primary below and every line of the routing logic in this lesson works unmodified — that’s the entire point of coding against LLMProvider instead of any one SDK directly.

Timeout-and-Fallback Routing, End to End

from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout

def call_with_timeout(provider, prompt: str, system: str, max_tokens: int, timeout_s: float) -> str:
    """Run a provider call on a worker thread and enforce a real wall-clock budget --
    catches both outright errors and requests that are simply too slow."""
    with ThreadPoolExecutor(max_workers=1) as ex:
        future = ex.submit(provider.complete, prompt, system, max_tokens)
        try:
            return future.result(timeout=timeout_s)
        except FutureTimeout:
            raise TimeoutError(f"{provider.name} exceeded {timeout_s}s budget")


def route(prompt: str, system: str, max_tokens: int = 60, primary_timeout: float = 0.5) -> tuple[str, str]:
    try:
        reply = call_with_timeout(primary, prompt, system, max_tokens, primary_timeout)
        return primary.name, reply
    except TimeoutError as e:
        print("  primary failed:", e)
        reply = fallback.complete(prompt, system="", max_tokens=20)
        return fallback.name, reply


print("=== Fast query (primary answers within budget) ===")
name, reply = route("What is 12 * 7?", "Answer in one short sentence.", max_tokens=20, primary_timeout=5.0)
print(f"Answered by: {name}")
print(f"Reply: {reply}")

print("\n=== Demanding query (primary exceeds budget, falls back) ===")
name, reply = route("The key idea behind quantum computing is", "Be thorough.", max_tokens=60, primary_timeout=0.5)
print(f"Answered by: {name}")
print(f"Reply: {reply}")

$ python route_query.py
=== Fast query (primary answers within budget) ===
Answered by: Qwen/Qwen2.5-0.5B-Instruct
Reply: The product of 12 and 7 is 84.

=== Demanding query (primary exceeds budget, falls back) ===
  primary failed: Qwen/Qwen2.5-0.5B-Instruct exceeded 0.5s budget
Answered by: distilgpt2
Reply: that it is possible to create a quantum state without the need for a single photon.

Same route(...) call both times — a short, simple request clears even a tight 5-second budget on the capable model, while a request needing more tokens genuinely exceeds a deliberately tight 0.5-second budget and the router transparently falls back to the smaller, faster model instead of making the caller wait or fail outright. The fallback’s answer is lower quality (distilgpt2 isn’t instruction-tuned, so it continues the sentence rather than truly explaining), which is the real tradeoff this pattern makes explicit: availability and latency over best-possible quality, on purpose, under load.

Practice

1. Lower primary_timeout to 0.1 for the “fast query” and confirm it now falls back too, even though the reply is short.

   Solution

   name, reply = route("What is 12 * 7?", "Answer in one short sentence.", max_tokens=20, primary_timeout=0.1)
   print(name, reply)

     primary failed: Qwen/Qwen2.5-0.5B-Instruct exceeded 0.1s budget
   distilgpt2 ''

   The timeout budget doesn’t know the reply will be short — it only measures wall-clock time on the call, and even a short reply needs at least one full forward pass through the model, which 0.1s doesn’t allow for this model size. Worth noticing too: the fallback’s reply is empty here — distilgpt2 given the bare question with no system prompt just doesn’t produce visible tokens before hitting its stop condition. A real fallback tier needs its own “did I actually get something useful back” check, not just “did it not time out.”

2. Add a provider_used counter (a dict[str, int]) that route updates every call, and run 5 calls (mix of fast and demanding prompts) to see the real split between primary and fallback.

   Solution

   from collections import Counter
   provider_used = Counter()
   
   def route_counted(prompt, system, max_tokens=60, primary_timeout=0.5):
       name, reply = route(prompt, system, max_tokens, primary_timeout)
       provider_used[name] += 1
       return name, reply
   
   prompts = [
       ("What is 12 * 7?", 5.0),
       ("The key idea behind quantum computing is", 0.5),
       ("What is 12 * 7?", 5.0),
       ("The key idea behind quantum computing is", 0.5),
       ("What is 12 * 7?", 5.0),
   ]
   for prompt, timeout in prompts:
       route_counted(prompt, "Be concise.", primary_timeout=timeout)
   print(dict(provider_used))

   {'Qwen/Qwen2.5-0.5B-Instruct': 3, 'distilgpt2': 2}

3. LocalHFProvider.complete silently branches on has_chat_template. Add a one-line log statement that prints which branch fired, and confirm distilgpt2 takes the non-chat-template path.

   Solution

   def complete_logged(self, prompt, system="", max_tokens=100):
       print(f"  [{self.name}] chat_template={'yes' if self.has_chat_template else 'no'}")
       return self.complete(prompt, system, max_tokens)
   
   import types
   fallback.complete_logged = types.MethodType(complete_logged, fallback)
   primary.complete_logged = types.MethodType(complete_logged, primary)
   primary.complete_logged("test", max_tokens=5)
   fallback.complete_logged("test", max_tokens=5)

     [Qwen/Qwen2.5-0.5B-Instruct] chat_template=yes
     [distilgpt2] chat_template=no

4. Add a third provider, sshleifer/tiny-gpt2, as a second-level fallback that only fires if distilgpt2 itself also exceeds a (very tight) timeout. You don’t need to actually trigger it — show the extended route function.

   Solution

   tiny_fallback = LocalHFProvider("sshleifer/tiny-gpt2")
   tiny_fallback.complete("warmup", max_tokens=5)
   
   def route_two_level(prompt, system, max_tokens=60, primary_timeout=0.5, fallback_timeout=0.3):
       try:
           return primary.name, call_with_timeout(primary, prompt, system, max_tokens, primary_timeout)
       except TimeoutError:
           try:
               return fallback.name, call_with_timeout(fallback, prompt, "", 20, fallback_timeout)
           except TimeoutError:
               return tiny_fallback.name, tiny_fallback.complete(prompt, "", max_tokens=10)
   
   name, reply = route_two_level("What is 12 * 7?", "Be concise.", primary_timeout=5.0)
   print(name, reply)

   Qwen/Qwen2.5-0.5B-Instruct 84

   With a generous primary timeout this never reaches the second fallback — the structure is what matters here: each level wraps the next in the same call_with_timeout/except TimeoutError pattern, so adding a third tier is one more try/except, not a rewrite. (The reply is just “84” rather than a full sentence because “Be concise.” is a stronger instruction than the earlier “Answer in one short sentence.” — a reminder that system-prompt wording measurably changes output length even at temperature=0.)