From c31a57bc09217a08bbf78c58270e04ccc51a5baa Mon Sep 17 00:00:00 2001 From: Gideon Jones <109116843+Gideonjon@users.noreply.github.com> Date: Fri, 10 Jul 2026 15:50:18 +0100 Subject: [PATCH 1/4] feat(vllm-realtime): add realtime speech transcription example Implements GitHub issue #2: audio ingest via Trickle, transcript events streamed back to the client over an orchestrator-proxied WebSocket, live settings via mid-stream session.update, mock backend (no GPU), and the real vLLM Voxtral backend. Verified end-to-end offchain with both backends: mock on a GPU-free Docker host, and vLLM 0.24 serving Voxtral-Mini-4B-Realtime-2602 on an RTX 4090 with live speech transcribed as streaming word deltas. Key implementation notes: - Runner subscribes to Trickle via the channel's internal_url; the public url goes to the client (the advertised serviceAddr is not reachable from inside the runner's network). - vLLM /v1/realtime protocol (v0.24): model at the top level of session.update, generation started/flushed via input_audio_buffer .commit final=false/true, transcription.delta/done events, empty-text deltas filtered. - vllm compose service caps --max-model-len at 16384 so the KV cache fits next to the weights on 24 GB cards. Also adds platform: linux/amd64 to compose.orchestrator.yml for Apple Silicon Mac compatibility. --- README.md | 17 +- compose.orchestrator.yml | 1 + vllm-realtime/.env.example | 32 +++ vllm-realtime/Dockerfile | 22 ++ vllm-realtime/FEEDBACK.md | 84 +++++++ vllm-realtime/README.md | 100 ++++++++ vllm-realtime/client.py | 226 +++++++++++++++++ vllm-realtime/docker-compose.onchain.yml | 33 +++ vllm-realtime/docker-compose.yml | 77 ++++++ vllm-realtime/pyproject.toml | 13 + vllm-realtime/runner.py | 297 ++++++++++++++++++++++ vllm-realtime/transcriber.py | 302 +++++++++++++++++++++++ 12 files changed, 1196 insertions(+), 8 deletions(-) create mode 100644 vllm-realtime/.env.example create mode 100644 vllm-realtime/Dockerfile create mode 100644 vllm-realtime/FEEDBACK.md create mode 100644 vllm-realtime/README.md create mode 100644 vllm-realtime/client.py create mode 100644 vllm-realtime/docker-compose.onchain.yml create mode 100644 vllm-realtime/docker-compose.yml create mode 100644 vllm-realtime/pyproject.toml create mode 100644 vllm-realtime/runner.py create mode 100644 vllm-realtime/transcriber.py diff --git a/README.md b/README.md index ed06368..bb2a3b3 100644 --- a/README.md +++ b/README.md @@ -30,8 +30,8 @@ The orchestrator is a **transparent reverse proxy**: every endpoint you expose i - **HTTP** request/response — the common case. (`hello-world`, `tiles`) - **HTTP + SSE** — streamed / token responses. (`vllm`) -- **Trickle** — continuous realtime video in/out. (`echo`) -- **WebSocket** — long-lived bidirectional sessions. (external: `scope`) +- **Trickle** — continuous realtime media in/out. (`echo`, `vllm-realtime`) +- **WebSocket** — long-lived bidirectional sessions. (`vllm-realtime`; external: `scope`) Need a schema that isn't here? [Open an issue](https://github.com/livepeer/live-runner-example-apps/issues). @@ -40,12 +40,13 @@ Need a schema that isn't here? [Open an issue](https://github.com/livepeer/live- ## Examples -| Example | Goal | Registration | Mode | Transport | -| ------------------------------ | ----------------------------------------------- | ------------ | ---------------------------------- | ----------------- | -| [`hello-world`](./hello-world) | The simplest app: one request, one response | dynamic | persistent (single-shot by nature) | HTTP (JSON) | -| [`tiles`](./tiles) | Capacity fan-out — one session per tile | dynamic | persistent (single-shot by nature) | HTTP (base64 PNG) | -| [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | -| [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | persistent (single-shot by nature) | HTTP + SSE | +| Example | Goal | Registration | Mode | Transport | +| ---------------------------------- | ----------------------------------------------- | ------------ | ---------------------------------- | ---------------------------- | +| [`hello-world`](./hello-world) | The simplest app: one request, one response | dynamic | persistent (single-shot by nature) | HTTP (JSON) | +| [`tiles`](./tiles) | Capacity fan-out — one session per tile | dynamic | persistent (single-shot by nature) | HTTP (base64 PNG) | +| [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | +| [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | persistent (single-shot by nature) | HTTP + SSE | +| [`vllm-realtime`](./vllm-realtime) | Realtime speech-to-text with live metrics | dynamic | persistent | trickle in + WebSocket out | Start with `hello-world` (the smallest end-to-end path); the others each layer on one new idea. More will follow, including a full example that exercises every feature. Each is self-contained and runs **offchain** (free, no wallet); most also run **on-chain** (paid) — see each README. diff --git a/compose.orchestrator.yml b/compose.orchestrator.yml index 3a985d2..8ee5196 100644 --- a/compose.orchestrator.yml +++ b/compose.orchestrator.yml @@ -14,6 +14,7 @@ services: orchestrator: image: livepeer/go-livepeer:ja-live-runner + platform: linux/amd64 container_name: example_apps_orchestrator # Two distinct addresses: # serviceAddr advertised to external clients in discovery responses, diff --git a/vllm-realtime/.env.example b/vllm-realtime/.env.example new file mode 100644 index 0000000..79ec606 --- /dev/null +++ b/vllm-realtime/.env.example @@ -0,0 +1,32 @@ +# Copy to .env (gitignored) and fill in. Never commit secrets. +# Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only. + +# Transcriber backend: mock (CPU, laptop-friendly) | vllm (needs an NVIDIA GPU +# and the `vllm` compose profile). +TRANSCRIBER=mock +# Realtime model vLLM serves (only used when TRANSCRIBER=vllm). +VLLM_REALTIME_MODEL=mistralai/Voxtral-Mini-4B-Realtime-2602 +# Required for the gated Voxtral model on real runs. +HUGGING_FACE_HUB_TOKEN= + +# --- On-chain (paid) only below; offchain ignores these. --- + +NETWORK=arbitrum-one-mainnet +ETH_RPC_URL=https://arb1.arbitrum.io/rpc + +# Signer (payer): needs an on-chain deposit + reserve. +SIGNER_KEYSTORE_DIR=/absolute/path/to/signer-keystore +SIGNER_ETH_ACCT=0xYourSignerAddress +SIGNER_ETH_PASSWORD=your-signer-keystore-password + +# Orchestrator operating key (split-key): needs ETH for gas to redeem tickets. +ORCH_KEYSTORE_DIR=/absolute/path/to/operator-keystore +ORCH_ETH_ACCT=0xYourOperatorAddress +ORCH_ETH_PASSWORD=your-operator-keystore-password +# Registered orch = ticket recipient (-ethOrchAddr); empty = use the operating key. +ORCH_ONCHAIN_ADDR=0xYourRegisteredOrchestrator + +# Pricing (on-chain), in USD, converted to wei via the price feed. +PRICE_PER_UNIT=1 +PIXELS_PER_UNIT=1000 +MAX_PRICE_PER_UNIT=0.10USD diff --git a/vllm-realtime/Dockerfile b/vllm-realtime/Dockerfile new file mode 100644 index 0000000..142d165 --- /dev/null +++ b/vllm-realtime/Dockerfile @@ -0,0 +1,22 @@ +# vllm-realtime example app (the transcription bridge, not vLLM itself). +FROM python:3.12-slim + +RUN apt-get update \ + && apt-get install -y --no-install-recommends git \ + && rm -rf /var/lib/apt/lists/* + +# livepeer-gateway SDK isn't on PyPI yet; install from Git. Trickle lives on the +# ja/live-runner branch. `websockets` is for the vLLM realtime backend; `av`, +# `aiohttp` come transitively with the SDK. +RUN pip install --no-cache-dir \ + "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@ja/live-runner" \ + websockets + +WORKDIR /app +COPY runner.py transcriber.py client.py ./ + +EXPOSE 5000 + +# ENTRYPOINT (not CMD) so the compose `command:` passes runner flags straight +# through, matching the other examples in this repo. +ENTRYPOINT ["python", "runner.py"] diff --git a/vllm-realtime/FEEDBACK.md b/vllm-realtime/FEEDBACK.md new file mode 100644 index 0000000..0eab758 --- /dev/null +++ b/vllm-realtime/FEEDBACK.md @@ -0,0 +1,84 @@ +# Developer experience feedback — building `vllm-realtime` + +Notes from building this example against the `ja/live-runner` branch of +go-livepeer and `livepeer-python-gateway`. Goal: capture what was smooth, what +was confusing, and the open questions a second builder would hit. + +## What worked well + +- **The example apps are a great on-ramp.** `hello-world` (dynamic registration) + and `echo` (Trickle in/out) together gave a near-complete template. + `create_trickle_channels(request, [...])` returning `{name, url, internal_url, + ...}` dicts is clean, and raw WebSocket passthrough made the transcript output + path a plain aiohttp handler. +- **Mock-first development is viable.** Because the app only needs raw PCM in and + text out, a GPU-free mock backend exercises the entire Trickle path on a laptop. + This decoupled "is my Livepeer plumbing correct?" from "is vLLM configured?", + which made iteration fast. +- **Offchain → onchain is a thin overlay.** Reusing the shared + `compose.orchestrator.yml` / `compose.onchain.yml` via `extends` meant the paid + path was just a price flag + signer service, with zero app code changes. + +## Friction / confusion + +1. **Use `internal_url`, not `url`, for the runner-side subscribe.** The + orchestrator advertises its external `serviceAddr` (e.g. `127.0.0.1:8935`) in + the public channel `url`; from inside the app container that loops back to the + app itself and the subscriber dies with connection-refused after max retries. + `create_trickle_channels` also returns a runner-reachable `internal_url` — + subscribe on that, hand `url` to the client. `echo` does this correctly, but + nothing shouts about it and the failure mode (a subscriber that never sees a + segment) is silent until you read orchestrator logs. +2. **Trickle streams are deleted when the publisher closes — unread segments are + dropped.** Publishing faster than realtime (`--no-realtime`) pushes all + segments and deletes the stream within milliseconds; a subscriber that + connects even a beat later gets nothing. Paced (realtime) publishing works + because pub and sub overlap. Worth documenting whether the orchestrator can + buffer/retain segments for late subscribers. +3. **Importing the SDK pulls in PyAV.** `livepeer_gateway/__init__.py` imports the + media modules at top level, so `import livepeer_gateway` (even just for + Trickle) requires `av`. It installs transitively, but a from-source build on a + minimal image can be surprised by it. +4. **Trickle mime types are undocumented for non-video.** `echo` uses + `video/mp2t`. It's unclear whether the orchestrator validates mime or treats it + as opaque relay metadata. This example uses `audio/raw` in and it relays fine; + the contract should still be documented. +5. **vLLM `/v1/realtime` has its own protocol — it is not OpenAI Realtime.** + Validated against vLLM 0.24.0: `session.update` requires `model` at the *top + level* of the event (a nested `session` object silently fails validation and + every later commit errors with `model_not_validated`); generation starts on + `input_audio_buffer.commit` with `final: false` and the stream is flushed with + `final: true`; output events are `transcription.delta` / `transcription.done`. + Voxtral also interleaves empty-text deltas between words — filter them. +6. **Payment granularity for long streams is unclear.** `call_runner`'s 402 + challenge pays once at session reservation. Whether one payment authorizes an + arbitrarily long transcription stream, or the orchestrator expects ongoing + per-segment payment (as `lv2v.py` does with a periodic `send_payment` loop + against the output channel), needs confirmation. This example uses + single-payment-at-reserve; if long sessions get dropped for non-payment, port + the `lv2v` per-segment payment loop into the client. *(Plan R4)* +7. **Serving Voxtral on a 24 GB card needs `--max-model-len`.** The model's + native 131k context wants a ~4 GiB KV cache that doesn't fit beside the + weights on an RTX 4090; the engine refuses to start. `--max-model-len 16384` + is plenty for realtime transcription. Also: the PyPI `vllm` wheel ships + CUDA-13 torch — on hosts with a 12.x driver (common on cloud GPU rentals) + install the `+cu129` wheel from GitHub releases instead, and + `mistral-common[soundfile]` is required at runtime for audio decode. + +## Suggestions + +- Add a "transports" page: HTTP / SSE / **Trickle** with a one-paragraph "when to + use each" and the raw-bytes (`TricklePublisher`/`TrickleSubscriber`) vs AV + (`MediaPublish`/`MediaOutput`) distinction. +- Document the trickle channel mime-type contract (validated vs opaque). +- Document the streaming payment model (one-shot vs per-segment) end to end. +- Consider splitting the `av` import so Trickle-only apps don't need PyAV. + +## Environment + +- Mock path: built on macOS, verified end-to-end on a Linux Docker host. +- Real path: verified end-to-end with vLLM 0.24.0 (`+cu129` wheel) serving + `mistralai/Voxtral-Mini-4B-Realtime-2602` on an RTX 4090 (24 GB, driver + 570.x/CUDA 12.8). Live speech was transcribed with word-by-word deltas + streaming over the orchestrator-proxied WebSocket while audio was still + being published. diff --git a/vllm-realtime/README.md b/vllm-realtime/README.md new file mode 100644 index 0000000..ab79ee1 --- /dev/null +++ b/vllm-realtime/README.md @@ -0,0 +1,100 @@ +# vllm-realtime app + +Realtime speech transcription on the Livepeer network. The client streams audio +in over **Trickle** and receives the live transcript back over an +orchestrator-proxied **WebSocket**, plus derived metrics (word count + a simple +sentiment label) computed on the running text. Settings (e.g. language) can be +adjusted live over the same WebSocket without restarting the stream. + +| | | +| ------------ | ------------------------------------------------------ | +| App id | `livepeer-sample/vllm-realtime` | +| Transport | Trickle in (PCM audio), WebSocket out (JSON events) | +| Registration | dynamic (self-registers via the SDK) | +| Ports | 5000 (app), 8000 (vLLM) | + +Prerequisites (Docker, `uv`, and the not-yet-released `livepeer-gateway` SDK — +pinned in `pyproject.toml`) and the shared on-chain/payment setup live in the +[repo README](../README.md). + +## How it works + +``` +client orchestrator app (runner.py) vLLM | mock +publish PCM ─> in channel ─── trickle ───> subscribe (internal_url) + feed audio ─ local ws ─> /v1/realtime (GPU) + transcript + metrics or mock (no GPU) + <── JSON events ─────────── websocket ── /ws (bidirectional) + ──> {"type":"session.update", ...} ────> live settings, forwarded to vLLM +``` + +`POST /transcribe` mints a Trickle `in` channel. The client publishes PCM16/16 kHz +audio to it; the app subscribes (via the channel's runner-reachable +`internal_url`), bridges to a **local** vLLM `/v1/realtime` WebSocket +(co-located in the same compose, never proxied), and streams transcript deltas + +metrics to the client over `GET /ws` — a raw WebSocket the orchestrator proxies +in both directions. The client can send `{"type": "session.update", "session": +{...}}` on that socket at any time to adjust settings mid-stream. See +[runner.py](runner.py) and [transcriber.py](transcriber.py). + +### Backends + +- **`mock`** (default) — no GPU, no vLLM. Fabricates plausible, time-paced + transcription events so the whole Trickle pipeline runs on a laptop. +- **`vllm`** — opens the real vLLM realtime WebSocket. Needs an NVIDIA GPU + (≥16 GB) and the `vllm` compose profile. + +Switch with the `TRANSCRIBER` env var. + +## Run offchain (free) + +GPU-free, using the mock backend — only the orchestrator and app start: + +```sh +docker compose up -d --build +uv run client.py --discovery https://localhost:8935/discovery +# [delta] +'hello' words=1 sentiment=neu +# ... +# [done] 'hello and welcome to the livepeer realtime transcription demo ...' words=... sentiment=pos +docker compose down +``` + +The client synthesizes a few seconds of audio by default; pass a 16 kHz mono +16-bit WAV with `--input path.wav` to stream a real file. Pass `--language en` +to demonstrate a live `session.update` over the WebSocket after it connects. + +### Real transcription (GPU box) + +```sh +TRANSCRIBER=vllm docker compose --profile vllm up -d --build +docker compose logs -f vllm # wait for /health to pass (model download) +uv run client.py --input speech-16k.wav --discovery https://localhost:8935/discovery +``` + +Validated on an RTX 4090 (24 GB): the compose file caps `--max-model-len` at +16384 so the KV cache fits next to the weights on 24 GB cards (the model's +native 131k context does not). The Voxtral model is public on HuggingFace — no +token needed. + +## Run on-chain (paid) + +Layer `docker-compose.onchain.yml` to add a remote signer and run the +orchestrator on-chain, so the app advertises a price and the SDK pays per +session. Needs an Ethereum RPC, a funded signer wallet (deposit + reserve), and +an orchestrator wallet — see [On-chain (paid) setup](../README.md#on-chain-paid-setup). + +```sh +cp .env.example .env # fill in RPC, network, keystore paths, accounts, pricing +docker compose -f docker-compose.yml -f docker-compose.onchain.yml up -d --build +# confirm the price is advertised (price_per_unit != 0): +curl -sk https://localhost:8935/discovery | jq +uv run client.py --discovery https://localhost:8935/discovery --signer http://localhost:7936 +docker compose -f docker-compose.yml -f docker-compose.onchain.yml down +``` + +Add `--profile vllm` to the `up` command on a GPU box for real transcription. + +> [!NOTE] +> Payment is settled once at session reservation via the SDK's 402 challenge. +> For very long streams the orchestrator may expect ongoing per-segment payment; +> see [FEEDBACK.md](FEEDBACK.md) (R4) for the open question and the fallback. diff --git a/vllm-realtime/client.py b/vllm-realtime/client.py new file mode 100644 index 0000000..9acc7f9 --- /dev/null +++ b/vllm-realtime/client.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""vllm-realtime client: stream audio in over Trickle, receive transcript over WebSocket. + +Flow +---- +1. reserve_session via discovery (optionally on-chain with --signer) +2. POST /transcribe {language} → {in: , ws: "/ws"} +3. Connect WebSocket to wss://orchestrator/ws to receive transcript events +4. Optionally send {"type": "session.update", "session": {...}} mid-stream + to adjust settings (language etc.) without stopping the stream +5. Publish PCM16/16 kHz audio to the Trickle in channel, paced to real time +6. Drain transcript events from the WebSocket until "done" +7. Release the session + +Pass --signer for the on-chain (paid) path. +Pass --language to set the transcription language (e.g. "en", "fr"). +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import math +import ssl +import struct +import wave +from contextlib import suppress + +import aiohttp + +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.http import post_json +from livepeer_gateway.live_runner import stop_runner_session +from livepeer_gateway.selection import reserve_session +from livepeer_gateway.trickle_publisher import TricklePublisher + +DEFAULT_DISCOVERY = "http://localhost:8935/discovery" +APP_ID = "livepeer-sample/vllm-realtime" + +SAMPLE_RATE = 16000 +BYTES_PER_SECOND = SAMPLE_RATE * 2 # PCM16 mono +FRAME_SECONDS = 0.04 # 40 ms frames per Trickle write +SEGMENT_SECONDS = 0.5 # one Trickle segment per ~0.5 s of audio +IN_MIME = "audio/raw" + +log = logging.getLogger("vllm-realtime-client") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Run the vllm-realtime Live Runner demo.") + parser.add_argument("--discovery", default=DEFAULT_DISCOVERY) + parser.add_argument( + "--input", default="", + help="Path to a 16 kHz mono 16-bit PCM WAV. Omit to synthesize a test tone.", + ) + parser.add_argument( + "--seconds", type=float, default=6.0, + help="Seconds of synthesized audio when --input is omitted.", + ) + parser.add_argument( + "--no-realtime", action="store_true", + help="Publish as fast as possible instead of pacing to wall clock.", + ) + parser.add_argument( + "--language", default="", + help="Transcription language code (e.g. 'en', 'fr'). Sent as a live " + "session.update after connecting — demonstrates mid-stream settings.", + ) + parser.add_argument( + "--signer", default="", + help="Remote signer base URL for the on-chain (paid) path.", + ) + return parser.parse_args() + + +def _load_pcm(path: str) -> bytes: + """Read a WAV as raw PCM16 mono bytes; warn if format differs from 16 kHz/mono/16-bit.""" + with wave.open(path, "rb") as wav: + channels = wav.getnchannels() + width = wav.getsampwidth() + rate = wav.getframerate() + frames = wav.readframes(wav.getnframes()) + if (channels, width, rate) != (1, 2, SAMPLE_RATE): + log.warning( + "input is %d ch / %d-bit / %d Hz; expected mono/16-bit/16 kHz. " + "Fine for mock; resample for real vLLM.", + channels, width * 8, rate, + ) + return frames + + +def _synthesize_pcm(seconds: float) -> bytes: + """A quiet 220 Hz tone — gives the pipeline real bytes to move (mock-friendly).""" + total = int(seconds * SAMPLE_RATE) + out = bytearray() + for n in range(total): + sample = int(3000 * math.sin(2 * math.pi * 220 * n / SAMPLE_RATE)) + out += struct.pack(" ssl.SSLContext: + """TLS context that skips verification — orchestrator uses a self-signed cert.""" + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + return ctx + + +async def _read_transcript( + ws: aiohttp.ClientWebSocketResponse, + done: asyncio.Event, +) -> None: + """Print transcript + metrics as JSON events arrive on the WebSocket.""" + try: + async for msg in ws: + if msg.type != aiohttp.WSMsgType.TEXT: + break + try: + data = json.loads(msg.data) + except ValueError: + continue + kind = data.get("type") + transcript = data.get("transcript", "") + if kind == "done": + print( + f"\n[done] {transcript!r} " + f"words={data.get('word_count')} sentiment={data.get('sentiment')}" + ) + break + print( + f"[delta] +{data.get('delta', '').strip()!r} " + f"words={data.get('word_count')} sentiment={data.get('sentiment')}" + ) + except Exception as exc: + log.warning("transcript reader ended: %s", exc) + finally: + done.set() + + +async def _publish_pcm(in_url: str, pcm: bytes, realtime: bool) -> None: + """Publish PCM to the Trickle input channel, ~0.5 s per segment, 40 ms per frame.""" + frame_bytes = int(FRAME_SECONDS * BYTES_PER_SECOND) + seg_bytes = int(SEGMENT_SECONDS * BYTES_PER_SECOND) + async with TricklePublisher(in_url, IN_MIME) as pub: + for seg_start in range(0, len(pcm), seg_bytes): + segment = pcm[seg_start : seg_start + seg_bytes] + writer = await pub.next() + async with writer: + for off in range(0, len(segment), frame_bytes): + await writer.write(segment[off : off + frame_bytes]) + if realtime: + await asyncio.sleep(FRAME_SECONDS) + + +async def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + args = _parse_args() + signer_url = args.signer.strip() or None + + pcm = _load_pcm(args.input) if args.input else _synthesize_pcm(args.seconds) + log.info("audio: %.1f s (%d bytes)", len(pcm) / BYTES_PER_SECOND, len(pcm)) + + ssl_ctx = _make_ssl_ctx() + session = None + try: + session = await reserve_session( + discovery_url=args.discovery, + app=APP_ID, + signer_url=signer_url, + ) + log.info("session_id=%s app_url=%s", session.session_id, session.app_url) + + # POST /transcribe with optional language — runner mints the Trickle in channel. + result = await post_json( + session.app_url.rstrip("/") + "/transcribe", + {"language": args.language}, + ) + in_url = result["in"] + + # Build the WebSocket URL from the session's app_url (orchestrator proxies it). + ws_url = ( + session.app_url + .replace("https://", "wss://") + .replace("http://", "ws://") + .rstrip("/") + "/ws" + ) + log.info("trickle in=%s ws=%s", in_url, ws_url) + + async with aiohttp.ClientSession() as http: + async with http.ws_connect(ws_url, ssl=ssl_ctx, heartbeat=20) as ws: + # Send a live session.update immediately after connecting. + # This demonstrates mid-stream settings adjustment: language can be + # changed here (or at any point while audio is still flowing) without + # restarting the stream. + if args.language: + await ws.send_str( + json.dumps({ + "type": "session.update", + "session": {"language": args.language}, + }) + ) + log.info("sent live session.update language=%r", args.language) + + done = asyncio.Event() + reader = asyncio.create_task(_read_transcript(ws, done)) + + await _publish_pcm(in_url, pcm, realtime=not args.no_realtime) + + with suppress(asyncio.TimeoutError): + await asyncio.wait_for(done.wait(), timeout=30) + reader.cancel() + with suppress(Exception): + await reader + + except LivepeerGatewayError as exc: + raise SystemExit(f"ERROR: {exc}") from exc + finally: + if session is not None: + with suppress(Exception): + await stop_runner_session(session) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/vllm-realtime/docker-compose.onchain.yml b/vllm-realtime/docker-compose.onchain.yml new file mode 100644 index 0000000..d6b9c1a --- /dev/null +++ b/vllm-realtime/docker-compose.onchain.yml @@ -0,0 +1,33 @@ +# On-chain payment overlay for vllm-realtime. Layer it on the offchain base: +# docker compose -f docker-compose.yml -f docker-compose.onchain.yml up -d --build +# # add --profile vllm on a GPU box for real transcription +# +# Adds the shared remote signer, re-points the orchestrator on-chain (see +# ../compose.onchain.yml), and registers the app with a price so the +# orchestrator issues a payment challenge. Requires a local .env (gitignored); +# copy .env.example and fill it in. Then pay through the signer: +# uv run client.py --discovery https://localhost:8935/discovery \ +# --signer http://localhost:7936 + +services: + signer: + extends: + file: ../compose.onchain.yml + service: signer + ports: + - "7936:7936" + + orchestrator: + extends: + file: ../compose.onchain.yml + service: orchestrator + + # Re-declare the command to advertise a price (base file registers free). + app: + command: + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:5000 + - --price=${PRICE_PER_UNIT} + - --pixels-per-unit=${PIXELS_PER_UNIT} diff --git a/vllm-realtime/docker-compose.yml b/vllm-realtime/docker-compose.yml new file mode 100644 index 0000000..97db8f2 --- /dev/null +++ b/vllm-realtime/docker-compose.yml @@ -0,0 +1,77 @@ +# End-to-end offchain demo: orchestrator + the vllm-realtime app. +# +# The orchestrator is defined once in ../compose.orchestrator.yml and pulled in +# with `extends`; this file adds the app and (optionally) a vLLM server. +# +# Default run is GPU-FREE: the app uses the mock transcriber, so only the +# orchestrator + app start. The real vLLM server is gated behind the `vllm` +# profile because it needs an NVIDIA GPU. +# +# # mock (no GPU, e.g. a laptop): +# docker compose up -d --build +# uv run client.py --discovery https://localhost:8935/discovery +# +# # real transcription (GPU box): +# TRANSCRIBER=vllm docker compose --profile vllm up -d --build + +services: + orchestrator: + extends: + file: ../compose.orchestrator.yml + service: orchestrator + + app: + build: . + container_name: example_apps_vllm_realtime_app + # Wait for the orchestrator's healthcheck so registration doesn't race its boot. + depends_on: + orchestrator: + condition: service_healthy + environment: + # mock (default, GPU-free) | vllm (needs the `vllm` profile + a GPU) + - TRANSCRIBER=${TRANSCRIBER:-mock} + - VLLM_REALTIME_URL=ws://vllm:8000/v1/realtime + - VLLM_REALTIME_MODEL=${VLLM_REALTIME_MODEL:-mistralai/Voxtral-Mini-4B-Realtime-2602} + command: + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:5000 + + vllm: + # Real realtime transcription server. Started only with `--profile vllm`. + image: vllm/vllm-openai:latest + container_name: example_apps_vllm_realtime + profiles: ["vllm"] + command: + - "--model" + - "${VLLM_REALTIME_MODEL:-mistralai/Voxtral-Mini-4B-Realtime-2602}" + - "--port" + - "8000" + # Launch flags validated on an RTX 4090 (24 GB). The model's native 131k + # context needs a ~4 GiB KV cache that does not fit next to the weights + # on a 24 GB card — cap the context or the engine refuses to start. + # 16k tokens is far more than a realtime transcription session needs. + - "--max-model-len" + - "${VLLM_MAX_MODEL_LEN:-16384}" + - "--compilation_config" + - '{"cudagraph_mode": "PIECEWISE"}' + - "--gpu-memory-utilization" + - "${VLLM_GPU_MEM_UTIL:-0.9}" + environment: + # Voxtral is gated on HuggingFace; a token is required for real runs. + - HUGGING_FACE_HUB_TOKEN=${HUGGING_FACE_HUB_TOKEN:-} + - VLLM_DISABLE_COMPILE_CACHE=1 + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: 1 + capabilities: [gpu] + healthcheck: + test: ["CMD", "curl", "-fsS", "http://localhost:8000/health"] + interval: 5s + timeout: 3s + retries: 60 # model download + load can take a while + start_period: 20s diff --git a/vllm-realtime/pyproject.toml b/vllm-realtime/pyproject.toml new file mode 100644 index 0000000..e68f1da --- /dev/null +++ b/vllm-realtime/pyproject.toml @@ -0,0 +1,13 @@ +[project] +name = "livepeer-vllm-realtime" +version = "0.1.0" +description = "Realtime speech transcription example app for the Livepeer network." +requires-python = ">=3.12" +dependencies = [ + "livepeer-gateway", + "websockets", +] + +# livepeer-gateway is not on PyPI yet; pull it from the branch (Trickle lives here). +[tool.uv.sources] +livepeer-gateway = { git = "https://github.com/livepeer/livepeer-python-gateway", branch = "ja/live-runner" } diff --git a/vllm-realtime/runner.py b/vllm-realtime/runner.py new file mode 100644 index 0000000..1d34496 --- /dev/null +++ b/vllm-realtime/runner.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python3 +"""vllm-realtime runner: realtime speech transcription over Trickle (in) + WebSocket (out). + +Endpoints +--------- +POST /transcribe + Mint a Trickle audio-input channel, start the transcription pipeline. + Returns {"in": , "ws": "/ws"}. + +GET /ws + WebSocket: the client receives transcript + metrics as JSON events and may + send {"type": "session.update", "session": {...}} mid-stream to adjust + settings (language, model, etc.) without stopping the stream. + +Architecture +------------ +The orchestrator proxies both Trickle segments and WebSocket upgrades, so the +client uses Trickle only for the audio upload path (where chunked, paced +segments are natural) and a standard WebSocket for the output path (where +bidirectional, event-oriented framing fits transcript deltas and settings +updates). + + client ──PCM16 Trickle──> orchestrator ──> TrickleSubscriber + │ + Transcriber (local vLLM ws | mock) + │ + client <──WS JSON events── orchestrator <── event_queue +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +from contextlib import suppress +from dataclasses import dataclass, field +from typing import Optional + +from aiohttp import web + +from livepeer_gateway.live_runner import register_runner +from livepeer_gateway.trickle_subscriber import TrickleSubscriber + +from transcriber import Transcriber, make_transcriber, simple_sentiment, word_count + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 5000 +APP_ID = "livepeer-sample/vllm-realtime" + +IN_MIME = "audio/raw" + +# Sentinel pushed to the event queue when the pipeline is finished. +_WS_DONE = object() + +log = logging.getLogger("vllm-realtime") + + +@dataclass +class TranscribeSession: + session_id: str + in_url: str + event_queue: asyncio.Queue = field(default_factory=asyncio.Queue) + task: Optional[asyncio.Task] = None + transcriber: Optional[Transcriber] = None + + def to_json(self) -> dict: + return {"session": self.session_id, "in": self.in_url, "ws": "/ws"} + + +# Single active session (capacity 1). +state: Optional[TranscribeSession] = None + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Live Runner vllm-realtime demo.") + parser.add_argument("--orchestrator", default="http://localhost:8935") + parser.add_argument("--orchSecret", default="abcdef") + parser.add_argument("--runner-url", default=f"http://{DEFAULT_HOST}:{DEFAULT_PORT}") + parser.add_argument("--host", default=DEFAULT_HOST, help="Bind address (use 0.0.0.0 in containers).") + parser.add_argument("--price", type=int, default=0) + parser.add_argument("--pixels-per-unit", type=int, default=1) + return parser.parse_args() + + +def _session_id(request: web.Request) -> str: + return request.headers.get("Livepeer-Session-Id", "").strip() + + +async def _pump_audio(in_url: str, transcriber: Transcriber) -> None: + """Read PCM segments off the Trickle input channel and feed the transcriber.""" + async with TrickleSubscriber(in_url) as sub: + while True: + segment = await sub.next() + if segment is None: + break + reader = segment.make_reader() + while True: + chunk = await reader.read() + if chunk is None: + break + await transcriber.feed(chunk) + with suppress(Exception): + await segment.close() + + +async def _drain_events(transcriber: Transcriber, event_queue: asyncio.Queue) -> None: + """Read normalized transcriber events, compute metrics, push to the WS queue.""" + transcript = "" + async for event in transcriber.events(): + if event["type"] == "delta": + if not event.get("text"): + # Voxtral emits empty-text deltas between words; skip them + # rather than spamming clients with no-op events. + continue + transcript += event.get("text", "") + else: + transcript = event.get("text", "") or transcript + clean = transcript.strip() + await event_queue.put( + { + "type": event["type"], + "delta": event.get("text", "") if event["type"] == "delta" else "", + "transcript": clean, + "word_count": word_count(clean), + "sentiment": simple_sentiment(clean), + } + ) + + +async def _run_pipeline( + in_url: str, + transcriber: Transcriber, + event_queue: asyncio.Queue, +) -> None: + """Wire input audio → transcriber → metrics → WS event queue.""" + drain: Optional[asyncio.Task] = None + try: + await transcriber.open() + drain = asyncio.create_task(_drain_events(transcriber, event_queue)) + await _pump_audio(in_url, transcriber) + except Exception: + log.exception("transcription pipeline failed") + finally: + with suppress(Exception): + await transcriber.close() + if drain is not None: + try: + await drain + except Exception: + # A dead drain means transcript events silently stop reaching + # clients — surface it instead of swallowing it. + log.exception("event drain task failed") + # Signal the WS drain coroutine that no more events are coming. + await event_queue.put(_WS_DONE) + log.info("transcription pipeline finished") + + +async def _drain_queue_to_ws(queue: asyncio.Queue, ws: web.WebSocketResponse) -> None: + """Forward queued transcript events to the WebSocket client.""" + try: + while True: + event = await queue.get() + if event is _WS_DONE: + await ws.close() + return + with suppress(Exception): + await ws.send_json(event) + except asyncio.CancelledError: + pass + + +async def _handle_transcribe(request: web.Request) -> web.Response: + global state + session_id = _session_id(request) + + if state is not None: + if state.session_id != session_id: + raise web.HTTPConflict(text="vllm-realtime runner already has an active session") + return web.json_response(state.to_json()) + + try: + body = await request.json() + except Exception: + body = {} + language: str = body.get("language", "") + + # Pass the request so the SDK opens channels using the orchestrator's + # Session-Control header, whose URLs are reachable from the runner's network. + channels = await request.app["registration"].create_trickle_channels( + request, + [{"name": "in", "mime_type": IN_MIME}], + ) + if not channels or "url" not in channels[0]: + raise web.HTTPInternalServerError(text="orchestrator did not return in channel") + # Public URL goes to the client; internal_url is the runner-reachable + # address (== public url when runner and orchestrator share a network). + in_url = channels[0]["url"] + in_internal_url = channels[0].get("internal_url", in_url) + + event_queue: asyncio.Queue = asyncio.Queue() + transcriber = make_transcriber(language=language) + + session = TranscribeSession( + session_id=session_id, + in_url=in_url, + event_queue=event_queue, + transcriber=transcriber, + ) + task = asyncio.create_task(_run_pipeline(in_internal_url, transcriber, event_queue)) + task.add_done_callback(_clear_state) + session.task = task + state = session + + log.info("started transcription session %s language=%r", session_id, language) + return web.json_response(state.to_json()) + + +async def _handle_ws(request: web.Request) -> web.WebSocketResponse: + """WebSocket output: stream transcript events; accept session.update settings.""" + if state is None: + raise web.HTTPNotFound(text="no active transcription session") + + ws = web.WebSocketResponse(heartbeat=20) + await ws.prepare(request) + + session = state + drain_task = asyncio.create_task(_drain_queue_to_ws(session.event_queue, ws)) + + try: + async for msg in ws: + if msg.type == web.WSMsgType.TEXT: + log.info("ws command received: %.200s", msg.data) + with suppress(Exception): + cmd = json.loads(msg.data) + if cmd.get("type") == "session.update" and session.transcriber is not None: + await session.transcriber.apply_settings(cmd.get("session", {})) + elif msg.type in (web.WSMsgType.ERROR, web.WSMsgType.CLOSE): + break + finally: + drain_task.cancel() + with suppress(Exception): + await drain_task + + return ws + + +def _clear_state(_task: asyncio.Task) -> None: + global state + state = None + + +async def _handle_health(_request: web.Request) -> web.Response: + return web.json_response({"ok": True}) + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + args = _parse_args() + + async def _on_startup(app: web.Application) -> None: + app["registration"] = await register_runner( + args.orchestrator, + secret=args.orchSecret, + runner_url=args.runner_url, + app=APP_ID, + price_per_unit=args.price, + pixels_per_unit=args.pixels_per_unit, + ) + log.info( + "registered runner_id=%s orchestrator=%s", + app["registration"].runner_id, + app["registration"].orchestrator_url, + ) + + async def _on_cleanup(app: web.Application) -> None: + if state is not None: + if state.task is not None: + state.task.cancel() + with suppress(Exception): + await state.task + if state.transcriber is not None: + with suppress(Exception): + await state.transcriber.close() + with suppress(Exception): + await app["registration"].close() + + app = web.Application() + app.router.add_post("/transcribe", _handle_transcribe) + app.router.add_get("/ws", _handle_ws) + app.router.add_get("/healthz", _handle_health) + app.on_startup.append(_on_startup) + app.on_cleanup.append(_on_cleanup) + web.run_app(app, host=args.host, port=DEFAULT_PORT, print=None) + + +if __name__ == "__main__": + main() diff --git a/vllm-realtime/transcriber.py b/vllm-realtime/transcriber.py new file mode 100644 index 0000000..270fd88 --- /dev/null +++ b/vllm-realtime/transcriber.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +"""Transcriber backends for the vllm-realtime example. + +The runner stays backend-agnostic: it feeds raw PCM in, reads normalized +events out, and can push live settings updates mid-stream. + +Shared interface (Protocol) +--------------------------- + open() # connect / start + feed(pcm: bytes) # PCM16LE mono 16 kHz audio chunk + events() -> AsyncIterator # yields {"type": "delta"|"done", "text": str} + apply_settings(dict) # forward a session.update mid-stream + close() # no more audio; flush final transcript, end events() + +Backends +-------- +- ``MockTranscriber`` — no GPU, no vLLM. Fabricates plausible, time-paced + transcription events from canned text so the whole + Trickle + WebSocket pipeline is exercisable on a laptop. + This is the default (``TRANSCRIBER=mock``). +- ``VllmRealtimeTranscriber`` — opens a *local* WebSocket to vLLM's ``/v1/realtime`` + endpoint (co-located in the same compose, never proxied + by the orchestrator) and translates its events into ours. + Use on a GPU box with ``TRANSCRIBER=vllm``. + +NOTE (R1): vLLM's realtime event names are still settling. The vLLM backend +checks several candidate event keys. Validate against your vLLM build. +""" +from __future__ import annotations + +import asyncio +import base64 +import json +import logging +import os +from contextlib import suppress +from typing import AsyncIterator, Optional, Protocol, runtime_checkable + +log = logging.getLogger("vllm-realtime.transcriber") + +# PCM16LE mono @ 16 kHz: 2 bytes/sample × 16000 samples/s = 32000 bytes/s. +PCM_BYTES_PER_SECOND = 16000 * 2 + +_END = object() # sentinel to close events() + + +@runtime_checkable +class Transcriber(Protocol): + async def open(self) -> None: ... + async def feed(self, pcm: bytes) -> None: ... + def events(self) -> AsyncIterator[dict]: ... + async def apply_settings(self, settings: dict) -> None: ... + async def close(self) -> None: ... + + +def make_transcriber(language: str = "") -> Transcriber: + """Pick a backend from the TRANSCRIBER env var (default: mock).""" + backend = os.getenv("TRANSCRIBER", "mock").strip().lower() + if backend == "vllm": + url = os.getenv("VLLM_REALTIME_URL", "ws://vllm:8000/v1/realtime") + model = os.getenv( + "VLLM_REALTIME_MODEL", "mistralai/Voxtral-Mini-4B-Realtime-2602" + ) + log.info("using vLLM realtime backend url=%s model=%s language=%r", url, model, language) + return VllmRealtimeTranscriber(url=url, model=model, language=language) + log.info("using mock transcriber (no GPU required)") + return MockTranscriber() + + +# --------------------------------------------------------------------------- # +# Mock backend +# --------------------------------------------------------------------------- # + +_MOCK_SCRIPT = ( + "hello and welcome to the livepeer realtime transcription demo " + "this transcript is produced by a mock backend so you can test the " + "trickle pipeline without a gpu it streams words as your audio is " + "consumed and reports a final result when the stream ends" +).split() + +_MOCK_WORDS_PER_SECOND = 3.0 + + +class MockTranscriber: + """Fake transcriber: emits canned words paced to the audio duration fed.""" + + def __init__(self, script: Optional[list[str]] = None) -> None: + self._script = list(script) if script is not None else list(_MOCK_SCRIPT) + self._queue: asyncio.Queue = asyncio.Queue() + self._bytes_fed = 0 + self._words_emitted = 0 + self._closed = False + + async def open(self) -> None: + pass + + async def feed(self, pcm: bytes) -> None: + if self._closed: + return + self._bytes_fed += len(pcm) + seconds = self._bytes_fed / PCM_BYTES_PER_SECOND + target = min(len(self._script), int(seconds * _MOCK_WORDS_PER_SECOND)) + while self._words_emitted < target: + word = self._script[self._words_emitted] + self._words_emitted += 1 + await self._queue.put({"type": "delta", "text": word + " "}) + + async def apply_settings(self, settings: dict) -> None: + # Mock ignores settings; log so the WS path is visibly exercised. + log.debug("mock: ignoring session.update %s", settings) + + async def close(self) -> None: + if self._closed: + return + self._closed = True + while self._words_emitted < len(self._script): + word = self._script[self._words_emitted] + self._words_emitted += 1 + await self._queue.put({"type": "delta", "text": word + " "}) + full = " ".join(self._script[: self._words_emitted]) + await self._queue.put({"type": "done", "text": full}) + await self._queue.put(_END) + + async def events(self) -> AsyncIterator[dict]: + while True: + item = await self._queue.get() + if item is _END: + return + yield item + + +# --------------------------------------------------------------------------- # +# vLLM realtime backend +# --------------------------------------------------------------------------- # + +# Candidate event names across vLLM realtime builds (R1). +_VLLM_DELTA_TYPES = { + "transcription.delta", + "response.audio_transcript.delta", + "conversation.item.input_audio_transcription.delta", +} +_VLLM_DONE_TYPES = { + "transcription.done", + "transcription.completed", + "response.audio_transcript.done", + "conversation.item.input_audio_transcription.completed", +} + + +def _extract_text(event: dict) -> str: + for key in ("delta", "text", "transcript"): + value = event.get(key) + if isinstance(value, str): + return value + return "" + + +class VllmRealtimeTranscriber: + """Bridge to a local vLLM ``/v1/realtime`` WebSocket. + + The connection is local to the compose network and never proxied by the + orchestrator, so the lack of external WebSocket passthrough is not a blocker. + """ + + def __init__(self, *, url: str, model: str, language: str = "") -> None: + self._url = url + self._model = model + self._language = language + self._ws = None + self._queue: asyncio.Queue = asyncio.Queue() + self._recv_task: Optional[asyncio.Task] = None + self._closed = False + self._started = False + self._done_event = asyncio.Event() + self._opened = asyncio.Event() + + async def open(self) -> None: + import websockets # lazy import so mock path needs no ws dep + + self._ws = await websockets.connect(self._url, max_size=None) + # vLLM validates the model from the TOP LEVEL of the session.update + # event (not a nested "session" object); unknown fields are ignored. + update: dict = {"type": "session.update", "model": self._model} + if self._language: + update["language"] = self._language # hint; ignored by vLLM 0.24 + await self._ws.send(json.dumps(update)) + self._recv_task = asyncio.create_task(self._recv_loop()) + self._opened.set() + + async def _recv_loop(self) -> None: + try: + async for raw in self._ws: + try: + event = json.loads(raw) + except (ValueError, TypeError): + continue + etype = event.get("type", "") + if etype in _VLLM_DELTA_TYPES: + await self._queue.put({"type": "delta", "text": _extract_text(event)}) + elif etype in _VLLM_DONE_TYPES: + await self._queue.put({"type": "done", "text": _extract_text(event)}) + self._done_event.set() + elif etype == "error": + log.warning("vLLM realtime error: %s", event) + except Exception as exc: + log.warning("vLLM recv loop ended: %s", exc) + finally: + await self._queue.put(_END) + + async def feed(self, pcm: bytes) -> None: + if self._closed or self._ws is None: + return + await self._ws.send( + json.dumps( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(pcm).decode("ascii"), + } + ) + ) + if not self._started: + # A non-final commit starts generation; vLLM then streams deltas + # while continuing to consume appended audio. + self._started = True + await self._ws.send( + json.dumps({"type": "input_audio_buffer.commit", "final": False}) + ) + + async def apply_settings(self, settings: dict) -> None: + """Forward a session.update from the client mid-stream to vLLM.""" + # The client can send settings the instant its WS connects, which may + # beat open() finishing — wait briefly instead of dropping the update. + try: + await asyncio.wait_for(self._opened.wait(), timeout=5) + except asyncio.TimeoutError: + log.warning("dropping session.update; vLLM connection not open: %s", settings) + return + if self._closed or self._ws is None: + return + # vLLM requires "model" on every session.update; client settings ride + # along at the top level (unknown fields are ignored by the server). + await self._ws.send( + json.dumps({"type": "session.update", "model": self._model, **settings}) + ) + log.info("forwarded session.update to vLLM: %s", settings) + + async def events(self) -> AsyncIterator[dict]: + while True: + item = await self._queue.get() + if item is _END: + return + yield item + + async def close(self) -> None: + if self._closed: + return + self._closed = True + if self._ws is not None: + with suppress(Exception): + # final=True ends the audio stream; vLLM flushes the remaining + # transcript and emits transcription.done. + await self._ws.send( + json.dumps({"type": "input_audio_buffer.commit", "final": True}) + ) + with suppress(Exception): + await asyncio.wait_for(self._done_event.wait(), timeout=60) + with suppress(Exception): + await self._ws.close() + + +# --------------------------------------------------------------------------- # +# Derived metrics +# --------------------------------------------------------------------------- # + +_POSITIVE = { + "good", "great", "love", "welcome", "happy", "nice", "thanks", "thank", + "excellent", "awesome", "wonderful", "fast", "easy", "win", "best", +} +_NEGATIVE = { + "bad", "hate", "slow", "broken", "error", "fail", "failed", "wrong", + "terrible", "awful", "hard", "bug", "crash", "worst", +} + + +def word_count(text: str) -> int: + return len(text.split()) + + +def simple_sentiment(text: str) -> str: + """Tiny lexicon sentiment: 'pos' | 'neg' | 'neu'.""" + score = 0 + for token in text.lower().split(): + word = token.strip(".,!?;:\"'") + if word in _POSITIVE: + score += 1 + elif word in _NEGATIVE: + score -= 1 + if score > 0: + return "pos" + if score < 0: + return "neg" + return "neu" From b4761490c2ce64d906554fa9dcd7bfd96c27627f Mon Sep 17 00:00:00 2001 From: Gideon Jones <109116843+Gideonjon@users.noreply.github.com> Date: Wed, 15 Jul 2026 23:42:28 +0100 Subject: [PATCH 2/4] feat(vllm-realtime): report Trickle + WebSocket performance stats The SDK meters Trickle for free (TrickleSubscriber.get_stats()) but does not meter WebSockets, so an app that streams its results over one has no observability unless it counts them itself. Add stats.py with the missing half, shaped like the SDK's stats: WebSocketMeter/WebSocketStats for the transcript socket, and TranscriptionMeter/TranscriptionStats for latency and throughput. The runner folds in the SDK's Trickle counters and emits a final {"type": "stats"} event; the client prints it as a summary. Note that the SDK's frame counters (MediaOutputStats.audio_frames_decoded) belong to the AV decode path. This example publishes raw PCM16, so there is no decoder and no frame count -- audio duration is derived from byte totals. Also correct the framing around --no-realtime. It cannot be used to measure throughput: Trickle deletes unread segments when the publisher closes, so an unpaced run drops most of its audio and reports confident nonsense (1 of 14 segments in practice). Under realtime pacing the real-time factor is pinned near 1.0 by construction and only answers "did the pipeline keep up"; the honest latency signals are time-to-first-word and the finalize tail. Verified end-to-end on an RTX 4090 serving Voxtral-Mini-4B-Realtime: 6.56 s of speech, finalize tail 0.38 s (reproducible within ~10 ms), RTF 1.12x, 14 segments with zero sequence gaps, retries or failures. --- vllm-realtime/Dockerfile | 2 +- vllm-realtime/FEEDBACK.md | 29 +++++ vllm-realtime/README.md | 39 +++++++ vllm-realtime/client.py | 47 +++++++- vllm-realtime/runner.py | 117 ++++++++++++++------ vllm-realtime/stats.py | 226 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 422 insertions(+), 38 deletions(-) create mode 100644 vllm-realtime/stats.py diff --git a/vllm-realtime/Dockerfile b/vllm-realtime/Dockerfile index 142d165..97eca07 100644 --- a/vllm-realtime/Dockerfile +++ b/vllm-realtime/Dockerfile @@ -13,7 +13,7 @@ RUN pip install --no-cache-dir \ websockets WORKDIR /app -COPY runner.py transcriber.py client.py ./ +COPY runner.py transcriber.py stats.py client.py ./ EXPOSE 5000 diff --git a/vllm-realtime/FEEDBACK.md b/vllm-realtime/FEEDBACK.md index 0eab758..4447e22 100644 --- a/vllm-realtime/FEEDBACK.md +++ b/vllm-realtime/FEEDBACK.md @@ -65,6 +65,26 @@ was confusing, and the open questions a second builder would hit. install the `+cu129` wheel from GitHub releases instead, and `mistral-common[soundfile]` is required at runtime for audio decode. +8. **The SDK meters Trickle, but nothing meters a WebSocket.** `get_stats()` on + `TrickleSubscriber` / `TricklePublisher` / `MediaOutput` gives transport and + frame counters for free, so a Trickle-in/Trickle-out app is observable without + writing any code. An app whose *output* is a WebSocket gets none of it and has + to hand-roll the whole half (here: `stats.py`). Two things make that sharper + than it first looks. First, the frame counters everyone points you at + (`MediaOutputStats.audio_frames_decoded`) only exist if you go through the AV + decode path — this example publishes raw PCM16, so there is no decoder and no + frame count, and audio duration has to be derived from byte totals instead. + Second, there is no suggested shape for app-authored stats, so every WS app + will invent its own field names for the same quantities. +9. **Realtime pacing makes a real-time factor structurally meaningless.** Wall + clock can't drop below the audio duration when the client paces to real time, + so RTF is pinned just above 1.0 regardless of backend speed — it answers "did + the pipeline keep up?", not "how fast is the model?". The obvious workaround + (publish unpaced) collides with #2: the stream is deleted before the subscriber + drains it, so an unpaced run silently transcribes ~1 segment of audio and + reports confident nonsense. There is currently no way to measure true + throughput end-to-end through Trickle without racing the deletion. + ## Suggestions - Add a "transports" page: HTTP / SSE / **Trickle** with a one-paragraph "when to @@ -73,6 +93,9 @@ was confusing, and the open questions a second builder would hit. - Document the trickle channel mime-type contract (validated vs opaque). - Document the streaming payment model (one-shot vs per-segment) end to end. - Consider splitting the `av` import so Trickle-only apps don't need PyAV. +- Ship a `WebSocketStats` counter helper next to the Trickle ones, so apps that + stream results over a WebSocket report the same shape instead of each inventing + one (see #8). ## Environment @@ -82,3 +105,9 @@ was confusing, and the open questions a second builder would hit. 570.x/CUDA 12.8). Live speech was transcribed with word-by-word deltas streaming over the orchestrator-proxied WebSocket while audio was still being published. +- Measured on that setup, 6.56 s of speech published at realtime pace: time to + first word 2.4 s (inflated by lead-in silence in the clip), finalize tail + 0.38 s, real-time factor 1.12x, 28 deltas / 15 words, and zero Trickle + sequence gaps, retries or failures across 14 segments. The finalize tail is + the stable figure — it reproduced within ~10 ms across runs, while first-word + moves with whatever silence precedes speech. diff --git a/vllm-realtime/README.md b/vllm-realtime/README.md index ab79ee1..5a39d47 100644 --- a/vllm-realtime/README.md +++ b/vllm-realtime/README.md @@ -76,6 +76,45 @@ Validated on an RTX 4090 (24 GB): the compose file caps `--max-model-len` at native 131k context does not). The Voxtral model is public on HuggingFace — no token needed. +## Performance + +The last event of every session is `{"type": "stats", ...}`, which the client +prints as a summary. It has three parts: + +``` +──── performance ──── + audio 6.56 s + wall clock 7.355 s + real-time factor 1.121x + time to first word 2.364 s + finalize tail 0.379 s + words / deltas 15 / 28 + + trickle in (SDK) segments=14 seq_gaps=0 retries=0 failures=0 stall=6970ms + websocket out (app) events=29 deltas=28 bytes=3779 failures=0 cmds_in=1 +``` + +**Trickle in** comes free from the SDK (`TrickleSubscriber.get_stats()`) and is +transport health: `seq_gaps` above 0 means audio was dropped. **WebSocket out** +and the latency numbers are metered by this app in [stats.py](stats.py) — the SDK +does not meter WebSockets, so an app that streams results over one has to count +them itself. + +Two caveats worth reading before quoting any of it: + +- **Real-time factor is pinned near 1.0 by design.** It is wall clock ÷ audio, and + the client paces audio to real time, so wall clock can never drop below the + audio duration however fast the GPU is. Read it as *"the pipeline kept up"* — it + climbs only when the backend falls behind. It is not model speed. Publishing + unpaced does not fix this: Trickle deletes unread segments when the publisher + closes, so `--no-realtime` drops most of the audio and reports confident + nonsense (see [FEEDBACK.md](FEEDBACK.md) #2, #9). +- **Time to first word includes lead-in silence.** It moves with whatever quiet + precedes speech in the clip. The **finalize tail** — last audio byte to final + transcript — is the stable figure, reproducing within ~10 ms across runs. + +Numbers above: RTX 4090, Voxtral-Mini-4B-Realtime, realtime-paced speech. + ## Run on-chain (paid) Layer `docker-compose.onchain.yml` to add a remote signer and run the diff --git a/vllm-realtime/client.py b/vllm-realtime/client.py index 9acc7f9..c759719 100644 --- a/vllm-realtime/client.py +++ b/vllm-realtime/client.py @@ -9,7 +9,8 @@ 4. Optionally send {"type": "session.update", "session": {...}} mid-stream to adjust settings (language etc.) without stopping the stream 5. Publish PCM16/16 kHz audio to the Trickle in channel, paced to real time -6. Drain transcript events from the WebSocket until "done" +6. Drain transcript events from the WebSocket until "done", then print the + "stats" event the runner sends last (see stats.py) 7. Release the session Pass --signer for the on-chain (paid) path. @@ -60,7 +61,10 @@ def _parse_args() -> argparse.Namespace: ) parser.add_argument( "--no-realtime", action="store_true", - help="Publish as fast as possible instead of pacing to wall clock.", + help="Publish as fast as possible instead of pacing to wall clock. " + "Trickle deletes unread segments when the publisher closes, so an " + "unpaced run drops most of its audio — smoke tests only, never " + "measurement (see FEEDBACK.md).", ) parser.add_argument( "--language", default="", @@ -108,6 +112,39 @@ def _make_ssl_ctx() -> ssl.SSLContext: return ctx +def _print_stats(data: dict) -> None: + """Render the runner's final stats event.""" + t = data.get("transcription") or {} + w = data.get("websocket") or {} + k = data.get("trickle") or {} + + print("\n──── performance ────") + print(f" audio {t.get('audio_seconds')} s") + print(f" wall clock {t.get('wall_seconds')} s") + print(f" real-time factor {t.get('realtime_factor')}x") + print(f" time to first word {t.get('time_to_first_word_s')} s") + print(f" finalize tail {t.get('finalize_tail_s')} s") + print(f" words / deltas {t.get('words')} / {t.get('deltas')}") + print( + "\n trickle in (SDK) " + f"segments={k.get('segments_delivered')} seq_gaps={k.get('seq_gap_events')} " + f"retries={k.get('get_retries')} failures={k.get('get_failures')} " + f"stall={k.get('wait_ms_total')}ms" + ) + print( + " websocket out (app) " + f"events={w.get('events_sent')} deltas={w.get('deltas_sent')} " + f"bytes={w.get('bytes_sent')} failures={w.get('send_failures')} " + f"cmds_in={w.get('commands_received')}" + ) + print( + "\n note: audio is paced to real time, so wall clock cannot drop below the\n" + " audio duration — the real-time factor is pinned near 1.0 and reads as\n" + ' "the pipeline kept up", not as GPU speed. It climbs only if the backend\n' + " falls behind. The latency signals are first word and finalize tail." + ) + + async def _read_transcript( ws: aiohttp.ClientWebSocketResponse, done: asyncio.Event, @@ -123,12 +160,16 @@ async def _read_transcript( continue kind = data.get("type") transcript = data.get("transcript", "") + if kind == "stats": + # Sent after "done"; last event of the session. + _print_stats(data) + break if kind == "done": print( f"\n[done] {transcript!r} " f"words={data.get('word_count')} sentiment={data.get('sentiment')}" ) - break + continue print( f"[delta] +{data.get('delta', '').strip()!r} " f"words={data.get('word_count')} sentiment={data.get('sentiment')}" diff --git a/vllm-realtime/runner.py b/vllm-realtime/runner.py index 1d34496..ab5f86b 100644 --- a/vllm-realtime/runner.py +++ b/vllm-realtime/runner.py @@ -12,6 +12,10 @@ send {"type": "session.update", "session": {...}} mid-stream to adjust settings (language, model, etc.) without stopping the stream. + The last event before close is {"type": "stats", ...}: Trickle transport + counters (from the SDK) plus WebSocket and latency counters (metered by this + app — see stats.py, the SDK does not meter WebSockets). + Architecture ------------ The orchestrator proxies both Trickle segments and WebSocket upgrades, so the @@ -33,14 +37,15 @@ import json import logging from contextlib import suppress -from dataclasses import dataclass, field +from dataclasses import asdict, dataclass, field from typing import Optional from aiohttp import web from livepeer_gateway.live_runner import register_runner -from livepeer_gateway.trickle_subscriber import TrickleSubscriber +from livepeer_gateway.trickle_subscriber import TrickleSubscriber, TrickleSubscriberStats +from stats import TranscriptionMeter, WebSocketMeter from transcriber import Transcriber, make_transcriber, simple_sentiment, word_count DEFAULT_HOST = "127.0.0.1" @@ -62,6 +67,9 @@ class TranscribeSession: event_queue: asyncio.Queue = field(default_factory=asyncio.Queue) task: Optional[asyncio.Task] = None transcriber: Optional[Transcriber] = None + # The SDK meters Trickle for us; the WS and the latency numbers are ours. + meter: TranscriptionMeter = field(default_factory=TranscriptionMeter) + ws_meter: WebSocketMeter = field(default_factory=WebSocketMeter) def to_json(self) -> dict: return {"session": self.session_id, "in": self.in_url, "ws": "/ws"} @@ -86,37 +94,42 @@ def _session_id(request: web.Request) -> str: return request.headers.get("Livepeer-Session-Id", "").strip() -async def _pump_audio(in_url: str, transcriber: Transcriber) -> None: +async def _pump_audio( + sub: TrickleSubscriber, transcriber: Transcriber, meter: TranscriptionMeter +) -> None: """Read PCM segments off the Trickle input channel and feed the transcriber.""" - async with TrickleSubscriber(in_url) as sub: + while True: + segment = await sub.next() + if segment is None: + break + reader = segment.make_reader() while True: - segment = await sub.next() - if segment is None: + chunk = await reader.read() + if chunk is None: break - reader = segment.make_reader() - while True: - chunk = await reader.read() - if chunk is None: - break - await transcriber.feed(chunk) - with suppress(Exception): - await segment.close() + meter.mark_audio(len(chunk)) + await transcriber.feed(chunk) + with suppress(Exception): + await segment.close() -async def _drain_events(transcriber: Transcriber, event_queue: asyncio.Queue) -> None: +async def _drain_events(session: TranscribeSession) -> None: """Read normalized transcriber events, compute metrics, push to the WS queue.""" transcript = "" - async for event in transcriber.events(): + async for event in session.transcriber.events(): if event["type"] == "delta": if not event.get("text"): # Voxtral emits empty-text deltas between words; skip them # rather than spamming clients with no-op events. continue transcript += event.get("text", "") + session.meter.mark_delta() else: transcript = event.get("text", "") or transcript clean = transcript.strip() - await event_queue.put( + if event["type"] != "delta": + session.meter.mark_done(clean) + await session.event_queue.put( { "type": event["type"], "delta": event.get("text", "") if event["type"] == "delta" else "", @@ -127,22 +140,50 @@ async def _drain_events(transcriber: Transcriber, event_queue: asyncio.Queue) -> ) -async def _run_pipeline( - in_url: str, - transcriber: Transcriber, - event_queue: asyncio.Queue, -) -> None: +def _stats_event( + session: TranscribeSession, trickle_stats: Optional[TrickleSubscriberStats] +) -> dict: + """Final event of a session: Trickle in (from the SDK) + WS out (ours).""" + transcription = session.meter.get_stats() + websocket = session.ws_meter.get_stats() + trickle = None + if trickle_stats is not None: + trickle = asdict(trickle_stats) + trickle["elapsed_s"] = round(trickle["elapsed_s"], 3) + log.info("stats %s", transcription) + log.info("stats %s", websocket) + if trickle_stats is not None: + log.info("stats %s", trickle_stats) + return { + "type": "stats", + "transcription": transcription.to_json(), + "websocket": websocket.to_json(), + "trickle": trickle, + } + + +async def _run_pipeline(session: TranscribeSession, in_url: str) -> None: """Wire input audio → transcriber → metrics → WS event queue.""" drain: Optional[asyncio.Task] = None + trickle_stats: Optional[TrickleSubscriberStats] = None try: - await transcriber.open() - drain = asyncio.create_task(_drain_events(transcriber, event_queue)) - await _pump_audio(in_url, transcriber) + await session.transcriber.open() + drain = asyncio.create_task(_drain_events(session)) + async with TrickleSubscriber(in_url) as sub: + try: + await _pump_audio(sub, session.transcriber, session.meter) + finally: + # Read the transport counters while the subscriber is still open, + # so a mid-stream failure still reports how far the audio got. + trickle_stats = sub.get_stats() except Exception: log.exception("transcription pipeline failed") finally: + # Audio is done; the finalize tail is measured from here to the final + # transcript, so mark it before close() triggers the flush. + session.meter.mark_audio_end() with suppress(Exception): - await transcriber.close() + await session.transcriber.close() if drain is not None: try: await drain @@ -150,21 +191,28 @@ async def _run_pipeline( # A dead drain means transcript events silently stop reaching # clients — surface it instead of swallowing it. log.exception("event drain task failed") + await session.event_queue.put(_stats_event(session, trickle_stats)) # Signal the WS drain coroutine that no more events are coming. - await event_queue.put(_WS_DONE) + await session.event_queue.put(_WS_DONE) log.info("transcription pipeline finished") -async def _drain_queue_to_ws(queue: asyncio.Queue, ws: web.WebSocketResponse) -> None: - """Forward queued transcript events to the WebSocket client.""" +async def _drain_queue_to_ws(session: TranscribeSession, ws: web.WebSocketResponse) -> None: + """Forward queued transcript events to the WebSocket client, metering as we go.""" try: while True: - event = await queue.get() + event = await session.event_queue.get() if event is _WS_DONE: await ws.close() return - with suppress(Exception): - await ws.send_json(event) + # Serialize here rather than send_json so the payload size is + # measurable — the SDK counts Trickle bytes for us, not these. + payload = json.dumps(event) + try: + await ws.send_str(payload) + session.ws_meter.record_sent(event, len(payload)) + except Exception: + session.ws_meter.record_send_failure() except asyncio.CancelledError: pass @@ -206,7 +254,7 @@ async def _handle_transcribe(request: web.Request) -> web.Response: event_queue=event_queue, transcriber=transcriber, ) - task = asyncio.create_task(_run_pipeline(in_internal_url, transcriber, event_queue)) + task = asyncio.create_task(_run_pipeline(session, in_internal_url)) task.add_done_callback(_clear_state) session.task = task state = session @@ -224,12 +272,13 @@ async def _handle_ws(request: web.Request) -> web.WebSocketResponse: await ws.prepare(request) session = state - drain_task = asyncio.create_task(_drain_queue_to_ws(session.event_queue, ws)) + drain_task = asyncio.create_task(_drain_queue_to_ws(session, ws)) try: async for msg in ws: if msg.type == web.WSMsgType.TEXT: log.info("ws command received: %.200s", msg.data) + session.ws_meter.record_command() with suppress(Exception): cmd = json.loads(msg.data) if cmd.get("type") == "session.update" and session.transcriber is not None: diff --git a/vllm-realtime/stats.py b/vllm-realtime/stats.py new file mode 100644 index 0000000..618e2a9 --- /dev/null +++ b/vllm-realtime/stats.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""Performance stats for the vllm-realtime example. + +Why this module exists +---------------------- +The SDK meters the **Trickle** path for free — ``TrickleSubscriber.get_stats()`` +returns a ``TrickleSubscriberStats`` (segments delivered, sequence gaps, retries, +stall time). There is no equivalent for a **WebSocket**: an app that streams its +results out over one has to meter that path itself. + +This module is the missing half, shaped like the SDK's so the two read alike: + + WebSocketMeter.get_stats() -> WebSocketStats # transport, the WS out path + TranscriptionMeter.get_stats() -> TranscriptionStats # domain, latency + throughput + +Two things worth knowing if you copy this into your own app: + +1. The SDK's frame counters (``MediaOutputStats.audio_frames_decoded``) belong to + the media/AV pipeline, which decodes encoded media into frames. This example + publishes *raw* PCM16, so there is no decoder in the path and those counters + do not exist for us. Audio duration is derived from bytes instead. + +2. ``realtime_factor`` is wall time / audio duration, so it measures the whole + pipeline, not the model. The client paces audio to real time, so wall time can + never drop below the audio duration: the factor is pinned just above 1.0 no + matter how fast the GPU is. Read it as "did the pipeline keep up?" — it holds + near 1.0 while the backend keeps pace and climbs once it falls behind. + Publishing unpaced does *not* turn it into a throughput measurement: Trickle + deletes unread segments when the publisher closes, so an unpaced run drops most + of its audio and measures nothing (see FEEDBACK.md). The honest latency signals + are ``time_to_first_word_s`` and ``finalize_tail_s``. +""" +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Optional + +# PCM16LE mono @ 16 kHz: 2 bytes/sample × 16000 samples/s. +PCM_BYTES_PER_SECOND = 16000 * 2 + + +def _round(value: Optional[float], places: int = 3) -> Optional[float]: + return None if value is None else round(value, places) + + +# --------------------------------------------------------------------------- # +# WebSocket transport stats (the half the SDK does not provide) +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True) +class WebSocketStats: + """Counters for the transcript WebSocket, mirroring the SDK's Trickle stats.""" + + elapsed_s: float + events_sent: int + deltas_sent: int + bytes_sent: int + send_failures: int + commands_received: int + + @property + def send_rate_hz(self) -> float: + return self.events_sent / self.elapsed_s if self.elapsed_s > 0 else 0.0 + + def to_json(self) -> dict: + return { + "elapsed_s": _round(self.elapsed_s), + "events_sent": self.events_sent, + "deltas_sent": self.deltas_sent, + "bytes_sent": self.bytes_sent, + "send_failures": self.send_failures, + "commands_received": self.commands_received, + "send_rate_hz": _round(self.send_rate_hz), + } + + def __str__(self) -> str: + return ( + "WebSocketStats(" + f"elapsed_s={self.elapsed_s:.1f}, " + f"events_sent={self.events_sent}, " + f"deltas_sent={self.deltas_sent}, " + f"bytes_sent={self.bytes_sent}, " + f"send_failures={self.send_failures}, " + f"commands_received={self.commands_received}" + ")" + ) + + +class WebSocketMeter: + """Tracks what goes out over — and comes back in on — the transcript WebSocket.""" + + def __init__(self) -> None: + self._started_at = time.time() + self._events_sent = 0 + self._deltas_sent = 0 + self._bytes_sent = 0 + self._send_failures = 0 + self._commands_received = 0 + + def record_sent(self, event: dict, nbytes: int) -> None: + self._events_sent += 1 + self._bytes_sent += nbytes + if event.get("type") == "delta": + self._deltas_sent += 1 + + def record_send_failure(self) -> None: + self._send_failures += 1 + + def record_command(self) -> None: + """A client → runner message (e.g. a live session.update).""" + self._commands_received += 1 + + def get_stats(self) -> WebSocketStats: + return WebSocketStats( + elapsed_s=max(0.0, time.time() - self._started_at), + events_sent=self._events_sent, + deltas_sent=self._deltas_sent, + bytes_sent=self._bytes_sent, + send_failures=self._send_failures, + commands_received=self._commands_received, + ) + + +# --------------------------------------------------------------------------- # +# Transcription (domain) stats +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True) +class TranscriptionStats: + """End-to-end latency and throughput for one transcription session.""" + + audio_seconds: float + wall_seconds: float + realtime_factor: float + time_to_first_word_s: Optional[float] + finalize_tail_s: Optional[float] + deltas: int + words: int + + def to_json(self) -> dict: + return { + "audio_seconds": _round(self.audio_seconds), + "wall_seconds": _round(self.wall_seconds), + "realtime_factor": _round(self.realtime_factor), + "time_to_first_word_s": _round(self.time_to_first_word_s), + "finalize_tail_s": _round(self.finalize_tail_s), + "deltas": self.deltas, + "words": self.words, + } + + def __str__(self) -> str: + return ( + "TranscriptionStats(" + f"audio_s={self.audio_seconds:.2f}, " + f"wall_s={self.wall_seconds:.2f}, " + f"rtf={self.realtime_factor:.2f}, " + f"first_word_s={self.time_to_first_word_s}, " + f"finalize_tail_s={self.finalize_tail_s}, " + f"deltas={self.deltas}, " + f"words={self.words}" + ")" + ) + + +class TranscriptionMeter: + """Times the audio-in → transcript-out path. + + Timestamps are taken at the runner, so they exclude the client↔orchestrator + hop; they measure what this app is responsible for. + """ + + def __init__(self) -> None: + self._t_first_audio: Optional[float] = None + self._t_first_delta: Optional[float] = None + self._t_audio_end: Optional[float] = None + self._t_done: Optional[float] = None + self._bytes_in = 0 + self._deltas = 0 + self._words = 0 + + def mark_audio(self, nbytes: int) -> None: + if self._t_first_audio is None: + self._t_first_audio = time.time() + self._bytes_in += nbytes + + def mark_delta(self) -> None: + """A non-empty transcript delta reached the client path.""" + if self._t_first_delta is None: + self._t_first_delta = time.time() + self._deltas += 1 + + def mark_audio_end(self) -> None: + """Last audio byte consumed — the clock for the finalize tail starts here.""" + if self._t_audio_end is None: + self._t_audio_end = time.time() + + def mark_done(self, transcript: str) -> None: + if self._t_done is None: + self._t_done = time.time() + self._words = len(transcript.split()) + + def get_stats(self) -> TranscriptionStats: + audio_seconds = self._bytes_in / PCM_BYTES_PER_SECOND + end = self._t_done or self._t_audio_end or time.time() + wall = (end - self._t_first_audio) if self._t_first_audio is not None else 0.0 + + first_word = None + if self._t_first_delta is not None and self._t_first_audio is not None: + first_word = self._t_first_delta - self._t_first_audio + + tail = None + if self._t_done is not None and self._t_audio_end is not None: + tail = self._t_done - self._t_audio_end + + return TranscriptionStats( + audio_seconds=audio_seconds, + wall_seconds=max(0.0, wall), + realtime_factor=(wall / audio_seconds) if audio_seconds > 0 else 0.0, + time_to_first_word_s=first_word, + finalize_tail_s=tail, + deltas=self._deltas, + words=self._words, + ) From 889b7b8d958ef4d02af2e035607e84328e12211a Mon Sep 17 00:00:00 2001 From: Gideon Jones <109116843+Gideonjon@users.noreply.github.com> Date: Thu, 16 Jul 2026 00:11:44 +0100 Subject: [PATCH 3/4] fix(vllm-realtime): stop losing audio when publishing outruns the runner A Trickle channel keeps no backlog. TrickleSubscriber defaults to start_seq=-2 (join at the live edge) and an earlier index answers 470 "no data at this index" rather than replaying, so any segment published before the runner reads the previous one is destroyed, not queued. That is the right default for video -- drop stale frames, stay current -- but audio is not droppable, and nothing reports the loss: publishing 6.5s of speech unpaced delivered 1 of 14 segments with seq_gap_events=0, no error, and a transcript that was simply short. Subscribing at start_seq=0 is not a fix; it delivers 0 segments and resets to the live edge. Fix it with app-level backpressure over the WebSocket we already have. The runner emits {"type": "progress", "audio_bytes": N} after each segment it consumes; the client publishes one segment, waits for the runner to acknowledge it, then publishes the next. At most one segment is ever in flight, so the live edge is always the unread segment. This makes --no-realtime measure real throughput instead of shredding the audio, and the two modes now report different, honest numbers on the same 6.56s clip (RTX 4090, Voxtral-Mini-4B-Realtime): paced finalize tail 0.37s, RTF 1.13x -- live latency unpaced wall 1.78s, RTF 0.27x -- ~3.7x realtime throughput Both transcribe identically, 14/14 segments, no gaps. Verified against the mock backend too, paced and unpaced. --- vllm-realtime/FEEDBACK.md | 59 ++++++++++++++++--------- vllm-realtime/README.md | 49 +++++++++++++-------- vllm-realtime/client.py | 90 +++++++++++++++++++++++++++++++++------ vllm-realtime/runner.py | 26 ++++++++--- vllm-realtime/stats.py | 21 +++++---- 5 files changed, 181 insertions(+), 64 deletions(-) diff --git a/vllm-realtime/FEEDBACK.md b/vllm-realtime/FEEDBACK.md index 4447e22..fee0211 100644 --- a/vllm-realtime/FEEDBACK.md +++ b/vllm-realtime/FEEDBACK.md @@ -29,12 +29,22 @@ was confusing, and the open questions a second builder would hit. subscribe on that, hand `url` to the client. `echo` does this correctly, but nothing shouts about it and the failure mode (a subscriber that never sees a segment) is silent until you read orchestrator logs. -2. **Trickle streams are deleted when the publisher closes — unread segments are - dropped.** Publishing faster than realtime (`--no-realtime`) pushes all - segments and deletes the stream within milliseconds; a subscriber that - connects even a beat later gets nothing. Paced (realtime) publishing works - because pub and sub overlap. Worth documenting whether the orchestrator can - buffer/retain segments for late subscribers. +2. **Trickle channels keep no backlog: the subscriber reads the live edge, so + publishing faster than it consumes destroys audio.** `TrickleSubscriber` + defaults to `start_seq=-2` — join at the live edge — and an earlier index + answers `470` ("no data at this index") instead of replaying, so segments + published while you were behind are simply gone. Measured here: publishing + 6.5 s of speech unpaced delivered **1 of 14 segments** + (`segments_delivered=1, latest_seq=14`), and asking for `start_seq=0` + delivered **0** and reset to the edge. This is the right default for video — + drop stale frames, stay current — but audio is not droppable, and nothing in + the API signals the difference: there is no error, no gap event + (`seq_gap_events=0`), just a transcript that comes back short. Apps that + cannot lose bytes need app-level backpressure; this example publishes one + segment, waits for the runner to report it consumed it over the WebSocket, + then publishes the next. A documented "lossless vs live-edge" subscribe mode + (or any retention window) would spare every non-video app from rediscovering + this the hard way. 3. **Importing the SDK pulls in PyAV.** `livepeer_gateway/__init__.py` imports the media modules at top level, so `import livepeer_gateway` (even just for Trickle) requires `av`. It installs transitively, but a from-source build on a @@ -76,14 +86,16 @@ was confusing, and the open questions a second builder would hit. frame count, and audio duration has to be derived from byte totals instead. Second, there is no suggested shape for app-authored stats, so every WS app will invent its own field names for the same quantities. -9. **Realtime pacing makes a real-time factor structurally meaningless.** Wall - clock can't drop below the audio duration when the client paces to real time, - so RTF is pinned just above 1.0 regardless of backend speed — it answers "did - the pipeline keep up?", not "how fast is the model?". The obvious workaround - (publish unpaced) collides with #2: the stream is deleted before the subscriber - drains it, so an unpaced run silently transcribes ~1 segment of audio and - reports confident nonsense. There is currently no way to measure true - throughput end-to-end through Trickle without racing the deletion. +9. **A real-time factor only means something once the client has backpressure.** + Wall clock cannot drop below the audio duration while the client paces to real + time, so RTF is pinned just above 1.0 regardless of backend speed and only + answers "did the pipeline keep up?". Publishing unpaced to measure real + throughput runs straight into #2 and reports confident nonsense (1 segment of + 14, zero words, RTF 1.89 — all with no error raised anywhere). Once the client + holds the live edge until the runner acknowledges each segment, the same clip + reports **RTF 0.27 (~3.7x realtime)** and the figure is finally real. Worth + noting that the two modes measure different things and both are useful: paced + gives live latency (finalize tail 0.37 s), unpaced gives throughput headroom. ## Suggestions @@ -96,6 +108,10 @@ was confusing, and the open questions a second builder would hit. - Ship a `WebSocketStats` counter helper next to the Trickle ones, so apps that stream results over a WebSocket report the same shape instead of each inventing one (see #8). +- Give `TrickleSubscriber` a lossless subscribe mode, or document the live-edge + contract loudly (see #2). Right now the safe default for video is a silent + data-loss trap for every other media type, and the counters report success + (`seq_gap_events=0`) while audio disappears. ## Environment @@ -105,9 +121,12 @@ was confusing, and the open questions a second builder would hit. 570.x/CUDA 12.8). Live speech was transcribed with word-by-word deltas streaming over the orchestrator-proxied WebSocket while audio was still being published. -- Measured on that setup, 6.56 s of speech published at realtime pace: time to - first word 2.4 s (inflated by lead-in silence in the clip), finalize tail - 0.38 s, real-time factor 1.12x, 28 deltas / 15 words, and zero Trickle - sequence gaps, retries or failures across 14 segments. The finalize tail is - the stable figure — it reproduced within ~10 ms across runs, while first-word - moves with whatever silence precedes speech. +- Measured on that setup with 6.56 s of speech, 14 segments, no sequence gaps, + retries or failures in either mode: + - *Realtime-paced* (live latency): finalize tail **0.37 s**, real-time factor + 1.13x, time to first word 2.4 s. The tail is the stable figure — it + reproduced within ~10 ms across runs; first-word moves with whatever silence + precedes speech in the clip. + - *Unpaced with backpressure* (throughput): wall clock **1.78 s** for 6.56 s of + audio — real-time factor **0.27x**, i.e. ~3.7x faster than realtime, with an + identical transcript. diff --git a/vllm-realtime/README.md b/vllm-realtime/README.md index 5a39d47..9d07016 100644 --- a/vllm-realtime/README.md +++ b/vllm-realtime/README.md @@ -84,14 +84,14 @@ prints as a summary. It has three parts: ``` ──── performance ──── audio 6.56 s - wall clock 7.355 s - real-time factor 1.121x - time to first word 2.364 s - finalize tail 0.379 s + wall clock 7.395 s + real-time factor 1.127x + time to first word 2.418 s + finalize tail 0.374 s words / deltas 15 / 28 - trickle in (SDK) segments=14 seq_gaps=0 retries=0 failures=0 stall=6970ms - websocket out (app) events=29 deltas=28 bytes=3779 failures=0 cmds_in=1 + trickle in (SDK) segments=14 seq_gaps=0 retries=0 failures=0 stall=7009ms + websocket out (app) events=43 deltas=28 bytes=4375 failures=0 cmds_in=1 ``` **Trickle in** comes free from the SDK (`TrickleSubscriber.get_stats()`) and is @@ -100,20 +100,33 @@ and the latency numbers are metered by this app in [stats.py](stats.py) — the does not meter WebSockets, so an app that streams results over one has to count them itself. -Two caveats worth reading before quoting any of it: +### Two modes, two different numbers -- **Real-time factor is pinned near 1.0 by design.** It is wall clock ÷ audio, and - the client paces audio to real time, so wall clock can never drop below the - audio duration however fast the GPU is. Read it as *"the pipeline kept up"* — it - climbs only when the backend falls behind. It is not model speed. Publishing - unpaced does not fix this: Trickle deletes unread segments when the publisher - closes, so `--no-realtime` drops most of the audio and reports confident - nonsense (see [FEEDBACK.md](FEEDBACK.md) #2, #9). -- **Time to first word includes lead-in silence.** It moves with whatever quiet - precedes speech in the clip. The **finalize tail** — last audio byte to final - transcript — is the stable figure, reproducing within ~10 ms across runs. +By default the client paces audio to real time, so wall clock can never drop +below the audio duration — the **real-time factor is pinned near 1.0 however fast +the backend is**. In that mode it answers *"did the pipeline keep up?"*, and the +latency signal is the **finalize tail** (last audio byte → final transcript), +which reproduces within ~10 ms across runs. Time to first word is *not* a stable +figure: it moves with whatever silence precedes speech in the clip. -Numbers above: RTX 4090, Voxtral-Mini-4B-Realtime, realtime-paced speech. +Run with `--no-realtime` to remove the pacing floor and measure real throughput: + +``` +audio 6.56 s · wall clock 1.78 s · real-time factor 0.271x → ~3.7x realtime +``` + +Both are honest; they measure different things. Paced gives live latency, unpaced +gives throughput headroom. + +`--no-realtime` is safe because the client applies **backpressure**: a Trickle +channel keeps no backlog, so a segment published before the runner reads the +previous one is destroyed, not queued. The client publishes one segment, waits +for the runner's `progress` event, then publishes the next — so "as fast as +possible" means "as fast as the runner consumes". Without that, an unpaced run +delivers 1 segment of 14 and reports confident nonsense, with no error raised +anywhere. See [FEEDBACK.md](FEEDBACK.md) #2 and #9. + +Numbers above: RTX 4090, Voxtral-Mini-4B-Realtime. ## Run on-chain (paid) diff --git a/vllm-realtime/client.py b/vllm-realtime/client.py index c759719..533efe2 100644 --- a/vllm-realtime/client.py +++ b/vllm-realtime/client.py @@ -45,6 +45,10 @@ SEGMENT_SECONDS = 0.5 # one Trickle segment per ~0.5 s of audio IN_MIME = "audio/raw" +# How long to wait for the runner to finish reading what we published before +# closing the stream anyway. Generous: a cold backend can lag well behind. +DRAIN_TIMEOUT_SECONDS = 60.0 + log = logging.getLogger("vllm-realtime-client") @@ -61,10 +65,9 @@ def _parse_args() -> argparse.Namespace: ) parser.add_argument( "--no-realtime", action="store_true", - help="Publish as fast as possible instead of pacing to wall clock. " - "Trickle deletes unread segments when the publisher closes, so an " - "unpaced run drops most of its audio — smoke tests only, never " - "measurement (see FEEDBACK.md).", + help="Publish as fast as the runner can consume instead of pacing to " + "wall clock. Removes the pacing floor, so the real-time factor " + "measures actual backend throughput.", ) parser.add_argument( "--language", default="", @@ -112,6 +115,39 @@ def _make_ssl_ctx() -> ssl.SSLContext: return ctx +class _Progress: + """How much of our published audio the runner reports it has consumed. + + Trickle deletes unread segments the instant the publisher closes, so + publishing faster than the runner consumes silently loses audio. The runner + reports its ingest position over the WebSocket; we wait for it to catch up + before closing the stream. That turns "publish and hope" into backpressure. + """ + + def __init__(self) -> None: + self.consumed = 0 + self._bump = asyncio.Event() + + def update(self, consumed: int) -> None: + self.consumed = consumed + self._bump.set() + + async def wait_for(self, target: int, timeout: float) -> bool: + """Block until the runner has consumed `target` bytes. False on timeout.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while self.consumed < target: + remaining = deadline - loop.time() + if remaining <= 0: + return False + self._bump.clear() + if self.consumed >= target: + return True + with suppress(asyncio.TimeoutError): + await asyncio.wait_for(self._bump.wait(), timeout=remaining) + return True + + def _print_stats(data: dict) -> None: """Render the runner's final stats event.""" t = data.get("transcription") or {} @@ -138,16 +174,17 @@ def _print_stats(data: dict) -> None: f"cmds_in={w.get('commands_received')}" ) print( - "\n note: audio is paced to real time, so wall clock cannot drop below the\n" - " audio duration — the real-time factor is pinned near 1.0 and reads as\n" - ' "the pipeline kept up", not as GPU speed. It climbs only if the backend\n' - " falls behind. The latency signals are first word and finalize tail." + "\n note: audio is paced to real time by default, so wall clock cannot drop\n" + ' below the audio duration — the real-time factor reads as "the pipeline\n' + ' kept up", not as backend speed. Re-run with --no-realtime to remove the\n' + " pacing floor and measure actual throughput." ) async def _read_transcript( ws: aiohttp.ClientWebSocketResponse, done: asyncio.Event, + progress: _Progress, ) -> None: """Print transcript + metrics as JSON events arrive on the WebSocket.""" try: @@ -160,6 +197,10 @@ async def _read_transcript( continue kind = data.get("type") transcript = data.get("transcript", "") + if kind == "progress": + # Ingest position, not transcript — drives backpressure only. + progress.update(data.get("audio_bytes", 0)) + continue if kind == "stats": # Sent after "done"; last event of the session. _print_stats(data) @@ -180,10 +221,23 @@ async def _read_transcript( done.set() -async def _publish_pcm(in_url: str, pcm: bytes, realtime: bool) -> None: - """Publish PCM to the Trickle input channel, ~0.5 s per segment, 40 ms per frame.""" +async def _publish_pcm( + in_url: str, pcm: bytes, realtime: bool, progress: _Progress +) -> None: + """Publish PCM to the Trickle input channel, ~0.5 s per segment, 40 ms per frame. + + A Trickle channel keeps no backlog — a subscriber reads the live edge, and an + earlier index answers 470 rather than replaying. So publishing a segment + before the runner has read the previous one does not queue it, it destroys + it. After each segment we wait for the runner to report that it consumed + everything so far, which keeps at most one segment in flight and makes the + runner's ingest rate the publish rate. Under realtime pacing the wait is a + no-op (we are already slower than the backend); it is what lets --no-realtime + run flat out without shredding the audio. + """ frame_bytes = int(FRAME_SECONDS * BYTES_PER_SECOND) seg_bytes = int(SEGMENT_SECONDS * BYTES_PER_SECOND) + published = 0 async with TricklePublisher(in_url, IN_MIME) as pub: for seg_start in range(0, len(pcm), seg_bytes): segment = pcm[seg_start : seg_start + seg_bytes] @@ -193,6 +247,15 @@ async def _publish_pcm(in_url: str, pcm: bytes, realtime: bool) -> None: await writer.write(segment[off : off + frame_bytes]) if realtime: await asyncio.sleep(FRAME_SECONDS) + published += len(segment) + + if not await progress.wait_for(published, timeout=DRAIN_TIMEOUT_SECONDS): + log.warning( + "runner stalled at %d/%d bytes after %.0fs; stopping publish " + "rather than overwriting audio it has not read", + progress.consumed, published, DRAIN_TIMEOUT_SECONDS, + ) + break async def main() -> None: @@ -245,9 +308,12 @@ async def main() -> None: log.info("sent live session.update language=%r", args.language) done = asyncio.Event() - reader = asyncio.create_task(_read_transcript(ws, done)) + progress = _Progress() + reader = asyncio.create_task(_read_transcript(ws, done, progress)) - await _publish_pcm(in_url, pcm, realtime=not args.no_realtime) + await _publish_pcm( + in_url, pcm, realtime=not args.no_realtime, progress=progress + ) with suppress(asyncio.TimeoutError): await asyncio.wait_for(done.wait(), timeout=30) diff --git a/vllm-realtime/runner.py b/vllm-realtime/runner.py index ab5f86b..f6b1966 100644 --- a/vllm-realtime/runner.py +++ b/vllm-realtime/runner.py @@ -12,6 +12,10 @@ send {"type": "session.update", "session": {...}} mid-stream to adjust settings (language, model, etc.) without stopping the stream. + {"type": "progress", "audio_bytes": N} reports how much audio has actually + been read off the Trickle channel. The client uses it as backpressure: the + channel keeps no backlog, so it must not publish past what we have consumed. + The last event before close is {"type": "stats", ...}: Trickle transport counters (from the SDK) plus WebSocket and latency counters (metered by this app — see stats.py, the SDK does not meter WebSockets). @@ -94,9 +98,7 @@ def _session_id(request: web.Request) -> str: return request.headers.get("Livepeer-Session-Id", "").strip() -async def _pump_audio( - sub: TrickleSubscriber, transcriber: Transcriber, meter: TranscriptionMeter -) -> None: +async def _pump_audio(sub: TrickleSubscriber, session: TranscribeSession) -> None: """Read PCM segments off the Trickle input channel and feed the transcriber.""" while True: segment = await sub.next() @@ -107,10 +109,17 @@ async def _pump_audio( chunk = await reader.read() if chunk is None: break - meter.mark_audio(len(chunk)) - await transcriber.feed(chunk) + session.meter.mark_audio(len(chunk)) + await session.transcriber.feed(chunk) with suppress(Exception): await segment.close() + # Report ingest progress after every segment. Trickle deletes unread + # segments the moment the publisher closes, so a client that publishes + # faster than we consume would lose most of its audio. This lets the + # client hold the stream open until we have actually caught up. + await session.event_queue.put( + {"type": "progress", "audio_bytes": session.meter.bytes_in} + ) async def _drain_events(session: TranscribeSession) -> None: @@ -169,9 +178,14 @@ async def _run_pipeline(session: TranscribeSession, in_url: str) -> None: try: await session.transcriber.open() drain = asyncio.create_task(_drain_events(session)) + # Default start_seq joins at the live edge, which is the only option: + # the channel keeps no backlog, so an earlier index answers 470 (no data) + # rather than replaying. Audio is not droppable the way stale video + # frames are, so the client must not move the edge forward until we have + # read the current segment — see the progress events below. async with TrickleSubscriber(in_url) as sub: try: - await _pump_audio(sub, session.transcriber, session.meter) + await _pump_audio(sub, session) finally: # Read the transport counters while the subscriber is still open, # so a mid-stream failure still reports how far the audio got. diff --git a/vllm-realtime/stats.py b/vllm-realtime/stats.py index 618e2a9..6bcf91f 100644 --- a/vllm-realtime/stats.py +++ b/vllm-realtime/stats.py @@ -21,14 +21,14 @@ do not exist for us. Audio duration is derived from bytes instead. 2. ``realtime_factor`` is wall time / audio duration, so it measures the whole - pipeline, not the model. The client paces audio to real time, so wall time can - never drop below the audio duration: the factor is pinned just above 1.0 no - matter how fast the GPU is. Read it as "did the pipeline keep up?" — it holds - near 1.0 while the backend keeps pace and climbs once it falls behind. - Publishing unpaced does *not* turn it into a throughput measurement: Trickle - deletes unread segments when the publisher closes, so an unpaced run drops most - of its audio and measures nothing (see FEEDBACK.md). The honest latency signals - are ``time_to_first_word_s`` and ``finalize_tail_s``. + pipeline, not the model. Under the client's default realtime pacing, wall time + can never drop below the audio duration and the factor is pinned just above + 1.0 however fast the backend is — there, read it as "did the pipeline keep + up?". Run the client with ``--no-realtime`` to remove the pacing floor and let + it measure real throughput; the client applies backpressure from this app's + ``progress`` events, so publishing unpaced no longer outruns Trickle's segment + retention. ``time_to_first_word_s`` moves with lead-in silence in the audio; + ``finalize_tail_s`` is the stable latency figure. """ from __future__ import annotations @@ -181,6 +181,11 @@ def __init__(self) -> None: self._deltas = 0 self._words = 0 + @property + def bytes_in(self) -> int: + """Audio bytes consumed so far — reported to the client as backpressure.""" + return self._bytes_in + def mark_audio(self, nbytes: int) -> None: if self._t_first_audio is None: self._t_first_audio = time.time() From d4c531f5e7d982c4db47a7bd2177e6e3df382f30 Mon Sep 17 00:00:00 2001 From: Gideon Jones <109116843+Gideonjon@users.noreply.github.com> Date: Thu, 16 Jul 2026 00:20:58 +0100 Subject: [PATCH 4/4] feat(vllm-realtime): report the Trickle publish side too The performance summary metered the runner's ingest (TrickleSubscriber) and the WebSocket, but never the client's own publish path, even though TricklePublisher.get_stats() offers it for free: bytes submitted, segments started/completed/failed, post successes and retries. Report it, and order the three transport lines along the audio's actual path -- publish, ingest, websocket out. Pairing the two SDK lines is the point: publish segments=14/14 against ingest segments=14 proves nothing was dropped between them. That check is worth surfacing because the transport will not raise on loss -- a run that shredded 13 of 14 segments still reported seq_gap_events=0. The publish rate also independently confirms what each mode measures. Paced sustains 30 kB/s, exactly PCM16 mono @16kHz realtime. Unpaced sustains 1159 kB/s -- 38x faster -- and still lands 14/14, which is the backpressure doing its job at full tilt. Verified on GPU (paced, unpaced, --language en) and against the mock backend. --- vllm-realtime/FEEDBACK.md | 12 ++++++---- vllm-realtime/README.md | 33 ++++++++++++++++---------- vllm-realtime/client.py | 50 +++++++++++++++++++++++++++++++-------- 3 files changed, 69 insertions(+), 26 deletions(-) diff --git a/vllm-realtime/FEEDBACK.md b/vllm-realtime/FEEDBACK.md index fee0211..df9d9a6 100644 --- a/vllm-realtime/FEEDBACK.md +++ b/vllm-realtime/FEEDBACK.md @@ -44,7 +44,10 @@ was confusing, and the open questions a second builder would hit. segment, waits for the runner to report it consumed it over the WebSocket, then publishes the next. A documented "lossless vs live-edge" subscribe mode (or any retention window) would spare every non-video app from rediscovering - this the hard way. + this the hard way. Worth noting the loss *is* detectable, but only by pairing + both sides: `TricklePublisher.get_stats().segments_completed` against + `TrickleSubscriber.get_stats().segments_delivered`. Neither number is alarming + alone. Nothing in the API suggests you should compare them. 3. **Importing the SDK pulls in PyAV.** `livepeer_gateway/__init__.py` imports the media modules at top level, so `import livepeer_gateway` (even just for Trickle) requires `av`. It installs transitively, but a from-source build on a @@ -127,6 +130,7 @@ was confusing, and the open questions a second builder would hit. 1.13x, time to first word 2.4 s. The tail is the stable figure — it reproduced within ~10 ms across runs; first-word moves with whatever silence precedes speech in the clip. - - *Unpaced with backpressure* (throughput): wall clock **1.78 s** for 6.56 s of - audio — real-time factor **0.27x**, i.e. ~3.7x faster than realtime, with an - identical transcript. + - *Unpaced with backpressure* (throughput): wall clock **1.57 s** for 6.56 s of + audio — real-time factor **0.24x**, i.e. ~4x faster than realtime, with an + identical transcript. The publisher sustained 1159 kB/s (38x the paced + 30 kB/s) and still delivered 14/14 segments. diff --git a/vllm-realtime/README.md b/vllm-realtime/README.md index 9d07016..2eba08b 100644 --- a/vllm-realtime/README.md +++ b/vllm-realtime/README.md @@ -84,21 +84,28 @@ prints as a summary. It has three parts: ``` ──── performance ──── audio 6.56 s - wall clock 7.395 s - real-time factor 1.127x - time to first word 2.418 s - finalize tail 0.374 s + wall clock 7.405 s + real-time factor 1.129x + time to first word 2.392 s + finalize tail 0.359 s words / deltas 15 / 28 - trickle in (SDK) segments=14 seq_gaps=0 retries=0 failures=0 stall=7009ms - websocket out (app) events=43 deltas=28 bytes=4375 failures=0 cmds_in=1 + trickle publish (SDK) segments=14/14 bytes=209920 (30 kB/s) posts_ok=14 failed=0 retries=0 + trickle ingest (SDK) segments=14 seq_gaps=0 retries=0 failures=0 stall=7036ms + websocket out (app) events=43 deltas=28 bytes=4375 failures=0 cmds_in=1 ``` -**Trickle in** comes free from the SDK (`TrickleSubscriber.get_stats()`) and is -transport health: `seq_gaps` above 0 means audio was dropped. **WebSocket out** -and the latency numbers are metered by this app in [stats.py](stats.py) — the SDK -does not meter WebSockets, so an app that streams results over one has to count -them itself. +The three transport lines follow the audio's real path: the client publishes, the +runner ingests, the runner streams results back. + +- **trickle publish** and **trickle ingest** come free from the SDK + (`TricklePublisher.get_stats()` and `TrickleSubscriber.get_stats()`). Reading + them *together* is the useful part — `publish segments=14/14` against `ingest + segments=14` proves nothing was dropped in between. That check matters because + the transport will not tell you: dropped audio still reports `seq_gaps=0`. +- **websocket out** and every latency figure are metered by this app in + [stats.py](stats.py). The SDK does not meter WebSockets, so an app that streams + its results over one has to count them itself. ### Two modes, two different numbers @@ -112,7 +119,9 @@ figure: it moves with whatever silence precedes speech in the clip. Run with `--no-realtime` to remove the pacing floor and measure real throughput: ``` -audio 6.56 s · wall clock 1.78 s · real-time factor 0.271x → ~3.7x realtime +audio 6.56 s · wall clock 1.57 s · real-time factor 0.24x → ~4x realtime +trickle publish segments=14/14 bytes=209920 (1159 kB/s) → 38x the paced rate, +trickle ingest segments=14 seq_gaps=0 still zero loss ``` Both are honest; they measure different things. Paced gives live latency, unpaced diff --git a/vllm-realtime/client.py b/vllm-realtime/client.py index 533efe2..83cc6b0 100644 --- a/vllm-realtime/client.py +++ b/vllm-realtime/client.py @@ -27,6 +27,7 @@ import struct import wave from contextlib import suppress +from typing import Optional import aiohttp @@ -34,7 +35,7 @@ from livepeer_gateway.http import post_json from livepeer_gateway.live_runner import stop_runner_session from livepeer_gateway.selection import reserve_session -from livepeer_gateway.trickle_publisher import TricklePublisher +from livepeer_gateway.trickle_publisher import TricklePublisher, TricklePublisherStats DEFAULT_DISCOVERY = "http://localhost:8935/discovery" APP_ID = "livepeer-sample/vllm-realtime" @@ -148,8 +149,8 @@ async def wait_for(self, target: int, timeout: float) -> bool: return True -def _print_stats(data: dict) -> None: - """Render the runner's final stats event.""" +def _print_stats(data: dict, publisher: Optional[TricklePublisherStats]) -> None: + """Render the whole path: our publish side, then the runner's stats event.""" t = data.get("transcription") or {} w = data.get("websocket") or {} k = data.get("trickle") or {} @@ -161,14 +162,30 @@ def _print_stats(data: dict) -> None: print(f" time to first word {t.get('time_to_first_word_s')} s") print(f" finalize tail {t.get('finalize_tail_s')} s") print(f" words / deltas {t.get('words')} / {t.get('deltas')}") + + # Three hops, in order: we publish -> the runner ingests -> it streams back. + # The first two come free from the SDK; the third we count ourselves. + if publisher is not None: + rate = ( + publisher.bytes_submitted_to_transport / publisher.elapsed_s + if publisher.elapsed_s > 0 + else 0.0 + ) + print( + "\n trickle publish (SDK) " + f"segments={publisher.segments_completed}/{publisher.segments_started} " + f"bytes={publisher.bytes_submitted_to_transport} " + f"({rate / 1000:.0f} kB/s) posts_ok={publisher.post_success} " + f"failed={publisher.segments_failed} retries={publisher.post_retries_no_body_consumed}" + ) print( - "\n trickle in (SDK) " + " trickle ingest (SDK) " f"segments={k.get('segments_delivered')} seq_gaps={k.get('seq_gap_events')} " f"retries={k.get('get_retries')} failures={k.get('get_failures')} " f"stall={k.get('wait_ms_total')}ms" ) print( - " websocket out (app) " + " websocket out (app) " f"events={w.get('events_sent')} deltas={w.get('deltas_sent')} " f"bytes={w.get('bytes_sent')} failures={w.get('send_failures')} " f"cmds_in={w.get('commands_received')}" @@ -185,6 +202,7 @@ async def _read_transcript( ws: aiohttp.ClientWebSocketResponse, done: asyncio.Event, progress: _Progress, + stats_sink: dict, ) -> None: """Print transcript + metrics as JSON events arrive on the WebSocket.""" try: @@ -202,8 +220,9 @@ async def _read_transcript( progress.update(data.get("audio_bytes", 0)) continue if kind == "stats": - # Sent after "done"; last event of the session. - _print_stats(data) + # Sent after "done"; last event of the session. Hand it back to + # main, which pairs it with our publish-side stats and prints. + stats_sink.update(data) break if kind == "done": print( @@ -223,7 +242,7 @@ async def _read_transcript( async def _publish_pcm( in_url: str, pcm: bytes, realtime: bool, progress: _Progress -) -> None: +) -> TricklePublisherStats: """Publish PCM to the Trickle input channel, ~0.5 s per segment, 40 ms per frame. A Trickle channel keeps no backlog — a subscriber reads the live edge, and an @@ -257,6 +276,9 @@ async def _publish_pcm( ) break + # Read after close so the counters include the final segment flush. + return pub.get_stats() + async def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") @@ -309,9 +331,12 @@ async def main() -> None: done = asyncio.Event() progress = _Progress() - reader = asyncio.create_task(_read_transcript(ws, done, progress)) + stats_sink: dict = {} + reader = asyncio.create_task( + _read_transcript(ws, done, progress, stats_sink) + ) - await _publish_pcm( + publisher_stats = await _publish_pcm( in_url, pcm, realtime=not args.no_realtime, progress=progress ) @@ -321,6 +346,11 @@ async def main() -> None: with suppress(Exception): await reader + if stats_sink: + _print_stats(stats_sink, publisher_stats) + else: + log.warning("no stats event received from the runner") + except LivepeerGatewayError as exc: raise SystemExit(f"ERROR: {exc}") from exc finally: