Adapter Pattern

One interface, many models

The adapter pattern hides provider-specific shape behind a single interface. Your application code calls adapter.generate_stream(messages); the adapter knows which API to hit, how to format the request, and how to normalize the response.

Why bother

Swap models without changing app code — flip Gemini to OpenAI to local Ollama by changing one line.
Build fallback chains — if Gemini errors, try OpenAI, then local. The chain is a list of adapters.
A/B test providers — route 10% of traffic to a new model behind the same interface.
Translate prompt formats — abstract over per-provider quirks (system prompt placement, tool-call shape, role names).

Where to draw the abstraction line

Narrow. Abstract only the streaming generate call and (maybe) the tool loop. Don't try to abstract every provider feature — file APIs, caching, embeddings — into a unified interface. The cost of an over-broad abstraction is that every new provider requires shoehorning. Keep the seam at generate_stream and let provider-specific features stay provider-specific.

Code

ABC + chunk type — the contract·python

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Optional

@dataclass
class StreamChunk:
    text: str = ''
    function_call: Optional[dict] = None
    finish_reason: Optional[str] = None
    usage: Optional[dict] = None

class ModelAdapter(ABC):
    @abstractmethod
    async def generate_stream(
        self,
        messages: list[dict],
        tools: Optional[list[dict]] = None,
        **kw,
    ) -> AsyncIterator[StreamChunk]:
        ...

    @abstractmethod
    def name(self) -> str:
        ...

    @abstractmethod
    def supports_tools(self) -> bool:
        ...

    @abstractmethod
    def supports_vision(self) -> bool:
        ...

Gemini implementation·python

from google import genai
from google.genai import types

class GeminiAdapter(ModelAdapter):
    def __init__(self, api_key: str, model: str = 'gemini-2.5-flash'):
        self.client = genai.Client(api_key=api_key)
        self.model = model

    def name(self):           return f'gemini:{self.model}'
    def supports_tools(self): return True
    def supports_vision(self): return True

    async def generate_stream(self, messages, tools=None, **kw):
        contents = self._convert_messages(messages)
        config = types.GenerateContentConfig()
        if tools:
            config = types.GenerateContentConfig(
                tools=[types.Tool(function_declarations=tools)],
            )

        async for chunk in await self.client.aio.models.generate_content_stream(
            model=self.model, contents=contents, config=config,
        ):
            sc = StreamChunk(text=chunk.text or '')
            if chunk.usage_metadata:
                sc.usage = {
                    'prompt': chunk.usage_metadata.prompt_token_count,
                    'completion': chunk.usage_metadata.candidates_token_count,
                }
            cand = chunk.candidates[0] if chunk.candidates else None
            if cand:
                sc.finish_reason = cand.finish_reason.name if cand.finish_reason else None
            yield sc

    def _convert_messages(self, messages):
        # OpenAI-style messages -> Gemini contents
        out = []
        for m in messages:
            role = 'model' if m['role'] == 'assistant' else m['role']
            out.append(types.Content(
                role=role,
                parts=[types.Part(text=m['content'])],
            ))
        return out

Use the adapter from app code·python

import os
adapter: ModelAdapter = GeminiAdapter(api_key=os.environ['GEMINI_API_KEY'])

async for chunk in adapter.generate_stream(
    messages=[
        {'role': 'system', 'content': 'You are helpful.'},  # adapter handles role conversion
        {'role': 'user',   'content': 'Hello!'},
    ],
):
    if chunk.text:
        print(chunk.text, end='', flush=True)
    if chunk.usage:
        print(f'\n[usage: {chunk.usage}]')

Exercise

Build OpenAIAdapter alongside GeminiAdapter using OpenAI's SDK (or, if you don't have OpenAI access, build EchoAdapter that just echoes the user message). Wire both behind the same ModelAdapter ABC. Write a test that calls generate_stream on both and confirms the same caller code works against both providers.