Lesson 3: Consuming a Stream

Time-to-first-token versus total latency

A generation call producing 400 tokens takes about fourteen seconds. Without streaming, the user sees a spinner for fourteen seconds and then everything at once. With streaming, the first words appear in about half a second and the rest arrives as it is produced.

The total time is unchanged. The experience is completely different, and this is the point worth internalising: perceived speed is what users experience, and time-to-first-token is what determines it.

Two metrics, measured separately.

Time to first token is from request start to the first content arriving. It is dominated by the provider's prefill phase, which processes your input, so it grows with input length and is largely independent of output length.

Total latency is the whole response. It is dominated by generation, so it grows with output length.

The practical consequences follow directly. Shortening the prompt improves time-to-first-token. Requesting a shorter output improves total latency. And a long prompt with a short output has poor perceived speed even though the total time is small, which is a case people find surprising until they separate the two numbers.

[IMAGE PROMPT M8-4
Purpose: Show why streaming changes perceived speed without changing total time, and which phase each metric measures.
Visual type: Two-panel timeline comparison with annotated metrics.
Prompt: A clean educational comparison with two stacked panels sharing a horizontal time axis marked at 0s, 0.5s, and 14s. The upper panel is headed "Without streaming" and shows one long empty bar spanning 0s to 14s labelled "waiting, nothing visible", ending in a single block at 14s labelled "full response appears". Beneath it a bracket spans the whole width labelled "user sees nothing for 14 seconds". The lower panel is headed "With streaming" and shows a short segment from 0s to 0.5s labelled "prefill: processing input", followed by many small consecutive blocks from 0.5s to 14s labelled "tokens arrive continuously". Two brackets sit beneath: a short one from 0s to 0.5s labelled "time to first token", and a long one from 0s to 14s labelled "total latency". A caption beneath both panels reads "same total time, different experience".
Required elements: Shared time axis with the three marks, one long empty bar above, prefill plus streaming blocks below, three labelled brackets, the caption.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector.
Layout: Two panels stacked vertically aligned to a shared axis, brackets beneath each panel, caption at the bottom.
Text labels: "Without streaming", "With streaming", "0s", "0.5s", "14s", "waiting, nothing visible", "full response appears", "user sees nothing for 14 seconds", "prefill: processing input", "tokens arrive continuously", "time to first token", "total latency", "same total time, different experience".
Aspect ratio: 16:9
Accessibility: Distinguish the phases through block segmentation and bracket labels rather than colour alone.
Avoid: Screenshots, spinner or loading imagery, decorative icons, tiny text, logos, watermarks.
Alt text: Two timelines showing a non-streamed response where nothing is visible for fourteen seconds, against a streamed response where tokens begin arriving at half a second, with brackets marking time to first token and total latency.
END IMAGE PROMPT]

Server-Sent Events and chunked transfer encoding

Two mechanisms carry a stream over HTTP, and knowing which you are dealing with prevents a class of parsing confusion.

Chunked transfer encoding is an HTTP feature allowing a response body to be sent in pieces without declaring its total length in advance. It is the transport, and it carries anything.

Server-Sent Events is a text format layered on top, with a defined structure.

data: {"type":"content","text":"Spa"}

data: {"type":"content","text":"ghetti"}

data: [DONE]

The rules that matter. Each line has a field name, a colon, and a value, with data being the common one. A blank line terminates an event. Fields such as event, id, and retry exist and are used by browser clients for typing and reconnection. A line beginning with a colon is a comment, which is how heartbeats are usually sent.

That blank line is load-bearing. An SSE parser that splits on single newlines rather than double newlines will fragment multi-line events, which is a bug that only appears when a payload happens to contain a newline.

Note also that [DONE] is a convention rather than part of the specification, and not every provider uses it. Rely on the connection closing or on an explicit finish reason rather than on a sentinel string you assume is there.

Async generators as the natural primitive

A stream is a sequence of values arriving over time, which is exactly what an async generator expresses.

from collections.abc import AsyncIterator


async def stream_completion(messages: list[Message]) -> AsyncIterator[str]:
    """Yield text fragments as they arrive from the provider."""
    async with client.stream("POST", GENERATE_URL, json=payload) as response:
        response.raise_for_status()
        async for line in response.aiter_lines():
            if not line.startswith("data: "):
                continue
            payload_text = line.removeprefix("data: ").strip()
            if payload_text == "[DONE]":
                return
            chunk = json.loads(payload_text)
            text = extract_delta_text(chunk)
            if text:
                yield text

The consumer looks like an ordinary loop:

async for fragment in stream_completion(messages):
    print(fragment, end="", flush=True)

This is the generator pipeline from Module 4 with async added, and the property that mattered there matters here: nothing is held in memory beyond the current fragment.

Note async with client.stream(...), which keeps the connection open for the duration and closes it on exit, including on cancellation. Using the non-streaming form would buffer the entire response before returning, defeating the purpose.

Parsing provider deltas

A stream is not a sequence of text fragments. It is a sequence of events of several kinds, and text is one of them.

@dataclass
class StreamEvent:
    kind: Literal["start", "content", "tool_call_delta", "finish", "usage"]
    text: str = ""
    finish_reason: str | None = None
    usage: Usage | None = None

Typical event kinds, whatever a given provider calls them: a start event carrying the role and model, content deltas carrying text fragments, tool call deltas carrying argument fragments, a finish event carrying the stop reason, and a usage event carrying token counts.

[VOLATILE: event names, nesting, and ordering differ substantially across providers. The normalisation below is illustrative.]

This is Module 7's adapter pattern applied to streams. Each provider adapter converts its own event shape into your StreamEvent, and everything downstream handles one vocabulary.

class FastProviderAdapter:
    async def stream(self, messages: list[Message], *, model: str) -> AsyncIterator[StreamEvent]:
        async with self._transport.stream(FAST_URL, self._payload(messages, model)) as raw:
            async for chunk in raw:
                match chunk:
                    case {"type": "text_delta", "text": str(text)}:
                        yield StreamEvent(kind="content", text=text)
                    case {"type": "stopped", "because": str(reason)}:
                        yield StreamEvent(
                            kind="finish", finish_reason=self._normalize_finish(reason)
                        )
                    case {"type": "meta", "prompt_tokens": int(p), "completion_tokens": int(c)}:
                        yield StreamEvent(
                            kind="usage", usage=Usage(input_tokens=p, output_tokens=c)
                        )

The match statement from Module 2 is doing exactly what it is good at: dispatching on payload shape and extracting in one step.

Handle unknown event types by ignoring them. Providers add event kinds over time, and a stream parser that raises on an unrecognised type breaks the day they ship a new one. Log at debug level and continue.

Accumulating while streaming

You still need the complete text at the end: to store, to validate, to log, and to return to a caller who did not want the stream.

@dataclass
class StreamResult:
    text: str
    finish_reason: str | None
    usage: Usage | None


async def stream_and_collect(
    events: AsyncIterator[StreamEvent],
    on_text: Callable[[str], Awaitable[None]],
) -> StreamResult:
    """Forward fragments to a consumer while accumulating the full response."""
    parts: list[str] = []
    finish_reason: str | None = None
    usage: Usage | None = None

    async for event in events:
        if event.kind == "content":
            parts.append(event.text)
            await on_text(event.text)
        elif event.kind == "finish":
            finish_reason = event.finish_reason
        elif event.kind == "usage":
            usage = event.usage

    return StreamResult("".join(parts), finish_reason, usage)

Accumulate into a list and join once at the end. Repeated string concatenation in a loop builds a new string on every fragment, which is quadratic in the number of fragments and genuinely noticeable across thousands of tokens.

Handle the incomplete case. If the stream ends without a finish event, the response was cut off. Return what you have with finish_reason=None and let the caller decide, which is the honest degradation principle from Module 5 rather than pretending the response was complete.

Inter-chunk versus total timeouts

Module 5 established that every network call needs a timeout, and that read timeouts usually apply between chunks. Streaming is where that distinction becomes critical.

A stalled stream is a connection that is open, has sent some data, and has stopped. Nothing is closed and no error occurs. Without an inter-chunk timeout, you wait forever.

async def with_chunk_timeout(
    events: AsyncIterator[StreamEvent],
    *,
    per_chunk: float = 30.0,
    total: float = 300.0,
) -> AsyncIterator[StreamEvent]:
    """Fail fast if the stream stalls, or if the whole stream runs too long."""
    iterator = events.__aiter__()
    deadline = time.monotonic() + total

    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise ProviderTimeoutError(f"stream exceeded {total}s total")

        try:
            async with asyncio.timeout(min(per_chunk, remaining)):
                event = await iterator.__anext__()
        except StopAsyncIteration:
            return
        except TimeoutError as exc:
            raise ProviderTimeoutError(f"no chunk received for {per_chunk}s") from exc

        yield event

Both bounds are needed for the reason Module 5 gave: a stream trickling one chunk every twenty-nine seconds never trips a thirty second inter-chunk timeout and runs indefinitely without the total.

Set the inter-chunk timeout generously enough to survive a pause during a hard reasoning step, which can be many seconds, and tightly enough to detect a genuine stall.