Common AI Adapter Interface

Why an adapter

You want to swap local and cloud models without rewriting the app. The pattern: define a narrow interface every provider implements, and one universal chunk type every provider emits. Routes, store, and frontend talk to the interface; provider-specific code lives behind it.

The minimum interface

stream(messages, tools, **opts) — the only method that matters. Yields a sequence of universal chunks.
health_check() — fast yes/no for fallback routing.
list_models() — for UI and discovery.

The universal chunk

One StreamChunk shape covers everything you need to yield: text deltas, thinking deltas, tool calls, the final done flag with timing. Each adapter translates its provider's wire shape into this universal chunk.

Don't over-abstract

The adapter is narrow on purpose. Routes know about stream(), not about NDJSON vs SSE. Adapters know about NDJSON vs SSE, not about FastAPI routes. One thing at the boundary, and that thing is the chunk yield protocol.

And this is polymorphism actually doing a day's work. One interface, many vessels behind it, and the caller never learns which one answered. It's the thing OOP textbooks explain with shape classes — except here Claude and Ollama stand behind the same stream() and prove it. You don't get it from memorizing an inheritance diagram; you feel it the night the cloud goes down and the answers keep coming.

Code

The narrow interface·python

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import AsyncIterator, Any

@dataclass
class StreamChunk:
    """Universal chunk — every adapter yields this shape."""
    content: str = ""
    thinking: str = ""
    tool_calls: list = field(default_factory=list)
    done: bool = False
    tokens_per_sec: float = 0.0
    model: str = ""
    raw: Any = None  # provider-specific payload, debug-only

class AIAdapter(ABC):
    """Narrow contract every AI vessel implements."""

    @abstractmethod
    async def stream(
        self,
        messages: list[dict],
        tools: list[dict] | None = None,
        **opts: Any,
    ) -> AsyncIterator[StreamChunk]:
        """Stream a response as universal chunks."""
        ...

    @abstractmethod
    async def health_check(self) -> bool:
        """Fast yes/no — is the provider alive right now?"""
        ...

    @abstractmethod
    async def list_models(self) -> list[str]:
        """Available model names — for UI and discovery."""
        ...

Why an adapter

The minimum interface

The universal chunk

Don't over-abstract

Code

External links

Exercise

Progress

Comments 0