C.W.K.
Stream
Lesson 01 of 05 · published

Common AI Adapter Interface

~20 min · adapters, abstraction

Level 0Downloader
0 XP0/41 lessons0/11 achievements
0/120 XP to next level120 XP to go0% complete

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.

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."""
        ...

External links

Exercise

Define StreamChunk and AIAdapter in your own project. Resist adding more methods than the three above. List five things you might be tempted to add (token counters, conversation memory, retry logic, rate limiters, model selection) and write one sentence on why each belongs *outside* the adapter.

Progress

Progress is local-only — sign in to sync across devices.
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign inPlease sign in to comment.

No comments yet — be the first.