"""Cross-vendor streaming TTS latency benchmark (SpeechifyAI, round 2, October 2026). Every target is measured from the same client, with the same texts, the same clock and the same audio analysis. Vendor adapters differ only in how they send one utterance and how they pull PCM out of the response; timing, onset detection and statistics are shared. What it reports for every request: first byte the first response data for the utterance (an HTTP body byte or a WebSocket message) first audible when a player reaches the first audible sample, the first 10 ms window above -45 dBFS RMS (about 0.6% of full scale); the player starts on the first audio chunk and waits whenever the next chunk has not arrived yet leading silence how much audio precedes that first audible sample total when the response ended: a socket's completion message or the end of an HTTP body; ElevenLabs' socket sends no completion message, so there it is the last audio last audio when the last audio of the utterance arrived Warm requests reuse one open connection per target; cold requests open a new one inside the clock. Requests go round-robin across targets in a shuffled order each round, so every vendor shares the same time window. Deepgram's HTTP endpoint closes a connection that sits idle for under 20 seconds, so its warm requests run back-to-back after the round-robin instead. Run it: python3 -m venv .venv && . .venv/bin/activate && pip install httpx numpy websockets export SPEECHIFY_API_KEY=... CARTESIA_API_KEY=... DEEPGRAM_API_KEY=... OPENAI_API_KEY=... \ GEMINI_API_KEY=... ELEVENLABS_API_KEY=... # a target whose key is missing is skipped python bench.py net # resolved IP, TCP and TLS time per vendor host python bench.py probe # one request per target, a sanity check python bench.py run --vantage my-laptop --out results.csv python bench.py summarize results.csv Round 2 (9 October 2026) ran two windows, each from both vantages at the same time: python bench.py run --window main --vantage us-east4-a --out results.csv --only \ "speechify/simba-3.2 http" "speechify/simba-3.2 ws" "elevenlabs/flash-v2.5 http" "elevenlabs/flash-v2.5 ws" \ "cartesia/sonic-3.6 http" "cartesia/sonic-3.6 ws" "deepgram/aura-2 http" "deepgram/aura-2 ws" \ "openai/gpt-4o-mini-tts" "openai/tts-1" "google/gemini-3.8-flash-tts" "google/gemini-3.8-flash-lite-tts" python bench.py run --window second --vantage us-east4-a --out results-auto.csv --warm 60 --cold 6 --pause 6.5 \ --elevenlabs-char-cap 6500 --only "speechify/simba-3.2 http" "speechify/simba-3.2 ws" "elevenlabs/flash-v2.5 ws auto_mode" The second window's 6.5 s pause gives each connection about the idle time it had between turns in the first. Changed since the 9 October 2026 run, whose CSV is published beside this script: - last_audio_ms is new. Some streams stay open after their last audio, so total can be later. - first audible now follows the player through every late chunk. That run took the later of the first chunk plus the leading silence and the arrival of the chunk carrying the first audible sample, the same reading whenever that sample arrives in the first chunk. - --window names the run in every row, and summarize keeps windows apart. Keys are read from the environment and never logged. """ from __future__ import annotations import argparse import asyncio import base64 import csv import json import os import random import socket import ssl import statistics import sys import time import uuid from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path import httpx import numpy as np import websockets HERE = Path(__file__).resolve().parent SAMPLE_RATE = 24000 ONSET_WINDOW_MS = 10 ONSET_THRESHOLD_DBFS = -45.0 REQUEST_TIMEOUT_S = 30.0 WARMUP_REQUESTS = 2 WARM_REQUESTS = 60 COLD_REQUESTS = 12 PAUSE_BETWEEN_REQUESTS_S = 0.3 SEED = 7727 # Built once, outside any timer: loading the CA bundle per connection costs 20-40 ms. WS_SSL = ssl.create_default_context() # Response headers worth keeping for later debugging. Request headers are never stored. KEEP_RESPONSE_HEADERS = ( "content-type", "server-timing", "speechify-request-id", "x-request-id", "request-id", "openai-processing-ms", "dg-request-id", "cf-ray", "x-envoy-upstream-service-time", "x-region", "via", "server", ) def now_ms() -> float: return time.perf_counter_ns() / 1e6 def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds") def key(name: str) -> str | None: return os.environ.get(name) or None @dataclass class Row: window: str vantage: str target: str vendor: str model: str voice: str transport: str series: str seq: int text_id: str chars: int first_occurrence: bool started_utc: str = "" ok: bool = False http_status: int | None = None error: str = "" new_connection: bool = False peer_ip: str = "" connect_ms: float | None = None tls_ms: float | None = None headers_ms: float | None = None ttfb_ms: float | None = None first_audio_ms: float | None = None first_audible_ms: float | None = None onset_arrival_ms: float | None = None leading_silence_ms: float | None = None silence_at_m35_ms: float | None = None silence_at_m55_ms: float | None = None total_ms: float | None = None last_audio_ms: float | None = None audio_bytes: int = 0 audio_ms: float | None = None audio_chunks: int = 0 response_meta: str = "" note: str = "" class Capture: """Collects one utterance: raw response timing plus timestamped PCM chunks.""" def __init__(self, t0: float): self.t0 = t0 self.first_byte: float | None = None self.headers: float | None = None self.done: float | None = None self.chunks: list[tuple[float, bytes]] = [] self.meta: dict[str, str] = {} def body_bytes_arrived(self) -> None: if self.first_byte is None: self.first_byte = now_ms() def audio(self, pcm: bytes) -> None: if pcm: self.chunks.append((now_ms(), pcm)) def keep_headers(self, headers) -> None: for name in KEEP_RESPONSE_HEADERS: value = headers.get(name) if value: self.meta[name] = value def pcm_ms(n_bytes: int) -> float: return n_bytes / 2 / SAMPLE_RATE * 1000 def playback_onset(chunks: list[tuple[float, bytes]], onset_byte: int) -> tuple[float, float]: """When onset_byte's chunk arrived, and when a player that starts on the first chunk and waits on late ones reaches it.""" played = chunks[0][0] offset = 0 for t, c in chunks: played = max(played, t) if offset + len(c) > onset_byte: return t, played + pcm_ms(onset_byte - offset) played += pcm_ms(len(c)) offset += len(c) raise ValueError("onset_byte is past the end of the audio") def analyse(cap: Capture, row: Row, t_end: float) -> None: rel = lambda t: None if t is None else round(t - cap.t0, 2) # noqa: E731 row.headers_ms = rel(cap.headers) row.ttfb_ms = rel(cap.first_byte) row.total_ms = rel(cap.done if cap.done is not None else t_end) row.audio_chunks = len(cap.chunks) row.response_meta = json.dumps(cap.meta, sort_keys=True) if cap.meta else "" if not cap.chunks: return first_audio_t = cap.chunks[0][0] row.first_audio_ms = rel(first_audio_t) row.last_audio_ms = rel(cap.chunks[-1][0]) pcm = b"".join(c for _, c in cap.chunks) row.audio_bytes = len(pcm) samples = np.frombuffer(pcm[: len(pcm) - (len(pcm) % 2)], dtype=" threshold)[0] if len(hits): setattr(row, attr, round(int(hits[0]) * ONSET_WINDOW_MS, 1)) loud = np.nonzero(dbfs > ONSET_THRESHOLD_DBFS)[0] if len(loud) == 0: return onset_sample = int(loud[0]) * win onset_arrival_t, audible_t = playback_onset(cap.chunks, onset_sample * 2) row.leading_silence_ms = round(onset_sample / SAMPLE_RATE * 1000, 1) row.onset_arrival_ms = rel(onset_arrival_t) row.first_audible_ms = rel(audible_t) class StopVendor(Exception): """A quota or auth refusal: stop every target of this vendor instead of retrying.""" # -------------------------------------------------------------------------------------- # Transports # -------------------------------------------------------------------------------------- def http_client() -> httpx.AsyncClient: return httpx.AsyncClient( http1=True, http2=False, # Identity, as in every vendor's curl example: compression can hold back stream flushes. headers={"Accept-Encoding": "identity"}, timeout=httpx.Timeout(REQUEST_TIMEOUT_S, connect=10.0), limits=httpx.Limits(max_connections=1, max_keepalive_connections=1, keepalive_expiry=600), ) class Target: name = "" vendor = "" model = "" voice = "" host = "" transport = "" back_to_back = False note = "" def check_refusal(self, status: int | None, detail: str) -> None: """Raise StopVendor for a refusal that retrying would only repeat.""" class HttpTarget(Target): transport = "http" def __init__(self): self.client: httpx.AsyncClient | None = None async def open(self) -> None: self.client = http_client() async def close(self) -> None: if self.client: await self.client.aclose() self.client = None def request(self, text: str) -> tuple[str, dict, dict, dict]: """Return (url, headers, params, json_body).""" raise NotImplementedError async def consume(self, resp: httpx.Response, cap: Capture) -> None: async for chunk in resp.aiter_raw(): cap.body_bytes_arrived() cap.audio(chunk) async def synth(self, text: str, row: Row, fresh: bool) -> Capture: client = http_client() if fresh else self.client url, headers, params, body = self.request(text) phases: dict[str, float] = {} async def trace(event: str, info: dict) -> None: phases[event] = now_ms() t0 = now_ms() cap = Capture(t0) refusal = None try: async with client.stream( "POST", url, headers=headers, params=params, json=body, extensions={"trace": trace} ) as resp: cap.headers = now_ms() row.http_status = resp.status_code cap.keep_headers(resp.headers) stream = resp.extensions.get("network_stream") if stream is not None: addr = stream.get_extra_info("server_addr") row.peer_ip = addr[0] if addr else "" if resp.status_code != 200: detail = (await resp.aread())[:300].decode("utf-8", "replace") row.error = f"http {resp.status_code}: {detail}" refusal = (resp.status_code, detail) else: await self.consume(resp, cap) row.ok = bool(cap.chunks) and "vendor_error" not in cap.meta if not row.ok: row.error = row.error or cap.meta.get("vendor_error", "no audio in response") except Exception as exc: # noqa: BLE001 row.error = f"{type(exc).__name__}: {exc}"[:300] finally: t_end = now_ms() if fresh: await client.aclose() row.new_connection = "connection.connect_tcp.started" in phases if row.new_connection and "connection.connect_tcp.complete" in phases: row.connect_ms = round(phases["connection.connect_tcp.complete"] - phases["connection.connect_tcp.started"], 2) if "connection.start_tls.complete" in phases: row.tls_ms = round(phases["connection.start_tls.complete"] - phases["connection.start_tls.started"], 2) analyse(cap, row, t_end) if refusal: self.check_refusal(*refusal) return cap class SseMixin: """For HTTP targets whose body is server-sent events carrying base64 audio.""" async def consume(self, resp: httpx.Response, cap: Capture) -> None: buf = b"" async for chunk in resp.aiter_bytes(): cap.body_bytes_arrived() buf += chunk buf = buf.replace(b"\r\n", b"\n") while b"\n\n" in buf: raw_event, buf = buf.split(b"\n\n", 1) self.handle_event(raw_event.decode("utf-8", "replace"), cap) def handle_event(self, raw_event: str, cap: Capture) -> None: raise NotImplementedError class WsTarget(Target): transport = "websocket" # Renew a kept-open session before the clock starts once it is this old (None: never). max_session_age_s: float | None = None def __init__(self): self.ws = None self.opened_at = 0.0 def connect_args(self) -> tuple[str, dict]: raise NotImplementedError async def on_open(self, ws) -> None: """Anything the protocol needs before the first utterance (inside the clock when cold).""" async def _connect(self): url, headers = self.connect_args() ws = await websockets.connect( url, additional_headers=headers, ssl=WS_SSL, max_size=None, open_timeout=10, ping_interval=20, ping_timeout=20 ) await self.on_open(ws) return ws async def open(self) -> None: self.ws = await self._connect() self.opened_at = time.monotonic() async def close(self) -> None: if self.ws is not None: try: await self.ws.close() except Exception: # noqa: BLE001 pass self.ws = None async def drain(self, ws) -> int: """Discard anything left over from the previous utterance, before the clock starts.""" n = 0 while True: try: await asyncio.wait_for(ws.recv(), 0.02) n += 1 except (asyncio.TimeoutError, websockets.ConnectionClosed): return n async def exchange(self, ws, text: str, cap: Capture, row: Row) -> None: raise NotImplementedError async def synth(self, text: str, row: Row, fresh: bool) -> Capture: notes = [] ws = None if not fresh: stale = self.ws is None or self.ws.state != websockets.protocol.State.OPEN aged = self.max_session_age_s is not None and time.monotonic() - self.opened_at > self.max_session_age_s if stale or aged: await self.close() try: await self.open() except Exception as exc: # noqa: BLE001 row.error = f"{type(exc).__name__}: {exc}"[:300] analyse(Capture(now_ms()), row, now_ms()) self.check_refusal(getattr(getattr(exc, "response", None), "status_code", None), str(exc)) return Capture(now_ms()) notes.append("session renewed before the clock started" if aged and not stale else "reconnected before the clock started") drained = await self.drain(self.ws) if drained: notes.append(f"drained {drained} leftover messages before the clock started") t0 = now_ms() cap = Capture(t0) try: if fresh: ws = await self._connect() row.new_connection = True row.connect_ms = round(now_ms() - t0, 2) else: ws = self.ws peer = ws.transport.get_extra_info("peername") if ws.transport else None row.peer_ip = peer[0] if peer else "" resp = getattr(ws, "response", None) if resp is not None: cap.keep_headers(resp.headers) await asyncio.wait_for(self.exchange(ws, text, cap, row), REQUEST_TIMEOUT_S) row.ok = bool(cap.chunks) and not row.error row.http_status = 101 except Exception as exc: # noqa: BLE001 row.error = row.error or f"{type(exc).__name__}: {exc}"[:300] status = getattr(getattr(exc, "response", None), "status_code", None) row.http_status = status if not fresh: await self.close() finally: t_end = now_ms() if fresh and ws is not None: try: await ws.close() except Exception: # noqa: BLE001 pass row.note = "; ".join(notes) analyse(cap, row, t_end) if not row.ok and row.error: self.check_refusal(row.http_status, row.error) return cap # -------------------------------------------------------------------------------------- # Vendor adapters # -------------------------------------------------------------------------------------- SPEECHIFY_MODEL, SPEECHIFY_VOICE = "simba-3.2", "geffen_32" class SpeechifyStream(HttpTarget): vendor = "speechify" host = "api.speechify.ai" def __init__(self, model=SPEECHIFY_MODEL, voice=SPEECHIFY_VOICE): super().__init__() self.model, self.voice = model, voice self.key = key("SPEECHIFY_API_KEY") def request(self, text): return ( "https://api.speechify.ai/v1/audio/stream", {"Authorization": f"Bearer {self.key}", "Accept": "audio/pcm", "Content-Type": "application/json"}, {}, {"input": text, "voice_id": self.voice, "model": self.model}, ) class SpeechifyInputStream(WsTarget): """GET /v1/audio/stream/ws: send the sentence, flush, read until speech.flushed.""" vendor = "speechify" host = "api.speechify.ai" # A session stops taking text after 5 minutes; a client opens the next one ahead of time. max_session_age_s = 240 def __init__(self, model=SPEECHIFY_MODEL, voice=SPEECHIFY_VOICE): super().__init__() self.model, self.voice = model, voice self.key = key("SPEECHIFY_API_KEY") def connect_args(self): q = httpx.QueryParams({"voice_id": self.voice, "model": self.model, "output_format": f"pcm_{SAMPLE_RATE}"}) return f"wss://api.speechify.ai/v1/audio/stream/ws?{q}", {"Authorization": f"Bearer {self.key}"} async def exchange(self, ws, text, cap, row): await ws.send(json.dumps({"type": "input.text", "text": text + " "})) await ws.send(json.dumps({"type": "input.flush"})) while True: data = json.loads(await ws.recv()) cap.body_bytes_arrived() kind = data.get("type") if kind == "speech.chunk" and data.get("audio"): cap.audio(base64.b64decode(data["audio"])) elif kind == "speech.flushed": cap.done = now_ms() return elif kind in ("speech.error", "speech.done"): row.error = f"vendor {kind}: {json.dumps(data)[:250]}" return CARTESIA_VERSION = "2026-08-14" CARTESIA_MODEL, CARTESIA_VOICE = "sonic-3.6", "db6b0ed5-d5d3-463d-ae85-518a07d3c2b4" CARTESIA_FORMAT = {"container": "raw", "encoding": "pcm_s16le", "sample_rate": SAMPLE_RATE} class CartesiaBytes(HttpTarget): vendor = "cartesia" host = "api.cartesia.ai" def __init__(self, model=CARTESIA_MODEL, voice=CARTESIA_VOICE): super().__init__() self.model, self.voice = model, voice self.key = key("CARTESIA_API_KEY") def request(self, text): return ( "https://api.cartesia.ai/tts/bytes", {"Authorization": f"Bearer {self.key}", "Cartesia-Version": CARTESIA_VERSION}, {}, { "model_id": self.model, "transcript": text, "voice": {"mode": "id", "id": self.voice}, "language": "en", "output_format": CARTESIA_FORMAT, }, ) class CartesiaWs(WsTarget): vendor = "cartesia" host = "api.cartesia.ai" def __init__(self, model=CARTESIA_MODEL, voice=CARTESIA_VOICE): super().__init__() self.model, self.voice = model, voice self.key = key("CARTESIA_API_KEY") def connect_args(self): return f"wss://api.cartesia.ai/tts/websocket?cartesia_version={CARTESIA_VERSION}", {"X-API-Key": self.key} async def exchange(self, ws, text, cap, row): context_id = uuid.uuid4().hex await ws.send( json.dumps( { "model_id": self.model, "transcript": text, "voice": {"mode": "id", "id": self.voice}, "language": "en", "output_format": CARTESIA_FORMAT, "context_id": context_id, "continue": False, } ) ) while True: data = json.loads(await ws.recv()) if data.get("context_id") not in (None, context_id): continue cap.body_bytes_arrived() kind = data.get("type") if kind == "chunk" and data.get("data"): cap.audio(base64.b64decode(data["data"])) elif kind == "error": row.error = f"vendor error: {json.dumps(data)[:250]}" return if kind == "done" or data.get("done") is True: cap.done = now_ms() return DEEPGRAM_MODEL = "aura-2-thalia-en" DEEPGRAM_QUERY = {"model": DEEPGRAM_MODEL, "encoding": "linear16", "sample_rate": str(SAMPLE_RATE)} class DeepgramRest(HttpTarget): vendor = "deepgram" host = "api.deepgram.com" back_to_back = True note = "back-to-back: Deepgram closes an idle HTTP connection in under 20 s" def __init__(self, model=DEEPGRAM_MODEL): super().__init__() self.model = self.voice = model self.key = key("DEEPGRAM_API_KEY") def request(self, text): return ( "https://api.deepgram.com/v1/speak", {"Authorization": f"Token {self.key}", "Content-Type": "application/json"}, {**DEEPGRAM_QUERY, "model": self.model, "container": "none"}, {"text": text}, ) class DeepgramWs(WsTarget): vendor = "deepgram" host = "api.deepgram.com" def __init__(self, model=DEEPGRAM_MODEL): super().__init__() self.model = self.voice = model self.key = key("DEEPGRAM_API_KEY") def connect_args(self): q = httpx.QueryParams({**DEEPGRAM_QUERY, "model": self.model}) return f"wss://api.deepgram.com/v1/speak?{q}", {"Authorization": f"Token {self.key}"} async def exchange(self, ws, text, cap, row): await ws.send(json.dumps({"type": "Speak", "text": text})) await ws.send(json.dumps({"type": "Flush"})) while True: msg = await ws.recv() if isinstance(msg, bytes): cap.body_bytes_arrived() cap.audio(msg) continue data = json.loads(msg) kind = data.get("type") if kind == "Metadata": if data.get("request_id"): cap.meta["dg-request-id"] = data["request_id"] continue cap.body_bytes_arrived() if kind == "Flushed": cap.done = now_ms() return if kind == "Error" or (kind == "Warning" and not cap.chunks): row.error = f"vendor {kind}: {json.dumps(data)[:250]}" if kind == "Error": return ELEVENLABS_MODEL, ELEVENLABS_VOICE = "eleven_flash_v2_5", "JBFqnCBsd6RMkjVDRZzb" def elevenlabs_refusal(status: int | None, detail: str) -> None: text = detail.lower() if status in (401, 402, 429) or any(s in text for s in ("quota", "credits", "unauthorized", "invalid_api_key")): raise StopVendor(f"ElevenLabs refused ({status}): {detail[:200]}") class ElevenLabsStream(HttpTarget): """POST /v1/text-to-speech/{voice_id}/stream.""" vendor = "elevenlabs" host = "api.elevenlabs.io" def __init__(self, model=ELEVENLABS_MODEL, voice=ELEVENLABS_VOICE): super().__init__() self.model, self.voice = model, voice self.key = key("ELEVENLABS_API_KEY") def request(self, text): return ( f"https://api.elevenlabs.io/v1/text-to-speech/{self.voice}/stream", {"xi-api-key": self.key, "Content-Type": "application/json"}, {"output_format": f"pcm_{SAMPLE_RATE}"}, {"text": text, "model_id": self.model}, ) def check_refusal(self, status, detail): elevenlabs_refusal(status, detail) class ElevenLabsInputStream(WsTarget): """GET /v1/text-to-speech/{voice_id}/stream-input: send the sentence with flush, read its audio. The single-context socket marks no end of a flush, so the utterance ends when the alignment has covered every character sent, or after a quiet gap once audio has started. auto_mode=true, which ElevenLabs documents for text sent as whole sentences, skips the socket's buffering; without it the socket answered most flushed sentences in one message. """ vendor = "elevenlabs" host = "api.elevenlabs.io" quiet_gap_s = 1.5 def __init__(self, model=ELEVENLABS_MODEL, voice=ELEVENLABS_VOICE, auto_mode=False): super().__init__() self.model, self.voice, self.auto_mode = model, voice, auto_mode self.key = key("ELEVENLABS_API_KEY") def connect_args(self): params = {"model_id": self.model, "output_format": f"pcm_{SAMPLE_RATE}", "inactivity_timeout": "180"} if self.auto_mode: params["auto_mode"] = "true" q = httpx.QueryParams(params) return f"wss://api.elevenlabs.io/v1/text-to-speech/{self.voice}/stream-input?{q}", {"xi-api-key": self.key} async def on_open(self, ws): await ws.send(json.dumps({"text": " "})) async def exchange(self, ws, text, cap, row): await ws.send(json.dumps({"text": text + " ", "flush": True})) covered = 0 while True: try: msg = await asyncio.wait_for(ws.recv(), self.quiet_gap_s if cap.chunks else REQUEST_TIMEOUT_S) except asyncio.TimeoutError: cap.meta["end"] = "quiet gap" cap.done = cap.chunks[-1][0] return data = json.loads(msg) cap.body_bytes_arrived() if data.get("audio"): cap.audio(base64.b64decode(data["audio"])) alignment = data.get("alignment") or {} covered += len(alignment.get("chars") or []) if data.get("error") or (data.get("message") and not data.get("audio")): row.error = f"vendor error: {json.dumps(data)[:250]}" return if data.get("isFinal") or (cap.chunks and covered >= len(text)): cap.meta["end"] = "final" if data.get("isFinal") else "alignment" cap.done = cap.chunks[-1][0] if cap.chunks else now_ms() return def check_refusal(self, status, detail): elevenlabs_refusal(status, detail) class OpenAISpeech(HttpTarget): vendor = "openai" host = "api.openai.com" def __init__(self, model, voice): super().__init__() self.model, self.voice = model, voice self.key = key("OPENAI_API_KEY") def request(self, text): return ( "https://api.openai.com/v1/audio/speech", {"Authorization": f"Bearer {self.key}", "Content-Type": "application/json"}, {}, {"model": self.model, "input": text, "voice": self.voice, "response_format": "pcm"}, ) class GeminiInteractions(SseMixin, HttpTarget): vendor = "google" host = "generativelanguage.googleapis.com" transport = "http-sse" def __init__(self, model, voice="Kore"): super().__init__() self.model, self.voice = model, voice self.key = key("GEMINI_API_KEY") def request(self, text): return ( "https://generativelanguage.googleapis.com/v1beta/interactions", {"x-goog-api-key": self.key, "Content-Type": "application/json"}, {}, { "model": self.model, "input": [{"type": "user_input", "content": [{"type": "text", "text": text}]}], "response_format": {"type": "audio"}, "generation_config": {"speech_config": [{"voice": self.voice}]}, "stream": True, }, ) def handle_event(self, raw_event, cap): data_lines = [ln[5:].strip() for ln in raw_event.split("\n") if ln.startswith("data:")] if not data_lines: return payload = "\n".join(data_lines) if payload == "[DONE]": return try: data = json.loads(payload) except json.JSONDecodeError: return kind = data.get("event_type") cap.meta.setdefault("event_types", "") if kind and kind not in cap.meta["event_types"].split(","): cap.meta["event_types"] = ",".join(filter(None, [cap.meta["event_types"], kind])) # The first event is bookkeeping, not audio: first byte and first audio differ here. if kind == "step.delta": delta = data.get("delta") or {} if delta.get("type") == "audio" and delta.get("data"): if delta.get("mime_type"): cap.meta["audio_mime"] = delta["mime_type"] cap.audio(base64.b64decode(delta["data"])) elif kind == "error" or "error" in data: cap.meta["vendor_error"] = json.dumps(data)[:250] TARGETS = [ ("speechify/simba-3.2 http", SpeechifyStream, "SPEECHIFY_API_KEY"), ("speechify/simba-3.2 ws", SpeechifyInputStream, "SPEECHIFY_API_KEY"), ("elevenlabs/flash-v2.5 http", ElevenLabsStream, "ELEVENLABS_API_KEY"), ("elevenlabs/flash-v2.5 ws", ElevenLabsInputStream, "ELEVENLABS_API_KEY"), ("elevenlabs/flash-v2.5 ws auto_mode", lambda: ElevenLabsInputStream(auto_mode=True), "ELEVENLABS_API_KEY"), ("cartesia/sonic-3.6 http", CartesiaBytes, "CARTESIA_API_KEY"), ("cartesia/sonic-3.6 ws", CartesiaWs, "CARTESIA_API_KEY"), ("deepgram/aura-2 http", DeepgramRest, "DEEPGRAM_API_KEY"), ("deepgram/aura-2 ws", DeepgramWs, "DEEPGRAM_API_KEY"), ("openai/gpt-4o-mini-tts", lambda: OpenAISpeech("gpt-4o-mini-tts", "marin"), "OPENAI_API_KEY"), ("openai/tts-1", lambda: OpenAISpeech("tts-1", "alloy"), "OPENAI_API_KEY"), ("google/gemini-3.8-flash-tts", lambda: GeminiInteractions("gemini-3.8-flash-tts"), "GEMINI_API_KEY"), ("google/gemini-3.8-flash-lite-tts", lambda: GeminiInteractions("gemini-3.8-flash-lite-tts"), "GEMINI_API_KEY"), ] TARGET_ORDER = [name for name, _, _ in TARGETS] def build_targets(only: set[str] | None = None, quiet: bool = False) -> list[Target]: out = [] for name, make, env in TARGETS: if only and name not in only: continue if not key(env): if not quiet: print(f"skipping {name}: {env} is not set", file=sys.stderr) continue t = make() t.name = name out.append(t) return out # -------------------------------------------------------------------------------------- # Runner # -------------------------------------------------------------------------------------- # Short voice-agent replies, the same twelve in every round. The warm-up lines are not counted. WARMUP_TEXTS = [ "Thanks for waiting, I am pulling up your account details right now.", "Give me just a second while I check the latest status for that request.", ] TEXTS = [ "Sure, I can help with that. Could you tell me the order number printed on your receipt?", "Your appointment is confirmed for Thursday at three thirty in the afternoon.", "I found two flights that match, and the earlier one leaves at seven fifteen tomorrow morning.", "That payment went through successfully, and you should see a confirmation email within a few minutes.", "I'm sorry about the delay. Let me transfer you to someone who can sort this out right away.", "The package left our warehouse yesterday and should arrive by Friday evening.", "Before we continue, can you confirm the last four digits of the phone number on the account?", "It looks like your subscription renews on the fifth, so there is nothing you need to do today.", "Great question. The store opens at nine on weekdays and at ten on Saturdays and Sundays.", "I have updated your delivery address. Is there anything else I can help you with?", "Your balance is one hundred and twenty dollars, and the next statement closes at the end of the month.", "No problem at all. I will send you a text with the link so you can finish the form later.", ] def load_texts() -> tuple[list[str], list[str]]: return WARMUP_TEXTS, TEXTS class CsvSink: def __init__(self, path: Path): self.fh = path.open("w", newline="") self.w = csv.DictWriter(self.fh, fieldnames=list(Row.__dataclass_fields__)) self.w.writeheader() def write(self, row: Row) -> None: self.w.writerow(asdict(row)) self.fh.flush() def brief(row: Row) -> str: f = lambda v: "-" if v is None else f"{v:.0f}" # noqa: E731 status = "ok " if row.ok else "ERR" tail = "" if row.ok else f" {row.error[:120]}" return ( f"{status} {row.series:6} {row.seq:2} {row.target:34} ttfb {f(row.ttfb_ms):>5} " f"audio {f(row.first_audio_ms):>5} audible {f(row.first_audible_ms):>5} " f"sil {f(row.leading_silence_ms):>4} total {f(row.total_ms):>5} " f"new={int(row.new_connection)} ip={row.peer_ip}{tail}" ) async def run( out: Path, window: str, vantage: str, only: set[str] | None, warm_n: int, cold_n: int, char_cap: int, pause_s: float ) -> None: warmup_texts, texts = load_texts() targets = build_targets(only) sink = CsvSink(out) rng = random.Random(SEED) stopped: dict[str, str] = {} chars_sent: dict[str, int] = {} print(json.dumps({"window": window, "vantage": vantage, "started_utc": utc_now(), "targets": [t.name for t in targets]}), flush=True) for t in targets: try: await t.open() except Exception as exc: # noqa: BLE001 print(f"open failed for {t.name}: {type(exc).__name__}: {exc}", flush=True) seen: dict[str, set[str]] = {t.name: set() for t in targets} async def one(t: Target, series: str, seq: int, text_id: str, text: str, fresh: bool) -> None: if t.vendor in stopped: return if char_cap and chars_sent.get(t.vendor, 0) + len(text) > char_cap and t.vendor == "elevenlabs": stopped[t.vendor] = f"stopped at the {char_cap}-character budget" print(f"STOP {t.vendor}: {stopped[t.vendor]}", flush=True) return chars_sent[t.vendor] = chars_sent.get(t.vendor, 0) + len(text) first = text_id not in seen[t.name] seen[t.name].add(text_id) row = Row(window, vantage, t.name, t.vendor, t.model, t.voice, t.transport, series, seq, text_id, len(text), first) row.started_utc = utc_now() try: await t.synth(text, row, fresh) except StopVendor as exc: stopped[t.vendor] = str(exc) print(f"STOP {t.vendor}: {exc}", flush=True) if t.note: row.note = "; ".join(filter(None, [t.note, row.note])) sink.write(row) print(brief(row), flush=True) await asyncio.sleep(pause_s) # Warm-up and warm series: round-robin so every vendor shares the same time window; the order # inside a round is shuffled to avoid a fixed position effect. plan = [("warmup", i, f"w{i}", warmup_texts[i % len(warmup_texts)]) for i in range(WARMUP_REQUESTS)] plan += [("warm", i, f"m{i % len(texts)}", texts[i % len(texts)]) for i in range(warm_n)] rotating = [t for t in targets if not t.back_to_back] for series, seq, text_id, text in plan: order = rotating[:] rng.shuffle(order) for t in order: await one(t, series, seq, text_id, text, fresh=False) for t in (t for t in targets if t.back_to_back): for series, seq, text_id, text in plan: await one(t, series, seq, text_id, text, fresh=False) for t in targets: await t.close() for i in range(cold_n): order = targets[:] rng.shuffle(order) for t in order: await one(t, "cold", i, f"m{i % len(texts)}", texts[i % len(texts)], fresh=True) print(json.dumps({"finished_utc": utc_now(), "chars_sent": chars_sent, "stopped": stopped}), flush=True) def write_wav(path: Path, pcm: bytes) -> None: import wave with wave.open(str(path), "wb") as w: w.setnchannels(1) w.setsampwidth(2) w.setframerate(SAMPLE_RATE) w.writeframes(pcm[: len(pcm) - (len(pcm) % 2)]) async def probe(only: set[str] | None, text_ids: list[str] | None, audio_dir: Path | None, repeat: int, gap_s: float) -> None: """A few requests per target on one open connection; optional WAVs for listening.""" _, texts = load_texts() items = [(tid, texts[int(tid[1:])]) for tid in text_ids] if text_ids else [("probe", "Sure, I can check that for you. It will only take a moment.")] if audio_dir: audio_dir.mkdir(parents=True, exist_ok=True) stopped: set[str] = set() for t in build_targets(only): if t.vendor in stopped: continue try: await t.open() except Exception as exc: # noqa: BLE001 print(f"open failed for {t.name}: {type(exc).__name__}: {exc}") continue for rep in range(repeat): if t.vendor in stopped: break for seq, (tid, text) in enumerate(items): row = Row("probe", "probe", t.name, t.vendor, t.model, t.voice, t.transport, "probe", seq, tid, len(text), rep == 0) row.started_utc = utc_now() try: cap = await t.synth(text, row, fresh=False) except StopVendor as exc: stopped.add(t.vendor) print(f"STOP {t.vendor}: {exc}") break print(brief(row) + f" last_audio={row.last_audio_ms} chunks={row.audio_chunks} note={row.note!r} meta={row.response_meta[:160]}") if audio_dir and cap.chunks: slug = t.name.replace("/", "_").replace(" ", "_") write_wav(audio_dir / f"{slug}_{tid}_{rep}.wav", b"".join(c for _, c in cap.chunks)) await asyncio.sleep(gap_s) await t.close() def net() -> None: hosts = sorted({t.host for t in build_targets(quiet=True)}) ctx = ssl.create_default_context() for host in hosts: t = now_ms() infos = socket.getaddrinfo(host, 443, socket.AF_INET, socket.SOCK_STREAM) dns_ms = now_ms() - t ip = infos[0][4][0] rtts = [] for _ in range(5): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(5) t = now_ms() s.connect((ip, 443)) rtts.append(now_ms() - t) s.close() s = socket.create_connection((ip, 443), timeout=5) t = now_ms() tls = ctx.wrap_socket(s, server_hostname=host) tls_ms = now_ms() - t tls.close() print( f"{host:36} ip={ip:16} dns={dns_ms:6.1f}ms tcp_connect min={min(rtts):5.1f} " f"median={statistics.median(rtts):5.1f}ms tls_handshake={tls_ms:5.1f}ms" ) # -------------------------------------------------------------------------------------- # Summary # -------------------------------------------------------------------------------------- def pct(values: list[float], p: float) -> float | None: return float(np.percentile(values, p)) if values else None def summary_rows(paths: list[Path]) -> list[dict]: rows = [r for p in paths for r in csv.DictReader(p.open())] num = lambda r, k: float(r[k]) if r.get(k) not in (None, "", "None") else None # noqa: E731 out = [] groups = dict.fromkeys((r.get("window", ""), r["vantage"]) for r in rows) for window, vantage in groups: for series in ("warm", "cold"): for target in TARGET_ORDER + sorted({r["target"] for r in rows} - set(TARGET_ORDER)): sel = [ r for r in rows if r.get("window", "") == window and r["vantage"] == vantage and r["target"] == target and r["series"] == series ] if not sel: continue ok = [r for r in sel if r["ok"] == "True"] vals = lambda k: [v for v in (num(r, k) for r in ok) if v is not None] # noqa: E731 started = sorted(r["started_utc"] for r in sel) out.append( { "window": window, "vantage": vantage, "series": series, "target": target, "transport": sel[0]["transport"], "ok": len(ok), "sent": len(sel), "firstByteP50": pct(vals("ttfb_ms"), 50), "firstByteP90": pct(vals("ttfb_ms"), 90), "firstAudibleP50": pct(vals("first_audible_ms"), 50), "firstAudibleP90": pct(vals("first_audible_ms"), 90), "leadingSilenceP50": pct(vals("leading_silence_ms"), 50), "totalP50": pct(vals("total_ms"), 50), "lastAudioP50": pct(vals("last_audio_ms"), 50), "newConnections": sum(1 for r in sel if r["new_connection"] == "True"), "peerIps": sorted({r["peer_ip"] for r in sel if r["peer_ip"]}), "fromUtc": started[0], "toUtc": started[-1], "errors": [f"{r['series']} #{r['seq']}: {r['error'][:200]}" for r in sel if r["ok"] != "True"], } ) return out def summarize(paths: list[Path]) -> str: fmt = lambda v: "-" if v is None else f"{v:.0f}" # noqa: E731 out = [] srows = summary_rows(paths) for window, vantage in dict.fromkeys((s["window"], s["vantage"]) for s in srows): for series in ("warm", "cold"): sel = [s for s in srows if s["window"] == window and s["vantage"] == vantage and s["series"] == series] if not sel: continue span = f"{min(s['fromUtc'] for s in sel)} to {max(s['toUtc'] for s in sel)}" heading = ", ".join(filter(None, [window, vantage, series])) out.append(f"\n### {heading} ({span} UTC)\n") out.append( "| target | n ok/sent | first byte p50 / p90 | first audible p50 / p90 | leading silence p50 | total p50 " "| last audio p50 | new conns | IPs |" ) out.append("|---|---|---|---|---|---|---|---|---|") for s in sel: out.append( f"| {s['target']} | {s['ok']}/{s['sent']} | {fmt(s['firstByteP50'])} / {fmt(s['firstByteP90'])} | " f"{fmt(s['firstAudibleP50'])} / {fmt(s['firstAudibleP90'])} | {fmt(s['leadingSilenceP50'])} | " f"{fmt(s['totalP50'])} | {fmt(s['lastAudioP50'])} | {s['newConnections']} | {', '.join(s['peerIps'])} |" ) errs = [f"- {' '.join(filter(None, [s['window'], s['vantage'], s['target']]))} {e}" for s in srows for e in s["errors"]] out.append("\n### errors\n") out.extend(errs or ["none"]) return "\n".join(out) def main() -> None: ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("net") p = sub.add_parser("probe") p.add_argument("--only", nargs="*") p.add_argument("--texts", nargs="*", help="measured text ids, e.g. m0 m4") p.add_argument("--audio-dir") p.add_argument("--repeat", type=int, default=1) p.add_argument("--gap", type=float, default=PAUSE_BETWEEN_REQUESTS_S, help="seconds between requests") r = sub.add_parser("run") r.add_argument("--vantage", required=True, help="where this runs, e.g. zurich or us-east4") r.add_argument("--window", default="main", help="names this run's time window; summarize keeps windows apart") r.add_argument("--out", default=str(HERE / "results.csv")) r.add_argument("--only", nargs="*") r.add_argument("--warm", type=int, default=WARM_REQUESTS) r.add_argument("--cold", type=int, default=COLD_REQUESTS) r.add_argument("--elevenlabs-char-cap", type=int, default=20000, help="stop ElevenLabs past this many characters (0: no cap)") r.add_argument( "--pause", type=float, default=PAUSE_BETWEEN_REQUESTS_S, help="seconds after each request; raise it with --only to keep a full round's idle time" ) s = sub.add_parser("summarize") s.add_argument("paths", nargs="+") s.add_argument("--json", action="store_true") args = ap.parse_args() if args.cmd == "net": net() elif args.cmd == "probe": asyncio.run( probe( set(args.only) if args.only else None, args.texts, Path(args.audio_dir) if args.audio_dir else None, args.repeat, args.gap, ) ) elif args.cmd == "run": asyncio.run( run( Path(args.out), args.window, args.vantage, set(args.only) if args.only else None, args.warm, args.cold, args.elevenlabs_char_cap, args.pause, ) ) elif args.cmd == "summarize": paths = [Path(p) for p in args.paths] print(json.dumps(summary_rows(paths), indent=1) if args.json else summarize(paths)) if __name__ == "__main__": sys.exit(main())