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