AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build a streaming token accumulator

Build a Streaming Token Accumulator

LLM providers stream responses as a sequence of small delta chunks. Each chunk contains a fragment of the final response — sometimes a single character, sometimes a few words. Your application needs to both forward these deltas in real time and accumulate the complete final response for logging, caching, and downstream processing.

What to build

Implement a StreamAccumulator class that:

Accepts token deltas via push(delta: str) — each call represents one streamed chunk.
Tracks the accumulated text — text property returns the full response so far.
Counts tokens — token_count property returns the number of whitespace-separated tokens accumulated (a simple approximation).
Signals completion — finish(finish_reason: str) marks the stream as done and stores the reason.
Provides a final result — result() returns a StreamResult dataclass with: text: str, token_count: int, finish_reason: str, and raises RuntimeError if called before finish().

Also implement stream_to_accumulator(chunks: Iterator[dict]) -> StreamResult — a function that drives a StreamAccumulator from an iterator of {"delta": str, "done": bool, "finish_reason": str | None} dicts.

Why this matters

Every LLM streaming integration needs this pattern. Forwarding tokens to a WebSocket, a file, or a queue is the easy part. Knowing when the stream ended and capturing the complete response for logging and caching requires explicit accumulation. Without it, you end up re-implementing this logic in every streaming callsite.

Constraints

push() after finish() should raise RuntimeError.
result() before finish() should raise RuntimeError.
No external dependencies — standard library only.

Python Refresh / medium / Step 10 of 16

Practice stage

Core runtime habits

Hint

Normalize the boundary first, keep return shapes stable, and make the middle of the function boring on purpose.

Next drill

Implement token counting and cost tracking

Success criteria

- Returns one stable shape across branches
- Makes failure handling obvious
- Keeps the core logic readable in under a minute

Review checklist

- Did I make the input and output shapes explicit?
- Would a teammate know where validation happens?
- Did I avoid silent mutation or vague branching?

from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator

@dataclass
class StreamResult:
    text: str
    token_count: int
    finish_reason: str

class StreamAccumulator:
    def __init__(self) -> None:
        # TODO: initialize internal state
        raise NotImplementedError

def push(self, delta: str) -> None:
        # TODO: append delta, raise if already finished
        raise NotImplementedError

@property
    def text(self) -> str:
        # TODO: return accumulated text so far
        raise NotImplementedError

@property
    def token_count(self) -> int:
        # TODO: return whitespace-token count of accumulated text
        raise NotImplementedError

def finish(self, finish_reason: str) -> None:
        # TODO: mark stream as done, store finish_reason
        raise NotImplementedError

def result(self) -> StreamResult:
        # TODO: return StreamResult, raise if not finished
        raise NotImplementedError

def stream_to_accumulator(chunks: Iterator[dict]) -> StreamResult:
    # TODO: drive StreamAccumulator from chunk iterator
    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.