Lesson 3: Provider Abstraction

The problem

Your provider raises prices, or their reliability degrades, or a cheaper model appears that is good enough for extraction. Swapping should take an afternoon. Instead their SDK is imported in twenty-three files, their response shape is assumed in your parsers, and their exception types appear in your error handling.

This lesson makes that swap a configuration change.

Adapters: four incompatible SDKs behind one interface

An adapter converts one interface into another. Here, four providers with four different request shapes, response shapes, and error types are converted into one interface your code uses.

Start by defining what your code needs, in your own vocabulary:

from dataclasses import dataclass
from typing import Literal, Protocol

Role = Literal["system", "user", "assistant"]


@dataclass(frozen=True)
class Message:
    role: Role
    content: str


@dataclass(frozen=True)
class Usage:
    input_tokens: int
    output_tokens: int


@dataclass(frozen=True)
class Completion:
    text: str
    model: str
    usage: Usage
    finish_reason: Literal["stop", "length", "tool_use", "other"]


class LLMProvider(Protocol):
    name: str

    def generate(self, messages: list[Message], *, model: str, max_tokens: int) -> Completion: ...

    def embed(self, texts: list[str], *, model: str) -> list[list[float]]: ...

These types are yours. They are not any provider's shape, and that independence is the point: if every provider's response is converted into Completion, then nothing downstream knows or cares which provider produced it.

Note finish_reason normalised to a Literal with four values. Module 6 flagged that providers use different names for truncation, and this is where that difference is absorbed. The extraction code checks for "length" and never learns that one provider calls it max_tokens and another calls it something else.

An adapter for each provider:

class FastProviderAdapter:
    name = "fast"

    def __init__(self, api_key: str, transport: HTTPTransport) -> None:
        self._api_key = api_key
        self._transport = transport

    def generate(self, messages: list[Message], *, model: str, max_tokens: int) -> Completion:
        payload = {
            "model": model,
            "input": [{"speaker": m.role, "body": m.content} for m in messages],
            "limit": max_tokens,
        }
        data = self._transport.post_json(FAST_URL, payload, self._headers())
        return Completion(
            text=data["output"]["content"],
            model=data["model"],
            usage=Usage(
                input_tokens=data["meta"]["prompt_tokens"],
                output_tokens=data["meta"]["completion_tokens"],
            ),
            finish_reason=self._normalize_finish(data["meta"]["stopped_because"]),
        )

    @staticmethod
    def _normalize_finish(raw: str) -> Literal["stop", "length", "tool_use", "other"]:
        return {
            "complete": "stop",
            "limit_reached": "length",
            "tool_requested": "tool_use",
        }.get(raw, "other")

Every provider-specific detail is inside the adapter: the odd field names, the nested response structure, and the vocabulary of stop reasons. The adapter is the boundary from Module 2, applied to a provider rather than to a file.

[IMAGE PROMPT M7-2
Purpose: Show how adapters normalise four incompatible provider SDKs into one interface, and where provider knowledge is confined.
Visual type: Convergence architecture diagram with a marked boundary.
Prompt: A clean educational diagram reading right to left in three columns. The rightmost column shows four boxes stacked vertically labelled "Provider A SDK", "Provider B SDK", "Provider C SDK", and "Provider D SDK", each with a small sub-label indicating differing shapes: "input[]", "messages[]", "prompt", "turns[]". The middle column shows four corresponding adapter boxes labelled "AdapterA", "AdapterB", "AdapterC", "AdapterD", each connected by an arrow to its SDK. A vertical dashed line runs down the left edge of the adapter column, labelled "provider knowledge stops here". The leftmost column shows a single box labelled "LLMProvider (Protocol)" with a sub-label "generate, embed", receiving four converging arrows from the adapters, each arrow labelled "Completion". To the left of that, one box labelled "your application" with a sub-label "knows only Message, Completion, Usage".
Required elements: Four differing SDK boxes with differing shape sub-labels, four adapter boxes, a labelled boundary line, converging arrows all labelled with the same normalised type, the single protocol box and the application box.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, generous whitespace.
Layout: Three columns reading right to left, with the four provider rows converging leftward into one box.
Text labels: "Provider A SDK", "Provider B SDK", "Provider C SDK", "Provider D SDK", "input[]", "messages[]", "prompt", "turns[]", "AdapterA", "AdapterB", "AdapterC", "AdapterD", "provider knowledge stops here", "LLMProvider (Protocol)", "generate, embed", "Completion", "your application", "knows only Message, Completion, Usage".
Aspect ratio: 16:9
Accessibility: Convey the boundary with a labelled dashed line and text rather than colour, and label every arrow.
Avoid: Real vendor names or logos, screenshots, tiny text, watermarks, clutter.
Alt text: Diagram showing four provider SDKs with different request shapes each wrapped by an adapter, with a labelled boundary beyond which no provider knowledge passes, all converging into one LLMProvider protocol used by the application.
END IMAGE PROMPT]

Factory and registry patterns

With adapters written, something must choose one. Hardcoding the choice defeats the purpose.

from collections.abc import Callable


class ProviderRegistry:
    """Maps a provider name to a factory that builds it."""

    def __init__(self) -> None:
        self._factories: dict[str, Callable[[Settings], LLMProvider]] = {}

    def register(self, name: str, factory: Callable[[Settings], LLMProvider]) -> None:
        if name in self._factories:
            raise ValueError(f"provider {name!r} is already registered")
        self._factories[name] = factory

    def create(self, name: str, settings: Settings) -> LLMProvider:
        factory = self._factories.get(name)
        if factory is None:
            available = ", ".join(sorted(self._factories))
            raise ValueError(f"unknown provider {name!r}. Available: {available}")
        return factory(settings)

    def names(self) -> list[str]:
        return sorted(self._factories)


registry = ProviderRegistry()
registry.register("fast", lambda s: FastProviderAdapter(s.fast_api_key, build_transport(s)))
registry.register("cheap", lambda s: CheapProviderAdapter(s.cheap_api_key, build_transport(s)))
registry.register("local", lambda s: LocalProviderAdapter(s.local_url))

This is the same explicit-dictionary pattern as the tool registry from Module 6, and for the same reason: a name from configuration should reach only things you deliberately registered.

The application then selects by name:

provider = registry.create(settings.provider_name, settings)

Adding a fifth provider means writing one adapter file and one registration line. No existing file changes, which is the concrete test of whether the abstraction works.

A factory is the callable that builds one thing. The lambdas above are factories, and they exist so the registry stores a recipe rather than an instance. That matters because construction requires settings that may not be available when the registration runs, and because building all four providers when you need one wastes connections and may fail on a missing credential for a provider you were never going to use.

Dependency injection without a framework

Dependency injection means an object receives what it needs rather than constructing it. The name is heavier than the idea.

# Not injected: the class builds its own dependency
class ExtractionService:
    def __init__(self) -> None:
        self._provider = FastProviderAdapter(os.environ["FAST_KEY"], HTTPTransport(...))


# Injected: the dependency is given
class ExtractionService:
    def __init__(self, provider: LLMProvider, repository: DocumentRepository) -> None:
        self._provider = provider
        self._repository = repository

The second version can be tested with a fake provider and an in-memory repository, needs no environment variables to construct, and does not change when the provider changes. The first can do none of that.

Python needs no framework for this. Construct dependencies at the outermost layer and pass them inward:

def build_application(settings: Settings) -> ExtractionService:
    """Compose the object graph once, at the edge."""
    transport = HTTPTransport(
        timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0),
        breaker=CircuitBreaker(),
    )
    provider = registry.create(settings.provider_name, settings)
    repository = build_repository(settings)
    return ExtractionService(provider=provider, repository=repository)

This function is the composition root: the one place that knows how everything fits together. Everywhere else receives what it needs. When something is hard to test, the usual cause is that it constructs a dependency instead of accepting one, and moving that construction up to the composition root fixes it.

Two practical rules. Inject interfaces rather than concrete classes, so parameters are annotated LLMProvider rather than FastProviderAdapter. And do not inject the settings object everywhere: pass the specific values a class needs, so that its dependencies are visible in its signature rather than hidden behind a settings lookup.

Designing for capabilities that differ

Real providers are not interchangeable. One supports tool calling and another does not. One offers a strict structured output mode. One supports a two million token context and another eight thousand. Pretending otherwise produces an interface that lies.

Three approaches, in increasing order of honesty.

Lowest common denominator. Support only what every provider does. Simple, and it wastes the capabilities you are paying for.

Optional methods with a capability check. Ask before using.

@dataclass(frozen=True)
class Capabilities:
    max_context_tokens: int
    supports_tools: bool
    supports_structured_output: bool
    supports_streaming: bool


class LLMProvider(Protocol):
    name: str
    capabilities: Capabilities

    def generate(self, messages: list[Message], *, model: str, max_tokens: int) -> Completion: ...

if provider.capabilities.supports_structured_output:
    result = extract_with_native_schema(provider, document)
else:
    result = extract_with_prompted_schema(provider, document)

The branch is explicit and the fallback path is deliberate rather than accidental.

Degrade inside the adapter. The adapter emulates the missing capability, so callers do not branch at all. An adapter for a provider without a structured output mode can add schema instructions to the prompt and run the repair loop from Module 6, presenting a uniform interface.

This is the most convenient and the most dangerous, because it hides a real difference in reliability and cost behind an identical signature. If you do it, say so in the capabilities object anyway, so that code which cares can still find out.

One thing to check when adding a provider. Verify the whole path, not just that generate returns text. Token counting differs, so your budget checks may be wrong. Error responses differ, so your retryable classification from Module 5 needs its own translation per adapter. Rate limits differ. And output quality differs, which no interface can normalise, which is why the evaluation work in a later module is what actually tells you whether the swap was acceptable.