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..97eca07 --- /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 stats.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..df9d9a6 --- /dev/null +++ b/vllm-realtime/FEEDBACK.md @@ -0,0 +1,136 @@ +# 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 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. 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 + 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. + +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. **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 + +- 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. +- 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 + +- 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. +- 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.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 new file mode 100644 index 0000000..2eba08b --- /dev/null +++ b/vllm-realtime/README.md @@ -0,0 +1,161 @@ +# 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. + +## 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.405 s + real-time factor 1.129x + time to first word 2.392 s + finalize tail 0.359 s + words / deltas 15 / 28 + + 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 +``` + +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 + +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. + +Run with `--no-realtime` to remove the pacing floor and measure real throughput: + +``` +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 +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) + +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..83cc6b0 --- /dev/null +++ b/vllm-realtime/client.py @@ -0,0 +1,363 @@ +#!/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", then print the + "stats" event the runner sends last (see stats.py) +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 +from typing import Optional + +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, TricklePublisherStats + +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" + +# 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") + + +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 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="", + 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 + + +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, 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 {} + + 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')}") + + # 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( + " 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) " + 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 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, + stats_sink: dict, +) -> 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 == "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. Hand it back to + # main, which pairs it with our publish-side stats and prints. + stats_sink.update(data) + break + if kind == "done": + print( + f"\n[done] {transcript!r} " + f"words={data.get('word_count')} sentiment={data.get('sentiment')}" + ) + continue + 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, progress: _Progress +) -> 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 + 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] + 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) + 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 + + # 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") + 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() + progress = _Progress() + stats_sink: dict = {} + reader = asyncio.create_task( + _read_transcript(ws, done, progress, stats_sink) + ) + + publisher_stats = 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) + reader.cancel() + 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: + 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..f6b1966 --- /dev/null +++ b/vllm-realtime/runner.py @@ -0,0 +1,360 @@ +#!/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. + + {"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). + +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 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, TrickleSubscriberStats + +from stats import TranscriptionMeter, WebSocketMeter +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 + # 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"} + + +# 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(sub: TrickleSubscriber, session: TranscribeSession) -> None: + """Read PCM segments off the Trickle input channel and feed the transcriber.""" + 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 + 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: + """Read normalized transcriber events, compute metrics, push to the WS queue.""" + transcript = "" + 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() + 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 "", + "transcript": clean, + "word_count": word_count(clean), + "sentiment": simple_sentiment(clean), + } + ) + + +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 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) + 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 session.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") + await session.event_queue.put(_stats_event(session, trickle_stats)) + # Signal the WS drain coroutine that no more events are coming. + await session.event_queue.put(_WS_DONE) + log.info("transcription pipeline finished") + + +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 session.event_queue.get() + if event is _WS_DONE: + await ws.close() + return + # 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 + + +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(session, in_internal_url)) + 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, 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: + 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/stats.py b/vllm-realtime/stats.py new file mode 100644 index 0000000..6bcf91f --- /dev/null +++ b/vllm-realtime/stats.py @@ -0,0 +1,231 @@ +#!/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. 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 + +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 + + @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() + 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, + ) 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"