AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build a streaming response accumulator for LLM output

Build a Streaming Response Accumulator for LLM Output

Most production LLM APIs support streaming: instead of waiting for the full response, you receive tokens as they are generated. Streaming dramatically reduces time-to-first-token and lets you forward partial output to users in real time. But streaming also means your application needs to handle partial data, detect stream completion, and produce a coherent final result.

What to build

Implement a StreamAccumulator class that processes a stream of token chunks:

__init__() — initialize internal state.
feed(chunk: str | None) -> None — accept one chunk. None is the sentinel that signals end-of-stream.
is_complete() -> bool — return True after a None sentinel has been received.
full_text() -> str — return the concatenation of all received chunks (excluding the sentinel). Raise RuntimeError if the stream is not yet complete.
chunk_count() -> int — return the number of non-sentinel chunks received.
extract_json() -> dict | None — after the stream is complete, try to parse the full text as JSON. Return the parsed dict on success, None on failure. This is useful for tool call responses that stream as JSON.

Also implement simulate_stream(text: str, chunk_size: int = 5) -> list[str | None] — a helper that splits a text into chunks of chunk_size characters and appends a None sentinel. This simulates a provider stream for testing.

Why this matters

In a FastAPI streaming endpoint, you forward chunks to a client as they arrive. In a background job, you accumulate the full text and then validate it. The StreamAccumulator makes both use cases clean without duplicating the concatenation logic.

Constraints

full_text() must raise RuntimeError if called before the sentinel arrives.
feed(None) called multiple times should be idempotent (still just marks as complete).
Do not assume chunks are single tokens — they can be multi-character strings.

Data Transformation / medium / Step 13 of 16

Practice stage

Transform and summarize data safely

Hint

Think like an evaluation or trace pipeline: group clearly, preserve the data story, and make edge cases visible instead of clever.

Next drill

Compute per-model cost and usage summaries from request logs

Success criteria

- Handles missing or noisy records predictably
- Produces summary output that is easy to inspect
- Would be safe to rerun in a script workflow

Review checklist

- Do grouped metrics stay readable?
- Would malformed rows be debuggable?
- Did I choose names that explain the transformation?

from __future__ import annotations
import json

class StreamAccumulator:
    def __init__(self):
        # TODO: initialize state
        raise NotImplementedError

def feed(self, chunk: str | None) -> None:
        # TODO: accumulate chunk or mark complete on None
        raise NotImplementedError

def is_complete(self) -> bool:
        # TODO: return True after None sentinel
        raise NotImplementedError

def full_text(self) -> str:
        # TODO: return concatenated text; raise RuntimeError if not complete
        raise NotImplementedError

def chunk_count(self) -> int:
        # TODO: return number of non-sentinel chunks
        raise NotImplementedError

def extract_json(self) -> dict | None:
        # TODO: parse full_text() as JSON, return None on failure
        raise NotImplementedError

def simulate_stream(text: str, chunk_size: int = 5) -> list[str | None]:
    # TODO: split text into chunk_size pieces and append None sentinel
    raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.