Streaming¶

A model answer that takes eight seconds to generate feels slow if it arrives all at once and fast if it arrives word by word. Streaming is what makes the second possible — and it is also how a workflow reports progress while it runs.

Streamer is the primitive underneath all of it. You obtain one or more value streamers from it, each with a name, push partial chunks as they arrive, and mark each stream complete. The Streamer itself is an async iterator yielding StreamContent messages (type, name, value) until every active value streamer has finished.

You rarely construct one by hand — stream_prompt() and run_stream() hand you one already wired up. This notebook shows the mechanics underneath, which is what you need when writing a client, a stub, or a custom node.

Cells use top-level await (supported in Jupyter), so there is no asyncio.run(...) wrapper.

1. Basic streaming¶

Push partial chunks, then complete. By default each partial carries the cumulative value so far, and complete carries the final value.

from kavalai import Streamer

streamer = Streamer()
result = streamer.get_value_streamer("result")

await result.stream_partial("Hello,")
await result.stream_partial(" world!")
await result.stream_complete()  # marks the "result" stream complete

async for message in streamer:
    print(message.model_dump_json())

{"type":"partial","name":"result","value":"Hello,"}
{"type":"partial","name":"result","value":"Hello, world!"}
{"type":"complete","name":"result","value":"Hello, world!"}

Cumulative partials mean a consumer can render the latest value directly, with no buffer of its own to maintain.

2. Producing from a background task¶

Producer and consumer normally run concurrently: start the producer as a task and iterate the streamer to drain messages as they arrive. This is the shape a real client uses — the HTTP response feeds the producer while your loop renders.

import asyncio

streamer = Streamer()
result = streamer.get_value_streamer("result")


async def produce():
    await result.stream_partial("Hello,")
    await asyncio.sleep(0.05)      # pretend the network is thinking
    await result.stream_partial(" world!")
    await result.stream_complete()


task = asyncio.create_task(produce())

async for message in streamer:
    print(message.model_dump_json())

await task  # ensure the producer finished cleanly

{"type":"partial","name":"result","value":"Hello,"}
{"type":"partial","name":"result","value":"Hello, world!"}
{"type":"complete","name":"result","value":"Hello, world!"}

3. Delta mode¶

With stream_delta=True each partial carries only the new text rather than the whole value, and complete carries no value at all. The consumer appends as it goes.

Cumulative mode is simpler; delta mode is cheaper. A long answer streamed cumulatively re-sends its entire buffer on every chunk, which is quadratic in the length of the output — so prefer deltas for anything long.

streamer = Streamer(stream_delta=True)
result = streamer.get_value_streamer("result")

await result.stream_partial("Hello,")
await result.stream_partial(" world!")
await result.stream_complete()

async for message in streamer:
    print(message.model_dump_json())

{"type":"partial","name":"result","value":"Hello,"}
{"type":"partial","name":"result","value":" world!"}
{"type":"complete","name":"result","value":null}

4. Streaming a structured value¶

Streaming JSON is awkward: for most of the stream, what you have is not valid JSON. Pass a response_model and the value streamer parses the partial JSON safely as it accumulates, so every partial is the best-effort object so far. A UI can render a half-filled record without ever seeing a parse error.

from pydantic import BaseModel


class Resident(BaseModel):
    name: str
    birth_year: int


streamer = Streamer()
result = streamer.get_value_streamer("result", response_model=Resident)

await result.stream_partial('{"name": "Agnes Whit')
await result.stream_partial('low", ')
await result.stream_partial(' "birth_year": 1929}')
await result.stream_complete()

async for message in streamer:
    print(f"{message.type:<8} {message.value}")

partial  {"name": "Agnes Whit"}
partial  {"name": "Agnes Whitlow"}
partial  {"name": "Agnes Whitlow", "birth_year": 1929}
complete {"name": "Agnes Whitlow", "birth_year": 1929}

The final complete message holds fully-formed JSON, so it validates directly:

print(Resident.model_validate_json(message.value))

name='Agnes Whitlow' birth_year=1929

5. Restarting a stream after a retry¶

When an LLM call fails transiently the client retries it, and the retry regenerates its answer from scratch. Anything the consumer accumulated from the failed attempt is now wrong.

Two operations occur. reset_active() forgets the failed attempt’s value streamers — otherwise the completion accounting never reaches zero and the iterator hangs — and stream_restart(...) pushes a restart message telling consumers to discard what they have under that name.

Treat restart as “clear the buffer”, not as an error: the run is healthy.

streamer = Streamer()
result = streamer.get_value_streamer("result")

await result.stream_partial("Hel")  # first attempt, cut short

# The attempt failed: forget its streamers and tell consumers to start over.
streamer.reset_active()
await streamer.stream_restart("attempt 1: timeout")

# The retry re-registers its value streamers and re-sends from the beginning.
result = streamer.get_value_streamer("result")
await result.stream_partial("Hello, world!")
await result.stream_complete()

async for message in streamer:
    print(message.model_dump_json())

{"type":"partial","name":"result","value":"Hel"}
{"type":"restart","name":"response","value":"attempt 1: timeout"}
{"type":"partial","name":"result","value":"Hello, world!"}
{"type":"complete","name":"result","value":"Hello, world!"}

Note the restart message is named response — the stream the LLM clients push into — while the content messages keep their own name.

Where you meet this in practice¶

Nodes opt into content streaming with stream_output (and stream_delta to choose the mode); an agent node adds stream_instructions and stream_partials. See Workflows.

Where to next¶

• LLM clients — stream_prompt on a real model.

• Workflows — watching a run as it executes.

• Agent Server API — serving a stream over SSE, and the two consequences of SSE to plan for.