AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Stream LLM responses with progress callbacks

Stream LLM Responses with Progress Callbacks

Streaming is the single highest-impact UX improvement you can make to an LLM feature. Without streaming, a user waits 5–10 seconds in silence before seeing any output. With streaming, text starts appearing within 300ms. The engineering is not hard, but handling streaming correctly — accumulating partial results, handling errors mid-stream, reporting progress — requires care.

What you are building

Build a stream_response function that:

Accepts an Anthropic client, messages, model, and an optional on_chunk callback
Streams the response using client.messages.stream()
Calls on_chunk(text_delta: str, accumulated: str) after each text delta
Handles StopReason variants: end_turn (normal), max_tokens (response was cut off), stop_sequence
Returns a StreamResult with: the complete accumulated text, stop reason, total input/output tokens, and whether an error occurred mid-stream
If a StreamError occurs mid-stream, calls on_chunk("[STREAM_ERROR]", accumulated_so_far) and returns a partial result with error=True

Also build a StreamBuffer class that implements on_chunk as a method and stores chunks for consumers that cannot process them live (e.g., testing, batch processing).

Why streaming matters in production

Streaming changes the user experience from "waiting" to "reading." Users tolerate 10+ seconds of generation if text is appearing. The same generation with no streaming feels broken at 5 seconds. Additionally, streaming lets you implement early-stop behavior: if the user navigates away or presses escape, you cancel the stream rather than completing a full (expensive) generation.

Llm Foundations / medium / Step 6 of 6

Practice stage

General drill

Hint

Keep the solution explicit and reviewable.

Next drill

This is the end of the current mini-sequence.

Success criteria

Make the solution explicit, debuggable, and easy to explain.

Review checklist

Review where the boundary is, what gets validated, and what would be hard to debug later.

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Callable, Optional
import anthropic

@dataclass
class StreamResult:
    text: str
    stop_reason: str  # "end_turn", "max_tokens", "stop_sequence", "error"
    input_tokens: int
    output_tokens: int
    error: bool = False
    error_message: Optional[str] = None

class StreamBuffer:
    """Accumulates streaming chunks for batch or test consumers."""

def __init__(self) -> None:
        # TODO: store chunks list and full accumulated text
        raise NotImplementedError

def on_chunk(self, text_delta: str, accumulated: str) -> None:
        """Called by stream_response for each chunk. Stores the delta."""
        # TODO: append text_delta to chunks, update accumulated
        raise NotImplementedError

@property
    def full_text(self) -> str:
        """Return the full accumulated text."""
        # TODO: return joined chunks or cached accumulated
        raise NotImplementedError

@property
    def chunks(self) -> list[str]:
        """Return list of individual text deltas received."""
        raise NotImplementedError

def stream_response(
    client: anthropic.Anthropic,
    messages: list[dict],
    model: str = "claude-haiku-4-5",
    max_tokens: int = 1024,
    on_chunk: Optional[Callable[[str, str], None]] = None,
) -> StreamResult:
    """
    Stream an LLM response with progress callbacks.

Args:
        client: Anthropic client instance
        messages: Messages array for the API call
        model: Model to use
        max_tokens: Maximum tokens to generate
        on_chunk: Optional callback(text_delta, accumulated_text) called per chunk

Returns:
        StreamResult with accumulated text, stop reason, token counts, and error state
    """
    # TODO: use client.messages.stream() context manager
    # TODO: iterate over text_stream, accumulate, call on_chunk
    # TODO: get final message for token counts and stop reason
    # TODO: handle anthropic.APIError mid-stream
    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.