Streaming Results

---

If your function is an async generator (uses yield), the caller receives each value as soon as the worker produces it, without waiting for the generator to finish.

This makes functions a good fit for:

• Streaming LLM token output.
• Incremental ETL where downstream consumers start processing early.
• Long-running jobs that emit progress events.

​

Example: audio transcription

Transcribe an audio file and yield each segment as it’s decoded:

from collections.abc import AsyncIterator
from typing import TypedDict

import chalkcompute


class Segment(TypedDict):
    start: float
    end: float
    text: str


@chalkcompute.function(
    image=chalkcompute.Image.base("python:3.12-slim").run_commands(
        "pip install faster-whisper==1.2.1 av==18.1.0 httpx",
    ),
    cpu="4",
    memory="8Gi",
)
async def transcribe(audio_uri: str) -> AsyncIterator[Segment]:
    import httpx
    from faster_whisper import WhisperModel

    async with httpx.AsyncClient() as client:
        resp = await client.get(audio_uri)
        resp.raise_for_status()

    audio_path = "/tmp/audio"
    with open(audio_path, "wb") as f:
        f.write(resp.content)

    model = WhisperModel("small.en", compute_type="int8")
    segments, _info = model.transcribe(audio_path)

    for segment in segments:
        yield {
            "start": segment.start,
            "end": segment.end,
            "text": segment.text,
        }

transcribe needs a return type annotation: AsyncIterator[Segment] declares the type of each value it yields.

Calling .remote(...) on a generator function returns an async iterator of pyarrow RecordBatch objects. Because transcribe yields a TypedDict, each of its keys is a column, and batch.to_pylist() returns the yielded dictionaries. A generator that yields single values, such as strings, returns them in a column named result; see Results from generator functions.

The caller iterates over segments as they arrive:

import asyncio

async def main():
    url = "https://www.nasa.gov/wp-content/uploads/2015/01/590331main_ringtone_smallStep.mp3"
    async for batch in transcribe.remote(url):
        for segment in batch.to_pylist():
            print(f"[{segment['start']:.1f}s] {segment['text']}")

if __name__ == "__main__":
    asyncio.run(main())

Output streams in real time:

[0.0s]  I'm going to step off the land mount.
[4.0s]  That's one small step for man.
[9.0s]  One giant leap for mankind.

If the function has retries set, the caller receives an attempt’s output only after that attempt finishes. See Retries.