Configurable Unified Model Interface for Groq, Mistral, and Ollama

Hello! You now have three working smoke tests: Groq and Mistral through hosted APIs, and Ollama through a local service. Each takes chat messages, but each has its own client call and response shape. In this lesson, you’ll keep those direct SDK calls and put a small, application-owned interface around them.

The result will let the future document question-answering pipeline request an answer without knowing which provider was selected. Plan on about 40–50 minutes, mostly spent building and running the code. We’ll use completed responses for now; streaming comes later.

Decide what the rest of the application should see

Imagine the retrieval pipeline has found relevant passages and assembled a list of chat messages. It should be able to make one call such as model.generate(messages) and receive an answer with predictable fields. It should not need three branches to locate generated text or token counts.

The adapter pattern gives this boundary a useful name: an adapter wraps an existing service and presents the interface its client expects.

For our application, the contract will be deliberately small:

• Input: a list of messages, each with a role and text content.
• Output: generated text, selected provider and model, prompt and completion token counts when available, and elapsed SDK-call time.
• Selection: APP_PROVIDER in your existing configuration chooses the adapter.

A field that a provider does not report will be None, not an invented zero. Likewise, elapsed time will always mean time measured around our SDK call, rather than mixing that measurement with a provider-reported duration.

For a brief comparison with another approach, watch these clips from LiteLLM Crash Course | For Complete Beginners by Data Science Basics:

Python’s Protocol is a good fit for the calling side of this boundary: classes can satisfy it by implementing the required method, without inheriting from a shared base class. The Python documentation explains that distinction in “Nominal vs structural subtyping.”

Build the direct-SDK adapters

Work in your existing doc-qa project. The code below expects load_settings() from your earlier lessons to return these attributes:

provider, groq_api_key, groq_model, mistral_api_key, mistral_model, ollama_host, and ollama_model.

If your config.py uses different attribute names, change the factory at the bottom to match it. Keep credential validation conditional: choosing Ollama must not require hosted-provider keys.

Create model_gateway.py:

# model_gateway.py
from dataclasses import dataclass
from time import perf_counter
from typing import Literal, Protocol, TypedDict

from groq import Groq
from mistralai import Mistral
from ollama import Client as OllamaClient


class ChatMessage(TypedDict):
    role: Literal["system", "user", "assistant"]
    content: str


@dataclass(frozen=True)
class ModelResponse:
    text: str
    provider: str
    model: str
    prompt_tokens: int | None
    completion_tokens: int | None
    latency_seconds: float


class ChatModel(Protocol):
    def generate(self, messages: list[ChatMessage]) -> ModelResponse:
        ...


class ModelSettings(Protocol):
    provider: str
    groq_api_key: str | None
    groq_model: str
    mistral_api_key: str | None
    mistral_model: str
    ollama_host: str
    ollama_model: str


def require_text(value: object, provider: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{provider} returned no nonempty text answer")
    return value


class GroqAdapter:
    def __init__(self, *, api_key: str, model: str) -> None:
        self.client = Groq(api_key=api_key, timeout=60.0)
        self.model = model

    def generate(self, messages: list[ChatMessage]) -> ModelResponse:
        started = perf_counter()
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            max_tokens=96,
            stream=False,
        )
        elapsed = perf_counter() - started

        if not response.choices:
            raise ValueError("Groq returned no choices")

        usage = response.usage
        return ModelResponse(
            text=require_text(response.choices[0].message.content, "Groq"),
            provider="groq",
            model=self.model,
            prompt_tokens=getattr(usage, "prompt_tokens", None),
            completion_tokens=getattr(usage, "completion_tokens", None),
            latency_seconds=elapsed,
        )


class MistralAdapter:
    def __init__(self, *, api_key: str, model: str) -> None:
        self.client = Mistral(api_key=api_key)
        self.model = model

    def generate(self, messages: list[ChatMessage]) -> ModelResponse:
        started = perf_counter()
        response = self.client.chat.complete(
            model=self.model,
            messages=messages,
            max_tokens=96,
        )
        elapsed = perf_counter() - started

        if not response.choices:
            raise ValueError("Mistral returned no choices")

        usage = response.usage
        return ModelResponse(
            text=require_text(response.choices[0].message.content, "Mistral"),
            provider="mistral",
            model=self.model,
            prompt_tokens=getattr(usage, "prompt_tokens", None),
            completion_tokens=getattr(usage, "completion_tokens", None),
            latency_seconds=elapsed,
        )


class OllamaAdapter:
    def __init__(self, *, host: str, model: str) -> None:
        self.client = OllamaClient(host=host, timeout=120.0)
        self.model = model

    def generate(self, messages: list[ChatMessage]) -> ModelResponse:
        started = perf_counter()
        response = self.client.chat(
            model=self.model,
            messages=messages,
            options={"num_predict": 96},
            stream=False,
        )
        elapsed = perf_counter() - started

        return ModelResponse(
            text=require_text(response.message.content, "Ollama"),
            provider="ollama",
            model=self.model,
            prompt_tokens=response.prompt_eval_count,
            completion_tokens=response.eval_count,
            latency_seconds=elapsed,
        )


def make_model(settings: ModelSettings) -> ChatModel:
    provider = settings.provider.lower()

    if provider == "groq":
        if not settings.groq_api_key:
            raise ValueError("GROQ_API_KEY is required for Groq")
        return GroqAdapter(
            api_key=settings.groq_api_key,
            model=settings.groq_model,
        )

    if provider == "mistral":
        if not settings.mistral_api_key:
            raise ValueError("MISTRAL_API_KEY is required for Mistral")
        return MistralAdapter(
            api_key=settings.mistral_api_key,
            model=settings.mistral_model,
        )

    if provider == "ollama":
        return OllamaAdapter(
            host=settings.ollama_host,
            model=settings.ollama_model,
        )

    raise ValueError(
        f"Unsupported APP_PROVIDER: {settings.provider!r}; "
        "choose groq, mistral, or ollama"
    )

There are two different kinds of normalization here. All adapters accept the same messages and return ModelResponse. Inside each adapter, provider-specific code still extracts the answer and usage from the appropriate place. Notice particularly that Ollama reads response.message.content, while the hosted calls read the first choice’s message.

ChatModel describes what calling code needs; none of the adapter classes has to declare that it implements it. ModelSettings similarly documents what the factory expects from your existing settings object. These are type contracts, not runtime validators. The explicit key checks and require_text() handle important runtime failures.

The model field records the configured model requested by the application. latency_seconds measures the completed SDK call. The output-token limit is set with max_tokens for the hosted calls and num_predict for Ollama; those provider-specific details stay inside the adapters. Keep the full SDK response out of ModelResponse: later, prompts may contain private document text that you do not want casually printed or logged.

Verify that configuration really changes the provider

Create gateway_smoke.py. Unlike the earlier smoke scripts, this file must contain no provider-specific SDK code:

# gateway_smoke.py
from config import load_settings
from model_gateway import ChatMessage, make_model


def main() -> None:
    messages: list[ChatMessage] = [
        {
            "role": "system",
            "content": "You are a concise assistant. Answer in one sentence.",
        },
        {
            "role": "user",
            "content": (
                "Test note: The sample library opens at 9 a.m. on weekdays.\n"
                "Question: When does it open on weekdays?"
            ),
        },
    ]

    model = make_model(load_settings())
    answer = model.generate(messages)

    print(f"Provider: {answer.provider}")
    print(f"Requested model: {answer.model}")
    print(f"Answer: {answer.text}")
    print(f"Prompt tokens: {answer.prompt_tokens}")
    print(f"Generated tokens: {answer.completion_tokens}")
    print(f"Measured SDK call time (s): {answer.latency_seconds:.3f}")


if __name__ == "__main__":
    main()

Run it from the project root:

uv run --locked python gateway_smoke.py

Use your ignored local .env to select APP_PROVIDER=groq, then mistral, then ollama, running the same command each time. Keep each provider’s working model identifier and credentials from its earlier smoke test; for Ollama, keep its service running. You should get the weekday opening time on each run, though wording, usage counts, and latency can differ. Do not treat those differences as a controlled model-quality comparison.

If an individual smoke script works but this one does not, check the boundary between it and the factory first:

• Missing settings attribute: Align the factory’s attribute names with your config.py. Do not rewrite working SDK calls to compensate for a naming mismatch.
• Wrong provider selected: Check APP_PROVIDER and how load_settings() reads .env. An environment variable already set in your shell may take precedence over the file.
• Empty or unexpected response content: The adapter deliberately rejects anything other than nonempty text. Inspect the relevant provider smoke test rather than silently converting an unfamiliar response object to a string.
• Ollama connection or model error: Recheck the running service, OLLAMA_HOST, and the exact name shown by ollama list.

Once all three runs work, the important observation is in gateway_smoke.py: changing providers required no change to its messages, call, or answer-reading code. You have not removed provider differences; you have contained them in one place.

What this gives the project

You now have a configurable, direct-SDK model interface with a stable response shape. The adapters preserve provider-specific request and extraction logic, while future retrieval and UI code can depend on ChatModel and ModelResponse. Missing token counts remain explicitly unknown, and malformed text is rejected rather than passed downstream as an answer.

Next, you’ll build reusable system and user prompts for that same generate(messages) call, specifying the model’s role, how it should use evidence, and what its answer should look like.