Streaming Responses in the Common Model Interface

Hello! Your model gateway can already send the same prompt to Groq, Mistral, or Ollama and return a complete response. In the previous lesson, you used it with a reusable document-question-answering prompt. Now you’ll keep that request unchanged but receive the answer as it arrives, rather than waiting for generation to finish.

This is the last piece of the multi-provider generation core. You’ll add a stream() method to each adapter, keep one provider-independent event format, and run a live smoke test. Plan on about 40 minutes, mostly coding and inspection.

---

Decide what a stream promises

A streamed completion is not a list of finished answers. Providers send fragments that might be a word, part of a word, whitespace, or no text at all. The application must join text fragments in order and distinguish a fragment you can display from a response that has finished.

Give your common interface two event types:

• TextDelta(text) carries newly arrived text. Display its text without adding spaces or newlines.
• StreamDone(...) carries the assembled answer and available metadata. Do not display its answer a second time.

Keep generate() as it is; stream() is an additional method. Both accept the same ChatMessage list from build_messages().

Start a new streaming.py with the shared contract and small normalization helpers:

# streaming.py
from __future__ import annotations

from collections.abc import Iterator
from dataclasses import dataclass
from time import perf_counter
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from model_gateway import ChatMessage


@dataclass(frozen=True)
class TextDelta:
    text: str


@dataclass(frozen=True)
class StreamDone:
    provider: str
    model: str
    text: str                 # All emitted deltas, joined in order
    elapsed_s: float
    usage: dict[str, int] | None = None
    finish_reason: str | None = None


StreamEvent = TextDelta | StreamDone


def _field(value: Any, name: str) -> Any:
    if isinstance(value, dict):
        return value.get(name)
    return getattr(value, name, None)


def _text(value: Any) -> str | None:
    if value is None or value == "":
        return None
    if not isinstance(value, str):
        raise TypeError("Expected a text delta from the model provider")
    return value


def _token_usage(value: Any) -> dict[str, int] | None:
    counts = {
        "prompt_tokens": _field(value, "prompt_tokens"),
        "completion_tokens": _field(value, "completion_tokens"),
    }
    available = {key: count for key, count in counts.items()
                 if isinstance(count, int)}
    return available or None


def _done(
    provider: str,
    model: str,
    pieces: list[str],
    started: float,
    usage: dict[str, int] | None,
    finish_reason: str | None,
) -> StreamDone:
    return StreamDone(
        provider=provider,
        model=model,
        text="".join(pieces),
        elapsed_s=perf_counter() - started,
        usage=usage,
        finish_reason=finish_reason,
    )

The usage field is optional deliberately: streamed APIs do not always send token counts in the same way as non-streamed calls. An absent count is better than an invented one. Likewise, a finish_reason of length is useful information—it can mean your two-line answer was cut off by the output limit.

---

Implement the hosted-provider streams

Groq’s documentation shows the essential change: set stream=True, then read chunk.choices[0].delta.content rather than a completed message.content. Notice that some chunks can carry metadata instead of text.

Mistral exposes a stream of events rather than Groq’s chunk objects. Its SDK example uses mistral.chat.stream(...) and iterates over the resulting event stream. The page’s code rendering can be difficult to read; use it to identify the streaming method and event pattern, then use the implementation below for the gateway.

Add these functions to streaming.py. They use the same messages and 96-token demonstration limit as the earlier gateway examples. If your existing generate() methods get model IDs, token limits, or other request options from configuration, reuse those values here rather than maintaining a second set of defaults.

def stream_groq(
    client: Any,
    model: str,
    messages: list[ChatMessage],
) -> Iterator[StreamEvent]:
    started = perf_counter()  # Runs when iteration actually begins
    pieces: list[str] = []
    usage = None
    finish_reason = None

    raw_stream = client.chat.completions.create(
        model=model,
        messages=messages,
        max_completion_tokens=96,
        stream=True,
    )
    try:
        for chunk in raw_stream:
            usage = _token_usage(_field(chunk, "usage")) or usage

            for choice in chunk.choices or []:
                finish_reason = choice.finish_reason or finish_reason
                part = _text(_field(choice.delta, "content"))
                if part is not None:
                    pieces.append(part)
                    yield TextDelta(part)
    finally:
        close = getattr(raw_stream, "close", None)
        if callable(close):
            close()

    if finish_reason is None:
        raise RuntimeError("Groq stream ended without a finish reason")
    yield _done("groq", model, pieces, started, usage, finish_reason)


def stream_mistral(
    client: Any,
    model: str,
    messages: list[ChatMessage],
) -> Iterator[StreamEvent]:
    started = perf_counter()
    pieces: list[str] = []
    usage = None
    finish_reason = None

    with client.chat.stream(
        model=model,
        messages=messages,
        max_tokens=96,
    ) as events:
        for event in events:
            data = event.data
            usage = _token_usage(_field(data, "usage")) or usage

            for choice in _field(data, "choices") or []:
                finish_reason = _field(choice, "finish_reason") or finish_reason
                part = _text(_field(_field(choice, "delta"), "content"))
                if part is not None:
                    pieces.append(part)
                    yield TextDelta(part)

    yield _done("mistral", model, pieces, started, usage, finish_reason)

The two functions expose the same events despite different SDK objects. Neither converts an absent text fragment to the literal string "None". If an iteration raises an exception, it emits no StreamDone: any text displayed so far is a partial result, not a validated answer. Mistral’s finish_reason remains None if its event stream does not provide one; that is different from claiming the model stopped normally.

There is also a Python detail worth noticing: these are generator functions. Calling model.stream(messages) creates an iterator; the request begins when the caller iterates over it. That is why started = perf_counter() is inside each generator.

---

Add Ollama without changing the caller

Ollama also delivers incremental results. This video demonstrates its local /api/chat endpoint and shows why a streaming HTTP response must be processed piece by piece rather than treated as one JSON response. The implementation here uses the Ollama Python client if that is what your current adapter already uses; the video’s direct HTTP approach illustrates the underlying behavior.

For the Ollama Python client, chat(..., stream=True) provides successive responses whose message.content values hold text fragments. A final response indicates done and may include token counts under Ollama-specific field names. Add this third function:

def stream_ollama(
    client: Any,
    model: str,
    messages: list[ChatMessage],
) -> Iterator[StreamEvent]:
    started = perf_counter()
    pieces: list[str] = []
    usage = None
    finish_reason = None
    saw_done = False

    raw_stream = client.chat(
        model=model,
        messages=messages,
        stream=True,
        options={"num_predict": 96},
    )
    try:
        for chunk in raw_stream:
            part = _text(_field(_field(chunk, "message"), "content"))
            if part is not None:
                pieces.append(part)
                yield TextDelta(part)

            if _field(chunk, "done"):
                saw_done = True
                finish_reason = _field(chunk, "done_reason")
                counts = {
                    "prompt_tokens": _field(chunk, "prompt_eval_count"),
                    "completion_tokens": _field(chunk, "eval_count"),
                }
                available = {key: count for key, count in counts.items()
                             if isinstance(count, int)}
                usage = available or None
    finally:
        close = getattr(raw_stream, "close", None)
        if callable(close):
            close()

    if not saw_done:
        raise RuntimeError("Ollama stream ended without a done response")
    yield _done("ollama", model, pieces, started, usage, finish_reason)

Here the Ollama-specific prompt_eval_count and eval_count become the same prompt_tokens and completion_tokens keys used by the other adapters. This is the purpose of the gateway: its caller should not need to know which provider named the fields.

---

Wire the interface and watch it run

In model_gateway.py, import StreamEvent and the three streaming functions. Add stream(messages) -> Iterator[StreamEvent] to your existing common model protocol or abstract interface, without changing generate(). Each concrete adapter then delegates to its matching function. For example, assuming your adapter stores its SDK client as self.client and selected model ID as self.model:

# Add imports in model_gateway.py
from collections.abc import Iterator

from streaming import (
    StreamEvent,
    stream_groq,
    stream_mistral,
    stream_ollama,
)

# Add the appropriate method inside each existing adapter class.

# Groq adapter:
def stream(self, messages: list[ChatMessage]) -> Iterator[StreamEvent]:
    return stream_groq(self.client, self.model, messages)

# Mistral adapter:
def stream(self, messages: list[ChatMessage]) -> Iterator[StreamEvent]:
    return stream_mistral(self.client, self.model, messages)

# Ollama adapter:
def stream(self, messages: list[ChatMessage]) -> Iterator[StreamEvent]:
    return stream_ollama(self.client, self.model, messages)

Place each method inside its corresponding class, and substitute your actual client and model attribute names. If your Ollama adapter currently makes HTTP requests directly rather than using the Python client, keep that approach: iterate its streaming response, decode each JSON line, and map message.content, done, and the final counts into the same two event types. The interface contract does not depend on which Ollama transport you use.

Now create stream_smoke.py. It reuses the previous lesson’s prompt builder, so the only change at the call site is iterating over model.stream(messages).

# stream_smoke.py
from time import perf_counter

from config import load_settings
from model_gateway import make_model
from prompts import Evidence, build_messages
from streaming import StreamDone, TextDelta


def main() -> None:
    model = make_model(load_settings())
    messages = build_messages(
        "When does the sample library open on weekdays?",
        [Evidence(
            source_id="demo:p1",
            text="The sample library opens at 9 a.m. on weekdays.",
        )],
    )

    displayed: list[str] = []
    first_text_s = None
    started = perf_counter()

    try:
        for event in model.stream(messages):
            if isinstance(event, TextDelta):
                if first_text_s is None:
                    first_text_s = perf_counter() - started
                displayed.append(event.text)
                print(event.text, end="", flush=True)
            elif isinstance(event, StreamDone):
                assert "".join(displayed) == event.text
                print(
                    f"\nprovider={event.provider} model={event.model}"
                    f" first_text_s={first_text_s}"
                    f" total_s={event.elapsed_s:.2f}"
                    f" usage={event.usage}"
                    f" finish_reason={event.finish_reason}"
                )
    except Exception as exc:
        print(f"\nStream failed; displayed text is incomplete: {type(exc).__name__}")
        raise


if __name__ == "__main__":
    main()

Run it from the project root:

uv run --locked python stream_smoke.py

Try each configured provider by changing APP_PROVIDER, without editing the smoke test or prompt builder. Check three things: text appears progressively; the joined deltas equal StreamDone.text; and the answer follows the prompt’s evidence and citation instructions. Compare first_text_s with total time, but remember that a very fast response may look instantaneous even when it truly streamed. If a provider rejects an option or returns a different SDK object shape, inspect that provider’s actual event and adjust only its adapter.

Do not treat progressively displayed text as a final, grounded answer. The model could still finish with a missing citation, an unsupported claim, or a truncated Answer: line. Later application code will validate the completed answer before presenting citations as verified sources. Streaming changes when text becomes visible, not the standard of evidence it must meet.

---

Takeaways

Your gateway now supports both complete and streamed generation with the same prompt and provider selection. Provider-specific chunks become TextDelta events; a normally completed stream yields one StreamDone containing the assembled text, timing, and any available metadata. Errors leave partial text explicitly incomplete rather than disguising it as a successful answer.

Next, you’ll begin the raw-Python retrieval pipeline by extracting text and source metadata from PDF and plain-text documents—the material your prompt builder will eventually receive as evidence.