rohitg00/ai-engineering-from-scratch · error · ValueError

message batches do not provide progressive per-token output

Error message

message batches do not provide progressive per-token output

What it means

Contradictory-requirements guard in choose_access_pattern: the workload's facts select the Message Batches delivery while progressive per-token output is also demanded. Batch results arrive only when the entire batch completes, so promising streaming output is impossible and the chooser refuses to return a misleading pattern.

Solutions

1. Drop progressive_output if the latency tolerance allows batching
2. Keep progressive_output but relax the batch conditions (fewer independent requests or cannot wait) to get streaming
3. Split the workload: batch the background requests and stream the interactive one

When it happens

Trigger: Thrown at certifications/claude/lessons/08-messages-api-and-application-lifecycle/code/main.py:45 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of rohitg00/ai-engineering-from-scratch@39ea8a1c6d (2026-08-26). Data as JSON: /api/errors/f7c21ed0b8a0695b. Report an issue: GitHub.

Appendix: source

Thrown at certifications/claude/lessons/08-messages-api-and-application-lifecycle/code/main.py:45

@dataclass(frozen=True)
class AccessNeeds:
    """Workload facts used to choose a client and completion pattern."""

    supported_sdk: bool = True
    custom_transport: bool = False
    progressive_output: bool = False
    independent_requests: int = 1
    can_wait: bool = False


def choose_access_pattern(needs: AccessNeeds) -> dict[str, str]:
    """Choose SDK versus REST separately from sync, stream, or batch."""
    if needs.independent_requests < 1:
        raise ValueError("independent_requests must be positive")
    if needs.progressive_output and needs.can_wait and needs.independent_requests > 1:
        raise ValueError("message batches do not provide progressive per-token output")

    client = "raw-rest" if needs.custom_transport or not needs.supported_sdk else "sdk"
    if needs.can_wait and needs.independent_requests > 1:
        delivery = "message-batch"
    elif needs.progressive_output:
        delivery = "streaming"
    else:
        delivery = "synchronous"
    return {"client": client, "delivery": delivery}


class ScriptedTransport:
    """A stateless transport that returns scripted API responses."""

    def __init__(self, responses: list[dict[str, Any]]) -> None:
        self._responses = copy.deepcopy(responses)
        self.requests: list[list[dict[str, Any]]] = []

View on GitHub (pinned to 39ea8a1c6d)