Streaming results¶
lm15.result — Stream materialization.
One engine and two skins, all speaking the canonical StreamEvent vocabulary:
StreamAccumulator— the push-based engine. Feed it events, ask it for theResponse. It is the shape every port shares (Rust/Go/TS have no generators) and the only place accumulation logic lives.ResponseStream/AsyncResponseStream— lazy iteration sugar over a live stream: iterate for text as it arrives,.events()for the canonical typed events,.responseafterwards for the same Response a non-streaming call returns.materialize_response/amaterialize_response— the one-shot functional form.
Nothing here executes anything: tool calls are surfaced as data, and any execute-tools-until-done loop belongs to the layer above lm15.
StreamAccumulator
dataclass
¶
Accumulates canonical stream events into a complete Response.
Push-based so that sync iteration, async iteration, and one-shot materialization all share the same accumulation logic:
acc = StreamAccumulator(request)
for event in lm.stream(request):
acc.push(event)
response = acc.response()
push ignores error events — deciding whether to raise is the
caller's job (ResponseStream raises; a resumption layer might
not). response() materializes whatever has been accumulated so
far; callers normally push through the end event first.
push(event: StreamEvent) -> None
¶
Fold one canonical stream event into the accumulated state.
response() -> Response
¶
Build a complete Response from the accumulated state.
Raises :class:StreamAssemblyError when a tool call's fragments
never carried a name (MAP-9): the error carries everything else
that did assemble as partial.
ResponseStream
¶
Lazy stream-backed response assembler.
rs = ResponseStream(lm.stream(request), request)
for text in rs: # text as it arrives
print(text, end="")
rs.response # the same Response complete() returns
rs.events() yields the canonical StreamEvents instead — one
vocabulary for raw and assembled streaming. Accessors mirror
Response's own minimal set; everything richer is
rs.response.message.first(...) / .parts_of(...).
Tool calls are surfaced as data only; ResponseStream never executes anything.
AsyncResponseStream
¶
Async mirror of :class:ResponseStream, same accumulator engine.
rs = AsyncResponseStream(lm.stream(request), request)
async for text in rs:
print(text, end="")
response = await rs.response()
response() is a method (it may need to consume the stream, which
is an awaitable operation in async code).
materialize_response(events: Iterator[StreamEvent], request: Request) -> Response
¶
Consume a complete stream, requiring a final end event and closing its source.
amaterialize_response(events: AsyncIterator[StreamEvent], request: Request) -> Response
async
¶
Async mirror of :func:materialize_response.
response_to_events(response: Response) -> Iterator[StreamEvent]
¶
Convert a complete Response to stream events.
The conversion is intentionally lossless for the Delta vocabulary. If a response contains a valid Part that has no Delta representation, this function raises instead of silently dropping content.
coalesce_stream(events: Iterator[StreamEvent], *, model: str | None = None, adaptations: tuple = ()) -> Iterator[StreamEvent]
¶
Enforce MAP-3 and MAP-4: one final StreamEndEvent, one leading StreamStartEvent.
Adapters are stateless and may emit one end event per provider terminal
frame (finish_reason chunk, usage-only chunk, [DONE],
message_delta + message_stop). This wrapper passes delta and
error events through unchanged, absorbs every end event's fields —
a later non-None field replaces the accumulated value, a None field never
erases one — and emits the single merged end event once the underlying
iterator is exhausted. If no end event was seen (e.g. the stream errored
or was truncated), no end event is fabricated. provider_data follows
D9 (2026-09-06): the merged end carries the frame that supplied usage,
else the frame that supplied finish_reason; a later usage-bearing frame
replaces an earlier one, a finish-only frame never displaces a usage
frame.
Dialects without a start frame (chat completions, gemini SSE) get a
synthesized StreamStartEvent before the first delta or end event, so
every successful stream reads start → deltas → end (MAP-4). A provider
start passes through; duplicates after the first are dropped. Error
events never force a start: a stream that fails to open has no start.
adaptations (MAP-13) are stamped on the start event, provider-sent
or synthesized: they are known before the first byte.
See docs/mapping-rules.md MAP-3 and MAP-4.
acoalesce_stream(events: 'AsyncIterator[StreamEvent]', *, model: str | None = None, adaptations: tuple = ()) -> 'AsyncIterator[StreamEvent]'
async
¶
Async mirror of :func:coalesce_stream — same MAP-3/MAP-4 semantics.
Passes delta/error events through unchanged, absorbs every end event's fields (later non-None replaces, None never erases), and emits exactly one merged final StreamEndEvent once the source is exhausted. No end event is fabricated if none was seen. Synthesizes one leading StreamStartEvent for dialects without a start frame; duplicate starts are dropped; error events never force a start.
apply_client_side_stop(response: Response, stop: tuple[str, ...]) -> Response
¶
Cut the response's visible text at the first stop sequence.
The text parts are one stream in document order: a sequence that starts at the end of one part and finishes at the start of the next is a hit (a provider's own stop works on the token stream and knows no block boundary). The part holding the start is cut there; every later part is removed.