Universal-3.5 Pro Streaming API

Overview

This guide is for voice agents that connect AssemblyAI’s Universal-3.5 Pro Streaming WebSocket directly to a custom LLM and TTS, with no LiveKit, Pipecat, or other orchestrator in the loop. Universal-3.5 Pro Streaming is optimized for real-time audio under 10 seconds with low-latency turn detection, native multilingual code switching, and prompting support. The protocol is documented in detail on the Universal-3.5 Pro overview and message sequence pages. This guide focuses on the voice-agent loop and how to handle barge-in and interruptions correctly.

If you’re building on AssemblyAI’s Voice Agent API (a managed endpoint with built-in LLM and turn detection), see Turn detection and interruptions instead. Semantic interruption handling is built in there.

Quickstart

A minimal Python consumer that connects to the streaming WebSocket and reacts to Begin, Turn, SpeechStarted, and Termination events:

import json
import websocket
from urllib.parse import urlencode

API_KEY = "YOUR_API_KEY"
SAMPLE_RATE = 16000

CONNECTION_PARAMS = {
    "sample_rate": SAMPLE_RATE,
    "speech_model": "universal-3-5-pro",
    "min_turn_silence": 100,
    "max_turn_silence": 1000,
}

API_ENDPOINT = (
    "wss://streaming.assemblyai.com/v3/ws?" + urlencode(CONNECTION_PARAMS)
)


def on_message(ws, message):
    data = json.loads(message)
    msg_type = data.get("type")

    if msg_type == "Begin":
        print(f"Session started: {data.get('id')}")

    elif msg_type == "Turn":
        transcript = data.get("transcript", "")
        end_of_turn = data.get("end_of_turn", False)
        if end_of_turn:
            # Final transcript - send to your LLM
            print(f"Final: {transcript}")
        else:
            # Partial - optionally start pre-emptive LLM generation
            print(f"Partial: {transcript}")

    elif msg_type == "SpeechStarted":
        # User started speaking - interrupt the agent's TTS if it's playing
        print("Speech detected, interrupt agent if speaking")

    elif msg_type == "Termination":
        print("Session ended")


ws = websocket.WebSocketApp(
    API_ENDPOINT,
    header={"Authorization": API_KEY},
    on_message=on_message,
)
ws.run_forever()

For the full message protocol, including all event fields, audio framing, and termination, see the Universal-3.5 Pro message sequence reference.

Turn detection

Universal-3.5 Pro Streaming uses punctuation-based turn detection controlled by two parameters:

Parameter	Default	Description
`min_turn_silence`	`100` ms	Silence before a speculative end-of-turn check fires.
`max_turn_silence`	`1000` ms	Maximum silence before forcing the turn to end.

Lower values produce faster transcripts at the cost of occasional entity splits across turns. See the Universal-3.5 Pro overview for tuning guidance and the message sequence reference for the full event protocol.

Interruption handling

While the agent is speaking, users often produce backchannel utterances (“mhm”, “yeah”, “um”, “okay”) that you don’t want to treat as interruptions. A barge-in trigger that fires on every SpeechStarted (or every short Turn) will cause the agent to stop mid-sentence even though the user didn’t intend to interrupt. The recommended fix is a single combined filter applied to each Turn event during agent speech: skip the barge-in if the transcript is short or if every token is a known backchannel. Reset the filter once the agent has finished speaking.

import json
import string
import time
import websocket
from urllib.parse import urlencode


# "yes" / "no" deliberately omitted - in a booking flow a bare "yes"
# is a real confirmation. Edit for your domain.
BACKCHANNELS = frozenset({
    "mhm", "mm", "mmhm", "mmhmm",
    "uh", "uhhuh", "huh",
    "um", "umm", "uhm",
    "er", "erm",
    "hmm", "hm",
    "ah", "oh",
    "yeah", "yep", "yup",
    "okay", "ok",
    "right", "alright", "gotcha",
})

_PUNCT_STRIP = str.maketrans("", "", string.punctuation)
MIN_WORDS = 2          # Utterances below this are treated as filler
FILTER_GRACE_S = 1.0   # Keep filtering for 1s after agent stops speaking


# These flags are owned by your TTS layer.
agent_speaking = False
last_speaking_at = 0.0


def _is_all_backchannel(text: str) -> bool:
    tokens = text.lower().translate(_PUNCT_STRIP).split()
    return bool(tokens) and all(tok in BACKCHANNELS for tok in tokens)


def _should_suppress_interrupt(text: str) -> bool:
    now = time.monotonic()
    if agent_speaking:
        globals()["last_speaking_at"] = now
    elif now - last_speaking_at > FILTER_GRACE_S:
        return False

    word_count = len(text.split())
    return word_count < MIN_WORDS or _is_all_backchannel(text)


def on_message(ws, message):
    data = json.loads(message)
    msg_type = data.get("type")

    if msg_type == "Turn":
        transcript = data.get("transcript", "")
        end_of_turn = data.get("end_of_turn", False)

        if _should_suppress_interrupt(transcript):
            # Backchannel during agent speech - drop it.
            return

        if end_of_turn:
            handle_user_turn(transcript)  # send to LLM
        else:
            handle_partial(transcript)

    elif msg_type == "SpeechStarted":
        if agent_speaking:
            # Don't interrupt yet - wait for the Turn event,
            # which is gated by _should_suppress_interrupt above.
            return
        # Otherwise: normal barge-in path.

How it works:

While the agent is speaking (plus a 1-second grace window after speech ends), each Turn event is checked.
_should_suppress_interrupt returns True when the transcript has fewer than MIN_WORDS tokens or when every token is a known backchannel. Either condition drops the event.
Utterances with any non-filler content past the threshold (e.g., “yeah I’d like the suite”) always pass through.
SpeechStarted is gated through the same Turn-level check rather than firing barge-in directly. This prevents a race where a backchannel triggers SpeechStarted before the gating logic sees the transcript.

The BACKCHANNELS set is domain-customizable. “yes” and “no” are deliberately excluded because they often represent genuine confirmations in booking or IVR scenarios. Edit the set for your use case. MIN_WORDS = 2 is a reasonable default. Raise it if you see frequent two-word filler (“uh okay”, “yeah right”) slipping through.

If you’re building on LiveKit, prefer LiveKit’s adaptive interruption handling. See LiveKit > Interruption handling. The strategy on this page is the equivalent for direct-WebSocket integrations.

Recommended configuration

Three presets covering most voice-agent use cases:

# Fast - quick confirmations, IVR, yes/no questions
fast_params = {
    "speech_model": "universal-3-5-pro",
    "min_turn_silence": 100,
    "max_turn_silence": 800,
}

# Balanced - most voice agent conversations (recommended)
balanced_params = {
    "speech_model": "universal-3-5-pro",
    "min_turn_silence": 100,
    "max_turn_silence": 1000,
}

# Patient - entity dictation, complex instructions
patient_params = {
    "speech_model": "universal-3-5-pro",
    "min_turn_silence": 200,
    "max_turn_silence": 2000,
}

For cross-cutting topics like dynamic configuration updates, scaling, latency budgeting, and evals, see the voice agent best practices guide.

Getting started

Use cases & integrations

Trust & security

Universal-3.5 Pro Streaming API

Overview

Quickstart

Turn detection

Interruption handling

Recommended configuration

​Overview

​Quickstart

​Turn detection

​Interruption handling

​Recommended configuration

​Related

Overview

Quickstart

Turn detection

Interruption handling

Recommended configuration

Related