diff --git a/.github/workflows/images.yml b/.github/workflows/images.yml index 0915b21..b612c71 100644 --- a/.github/workflows/images.yml +++ b/.github/workflows/images.yml @@ -16,6 +16,7 @@ on: - "hello-world/**" - "echo/**" - "tiles/**" + - "realtime-transcription/**" - ".github/workflows/images.yml" pull_request: paths: *image_paths @@ -34,7 +35,7 @@ jobs: strategy: fail-fast: false matrix: - example: [hello-world, echo, tiles] + example: [hello-world, echo, tiles, realtime-transcription] steps: - uses: actions/checkout@v7 diff --git a/README.md b/README.md index 53b5bda..97a98cb 100644 --- a/README.md +++ b/README.md @@ -29,21 +29,23 @@ flowchart LR The orchestrator is a **transparent reverse proxy**: every endpoint you expose is passed through to your app unchanged, so you write an ordinary service and it runs on the network as-is. The transports supported today: - **HTTP** request/response — the common case. (`hello-world`, `tiles`, `api-proxy`) -- **HTTP + SSE** — streamed / token responses. (`vllm`) +- **HTTP + SSE** — streamed / token responses. (`vllm`, `ollama`) - **Trickle** — continuous realtime video in/out. (`echo`) -- **WebSocket** — long-lived bidirectional sessions. (external: `scope`) +- **WebSocket** — long-lived bidirectional sessions. (`realtime-transcription`) Need a schema that isn't here? [Open an issue](https://github.com/livepeer/runner-app-examples/issues). ## Examples -| Example | Goal | Registration | Mode | Transport | Pricing | -| ------------------------------ | ------------------------------------------------------------------------------------- | ------------ | ---------------------------------- | ----------------- | ------- | -| [`hello-world`](./hello-world) | The simplest app: one request, one response | dynamic | single-shot | HTTP (JSON) | fixed | -| [`tiles`](./tiles) | Capacity fan-out — one call per tile | dynamic | single-shot | HTTP (base64 PNG) | fixed | -| [`api-proxy`](./api-proxy) | Pass calls through to a hosted API — the operator holds the key, callers pay per call | static | single-shot | HTTP (JPEG bytes) | fixed | -| [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | hour | -| [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | persistent (single-shot by nature) | HTTP + SSE | hour | +| Example | Goal | Registration | Mode | Transport | Pricing | +| ---------------------------------------------------- | ------------------------------------------------------------------------------------- | ------------ | ----------- | ----------------- | ------- | +| [`hello-world`](./hello-world) | The simplest app: one request, one response | dynamic | single-shot | HTTP (JSON) | fixed | +| [`tiles`](./tiles) | Capacity fan-out — one call per tile | dynamic | single-shot | HTTP (base64 PNG) | fixed | +| [`api-proxy`](./api-proxy) | Pass calls through to a hosted API — the operator holds the key, callers pay per call | static | single-shot | HTTP (JPEG bytes) | fixed | +| [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | hour | +| [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | single-shot | HTTP + SSE | hour | +| [`realtime-transcription`](./realtime-transcription) | Audio up, transcripts back, on one socket | dynamic | persistent | WebSocket | hour | +| [`ollama`](./ollama) | One container, several models, each its own priced app | dynamic | single-shot | HTTP + SSE | hour | 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. @@ -53,9 +55,11 @@ This set stays **minimal and curated**: it covers each value of the axes above ( How the app attaches to the orchestrator: -- **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`) +- **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`, `realtime-transcription`, `ollama`) - **Static** — the orchestrator is configured with the app's URL in a `runners.json` and health-polls it; the app needs no SDK. Best for fixed, long-running deployments. (`vllm`, `api-proxy`) +Both forms also take an optional **`metadata`** string: up to 1 KB of app-controlled UTF-8, echoed back in `/discovery` and never read by the orchestrator. Use it to pass callers app-specific detail the protocol doesn't model, such as a context window or the languages a model handles; clients read it off the discovered runner, as `cursor.candidates[0].raw["metadata"]` after `runner_selector`, or `session.runner.raw["metadata"]` after `reserve_session`. Anything a caller **selects or pays differently for** belongs in the app id instead, which is the only key discovery can filter — none of the examples here need `metadata` for that reason. + The arrow flips — dynamic, the app announces itself; static, the orchestrator is told about a passive app: ```mermaid @@ -74,18 +78,15 @@ flowchart LR Chosen _at_ registration (above); **defaults to `persistent`** — set on both `register_runner(...)` and in `runners.json`. The examples set it explicitly. -- **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `vllm`) -- **Single-shot** — one request in, one response out; the orchestrator reserves a session per call and releases it when the response returns, so the client manages no session at all. Best for batch / request-response. (`hello-world`, `tiles`, `api-proxy`) - -> [!NOTE] -> The `vllm` example is single-shot by nature but stays **persistent** for now: it meters per second across a reserved session, and true per-token billing is brokerage for the gateway/signer layer. +- **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `realtime-transcription`) +- **Single-shot** — one request in, one response out; the orchestrator reserves a session per call and releases it when the response returns, so the client manages no session at all. Best for batch / request-response. With metered pricing the call pays for as long as it runs, so the work need not be short. (`hello-world`, `tiles`, `api-proxy`, `vllm`, `ollama`) ## Calling your app The client side depends on the runner's mode: -- **Single-shot** — **discover → call**: find the app via `runner_selector`, then one `call_runner`. The orchestrator reserves a session for the call and releases it when the response returns; on the paid path `call_runner` answers the 402 payment challenge inline. (`hello-world`, `tiles`, `api-proxy`) -- **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `vllm`) +- **Single-shot** — **discover → call**: find the app via `runner_selector`, then one `call_runner`. The orchestrator reserves a session for the call and releases it when the response returns; on the paid path `call_runner` answers the 402 payment challenge inline. (`hello-world`, `tiles`, `api-proxy`, `vllm`, `ollama`) +- **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `realtime-transcription`) Each example's `client.py` shows its exact calls — grep `# Livepeer:` to find them. diff --git a/ollama/.env.example b/ollama/.env.example new file mode 100644 index 0000000..8387f16 --- /dev/null +++ b/ollama/.env.example @@ -0,0 +1,37 @@ +# Copy to .env (gitignored) and fill in. Never commit secrets. +# Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only. + +# Models to pull, space separated. Which models exist is config; which models get +# advertised is discovered from Ollama by the registrar. +MODELS=qwen2.5:0.5b llama3.2:1b +# Concurrent generations the container can really run. The registrar splits this +# across the models, so the advertised total matches the hardware (go-livepeer#4015). +OLLAMA_NUM_PARALLEL=2 +OLLAMA_KEEP_ALIVE=1h + +# Per-model price in USD/hour: operator policy, so it is config rather than +# discovered. Models not listed here register free. +PRICES=qwen2.5:0.5b=0.01,llama3.2:1b=0.02 + +# --- 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 + +# The runner's price lives in runners.json (static runner): USD per hour, +# converted to wei via the price feed and metered per second. +# Signer's max-price cap (payer side) is per billing unit, here one second, so +# it must exceed the runners.json price / 3600 (0.000111USD is ~0.40 USD/hour). +MAX_PRICE_PER_UNIT=0.000111USD diff --git a/ollama/Dockerfile b/ollama/Dockerfile new file mode 100644 index 0000000..bcacd29 --- /dev/null +++ b/ollama/Dockerfile @@ -0,0 +1,20 @@ +# Registrar sidecar: the only Livepeer code in this example. Ollama itself is the +# stock upstream image with nothing added to it. +FROM python:3.12-slim + +# Flush stdout/stderr immediately so output isn't block-buffered in `docker logs`. +ENV PYTHONUNBUFFERED=1 + +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. +RUN pip install --no-cache-dir \ + aiohttp \ + "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@ja/live-runner" + +WORKDIR /app +COPY registrar.py ./ + +CMD ["python", "registrar.py"] diff --git a/ollama/README.md b/ollama/README.md new file mode 100644 index 0000000..767abf3 --- /dev/null +++ b/ollama/README.md @@ -0,0 +1,103 @@ +# Ollama app (one container, several priced models) + +One Ollama container serving several models, each registered as **its own Live Runner app** with its own price. This is the example about **multiple capabilities from one process**: every other example here registers exactly once. Ollama itself is the stock upstream image with no Livepeer code in it — a **registrar sidecar** does the registering, which is what wrapping software you did not write looks like. + +| | | +| ------------ | ------------------------------------------------ | +| App ids | `ollama/`, one per pulled model | +| Runner mode | single-shot (one session per call) | +| Registration | dynamic (a sidecar self-registers via the SDK) | +| Transport | HTTP + SSE (OpenAI `/v1/chat/completions`) | +| Pricing | hour (metered per second of the call), per model | +| Port | 11434 (Ollama), 8080 (gateway) | + +Runs on GPU or CPU — drop the `deploy` block in `compose.yml` for CPU-only, and expect it to be slow. Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared on-chain/payment setup are in the [repo README](../README.md). + +## How it's wired + +Three moving parts, and only one of them knows about Livepeer: + +- **Ollama** — the stock image. Never modified, never aware of anything. +- **`registrar.py`** — a sidecar that asks Ollama what it has (`GET /api/tags`) and calls `register_runner` once per model, each pointing at the same Ollama URL. +- **`gateway.py`** — a host-side OpenAI endpoint that maps the OpenAI `model` field onto an app id and forwards. Any OpenAI client works against it unchanged. + +**Which models exist is discovered; what they cost is configured.** The registrar never has a model list of its own: `ollama pull` something and it appears on the network at the next start. Prices are operator policy, so they come from `PRICES` in `.env`. + +**One model, one app id, one price.** A runner carries exactly one `mode` and one `price_info`, and discovery can only filter on `app`, so several capabilities means several registrations. `qwen2.5:0.5b` and `llama3.2:1b` are genuinely different products at different prices, and a caller picks between them the same way they pick between orchestrators. + +**The app id is the model name, verbatim.** `llama3.2:1b` registers as `ollama/llama3.2:1b` — go-livepeer only requires an app id be non-empty and trimmed, so there is no reason to slug it. Keeping it exact makes the mapping reversible in both directions, which is why nothing here needs a `metadata` field to restate the name: what you discover is what you send. + +**Listing models is free.** `GET /v1/models` is answered from `/discovery`, a plain GET with no session and no payment, so it reports what **the network** offers rather than what one container holds. Only the forward reserves a session — and because the runners are single-shot and metered, that call pays for as long as the generation runs. + +## Capacity, and why it is hand-sized + +`OLLAMA_NUM_PARALLEL` says how many generations the container will really run at once. The registrar divides that across the models it registers, so the **sum** of the advertised capacities equals what the hardware can do. + +That arithmetic is manual on purpose. Each registration carries its own capacity counter, and **the orchestrator does not know they share a GPU** — see [Improvements this example is waiting on](#improvements-this-example-is-waiting-on). Register two models at `capacity: 1` each on a box that can only run one generation, and the orchestrator will cheerfully admit two sessions; both callers pay and both get contention instead of the 503 that would have been correct. + +Both models stay resident under `OLLAMA_KEEP_ALIVE`, so a request never waits for a swap. + +## Run offchain (free) + +```sh +docker compose up -d --build # pulls the models, then registers one app per model +curl -sk https://localhost:8935/discovery | jq '.[].runners[] | {app, capacity, price_info}' +uv run gateway.py --discovery https://localhost:8935/discovery & # OpenAI endpoint on :8080 +uv run client.py --model qwen2.5:0.5b --prompt "In one sentence, what is Livepeer?" +uv run client.py --model llama3.2:1b --prompt "write a haiku about GPUs" +kill %1; docker compose down # stop the gateway, then the stack +``` + +Ask the network what it serves, with a stock OpenAI client: + +```sh +curl -s http://localhost:8080/v1/models | jq '.data[].id' +# "llama3.2:1b" +# "qwen2.5:0.5b" +``` + +Add a model without touching any config: + +```sh +docker compose exec ollama ollama pull gemma3:270m +docker compose restart registrar # picks it up from /api/tags +``` + +It registers at the fallback price, since `PRICES` does not mention it. + +## Run on-chain (paid) + +Layer `compose.onchain.yml` to add a remote signer and run the orchestrator on-chain. 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) in the repo README. + +```sh +cp .env.example .env # fill in RPC, network, keystore paths, accounts, prices +docker compose -f compose.yml -f compose.onchain.yml up -d --build +uv run gateway.py --signer http://localhost:7936 --discovery https://localhost:8935/discovery & +uv run client.py --model llama3.2:1b --prompt "In one sentence, what is Livepeer?" +kill %1; docker compose -f compose.yml -f compose.onchain.yml down +``` + +There is no `runners.json` here: registration is dynamic, so each model's price comes from the registrar's `PRICES`. Pick a costlier model and the caller pays more, which is the whole reason each model is its own app. + +## Improvements this example is waiting on + +Two gaps are tracked in [go-livepeer#4015](https://github.com/livepeer/go-livepeer/issues/4015), and both are visible here: + +- **Capacity is tracked per registration.** Nothing tells the orchestrator that these runners share one GPU, so the advertised total is whatever the registrations happen to add up to. This example sizes that sum by hand from `OLLAMA_NUM_PARALLEL`; a `pool` field on the heartbeat would let the orchestrator enforce it instead, and refuse the surplus with a 503. +- **Only one GPU can be described.** `LiveRunnerGPU` is a single struct, so a two-card host advertises one card and half its VRAM. + +A third piece is needed for the version that tracks capacity **dynamically** rather than by arithmetic — flipping every registration's `status` off a shared semaphore as work starts and finishes. That works today in principle, but the SDK exposes neither a public setter for `status` nor a way to force an immediate heartbeat, so it would mean reaching into private attributes. Not something an example should teach. + +## Run without Docker + +Start an orchestrator built from go-livepeer `v0.9.0` or newer (see [Build from source](https://docs.livepeer.org/v1/orchestrators/guides/install-go-livepeer#build-from-source)) and an Ollama server, then the registrar, gateway, and client directly: + +```sh +./livepeer -orchestrator -useLiveRunners -serviceAddr localhost:8935 -orchSecret abcdef -v 6 +ollama serve & +ollama pull qwen2.5:0.5b +uv run registrar.py --orchestrator https://localhost:8935 --orchSecret abcdef \ + --ollama-url http://localhost:11434 & +uv run gateway.py & +uv run client.py --model qwen2.5:0.5b --prompt "Hello!" +``` diff --git a/ollama/client.py b/ollama/client.py new file mode 100644 index 0000000..612fec9 --- /dev/null +++ b/ollama/client.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Stock OpenAI client — nothing Livepeer-specific in here. + +This is the whole point of the example: ordinary `openai` code with a fixed base_url and +api_key. It has no idea it's talking to Livepeer. The local gateway (gateway.py) sits at +base_url and does discovery + payment, so this works on-chain unchanged. + + docker compose up -d --build # the network side + uv run gateway.py & # the local gateway on :8080 + uv run client.py --prompt "Hello!" + +Any OpenAI tool (this script, another SDK, curl) can use the same base_url. +""" + +from __future__ import annotations + +import argparse + +from openai import OpenAI + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Chat with an Ollama Live Runner using the stock OpenAI library." + ) + parser.add_argument( + "--base-url", + default="http://localhost:8080/v1", + help="The gateway's OpenAI endpoint.", + ) + parser.add_argument( + "--api-key", + default="unused", + help="Ignored by the gateway; the OpenAI client requires a value.", + ) + parser.add_argument("--model", default="qwen2.5:0.5b") + parser.add_argument( + "--prompt", default="In one sentence, what is the Livepeer network?" + ) + parser.add_argument( + "--stream", action="store_true", help="Stream tokens as they arrive (SSE)." + ) + return parser.parse_args() + + +def main() -> None: + args = _parse_args() + client = OpenAI(base_url=args.base_url, api_key=args.api_key) + messages = [{"role": "user", "content": args.prompt}] + if args.stream: + stream = client.chat.completions.create( + model=args.model, messages=messages, stream=True + ) + for chunk in stream: + print(chunk.choices[0].delta.content or "", end="", flush=True) + print() + return + completion = client.chat.completions.create(model=args.model, messages=messages) + print(completion.choices[0].message.content) + + +if __name__ == "__main__": + main() diff --git a/ollama/compose.onchain.yml b/ollama/compose.onchain.yml new file mode 100644 index 0000000..fc0f1be --- /dev/null +++ b/ollama/compose.onchain.yml @@ -0,0 +1,23 @@ +# On-chain payment overlay for ollama. Layer it on the offchain base: +# docker compose -f compose.yml -f compose.onchain.yml up -d --build +# +# Adds the shared remote signer and re-points the orchestrator on-chain (see +# ../compose.onchain.yml). Registration is dynamic, so there is no runners.json to +# mount: the registrar advertises each model's price from PRICES. Requires a local +# .env (gitignored); copy .env.example and fill it in. Then the gateway pays per +# call through the signer: +# uv run gateway.py --signer http://localhost:7936 & +# uv run client.py --model qwen2.5:0.5b --prompt "Hello!" + +services: + signer: + extends: + file: ../compose.onchain.yml + service: signer + ports: + - "7936:7936" + + orchestrator: + extends: + file: ../compose.onchain.yml + service: orchestrator diff --git a/ollama/compose.yml b/ollama/compose.yml new file mode 100644 index 0000000..ac0d990 --- /dev/null +++ b/ollama/compose.yml @@ -0,0 +1,80 @@ +# Offchain demo: orchestrator + one Ollama container serving several models, each +# registered as its own Live Runner app by the registrar sidecar. +# +# docker compose up -d --build +# uv run gateway.py & # OpenAI endpoint on :8080 (host) +# uv run client.py --model qwen2.5:0.5b --prompt "Hello!" +# +# Ollama itself has no Livepeer code: the registrar sits beside it and declares the +# capabilities, which is what wrapping third-party software looks like. The gateway +# and OpenAI client run on the host, like the vllm example. + +services: + orchestrator: + extends: + file: ../compose.orchestrator.yml + service: orchestrator + + ollama: + image: ollama/ollama:latest + container_name: example_apps_ollama + environment: + # Concurrent generations the container will really run. The registrar splits + # this across the models it registers so the advertised total matches it. + - OLLAMA_NUM_PARALLEL=${OLLAMA_NUM_PARALLEL:-2} + # Keep both models resident so a request never waits for a swap. + - OLLAMA_KEEP_ALIVE=${OLLAMA_KEEP_ALIVE:-1h} + volumes: + - ollama:/root/.ollama # persist pulled models across runs + # Ollama runs on CPU too; drop this block to try it without a GPU. + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: 1 + capabilities: [gpu] + healthcheck: + test: ["CMD", "ollama", "list"] # talks to the server; no curl in the image + interval: 5s + timeout: 3s + retries: 20 + start_period: 5s + + # One-shot: pull the models, then exit. Which models to install is config; which + # models get advertised is discovered from Ollama by the registrar. + ollama-pull: + image: ollama/ollama:latest + environment: + - OLLAMA_HOST=http://ollama:11434 + # Must be in the container's env: the $$ below defers expansion to its shell. + - MODELS=${MODELS:-qwen2.5:0.5b llama3.2:1b} + depends_on: + ollama: + condition: service_healthy + entrypoint: ["/bin/sh", "-c"] + # Single-element list: the whole script must reach `sh -c` as one argument. + # A scalar string here gets tokenized, so `sh -c` would see only `ollama`. + command: + - "for m in $${MODELS:-qwen2.5:0.5b llama3.2:1b}; do ollama pull $$m; done" + restart: "no" + + registrar: + build: . + container_name: example_apps_ollama_registrar + depends_on: + orchestrator: + condition: service_healthy + ollama-pull: + condition: service_completed_successfully + command: + - python + - registrar.py + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --ollama-url=http://ollama:11434 + - --parallel=${OLLAMA_NUM_PARALLEL:-2} + - --prices=${PRICES:-} + +volumes: + ollama: diff --git a/ollama/gateway.py b/ollama/gateway.py new file mode 100644 index 0000000..63e8ef5 --- /dev/null +++ b/ollama/gateway.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +"""Minimal local OpenAI -> Livepeer gateway, multi-model edition. + +Same idea as the vllm example's gateway, but one Ollama container serves several +models and each is its own Live Runner app. So the gateway does one extra thing: +it maps the OpenAI `model` field onto an app id, and answers `GET /v1/models` from +**discovery** rather than from any one container -- on a network, "which models are +available" is a network question. + + uv run gateway.py --signer http://localhost:7936 & + export OPENAI_BASE_URL=http://localhost:8080/v1 OPENAI_API_KEY=unused + # then plain `openai`, curl, or any SDK, picking a model with `model=...` + +Livepeer integration (grep `# Livepeer:`): + 1. discover_runners() -- list the models the network advertises (free: no session) + 2. runner_selector() -- find runners for the chosen model's app + 3. call_runner() -- forward the request through the orchestrator (pays 402) + +Listing is free because /discovery is a plain GET. Only the forward reserves a +session, and because the runners are single-shot and metered, that call pays for as +long as the generation runs. +""" + +from __future__ import annotations + +import argparse +import logging + +from aiohttp import web + +from livepeer_gateway.discovery import discover_runners +from livepeer_gateway.errors import LivepeerHTTPError +from livepeer_gateway.live_runner import call_runner +from livepeer_gateway.selection import runner_selector + +APP_NAMESPACE = "ollama" +# A generation runs far longer than the SDK's 5s default, and a metered single-shot +# call is meant to pay for as long as it takes. +REQUEST_TIMEOUT = 300.0 + +log = logging.getLogger("ollama-gateway") + + +def _app_id(model: str) -> str: + # Mirrors registrar.py: `llama3.2:1b` -> `ollama/llama3.2:1b`, verbatim. + return f"{APP_NAMESPACE}/{model.strip()}" + + +def _model_of(app: str) -> str: + # ...and back again. Reversible because the id is not a slug. + return app.split("/", 1)[1] if "/" in app else app + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="OpenAI-compatible gateway in front of Ollama Live Runners." + ) + parser.add_argument("--discovery", default="https://localhost:8935/discovery") + parser.add_argument( + "--signer", + default="", + help="Remote signer base URL; omit for the offchain (free) path.", + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8080) + return parser.parse_args() + + +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 + + async def _discovered_models() -> list[str]: + # No app filter: discovery matches app ids exactly, with no prefix support, + # so the family is selected here instead. + entries = await discover_runners(discovery_url=args.discovery) # Livepeer: 1 + models: set[str] = set() + for entry in entries: + for runner in entry.get("runners", []): + app = runner.get("app") + if isinstance(app, str) and app.startswith(f"{APP_NAMESPACE}/"): + models.add(_model_of(app)) + return sorted(m for m in models if m) + + async def _list_models(request: web.Request) -> web.StreamResponse: + found = await _discovered_models() + return web.json_response( + { + "object": "list", + "data": [ + {"id": model, "object": "model", "owned_by": APP_NAMESPACE} + for model in found + ], + } + ) + + async def _forward(request: web.Request) -> web.StreamResponse: + payload = await request.json() if request.can_read_body else {} + runner_path = request.path # e.g. /v1/chat/completions + model = str(payload.get("model", "")).strip() + if not model: + raise web.HTTPBadRequest(text="request must name a model") + + cursor = await runner_selector( # Livepeer: 2 + discovery_url=args.discovery, # omit if the signer does discovery itself + app=_app_id(model), + ) + runner = cursor.candidates[0] + runner_url = runner.url.rstrip("/") + runner_path + + # When the OpenAI client asks for stream=True the runner replies with + # text/event-stream; pipe those chunks straight through so tokens reach the + # client as they arrive instead of buffering the blob. + if payload.get("stream"): + async with await call_runner( # Livepeer: 3 (streaming) + runner=runner, # discovery metadata tells call_runner the price unit + runner_url=runner_url, + payload=payload, + signer_url=signer_url, + timeout=REQUEST_TIMEOUT, + stream=True, + ) as stream: + resp = web.StreamResponse( + status=stream.status, + headers={ + "Content-Type": stream.content_type or "text/event-stream" + }, + ) + await resp.prepare(request) + async for ( + chunk + ) in stream.aiter_bytes(): # raw bytes -> keep SSE framing + await resp.write(chunk) + await resp.write_eof() + return resp + + result = await call_runner( # Livepeer: 3 + runner=runner, # discovery metadata tells call_runner the price unit + runner_url=runner_url, + payload=payload, + signer_url=signer_url, + timeout=REQUEST_TIMEOUT, + ) + return web.json_response(result.data) + + async def _forward_or_error(request: web.Request) -> web.StreamResponse: + # A single-shot call holds a capacity slot for its duration, so a busy runner + # answers 503. Hand that back as JSON an OpenAI client can read. + try: + return await _forward(request) + except LivepeerHTTPError as exc: + return web.json_response( + {"error": {"message": str(exc), "type": "livepeer_error"}}, + status=exc.status_code, + ) + + app = web.Application() + # /v1/models is answered from discovery, so listing costs nothing. Everything + # else is forwarded, and that is what reserves a session and pays. + app.router.add_get("/v1/models", _list_models) + app.router.add_route("*", "/v1/{tail:.*}", _forward_or_error) + log.info( + "gateway on http://%s:%d/v1 -> %s (signer=%s)", + args.host, + args.port, + args.discovery, + signer_url or "none", + ) + web.run_app(app, host=args.host, port=args.port, print=None) + + +if __name__ == "__main__": + main() diff --git a/ollama/pyproject.toml b/ollama/pyproject.toml new file mode 100644 index 0000000..ae7ce45 --- /dev/null +++ b/ollama/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "livepeer-ollama" +version = "0.1.0" +description = "Ollama multi-model example app for the Livepeer network." +requires-python = ">=3.12" +dependencies = [ + "aiohttp", # for gateway.py (the OpenAI -> Livepeer gateway) + "livepeer-gateway", + "openai", # for client.py (stock OpenAI library) +] + +# livepeer-gateway is not on PyPI yet; pull it from the branch. +[tool.uv.sources] +livepeer-gateway = { git = "https://github.com/livepeer/livepeer-python-gateway", branch = "ja/live-runner" } diff --git a/ollama/registrar.py b/ollama/registrar.py new file mode 100644 index 0000000..5459fcd --- /dev/null +++ b/ollama/registrar.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""Registrar sidecar: one Ollama container, one Live Runner app per model. + +Ollama has no Livepeer code in it -- it is the stock upstream image. This process +sits beside it and does the registering, which is the shape most real deployments +take: you wrap software you did not write. + +It asks Ollama what it has (`GET /api/tags`) and registers each model as its own +app, so every model is separately discoverable and separately priced. Nothing here +is hardcoded except operator policy (prices), because which models exist is a fact +about the container, not a choice. + +The app id is the model name verbatim, so a caller can send back exactly what it +discovered and nothing has to restate the name elsewhere. + +Livepeer integration (grep `# Livepeer:`): + 1. register_runner() x N -- one capability per model, all pointing at Ollama + +Capacity is the subtle part. Each registration carries its own counter and the +orchestrator does not know they share a GPU (go-livepeer#4015), so the numbers here +are sized to add up: the SUM across registrations is what the hardware must support, +which is why it is derived from OLLAMA_NUM_PARALLEL rather than set per model. +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import os + +import aiohttp + +from livepeer_gateway.live_runner import register_runner + +APP_NAMESPACE = "ollama" +log = logging.getLogger("ollama-registrar") + + +def _app_id(model: str) -> str: + # `llama3.2:1b` -> `ollama/llama3.2:1b`. The exact name, not a slug: go-livepeer + # only requires an app id be non-empty and trimmed, so keeping it verbatim means + # the mapping is reversible and nothing has to carry the real name separately. + return f"{APP_NAMESPACE}/{model.strip()}" + + +def _parse_prices(raw: str) -> dict[str, float]: + # "qwen2.5:0.5b=0.01,llama3.2:1b=0.02" -- operator policy, so it is config. + prices: dict[str, float] = {} + for item in raw.split(","): + if "=" in item: + name, _, value = item.partition("=") + prices[name.strip()] = float(value) + return prices + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Register Ollama models as Live Runners." + ) + parser.add_argument("--orchestrator", default="https://localhost:8935") + parser.add_argument("--orchSecret", default="abcdef") + parser.add_argument("--ollama-url", default="http://ollama:11434") + parser.add_argument( + "--parallel", + type=int, + default=int(os.environ.get("OLLAMA_NUM_PARALLEL", "2")), + help="Concurrent generations the container can really run; split across models.", + ) + parser.add_argument( + "--prices", + default=os.environ.get("PRICES", ""), + help="model=usd_per_hour pairs, comma separated. Unlisted models use --price.", + ) + parser.add_argument("--price", type=float, default=0.0, help="Fallback USD/hour.") + return parser.parse_args() + + +async def _installed_models(session: aiohttp.ClientSession, base_url: str) -> list[str]: + # Wait for Ollama and the puller: an empty list means nothing is pulled yet. + for _ in range(60): + try: + async with session.get(f"{base_url.rstrip('/')}/api/tags") as resp: + data = await resp.json() + models = [m["name"] for m in data.get("models", []) if m.get("name")] + if models: + return sorted(models) + except aiohttp.ClientError: + pass + await asyncio.sleep(5) + raise SystemExit(f"ERROR: no models pulled at {base_url} after 5 minutes") + + +async def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + prices = _parse_prices(args.prices) + + async with aiohttp.ClientSession() as session: + models = await _installed_models(session, args.ollama_url) + + # Split the container's real concurrency across the models it serves, so the + # advertised total matches the hardware instead of multiplying by model count. + # capacity 0 is not expressible (the orchestrator coerces it to 1), so with more + # models than parallel slots the total unavoidably overshoots -- say so loudly + # rather than quietly advertising capacity the GPU does not have. + per_model = max(1, args.parallel // len(models)) + advertised = per_model * len(models) + if advertised > args.parallel: + log.warning( + "advertising %d slots across %d models but the container runs %d at once: " + "raise OLLAMA_NUM_PARALLEL to at least the model count, or pull fewer " + "models. The orchestrator cannot know these share a GPU " + "(https://github.com/livepeer/go-livepeer/issues/4015).", + advertised, + len(models), + args.parallel, + ) + registrations = [] + for model in models: + registrations.append( + await register_runner( # Livepeer: 1 + args.orchestrator, + secret=args.orchSecret, + runner_url=args.ollama_url, + app=_app_id(model), + mode="single-shot", # one request in, one response out + price=prices.get(model, args.price), # USD/hour, metered while it runs + capacity=per_model, + ) + ) + log.info("registered %s as %s (capacity=%d)", model, _app_id(model), per_model) + + log.info("%d model(s) registered; %d slots advertised", len(models), advertised) + try: + await asyncio.Event().wait() # heartbeats run in the background + finally: + for registration in registrations: + await registration.close() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/realtime-transcription/.env.example b/realtime-transcription/.env.example new file mode 100644 index 0000000..4005e9b --- /dev/null +++ b/realtime-transcription/.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. + +# Device only: the model is fixed at large-v3-turbo, which needs a GPU to stay +# realtime. cpu/int8 loads but falls behind a live stream. +WHISPER_DEVICE=cuda +WHISPER_COMPUTE=float16 + +# --- 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 + +# Runner price (on-chain): USD per hour, metered per second while the socket is +# open. Keep under ~0.67: the signer signs at most 100 tickets per payment, and +# the demo orchestrator runs -ticketEV=1e10 (fee / ticketEV). +PRICE=0.01 +# Signer's max-price cap (payer side), per billing unit. Metered here, so the +# unit is one second and must exceed PRICE / 3600 (0.000111USD is ~0.40/hour). +MAX_PRICE_PER_UNIT=0.000111USD diff --git a/realtime-transcription/.gitignore b/realtime-transcription/.gitignore new file mode 100644 index 0000000..a6ebfbb --- /dev/null +++ b/realtime-transcription/.gitignore @@ -0,0 +1,5 @@ +# Local audio and demo assets: the README shows how to make sample.wav. +*.wav +*.mp4 +*.srt +media/ diff --git a/realtime-transcription/Dockerfile b/realtime-transcription/Dockerfile new file mode 100644 index 0000000..e7c377f --- /dev/null +++ b/realtime-transcription/Dockerfile @@ -0,0 +1,29 @@ +# Realtime-transcription app: an aiohttp WebSocket server wrapping faster-whisper +# that self-registers as a Live Runner. GPU only: large-v3-turbo is the model that +# is both accurate and fast enough to keep pace with a live stream, and it needs a +# CUDA runtime (CTranslate2 wants cuBLAS + cuDNN). +FROM python:3.12-slim + +# Flush stdout/stderr immediately so output isn't block-buffered in `docker logs`. +ENV PYTHONUNBUFFERED=1 + +RUN apt-get update \ + && apt-get install -y --no-install-recommends git \ + && rm -rf /var/lib/apt/lists/* + +# CTranslate2 loads cuBLAS/cuDNN from the nvidia pip wheels, so no CUDA base image +# is needed -- the host driver comes in via the compose `deploy` reservation. +# livepeer-gateway SDK isn't on PyPI yet; install from Git. +RUN pip install --no-cache-dir \ + faster-whisper numpy \ + nvidia-cublas-cu12 nvidia-cudnn-cu12 \ + "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@ja/live-runner" + +ENV LD_LIBRARY_PATH=/usr/local/lib/python3.12/site-packages/nvidia/cublas/lib:/usr/local/lib/python3.12/site-packages/nvidia/cudnn/lib + +WORKDIR /app +COPY runner.py client.py ./ + +EXPOSE 5005 + +CMD ["python", "runner.py"] diff --git a/realtime-transcription/README.md b/realtime-transcription/README.md new file mode 100644 index 0000000..f243705 --- /dev/null +++ b/realtime-transcription/README.md @@ -0,0 +1,76 @@ +# Realtime transcription app (WebSocket speech-to-text) + +Realtime speech-to-text on the Livepeer network over a **WebSocket** — the client streams audio _up_ and gets transcripts streamed _back_ on one socket. This is the example where WebSockets are genuinely required: HTTP can't stream audio upstream, and SSE is one-directional (server→client). The app is a small aiohttp server wrapping `faster-whisper` that **self-registers** (dynamic) — the dynamic, WebSocket counterpart to the static HTTP vLLM example. + +| | | +| ------------ | ----------------------------------------- | +| App id | `livepeer-example/realtime-transcription` | +| Runner mode | persistent (held-open WebSocket session) | +| Registration | dynamic (self-registers via the SDK) | +| Model | `large-v3-turbo` (faster-whisper, fixed) | +| Transport | WebSocket (`/transcribe`) | +| Port | 5005 | + +**Requires an NVIDIA GPU.** The model is fixed at `large-v3-turbo`: it swaps large-v3's 32-layer decoder for 4, so it runs far below realtime on a 3090 while staying near large-v3 quality. On CPU it loads but falls behind a live stream, which is the one thing this example is about. Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared on-chain/payment setup are in the [repo README](../README.md). + +## How it's wired + +The app is **dynamically registered**: it self-registers with the orchestrator via `register_runner` ([runner.py](runner.py)) and exposes a `GET /transcribe` WebSocket, whose upgrade the orchestrator proxies straight through — your app speaks standard WS, nothing Livepeer-specific in the socket. The client calls it with `reserve_session` → `ws_connect` → `stop_runner_session` ([client.py](client.py)) — reserve a session, open a `wss://` socket to the session URL, stream audio up / transcripts back, release. Grep `# Livepeer:` in either file to see the exact calls. + +On-chain, the reserved session is the billing unit: `reserve_session` pays at reserve and the meter runs while the socket is open — continuous connection = continuous billing, the right model for live audio. + +**Staying realtime is the constraint,** and it is why the model is a constant rather than a setting. The app degrades by stretching out partials rather than dropping audio, so a model that cannot keep pace buys accuracy with unbounded lag instead of failing loudly. Serving a different model is a different app, with its own price and its own app id — which is why the id names the capability (`realtime-transcription`) and not the technique. + +**Realtime design:** the receive loop only appends audio (never blocking on the model); a background worker transcribes the _current utterance_ (bounded to 15s) every ~0.5s, emits partials, and finalizes on trailing silence or max length — so cost stays bounded no matter how long the stream runs, instead of re-transcribing an ever-growing buffer. It uses the low-latency Whisper preset (`beam_size=1`, no cross-segment conditioning). For production-grade streaming you'd reach for a LocalAgreement approach (whisper_streaming / WhisperLive). + +Wire protocol on `/transcribe`: + +- client → server: binary frames of **16 kHz mono PCM (int16)** +- client → server: text `eos` to finish +- server → client: JSON `{"text": "...", "final": false|true}` + +## Audio + +Input must be **16 kHz mono WAV**. Convert any file you have, or record a few seconds of yourself talking: + +```sh +ffmpeg -i input.mp3 -ar 16000 -ac 1 sample.wav # convert +ffmpeg -f alsa -i default -ar 16000 -ac 1 -t 20 sample.wav # record (macOS: -f avfoundation -i :0) +``` + +Use a clip with a couple of sentences and a pause between them: the app finalizes on trailing silence, so that is what shows partials turning into finals more than once. + +## Run offchain (free) + +```sh +docker compose up -d --build # first run downloads the whisper model +curl -sk https://localhost:8935/discovery | jq '.[].runners[].app' # confirm livepeer-example/realtime-transcription registered +uv run client.py --discovery https://localhost:8935/discovery --file sample.wav +docker compose down +``` + +The client reserves a session, opens a WebSocket through the orchestrator, streams the WAV in real-time-paced chunks, and prints partial transcripts as they arrive plus a final one per utterance. + +## Run on-chain (paid) + +Layer `compose.onchain.yml` to add a remote signer and run the orchestrator on-chain. 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) in the repo README. + +```sh +cp .env.example .env # fill in RPC, network, keystore paths, accounts, pricing +docker compose -f compose.yml -f compose.onchain.yml up -d --build +uv run client.py --discovery https://localhost:8935/discovery \ + --signer http://localhost:7936 --file sample.wav +docker compose -f compose.yml -f compose.onchain.yml down +``` + +`reserve_session` pays for the session through the remote signer; the WebSocket then streams over it. Because the session is metered, keep clips short for the demo — a long-lived socket keeps billing for its duration. + +## Run without Docker + +Start an orchestrator built from go-livepeer `v0.9.0` or newer (see [Build from source](https://docs.livepeer.org/v1/orchestrators/guides/install-go-livepeer#build-from-source)), then the app and client directly (the app needs `faster-whisper` installed): + +```sh +./livepeer -orchestrator -useLiveRunners -serviceAddr localhost:8935 -orchSecret abcdef -v 6 +uv run runner.py --orchestrator https://localhost:8935 --orchSecret abcdef +uv run client.py --file sample.wav +``` diff --git a/realtime-transcription/client.py b/realtime-transcription/client.py new file mode 100644 index 0000000..37c2d70 --- /dev/null +++ b/realtime-transcription/client.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +"""realtime-transcription client: reserve a session, stream audio over WebSocket, settle up. + +One SDK call reserves a session; from there it's a standard WebSocket — the +orchestrator proxies the upgrade straight to the app. Audio is streamed up in +real-time-paced chunks; partial/final transcripts stream back. + +Livepeer integration (grep `# Livepeer:`): + 1. reserve_session() — discover the runner, reserve a session + 2. ws_connect() — open the proxied WebSocket to the session URL + 3. stop_runner_session() — end the session (settles payment on-chain) + +Audio must be 16 kHz mono. Convert anything with ffmpeg: + ffmpeg -i input.mp3 -ar 16000 -ac 1 sample.wav +""" +from __future__ import annotations + +import argparse +import asyncio +import logging +import ssl +import wave +from contextlib import suppress + +import aiohttp + +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.live_runner import stop_runner_session +from livepeer_gateway.selection import reserve_session + +DEFAULT_DISCOVERY = "https://localhost:8935/discovery" +APP_ID = "livepeer-example/realtime-transcription" +SAMPLE_RATE = 16000 +CHUNK_MS = 100 + +log = logging.getLogger("realtime-transcription-client") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Stream audio to a Whisper Live Runner over WebSocket." + ) + parser.add_argument("--discovery", default=DEFAULT_DISCOVERY) + parser.add_argument("--file", required=True, help="16 kHz mono WAV to stream.") + parser.add_argument( + "--signer", default="", help="Remote signer base URL (on-chain/paid path)." + ) + return parser.parse_args() + + +def _read_pcm(path: str) -> bytes: + with wave.open(path, "rb") as w: + if ( + w.getframerate() != SAMPLE_RATE + or w.getnchannels() != 1 + or w.getsampwidth() != 2 + ): + raise SystemExit( + f"ERROR: {path} must be 16 kHz mono 16-bit WAV (got {w.getframerate()}Hz, " + f"{w.getnchannels()}ch, {w.getsampwidth() * 8}-bit). Convert with ffmpeg." + ) + return w.readframes(w.getnframes()) + + +async def _send(ws: aiohttp.ClientWebSocketResponse, pcm: bytes) -> None: + step = SAMPLE_RATE * 2 * CHUNK_MS // 1000 # bytes per chunk + for i in range(0, len(pcm), step): + await ws.send_bytes(pcm[i : i + step]) + await asyncio.sleep(CHUNK_MS / 1000) # pace at real time + await ws.send_str("eos") + + +async def _recv(ws: aiohttp.ClientWebSocketResponse) -> None: + # Run to the end of the stream, not to the first final: a clip with several + # sentences produces one final per utterance, and the server closes after eos. + async for msg in ws: + if msg.type != aiohttp.WSMsgType.TEXT: + break + data = msg.json() + marker = "FINAL" if data.get("final") else "partial" + print(f"[{marker}] {data.get('text', '')}") + + +async def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + pcm = _read_pcm(args.file) + signer_url = args.signer.strip() or None + session = None + try: + session = await reserve_session( # Livepeer: 1 + discovery_url=args.discovery, # omit if the signer does discovery itself + app=APP_ID, + signer_url=signer_url, + ) + log.info("session_id=%s app_url=%s", session.session_id, session.app_url) + ws_url = ( + session.app_url.replace("https://", "wss://") + .replace("http://", "ws://") + .rstrip("/") + + "/transcribe" + ) + ctx = ssl.create_default_context() # orchestrator serves a self-signed cert + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + async with aiohttp.ClientSession() as cs: + async with cs.ws_connect( + ws_url, ssl=ctx, heartbeat=20 + ) as ws: # Livepeer: 2 + await asyncio.gather(_send(ws, pcm), _recv(ws)) + 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) # Livepeer: 3 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/realtime-transcription/compose.onchain.yml b/realtime-transcription/compose.onchain.yml new file mode 100644 index 0000000..1ca073d --- /dev/null +++ b/realtime-transcription/compose.onchain.yml @@ -0,0 +1,35 @@ +# On-chain payment overlay for the streaming-ASR example. Layer on the base: +# docker compose -f compose.yml -f compose.onchain.yml up -d --build +# +# Adds the shared remote signer, re-points the orchestrator on-chain, and +# registers the app with a price. A WebSocket rides a reserved (persistent, +# metered) session: reserve_session pays at reserve, the meter runs while the +# socket is open. Requires a local .env (gitignored). Then pay through the signer: +# uv run client.py --discovery https://localhost:8935/discovery \ +# --signer http://localhost:7936 --file sample.wav + +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: + - python + - runner.py + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:5005 + - --device=${WHISPER_DEVICE:-cuda} + - --compute-type=${WHISPER_COMPUTE:-float16} + - --price=${PRICE} diff --git a/realtime-transcription/compose.yml b/realtime-transcription/compose.yml new file mode 100644 index 0000000..6178b74 --- /dev/null +++ b/realtime-transcription/compose.yml @@ -0,0 +1,40 @@ +# Offchain demo: orchestrator + a streaming-ASR app that self-registers +# (dynamic) and serves speech-to-text over a WebSocket. +# +# docker compose up -d --build +# uv run client.py --discovery https://localhost:8935/discovery --file sample.wav +# +# The orchestrator is defined once in ../compose.orchestrator.yml and pulled in +# with `extends`. The app embeds the SDK and registers itself, like hello-world. +# Requires an NVIDIA GPU: the pinned model only keeps pace on one. + +services: + orchestrator: + extends: + file: ../compose.orchestrator.yml + service: orchestrator + + app: + build: . + container_name: example_apps_realtime_transcription + # Wait for the orchestrator's healthcheck so registration doesn't race its boot. + depends_on: + orchestrator: + condition: service_healthy + command: + - python + - runner.py + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:5005 + - --device=${WHISPER_DEVICE:-cuda} + - --compute-type=${WHISPER_COMPUTE:-float16} + # large-v3-turbo needs the GPU to stay ahead of a live stream. + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: 1 + capabilities: [gpu] diff --git a/realtime-transcription/pyproject.toml b/realtime-transcription/pyproject.toml new file mode 100644 index 0000000..8f23971 --- /dev/null +++ b/realtime-transcription/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "livepeer-realtime-transcription" +version = "0.1.0" +description = "Streaming speech-to-text (WebSocket) example app for the Livepeer network." +requires-python = ">=3.12" +# Base deps are what the *client* needs (host `uv run client.py`). The runner's +# ASR stack (faster-whisper) is the `runner` extra, installed in the Docker image. +dependencies = [ + "aiohttp", # client WebSocket + "livepeer-gateway", +] + +[project.optional-dependencies] +runner = [ + "faster-whisper", # runner: streaming Whisper ASR + "numpy", +] + +# livepeer-gateway is not on PyPI yet; pull it from the branch. +[tool.uv.sources] +livepeer-gateway = { git = "https://github.com/livepeer/livepeer-python-gateway", branch = "ja/live-runner" } diff --git a/realtime-transcription/runner.py b/realtime-transcription/runner.py new file mode 100644 index 0000000..5b92bb8 --- /dev/null +++ b/realtime-transcription/runner.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""realtime-transcription app: realtime speech-to-text over WebSocket — a Live Runner app. + +The WebSocket showcase: the client streams raw audio *up* and gets transcripts +streamed *back* over one socket — something HTTP can't do (no upstream stream) +and SSE can't do (one-directional). It self-registers (dynamic), so it embeds +the SDK and is its own server, like hello-world. + +Realtime design: the receive loop only appends audio, never blocking on the +model. A background worker transcribes the *current utterance* (bounded to +MAX_SEGMENT_SEC) every STEP_SEC, emits partials, and finalizes on trailing +silence (energy VAD) or max length — so cost stays bounded no matter how long +the stream runs (vs. re-transcribing an ever-growing buffer). + +Livepeer integration (grep `# Livepeer:`): + 1. register_runner() — announce the app to the orchestrator (startup) + 2. registration.close() — deregister (cleanup) + +/transcribe is an ordinary aiohttp WebSocket handler; the orchestrator proxies +the upgrade straight through — nothing Livepeer-specific in the socket itself. + +Wire protocol on /transcribe: + client -> server: binary frames of 16 kHz mono PCM (int16) + client -> server: text "eos" to finish + server -> client: JSON {"text": "...", "final": false|true} +""" +from __future__ import annotations + +import argparse +import asyncio +import logging +from contextlib import suppress + +import numpy as np +from aiohttp import web + +from livepeer_gateway.live_runner import register_runner + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 5005 +APP_ID = "livepeer-example/realtime-transcription" +# Fixed, not a knob: this example is about realtime transcription, so it pins the +# model that is both accurate and fast enough to keep pace. large-v3-turbo swaps +# large-v3's 32-layer decoder for 4, so it runs far below realtime on a 3090 while +# staying near large-v3 quality. Serving a different model is a different app, with +# its own price and app id, not a setting on this one. +WHISPER_MODEL = "large-v3-turbo" + +SAMPLE_RATE = 16000 +BYTES_PER_SEC = SAMPLE_RATE * 2 # int16 mono +STEP_SEC = 0.5 # emit a partial at most this often +MAX_SEGMENT_SEC = 15.0 # force-finalize a segment this long (bounds cost) +SILENCE_SEC = 0.5 # trailing silence that ends an utterance +SILENCE_RMS = 350.0 # int16 RMS below this = silence +MIN_SEGMENT_SEC = 0.3 # don't transcribe shorter than this + +_model = None + + +def _load_model(name: str, device: str, compute_type: str) -> None: + global _model + if _model is None: + from faster_whisper import WhisperModel # heavy import; defer until startup + + _model = WhisperModel(name, device=device, compute_type=compute_type) + log.info( + "loaded whisper model=%s device=%s compute=%s", name, device, compute_type + ) + + +def _rms(pcm: bytes) -> float: + if not pcm: + return 0.0 + a = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) + return float(np.sqrt(np.mean(a * a))) if a.size else 0.0 + + +def _transcribe(pcm: bytes) -> str: + if len(pcm) < int(BYTES_PER_SEC * MIN_SEGMENT_SEC): + return "" + audio = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0 + # Greedy (beam_size=1) + no cross-segment conditioning = the low-latency preset. + segments, _ = _model.transcribe( + audio, + language="en", + vad_filter=True, + beam_size=1, + condition_on_previous_text=False, + ) + return " ".join(s.text.strip() for s in segments).strip() + + +log = logging.getLogger("realtime-transcription") + + +async def _handle_transcribe(request: web.Request) -> web.WebSocketResponse: + ws = web.WebSocketResponse(heartbeat=20) + await ws.prepare(request) + log.info("transcription socket opened") + + seg = bytearray() # current utterance PCM; the worker trims finalized audio + spoke = False + + async def _worker() -> None: + nonlocal seg, spoke + while True: + await asyncio.sleep(STEP_SEC) + n = len(seg) + if n < int(BYTES_PER_SEC * MIN_SEGMENT_SEC): + continue + chunk = bytes(seg[:n]) + tail_silent = _rms(chunk[-int(BYTES_PER_SEC * SILENCE_SEC) :]) < SILENCE_RMS + text = await asyncio.to_thread(_transcribe, chunk) + finalize = (spoke and tail_silent) or n >= int( + BYTES_PER_SEC * MAX_SEGMENT_SEC + ) + if finalize: + if text: + await ws.send_json({"text": text, "final": True}) + del seg[ + :n + ] # drop finalized audio; keep anything appended during inference + spoke = False + elif text: + await ws.send_json({"text": text, "final": False}) + + worker = asyncio.create_task(_worker()) + try: + async for msg in ws: + if msg.type == web.WSMsgType.BINARY: + seg.extend(msg.data) + if _rms(msg.data) >= SILENCE_RMS: + spoke = True + elif msg.type == web.WSMsgType.TEXT and msg.data.strip() == "eos": + text = await asyncio.to_thread(_transcribe, bytes(seg)) + await ws.send_json({"text": text, "final": True}) + break + elif msg.type == web.WSMsgType.ERROR: + log.warning("ws error: %s", ws.exception()) + break + finally: + worker.cancel() + with suppress(asyncio.CancelledError, Exception): + await worker + log.info("transcription socket closed") + return ws + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Streaming Whisper ASR Live Runner.") + parser.add_argument("--orchestrator", default="https://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( + "--device", + default="cpu", + help="cpu (default, runs anywhere) or cuda (low latency).", + ) + parser.add_argument( + "--compute-type", default="int8", help="int8 (cpu) or float16 (cuda)." + ) + parser.add_argument( + "--price", + type=float, + default=0, + help="Runner price in USD per hour (0 = free).", + ) + return parser.parse_args() + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + _load_model( + WHISPER_MODEL, args.device, args.compute_type + ) # fail fast if the model is missing + + async def _on_startup(app: web.Application) -> None: + app["registration"] = await register_runner( # Livepeer: 1 + args.orchestrator, + secret=args.orchSecret, + runner_url=args.runner_url, + app=APP_ID, + mode="persistent", # the WebSocket is a held-open session + price=args.price, # decimal USD/hour + ) + log.info( + "registered runner_id=%s app=%s", app["registration"].runner_id, APP_ID + ) + + async def _on_cleanup(app: web.Application) -> None: + with suppress(Exception): + await app["registration"].close() # Livepeer: 2 + + app = web.Application() + app.router.add_get("/transcribe", _handle_transcribe) + 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/.env.example b/vllm/.env.example index 0e7db93..0d06823 100644 --- a/vllm/.env.example +++ b/vllm/.env.example @@ -1,8 +1,6 @@ # Copy to .env (gitignored) and fill in. Never commit secrets. # Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only. -# Model vLLM serves (must match the client's --model). -VLLM_MODEL=Qwen/Qwen2.5-0.5B-Instruct # Only needed for gated HuggingFace models. HF_TOKEN= diff --git a/vllm/README.md b/vllm/README.md index e78b10e..7c00f30 100644 --- a/vllm/README.md +++ b/vllm/README.md @@ -5,16 +5,13 @@ Runs an OpenAI-compatible LLM on the Livepeer network and consumes it with the * | | | | ------------ | ------------------------------------------ | | App id | `vllm/qwen2.5-0.5b-instruct` | -| Runner mode | persistent (single-shot by nature) | +| Runner mode | single-shot (one session per call) | | Registration | static (orchestrator config + health poll) | | Transport | HTTP + SSE (OpenAI `/v1/chat/completions`) | -| Pricing | hour (metered per second of session) | +| Pricing | hour (metered per second of the call) | | Port | 8000 (vLLM), 8080 (gateway) | -**Requires an NVIDIA GPU** for vLLM. The default model (`Qwen/Qwen2.5-0.5B-Instruct`) is tiny so it fits a modest card can be overridden with `VLLM_MODEL`. Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared on-chain/payment setup are in the [repo README](../README.md). - -> [!NOTE] -> This app is single-shot by nature but currently registers as **persistent**. It will switch to **single-shot** once [#5](https://github.com/livepeer/runner-app-examples/issues/5) lands. +**Requires an NVIDIA GPU** for vLLM. The model (`Qwen/Qwen2.5-0.5B-Instruct`) is tiny so it fits a modest card, and it is pinned rather than configurable: `runners.json` advertises it by name, and a static runner has no code to recompute that id, so serving a different model means editing `compose.yml` and `runners.json` together. That is the static registration bargain — the operator owns the contract, and the orchestrator health-polls the runner without ever checking that it serves what the config claims. A dynamic app can instead derive its app id from what it actually loaded (see [`realtime-transcription`](../realtime-transcription)). Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared on-chain/payment setup are in the [repo README](../README.md). ## How it's wired @@ -27,7 +24,9 @@ Two sides: The local gateway is a _client-side_ component, so it runs on the host like the client, not in the infra compose. -The gateway is the **only** Livepeer-aware piece in the whole path — and it's tiny: three SDK calls, `reserve_session` → `call_runner` → `stop_runner_session` (grep `# Livepeer:` in [gateway.py](gateway.py)). They exist _purely_ because an OpenAI client has no idea how to discover a runner or settle Livepeer's payments. Move that glue into the gateway and everything else — `client.py`, any OpenAI SDK, `curl` — stays 100% stock OpenAI, oblivious to Livepeer. +The gateway is the **only** Livepeer-aware piece in the whole path — and it's tiny: two SDK calls, `runner_selector` → `call_runner` (grep `# Livepeer:` in [gateway.py](gateway.py)). They exist _purely_ because an OpenAI client has no idea how to discover a runner or settle Livepeer's payments. Move that glue into the gateway and everything else — `client.py`, any OpenAI SDK, `curl` — stays 100% stock OpenAI, oblivious to Livepeer. + +The runner is **single-shot**, so the orchestrator reserves a session around each call and releases it when the response returns; the gateway never manages one, which is why there is no third SDK call. Pricing is still metered, so the call pays for as long as it runs and a long generation costs what it takes. With `capacity: 1`, a second request arriving while one is in flight gets a 503, which the gateway hands back as a JSON error rather than an opaque 500. ## Run offchain (free) @@ -39,7 +38,7 @@ uv run client.py --prompt "In one sentence, what is Livepeer?" kill %1; docker compose down # stop the gateway, then the stack ``` -`client.py` is stock `openai` with `base_url=http://localhost:8080/v1` (the `api_key` is ignored — it just needs _a_ value). Pass a `--model` that matches `VLLM_MODEL`, the name vLLM serves under. Any OpenAI tool works the same way — e.g. `curl`: +`client.py` is stock `openai` with `base_url=http://localhost:8080/v1` (the `api_key` is ignored — it just needs _a_ value). Pass a `--model` that matches the name vLLM serves under. Any OpenAI tool works the same way — e.g. `curl`: ```sh curl http://localhost:8080/v1/chat/completions \ @@ -76,4 +75,4 @@ kill %1; docker compose -f compose.yml -f compose.onchain.yml down The client is **unchanged** — only the gateway gets `--signer`; it pays per call through the remote signer, so the consumer never sees discovery or payment. The price is set in `runners.json`. -Pricing note: the orchestrator meters compute per **second**, not per token. Probabilistic payments are made up front, so token counts can't drive protocol pricing. Per-token billing is left to the signer/gateway layer, which sees `usage` in every response and can bill users per token while paying the orchestrator per second. +Pricing note: the orchestrator meters compute per **second** of the call, not per token. Probabilistic payments are made up front, so token counts can't drive protocol pricing. Per-token billing is left to the signer/gateway layer, which sees `usage` in every response and can bill users per token while paying the orchestrator per second. diff --git a/vllm/compose.yml b/vllm/compose.yml index 088a750..b9e3e54 100644 --- a/vllm/compose.yml +++ b/vllm/compose.yml @@ -29,11 +29,14 @@ services: vllm: image: vllm/vllm-openai:latest container_name: example_apps_vllm - # Tiny model (override with VLLM_MODEL); --gpu-memory-utilization caps VRAM so + # The model is pinned, not an env knob: runners.json advertises it by name, and + # a static runner has no code to recompute that id, so a knob here would make the + # app id a lie the orchestrator republishes. Serving another model means editing + # this command and runners.json together. --gpu-memory-utilization caps VRAM so # vLLM starts on a shared GPU (its 0.9 default can exceed free memory). command: - "--model" - - "${VLLM_MODEL:-Qwen/Qwen2.5-0.5B-Instruct}" + - "Qwen/Qwen2.5-0.5B-Instruct" - "--port" - "8000" - "--gpu-memory-utilization" diff --git a/vllm/gateway.py b/vllm/gateway.py index 9077afc..9f8c4ae 100644 --- a/vllm/gateway.py +++ b/vllm/gateway.py @@ -10,17 +10,17 @@ export OPENAI_BASE_URL=http://localhost:8080/v1 OPENAI_API_KEY=unused # then plain `openai`, curl, or any SDK just works -Each request: reserve a session, forward the body, release the session. call_runner does -the 402 payment challenge internally, so the client never sees discovery or payment. -(Release matters: the runner has capacity 1, so an unreleased session would block the -next call.) +Each request: discover the runner, forward the body. The runner is single-shot, so the +orchestrator reserves a session for the call and releases it when the response returns -- +the gateway manages no session at all. call_runner does the 402 payment challenge +internally, so the client never sees discovery or payment. Pricing is metered, so the +call keeps paying for as long as it runs, which for a long generation is the point. Livepeer integration (grep `# Livepeer:`): - 1. reserve_session() — discover the runner, reserve a session - 2. call_runner() — forward the request through the orchestrator (pays 402) - 3. stop_runner_session() — release the session + 1. runner_selector() — discover the runner advertising this app + 2. call_runner() — forward the request through the orchestrator (pays 402) -These three calls are the *entire* Livepeer surface. They live here, and only here, +These two calls are the *entire* Livepeer surface. They live here, and only here, because an OpenAI client can't do discovery or settle payments itself — so `client.py` (and any OpenAI SDK/curl) stays 100% stock, unaware of Livepeer. @@ -34,14 +34,17 @@ import argparse import logging -from contextlib import suppress from aiohttp import web -from livepeer_gateway.live_runner import call_runner, stop_runner_session -from livepeer_gateway.selection import reserve_session +from livepeer_gateway.errors import LivepeerHTTPError +from livepeer_gateway.live_runner import call_runner +from livepeer_gateway.selection import runner_selector APP_ID = "vllm/qwen2.5-0.5b-instruct" +# A generation runs far longer than the SDK's 5s default, and the whole point here +# is that a metered single-shot call pays for as long as it takes. +REQUEST_TIMEOUT = 300.0 log = logging.getLogger("vllm-gateway") @@ -69,51 +72,69 @@ def main() -> None: signer_url = args.signer.strip() or None async def _forward(request: web.Request) -> web.StreamResponse: - payload = await request.json() + # GET /v1/models carries no body; everything else posts JSON. + payload = await request.json() if request.can_read_body else {} runner_path = request.path # e.g. /v1/chat/completions - session = await reserve_session( + cursor = await runner_selector( # Livepeer: 1 discovery_url=args.discovery, # omit if the signer does discovery itself app=APP_ID, + ) + runner = cursor.candidates[0] + runner_url = runner.url.rstrip("/") + runner_path + + # When the OpenAI client asks for stream=True the runner replies with + # text/event-stream; pipe those chunks straight through with stream=True + # so tokens reach the client as they arrive instead of buffering the blob. + if payload.get("stream"): + async with await call_runner( # Livepeer: 2 (streaming) + runner=runner, # discovery metadata tells call_runner the price unit + runner_url=runner_url, + payload=payload, + signer_url=signer_url, + method=request.method, + timeout=REQUEST_TIMEOUT, + stream=True, + ) as stream: + resp = web.StreamResponse( + status=stream.status, + headers={ + "Content-Type": stream.content_type or "text/event-stream" + }, + ) + await resp.prepare(request) + async for ( + chunk + ) in stream.aiter_bytes(): # raw bytes -> keep SSE framing + await resp.write(chunk) + await resp.write_eof() + return resp + + result = await call_runner( # Livepeer: 2 + runner=runner, # discovery metadata tells call_runner the price unit + runner_url=runner_url, + payload=payload, signer_url=signer_url, - ) # Livepeer: 1 - + method=request.method, + timeout=REQUEST_TIMEOUT, + ) + return web.json_response(result.data) + + async def _forward_or_error(request: web.Request) -> web.StreamResponse: + # A single-shot call holds a capacity slot for its duration, so a busy runner + # answers 503. Hand that back as JSON an OpenAI client can read. try: - runner_url = session.app_url.rstrip("/") + runner_path - - # When the OpenAI client asks for stream=True the runner replies with - # text/event-stream; pipe those chunks straight through with stream=True - # so tokens reach the client as they arrive instead of buffering the blob. - if payload.get("stream"): - async with await call_runner( # Livepeer: 2 (streaming) - runner_url=runner_url, - payload=payload, - signer_url=signer_url, - stream=True, - ) as stream: - resp = web.StreamResponse( - status=stream.status, - headers={ - "Content-Type": stream.content_type or "text/event-stream" - }, - ) - await resp.prepare(request) - async for ( - chunk - ) in stream.aiter_bytes(): # raw bytes -> preserve SSE framing - await resp.write(chunk) - await resp.write_eof() - return resp - - result = await call_runner( - runner_url=runner_url, payload=payload, signer_url=signer_url - ) # Livepeer: 2 - return web.json_response(result.data) - finally: - with suppress(Exception): - await stop_runner_session(session) # Livepeer: 3 + return await _forward(request) + except LivepeerHTTPError as exc: + return web.json_response( + {"error": {"message": str(exc), "type": "livepeer_error"}}, + status=exc.status_code, + ) app = web.Application() - app.router.add_post("/v1/{tail:.*}", _forward) # forward every OpenAI path + # Every verb, not just POST: an OpenAI client lists models with GET /v1/models. + # That listing is a real single-shot call, so it reserves a session and, on-chain, + # pays for it -- a production gateway would cache it; an example should not hide it. + app.router.add_route("*", "/v1/{tail:.*}", _forward_or_error) log.info( "gateway on http://%s:%d/v1 -> %s (signer=%s)", args.host, diff --git a/vllm/runners.json b/vllm/runners.json index 664e134..bf58cca 100644 --- a/vllm/runners.json +++ b/vllm/runners.json @@ -5,7 +5,7 @@ "app": "vllm/qwen2.5-0.5b-instruct", "runner_url": "http://vllm:8000", "health_url": "/health", - "mode": "persistent", + "mode": "single-shot", "capacity": 1, "price_info": { "price": 0.01 } }