Skip to content
KugelAudioDocs

Streaming overview

Stream KugelAudio text-to-speech audio as it is generated: simple and LLM token streaming, the rules that keep latency low, audio playback and error handling.

On this page

Traditional TTS generates the entire audio before returning it. Streaming returns audio chunks as they're generated, providing:

  • Lower latency: Playback can begin before the full utterance finishes. See Latency for measurement guidance.
  • Better UX: Users hear audio immediately while more is being generated
  • LLM integration: Process text token-by-token as it arrives from language models

The four rules

Streaming integrations live or die by these. Each links to the page that explains it in depth:

  1. One turn per assistant reply, one session for the whole call. Keep the same streaming session open for the entire reply (never one session per sentence) and reuse it for the next reply. See Turn lifecycle.
  2. Send LLM tokens directly, without flushing. The server accumulates text and starts generating at natural sentence boundaries. On /ws/tts/stream a flush ends the turn, so flushing mid-reply splits it into several turns, each with its own model prefill. See Chunking & per-segment latency.
  3. Flush exactly once, at the end of the turn. This emits any trailing text, then ends the turn. endSession() / close() end the turn the same way. See Turn lifecycle.
  4. Pre-connect at startup. Don't pay the WebSocket handshake inside the first user interaction. See Latency.

Setup

Every snippet on this page uses a client created once:

from kugelaudio import KugelAudio

client = KugelAudio(api_key="YOUR_API_KEY")

Simple streaming

The simplest pattern, streaming a complete text:

Python
for chunk in client.tts.stream(
    text="Hello, this is streaming audio.",
    model_id="kugel-3.5",
    voice_id=1071,
):
    if hasattr(chunk, 'audio'):
        play_audio(chunk.audio)

LLM token streaming

Stream text token-by-token as it arrives from an LLM. Let the server handle chunking at sentence boundaries and flush once, when the reply ends.

Python
async def stream_from_llm(session, llm_response):
    async for token in llm_response:
        async for chunk in session.send(token):
            play_audio(chunk.audio)

    # Flush remaining text: ends the turn
    async for chunk in session.flush():
        play_audio(chunk.audio)

session comes from client.tts.streaming_session(...), opened once at startup (see Complete agent turn). In Python, send() returns only once the server has nothing more to deliver, so a fast LLM waits on it while audio arrives. The complete example below reads the LLM in a separate task so the two do not block each other.

Complete agent turn

The full shape of a Python agent: one session opened at startup and reused for every reply, and the LLM read in its own task so a slow send() never stalls it.

Python
import asyncio
from openai import AsyncOpenAI
from kugelaudio import KugelAudio

openai = AsyncOpenAI()
client = KugelAudio(api_key="YOUR_API_KEY")


async def speak_turn(session, user_message: str) -> None:
    llm = await openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_message}],
        stream=True,
    )
    tokens: asyncio.Queue = asyncio.Queue()

    async def read_llm() -> None:
        async for chunk in llm:
            token = chunk.choices[0].delta.content
            if token:
                await tokens.put(token)
        await tokens.put(None)  # end of reply

    reader = asyncio.create_task(read_llm())
    done = False
    while not done:
        # Forward everything the LLM produced since the last send.
        # No flush and no sentence buffering: the server chunks the text.
        parts = [await tokens.get()]
        while not tokens.empty():
            parts.append(tokens.get_nowait())
        if parts[-1] is None:
            parts.pop()
            done = True
        if parts:
            async for audio in session.send("".join(parts)):
                play_audio(audio.audio)
    await reader

    # One flush per reply: emits the trailing text and ends the turn.
    async for audio in session.flush():
        play_audio(audio.audio)


async def main() -> None:
    # Open once at startup; every reply reuses the same connection.
    async with client.tts.streaming_session(
        voice_id=1071,
        model_id="kugel-3.5",
        language="en",
    ) as session:
        await speak_turn(session, "Tell me a short story.")
        await speak_turn(session, "Now a shorter one.")


asyncio.run(main())

Spelling out text mid-stream

Use <spell> tags to spell out text letter by letter. Spell content bypasses text normalization automatically, while normalize: true still applies to the surrounding text. Set an explicit language for character names:

Python
text = "Contact us at <spell>hello@kugelaudio.com</spell> for help."

for chunk in client.tts.stream(
    text=text,
    model_id="kugel-3.5",
    voice_id=1071,
    normalize=True,
    language="en",
):
    if hasattr(chunk, 'audio'):
        play_audio(chunk.audio)

When streaming token-by-token, spell tags that span multiple chunks are handled automatically: the server buffers text until the closing </spell> tag arrives before generating audio, and auto-closes incomplete tags if the stream ends unexpectedly. See Text processing for the full spell-tag reference.

Audio playback

import { decodePCM16, type AudioChunk } from 'kugelaudio';

const audioContext = new AudioContext();
let scheduledTime = audioContext.currentTime;

function playChunk(chunk: AudioChunk) {
  const float32Data = decodePCM16(chunk.audio);

  const audioBuffer = audioContext.createBuffer(
    1, // mono
    float32Data.length,
    chunk.sampleRate
  );
  audioBuffer.copyToChannel(float32Data, 0);

  const source = audioContext.createBufferSource();
  source.buffer = audioBuffer;
  source.connect(audioContext.destination);

  // Schedule playback
  source.start(scheduledTime);
  scheduledTime += audioBuffer.duration;
}

Error handling

Python
import asyncio
from kugelaudio import KugelAudioConnectionError

async def robust_streaming():
    max_retries = 3

    for attempt in range(max_retries):
        try:
            async for chunk in client.tts.stream_async(
                text="Hello!",
                model_id="kugel-3.5",
                voice_id=1071,
            ):
                if hasattr(chunk, 'audio'):
                    play_audio(chunk.audio)
            break  # Success

        except KugelAudioConnectionError as e:
            if attempt < max_retries - 1:
                print(f"Connection closed, retrying... ({attempt + 1}/{max_retries})")
                await asyncio.sleep(1)
            else:
                raise

        except Exception as e:
            print(f"Streaming error: {e}")
            raise

Going deeper

Turn lifecycle

How turns start and end: flush, idle auto-flush, session reuse, usage

Chunking & per-segment latency

Flush granularity, buffer tuning, backpressure

Barge-in

Cancel the current turn when the user interrupts

Multi-context streaming

Up to 20 independent audio streams over one connection

Word timestamps

Word-level time alignments alongside streaming audio

WebSocket API reference

The full wire format: every message type, field by field