diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..654abe1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.py[cod] +.env +license.jwt +*.jwt +.ruff_cache/ diff --git a/README.md b/README.md index bc2fc59..0919c57 100644 --- a/README.md +++ b/README.md @@ -14,21 +14,30 @@ and, within each service, by **model**. - **Streaming** transcribes a live audio stream over a WebSocket connection. One stack serves multiple models; the client selects the model per session. See - [`streaming/README.md`](streaming/README.md). + [`streaming/docker/README.md`](streaming/docker/README.md). - **Sync** transcribes a complete file in a single HTTP request/response (audio ≤ 120 s by default). It is self-contained — a single GPU container plus the - license-and-usage-proxy, no load balancer. See [`sync/README.md`](sync/README.md). + license-and-usage-proxy, no load balancer. See [`sync/docker/README.md`](sync/docker/README.md). -Each service directory is self-contained: its compose file(s), `.env.example`, -example client, and `README.md` live together. Run commands from inside the -service directory. +Each stack is self-contained under `/docker/`: its compose file(s), +`.env.example`, example client, and `README.md` live together. Run compose +commands from inside that directory. + +Each stack can also run on serverless GPUs instead of hardware you manage. Every +stack is a self-contained Modal App deployed with a single `modal deploy` — see +[`sync/modal/`](sync/modal/) and +[`streaming/modal/`](streaming/modal/). ## Repository layout ``` . -├── streaming/ # WebSocket streaming ASR (Universal English/Multilingual, Universal-3.5 Pro) -└── sync/ # Synchronous full-file HTTP transcription (Universal-3.5 Pro) +├── streaming/ +│ ├── docker/ # compose stack (Universal English/Multilingual, Universal-3.5 Pro) +│ └── modal/ # serverless-GPU packages +└── sync/ + ├── docker/ # compose stack (Universal-3.5 Pro) + └── modal/ # serverless-GPU package ``` ## Prerequisites (all services) @@ -60,9 +69,9 @@ aws ecr get-login-password --region us-west-2 \ ### License file -Place your AssemblyAI `license.jwt` in the directory of the service you are -running (`streaming/` or `sync/`), or point the `LICENSE_FILE_PATH` environment -variable in that service's compose file at your license file's location. +Place your AssemblyAI `license.jwt` in the directory of the stack you are +running (`streaming/docker/` or `sync/docker/`), or point the `LICENSE_FILE_PATH` +environment variable in that stack's compose file at your license file's location. ## Shared component: license-and-usage-proxy @@ -143,7 +152,7 @@ This release introduces the **Sync self-hosted service** model. It transcribes a complete audio file (≤ 120 s) in a single `POST /transcribe` request/response — a single GPU container plus the license-and-usage-proxy, no load balancer. It exposes `GET /readyz` (200 once -the model is warm) for readiness probes. See [`sync/README.md`](sync/README.md). +the model is warm) for readiness probes. See [`sync/docker/README.md`](sync/docker/README.md). #### Streaming — U3 Pro replaced by Universal-3.5 Pro (BREAKING) @@ -165,9 +174,10 @@ upgrading from the v0.6.0 U3 Pro stack: #### Images -`release-v1.0.0` is published for `self-hosted-streaming-api`, -`self-hosted-streaming-license-and-usage-proxy`, -`self-hosted-streaming-asr-universal-3-5-pro`, and `self-hosted-sync-asr-u3-pro`. +`release-v1.0.1` is published for `self-hosted-streaming-api` (adds the +peer-aborted-handshake logging fix) and `self-hosted-streaming-asr-universal-3-5-pro`; +`release-v1.0.0` for `self-hosted-streaming-license-and-usage-proxy` and +`self-hosted-sync-asr-u3-pro`. The English and Multilingual ASR images are unchanged since v0.6.0 — keep `STREAMING_ASR_ENGLISH_IMAGE` and `STREAMING_ASR_MULTILANG_IMAGE` at `release-v0.6.0` (see `streaming/.env.example`). diff --git a/streaming/.env.example b/streaming/docker/.env.example similarity index 95% rename from streaming/.env.example rename to streaming/docker/.env.example index 0be23ba..df29f66 100644 --- a/streaming/.env.example +++ b/streaming/docker/.env.example @@ -1,5 +1,5 @@ # Required for every streaming stack: -STREAMING_API_IMAGE=344839248844.dkr.ecr.us-west-2.amazonaws.com/self-hosted-streaming-api:release-v1.0.0 +STREAMING_API_IMAGE=344839248844.dkr.ecr.us-west-2.amazonaws.com/self-hosted-streaming-api:release-v1.0.1 LICENSE_AND_USAGE_PROXY_IMAGE=344839248844.dkr.ecr.us-west-2.amazonaws.com/self-hosted-streaming-license-and-usage-proxy:release-v1.0.0 USAGE_TRACKING_API_KEY= # Required only when running the Universal stack (docker-compose.english-multilang.yml): diff --git a/streaming/README.md b/streaming/docker/README.md similarity index 95% rename from streaming/README.md rename to streaming/docker/README.md index cbd0722..445ea27 100644 --- a/streaming/README.md +++ b/streaming/docker/README.md @@ -5,7 +5,7 @@ Real-time transcription over a WebSocket connection. Run all commands from this > Prerequisites (license, Docker, GPU runtime, ECR auth) and the shared > license-and-usage-proxy (usage reporting, license status endpoint, proxy -> production recommendations) are documented in the [top-level README](../README.md). +> production recommendations) are documented in the [top-level README](../../README.md). ## Choosing a stack @@ -24,7 +24,7 @@ To switch between stacks, run `docker compose -f down` before starting th Both stacks include: - **streaming-api**: Gateway API service handling WebSocket connections. - **streaming-asr-lb**: nginx load balancer for ASR services with header-based routing. -- **license-and-usage-proxy**: License validation and usage reporting (see [top-level README](../README.md#shared-component-license-and-usage-proxy)). +- **license-and-usage-proxy**: License validation and usage reporting (see [top-level README](../../README.md#shared-component-license-and-usage-proxy)). ASR backends differ by stack: - Universal stack (`docker-compose.english-multilang.yml`): `streaming-asr-english` and `streaming-asr-multilang`. @@ -64,7 +64,7 @@ to an available backend. ## Setup -Complete the [shared prerequisites](../README.md#prerequisites-all-services) +Complete the [shared prerequisites](../../README.md#prerequisites-all-services) (GPU runtime, ECR authentication, license file) first. Copy the env reference and set the image variables for the stack you plan to run: @@ -188,7 +188,7 @@ python example_with_prerecorded_audio_file.py --help ### Usage reporting The license-and-usage-proxy's billing modes and behavior are documented in the -[top-level README](../README.md#usage-reporting). +[top-level README](../../README.md#usage-reporting). ## Monitoring & debugging @@ -215,9 +215,14 @@ docker compose -f docker-compose.english-multilang.yml restart streaming-asr-mul docker compose -f docker-compose.universal-3-5-pro.yml restart streaming-asr-universal-3-5-pro ``` +## Deploying on Modal (serverless GPU) + +Both streaming stacks also run on Modal's serverless GPUs as self-contained, +single-`modal deploy` Modal Apps. See [`../modal/`](../modal/). + ## Production deployment recommendations -See the [top-level README](../README.md#production-recommendations-license-and-usage-proxy) +See the [top-level README](../../README.md#production-recommendations-license-and-usage-proxy) for the license-and-usage-proxy. Streaming-specific services follow. ### streaming-api service diff --git a/streaming/docker-compose.english-multilang.yml b/streaming/docker/docker-compose.english-multilang.yml similarity index 100% rename from streaming/docker-compose.english-multilang.yml rename to streaming/docker/docker-compose.english-multilang.yml diff --git a/streaming/docker-compose.universal-3-5-pro.yml b/streaming/docker/docker-compose.universal-3-5-pro.yml similarity index 100% rename from streaming/docker-compose.universal-3-5-pro.yml rename to streaming/docker/docker-compose.universal-3-5-pro.yml diff --git a/streaming/example/example_audio_file.wav b/streaming/docker/example/example_audio_file.wav similarity index 100% rename from streaming/example/example_audio_file.wav rename to streaming/docker/example/example_audio_file.wav diff --git a/streaming/example/example_with_prerecorded_audio_file.py b/streaming/docker/example/example_with_prerecorded_audio_file.py similarity index 100% rename from streaming/example/example_with_prerecorded_audio_file.py rename to streaming/docker/example/example_with_prerecorded_audio_file.py diff --git a/streaming/example/requirements.txt b/streaming/docker/example/requirements.txt similarity index 100% rename from streaming/example/requirements.txt rename to streaming/docker/example/requirements.txt diff --git a/streaming/nginx_streaming_asr.conf b/streaming/docker/nginx_streaming_asr.conf similarity index 100% rename from streaming/nginx_streaming_asr.conf rename to streaming/docker/nginx_streaming_asr.conf diff --git a/streaming/modal/README.md b/streaming/modal/README.md new file mode 100644 index 0000000..fc4c8de --- /dev/null +++ b/streaming/modal/README.md @@ -0,0 +1,114 @@ +# Streaming stacks on Modal (serverless GPU) + +Each streaming stack runs on [Modal](https://modal.com) as a self-contained +Modal App: one `modal deploy` brings up every service and wires them together, +with no dependency on any other deployment. Compose equivalents live in +[`../docker/`](../docker/). + +| Stack | File | Servers | +|---|---|---| +| Universal-3.5 Pro | `modal_app_universal_3_5_pro.py` | `StreamingApi` (CPU), `Asr` (L40S), `LicenseProxy` (CPU) | +| English + Multilingual | `modal_app_english_multilang.py` | `StreamingApi` (CPU), `Lb` (CPU nginx), `AsrEnglish` (L40S), `AsrMultilang` (L40S), `LicenseProxy` (CPU) | + +`StreamingApi` resolves its backend and proxy URLs from the same App at startup, +so there is no manual wiring or two-phase deploy. The Universal-3.5 Pro stack +serves one model and needs no router, so nginx is dropped. The +English + Multilingual stack serves two models, so it keeps an nginx `Lb` that +routes the `x-model-version` gRPC metadata (from the client's `speech_model`) to +the matching backend, exactly as `streaming-asr-lb` does in compose. + +## Prerequisites and secrets + +Identical to the [sync stack](../../sync/modal/README.md#store-credentials-as-modal-secrets): +create the `aai-ecr-credentials` and `aai-license` Modal secrets once; all three +stacks share them. + +## Deploy + +```bash +modal deploy modal_app_universal_3_5_pro.py # or modal_app_english_multilang.py +``` + +Each GPU backend keeps one L40S warm (`min_containers=1`) and gates readiness on +`grpc_health_probe`, so the first deploy takes a few minutes to warm the model; +Modal then autoscales on concurrent sessions, with `target_concurrency` set to +each stack's `MAX_OPEN_STREAMS` (Universal-3.5 Pro 32, English + Multilingual 48, +matching compose). The endpoint URLs are printed, of the form +`https://---streamingapi..modal.direct`. + +## Verify + +The `streamingapi` endpoint is behind Modal proxy auth by default, so send a +proxy-auth token (`--modal-key` / `--modal-secret`, or `MODAL_KEY` / +`MODAL_SECRET`); a `licenseproxy` `/v1/status` check needs none. To probe +without a token, deploy the endpoint with `AAI_REQUIRE_MODAL_AUTH=0`. + +```bash +curl -fsS https://--aai-streaming-u3pro-licenseproxy..modal.direct/v1/status + +# Stream with the bundled sample client (it forwards Modal proxy-auth headers; +# the repo's example_with_prerecorded_audio_file.py does not, so it only works +# against an AAI_REQUIRE_MODAL_AUTH=0 endpoint): +python sample_streaming.py \ + --endpoint wss://--aai-streaming-u3pro-streamingapi..modal.direct \ + --audio ../docker/example/example_audio_file.wav \ + --speech-model universal-3-5-pro \ + --modal-key "$MODAL_KEY" --modal-secret "$MODAL_SECRET" +``` + +For the English + Multilingual stack use `--speech-model universal-streaming-english` +or `universal-streaming-multilingual`; the API maps these to the `en-default` / +`ml-default` routing keys and the `Lb` sends each to its backend. Or use the +[sample script](#sample-requests). + +## Authentication and security + +`StreamingApi` requires a Modal proxy-auth token by default +(`unauthenticated=False`); Modal enforces it on the WebSocket upgrade, so a +guessed URL alone gets `401`. Send the token as `Modal-Key` / `Modal-Secret` +headers, or deploy with `AAI_REQUIRE_MODAL_AUTH=0` for a throwaway public test +endpoint (any non-empty `Authorization` then connects, as behind your own +gateway). + +The internal hops (`StreamingApi` → `Asr`/`Lb`, and → `LicenseProxy`) cross +Modal's TLS edge, **not** a private bridge network as in compose: Modal has no +private inter-container network by default, so these `.modal.direct` endpoints +are public. The gRPC hop is encrypted — `h2_enabled` advertises ALPN h2 so the +API's default-TLS gRPC client connects with `AAI_USE_SECURE_CHANNEL_TO_ASR_SERVICE=True` +— but the backends and proxy are `unauthenticated=True`, because the API dials +them server-side and cannot attach Modal auth headers. Their URLs are +unguessable but reachable by anyone who learns them; a determined operator can +close that gap by co-locating the API and ASR in one container (localhost hop) +or by putting the backends on Modal's `i6pn` private network (an address +handshake via `modal.Dict`, same region). Treat the shipped topology as suitable +for evaluation, not untrusted public exposure of the backends. + +## Sample requests + +`sample_streaming.py` streams the audio at real time and prints turns live +(partial `…`, finalized `✓`); `--speech-model` picks the model and `--load N` +opens N concurrent sessions. + +```bash +pip install websockets +python sample_streaming.py \ + --endpoint wss://--aai-streaming-u3pro-streamingapi..modal.direct \ + --audio ../docker/example/example_audio_file.wav \ + --speech-model universal-3-5-pro +``` + +If the stack was deployed with the default proxy auth, pass `--modal-key` / +`--modal-secret` (or set `MODAL_KEY` / `MODAL_SECRET`). + +## Cost and teardown + +Each GPU backend holds an L40S while up (Modal bills it), scaling to at most +`max_containers` and down after `scaledown_window`. Tear a stack down when done: + +```bash +modal app stop aai-streaming-u3pro # or aai-streaming-english-multilang +``` + +Audio is processed on Modal's multi-tenant cloud in the configured region +(default `us-east`); pin `routing_region`/`compute_region` near your callers, +and note the data-residency difference from a self-hosted deployment. diff --git a/streaming/modal/modal_app_english_multilang.py b/streaming/modal/modal_app_english_multilang.py new file mode 100644 index 0000000..2e9b98f --- /dev/null +++ b/streaming/modal/modal_app_english_multilang.py @@ -0,0 +1,320 @@ +"""Run the self-hosted streaming English + Multilingual stack on Modal, standalone. + +`modal deploy modal_app_english_multilang.py` brings up the whole stack in one +command. This stack serves TWO ASR models, so unlike the single-model +Universal-3.5 Pro stack it keeps compose's routing layer: clients pick a model +with speech_model ("en-default" or "ml-default"), the API forwards it as the +gRPC metadata x-model-version, and an nginx load balancer routes to the matching +backend. Five Modal Servers in one App: + + streaming_api (CPU) -> WebSocket front door, the public entrypoint + lb (CPU) -> nginx, routes x-model-version to the two backends + asr_english (L40S) -> English gRPC backend + asr_multilang (L40S) -> Multilingual gRPC backend + license_proxy (CPU) -> license-and-usage-proxy + +Every backend hop crosses Modal's TLS edge (h2_enabled advertises ALPN h2 for +gRPC); the API and nginx dial with TLS. See README "Security". + +Deploy: modal deploy modal_app_english_multilang.py +Tear down: modal app stop aai-streaming-english-multilang +""" + +import os +import signal +import subprocess +import threading + +import modal + +APP_NAME = "aai-streaming-english-multilang" +REGISTRY = "344839248844.dkr.ecr.us-west-2.amazonaws.com" +# The English/Multilingual ASR images ship on their own release line, separate +# from the shared streaming-api and license-and-usage-proxy images. +ASR_MODEL_TAG = "release-v0.6.0" +# streaming-api carries the WARNING-not-ERROR handshake-logging fix at v1.0.1 +# (DeepLearning #19523); the license-and-usage-proxy has no v1.0.1. +API_TAG = "release-v1.0.1" +PROXY_TAG = "release-v1.0.0" +ASR_GRPC_PORT = 50051 + +REQUIRE_MODAL_AUTH = os.environ.get("AAI_REQUIRE_MODAL_AUTH", "1") != "0" + +ENGLISH_BIN = "/opt/assemblyai/engineering/projects/realtime/asr_server/asr_server_bin" +MULTILANG_BIN = "/opt/assemblyai/engineering/projects/realtime/asr_server/ml_asr_server_bin" +API_BIN = "/opt/assemblyai/engineering/projects/realtime/api_v2/bin" +PROXY_BIN = "/opt/assemblyai/engineering/projects/realtime/license_and_usage_proxy/bin" + +LICENSE_PATH = "/var/aai_license.jwt" + +ecr_secret = modal.Secret.from_name("aai-ecr-credentials") +license_secret = modal.Secret.from_name("aai-license") + +app = modal.App(APP_NAME) + + +def _vendor_image(repo: str, tag: str) -> modal.Image: + """A Modal-runnable image from an AssemblyAI ECR image (see sync/modal/modal_app.py).""" + return ( + modal.Image.from_aws_ecr( + f"{REGISTRY}/{repo}:{tag}", secret=ecr_secret, add_python="3.12" + ) + .entrypoint([]) + .pip_install(f"modal=={modal.__version__}") + ) + + +english_image = _vendor_image("self-hosted-streaming-asr-english", ASR_MODEL_TAG) +multilang_image = _vendor_image("self-hosted-streaming-asr-multilang", ASR_MODEL_TAG) +api_image = _vendor_image("self-hosted-streaming-api", API_TAG) +proxy_image = _vendor_image("self-hosted-streaming-license-and-usage-proxy", PROXY_TAG) +# nginx routes gRPC by x-model-version; no ECR pull needed. A Debian base gives +# Modal a detectable interpreter plus its client, alongside nginx. +lb_image = ( + modal.Image.debian_slim(python_version="3.12") + .apt_install("nginx") + .pip_install(f"modal=={modal.__version__}") +) + + +# Set by each @modal.exit stop() so the fate-share reaper can tell an intentional +# teardown from an unexpected vendor exit (one server per container). +_stopping = threading.Event() + + +def _launch(argv: list[str], env: dict[str, str]) -> subprocess.Popen: + """Start a binary and fate-share it with the container (see sync/modal/modal_app.py).""" + proc = subprocess.Popen(argv, env={**os.environ, **env}) + + def _reap() -> None: + proc.wait() + # An exit while we are not intentionally stopping means the vendor + # process died on its own; fail so Modal replaces the container. A clean + # @modal.exit teardown sets _stopping first, so stay quiet and let the + # exit handler finish (the container then exits 0). + if not _stopping.is_set(): + os._exit(proc.returncode if (proc.returncode or 0) > 0 else 1) + + threading.Thread(target=_reap, daemon=True).start() + return proc + + +def _wait_http_ok(url: str, timeout_s: int) -> None: + import time + import urllib.request + + deadline = time.monotonic() + timeout_s + last = "" + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(url, timeout=2) as resp: + if resp.status == 200: + return + last = f"HTTP {resp.status}" + except Exception as exc: # noqa: BLE001 + last = repr(exc) + time.sleep(3) + raise RuntimeError(f"{url} not ready after {timeout_s}s (last: {last})") + + +def _asr_ready(port: int, timeout_s: int) -> None: + import time + + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + if subprocess.run( + ["grpc_health_probe", f"-addr=:{port}"], capture_output=True + ).returncode == 0: + return + time.sleep(5) + raise RuntimeError(f"ASR not serving on :{port} after {timeout_s}s") + + +_ASR_KW = dict( + gpu="L40S", + cpu=4, + memory=16384, + port=ASR_GRPC_PORT, + h2_enabled=True, + unauthenticated=True, # dialed server-side (via nginx) over gRPC; see README "Security" + target_concurrency=48, # matches MAX_OPEN_STREAMS=48 (compose parity) + min_containers=1, + max_containers=4, + buffer_containers=1, + scaledown_window=600, + startup_timeout=900, + exit_grace_period=600, +) +_ASR_ENV = { + "SERVER_PORT": str(ASR_GRPC_PORT), + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "MAX_OPEN_STREAMS": os.environ.get("MAX_OPEN_STREAMS", "48"), + "VLLM_USE_FLASHINFER_SAMPLER": "0", +} + + +@app.server(image=proxy_image, port=8080, unauthenticated=True, cpu=1, memory=2048, + min_containers=1, max_containers=1, startup_timeout=180, exit_grace_period=30, + secrets=[license_secret], + env={"HTTP_PORT": "8080", "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", "LICENSE_FILE_PATH": LICENSE_PATH}) +class LicenseProxy: + @modal.enter() + def start(self) -> None: + token = os.environ.get("AAI_LICENSE_JWT") or os.environ["LICENSE_JWT"] + with open(LICENSE_PATH, "w") as fh: + fh.write(token.strip()) + self.proc = _launch([PROXY_BIN], {}) + _wait_http_ok("http://localhost:8080/health", 60) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=25) + except subprocess.TimeoutExpired: + self.proc.kill() + + +@app.server(image=english_image, env=_ASR_ENV, **_ASR_KW) +class AsrEnglish: + @modal.enter() + def start(self) -> None: + self.proc = _launch([ENGLISH_BIN], {}) + _asr_ready(ASR_GRPC_PORT, 840) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=570) + except subprocess.TimeoutExpired: + self.proc.kill() + + +@app.server(image=multilang_image, env=_ASR_ENV, **_ASR_KW) +class AsrMultilang: + @modal.enter() + def start(self) -> None: + self.proc = _launch([MULTILANG_BIN], {}) + _asr_ready(ASR_GRPC_PORT, 840) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=570) + except subprocess.TimeoutExpired: + self.proc.kill() + + +# nginx config: route the gRPC metadata x-model-version to the matching backend +# over Modal's TLS edge (grpcs, ALPN h2). Backend hosts are resolved from the +# App at startup and substituted in. Public DNS resolves the .modal.direct +# hosts, so a public resolver is used. +_NGINX_CONF = """ +events {{ worker_connections 1024; }} +http {{ + access_log /dev/stdout; + error_log /dev/stderr info; + resolver 1.1.1.1 8.8.8.8 valid=30s; + map $http_x_model_version $asr_backend {{ + default {english}:443; + en-default {english}:443; + ml-default {multilang}:443; + }} + keepalive_timeout 10h; + # Plain HTTP/1.1 readiness port for the startup probe; the gRPC listener below + # is h2-only and cannot answer an HTTP/1.1 GET. + server {{ + listen 8081; + location = /health {{ access_log off; return 200 "OK\\n"; }} + }} + server {{ + listen 8080 http2; + client_max_body_size 0; + location / {{ + grpc_pass grpcs://$asr_backend; + grpc_ssl_server_name on; + # This encrypts the LB->ASR hop but does not authenticate the backend: + # grpc_ssl_verify is left off because nginx cannot build a chain to Modal's + # edge certificate here (grpc_ssl_verify on fails with "unable to get local + # issuer certificate" and takes the backend down). The backend is + # unauthenticated by design anyway (see README "Authentication and + # security"); co-locate the API and ASR, or use i6pn, to remove the public + # hop entirely rather than only encrypt it. + grpc_connect_timeout 75s; + grpc_read_timeout 10h; + grpc_send_timeout 10h; + grpc_socket_keepalive on; + }} + }} +}} +""" + + +@app.server(image=lb_image, port=8080, h2_enabled=True, unauthenticated=True, + cpu=1, memory=1024, min_containers=1, max_containers=1, + startup_timeout=120, exit_grace_period=30) +class Lb: + @modal.enter() + def start(self) -> None: + english = modal.Server.from_name(APP_NAME, "AsrEnglish").get_url().split("://", 1)[1].rstrip("/") + multilang = modal.Server.from_name(APP_NAME, "AsrMultilang").get_url().split("://", 1)[1].rstrip("/") + with open("/etc/nginx/nginx.conf", "w") as fh: + fh.write(_NGINX_CONF.format(english=english, multilang=multilang)) + print(f"[startup] lb -> en={english} ml={multilang}", flush=True) + self.proc = _launch(["nginx", "-g", "daemon off;"], {}) + _wait_http_ok("http://localhost:8081/health", 30) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=25) + except subprocess.TimeoutExpired: + self.proc.kill() + + +@app.server(image=api_image, port=8080, unauthenticated=not REQUIRE_MODAL_AUTH, + cpu=1, memory=2048, target_concurrency=32, min_containers=1, + max_containers=4, nonpreemptible=True, scaledown_window=600, + startup_timeout=600, exit_grace_period=600, secrets=[license_secret]) +class StreamingApi: + @modal.enter() + def start(self) -> None: + lb_host = ( + os.environ.get("ASR_ENDPOINT") + or modal.Server.from_name(APP_NAME, "Lb").get_url().split("://", 1)[1].rstrip("/") + ) + proxy_url = os.environ.get("PROXY_ENDPOINT") or modal.Server.from_name( + APP_NAME, "LicenseProxy" + ).get_url().rstrip("/") + print(f"[startup] LB={lb_host}:443 proxy={proxy_url} require_auth={REQUIRE_MODAL_AUTH}", flush=True) + + self.proc = _launch( + [API_BIN], + { + "AAI_WSS_PORT": "8080", + "AAI_LOG_LEVEL": "INFO", + "AAI_USE_STRUCTURED_LOGGING": "False", + "AAI_ASR_ENDPOINT": f"{lb_host}:443", + "AAI_USE_SECURE_CHANNEL_TO_ASR_SERVICE": "True", + "AAI_LICENSE_AND_USAGE_PROXY_ENDPOINT": proxy_url, + }, + ) + _wait_http_ok("http://localhost:8080/v3/ws/health", 120) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=570) + except subprocess.TimeoutExpired: + self.proc.kill() diff --git a/streaming/modal/modal_app_universal_3_5_pro.py b/streaming/modal/modal_app_universal_3_5_pro.py new file mode 100644 index 0000000..973b5cc --- /dev/null +++ b/streaming/modal/modal_app_universal_3_5_pro.py @@ -0,0 +1,250 @@ +"""Run the self-hosted streaming Universal-3.5 Pro stack on Modal, standalone. + +`modal deploy modal_app_universal_3_5_pro.py` brings up the whole stack in one +command; nothing depends on any other deployment. Compose's four services become +three Modal Servers in one App: + + streaming_api (CPU) -> WebSocket front door, the public entrypoint + asr (L40S) -> Universal-3.5 Pro gRPC backend + license_proxy (CPU) -> license-and-usage-proxy + +nginx (streaming-asr-lb) is dropped: it only routes X-Model-Version across +several ASR backends, and this stack serves one model. streaming_api resolves +the ASR and proxy URLs from the same App at startup, so there is no manual +wiring step. + +The API dials the ASR over Modal's TLS edge (h2_enabled advertises ALPN h2, so a +standard gRPC client connects) with AAI_USE_SECURE_CHANNEL_TO_ASR_SERVICE=True, +replacing compose's private bridge network. See README "Security". + +Deploy: modal deploy modal_app_universal_3_5_pro.py +Tear down: modal app stop aai-streaming-u3pro +""" + +import os +import signal +import subprocess +import threading + +import modal + +APP_NAME = "aai-streaming-u3pro" +REGISTRY = "344839248844.dkr.ecr.us-west-2.amazonaws.com" +# streaming-api and the u3-5-pro ASR ship on release-v1.0.1 (the API image +# carries the WARNING-not-ERROR handshake-logging fix, DeepLearning #19523); +# the license-and-usage-proxy has no v1.0.1 and stays on v1.0.0. +API_TAG = "release-v1.0.1" +ASR_TAG = "release-v1.0.1" +PROXY_TAG = "release-v1.0.0" +ASR_GRPC_PORT = 50051 + +# See sync/modal/modal_app.py: the WebSocket API requires a Modal proxy-auth token by +# default. Set AAI_REQUIRE_MODAL_AUTH=0 for a throwaway test endpoint. +REQUIRE_MODAL_AUTH = os.environ.get("AAI_REQUIRE_MODAL_AUTH", "1") != "0" + +API_BIN = "/opt/assemblyai/engineering/projects/realtime/api_v2/bin" +ASR_BIN = "/opt/assemblyai/engineering/projects/realtime/asr_u3pro/self_hosted_bin" +PROXY_BIN = "/opt/assemblyai/engineering/projects/realtime/license_and_usage_proxy/bin" + +LICENSE_PATH = "/var/aai_license.jwt" + +ecr_secret = modal.Secret.from_name("aai-ecr-credentials") +license_secret = modal.Secret.from_name("aai-license") + +app = modal.App(APP_NAME) + + +def _vendor_image(repo: str, tag: str) -> modal.Image: + """A Modal-runnable image from an AssemblyAI ECR image (see sync/modal/modal_app.py).""" + return ( + modal.Image.from_aws_ecr( + f"{REGISTRY}/{repo}:{tag}", secret=ecr_secret, add_python="3.12" + ) + .entrypoint([]) + .pip_install(f"modal=={modal.__version__}") + ) + + +asr_image = _vendor_image("self-hosted-streaming-asr-universal-3-5-pro", ASR_TAG) +api_image = _vendor_image("self-hosted-streaming-api", API_TAG) +proxy_image = _vendor_image("self-hosted-streaming-license-and-usage-proxy", PROXY_TAG) + + +# Set by each @modal.exit stop() so the fate-share reaper can tell an intentional +# teardown from an unexpected vendor exit (one server per container). +_stopping = threading.Event() + + +def _launch(argv: list[str], env: dict[str, str]) -> subprocess.Popen: + """Start a vendor binary and fate-share it with the container (see sync/modal/modal_app.py).""" + proc = subprocess.Popen(argv, env={**os.environ, **env}) + + def _reap() -> None: + proc.wait() + # An exit while we are not intentionally stopping means the vendor + # process died on its own; fail so Modal replaces the container. A clean + # @modal.exit teardown sets _stopping first, so stay quiet and let the + # exit handler finish (the container then exits 0). + if not _stopping.is_set(): + os._exit(proc.returncode if (proc.returncode or 0) > 0 else 1) + + threading.Thread(target=_reap, daemon=True).start() + return proc + + +def _wait_http_ok(url: str, timeout_s: int) -> None: + import time + import urllib.request + + deadline = time.monotonic() + timeout_s + last = "" + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(url, timeout=2) as resp: + if resp.status == 200: + return + last = f"HTTP {resp.status}" + except Exception as exc: # noqa: BLE001 + last = repr(exc) + time.sleep(3) + raise RuntimeError(f"{url} not ready after {timeout_s}s (last: {last})") + + +@app.server( + image=proxy_image, + port=8080, + unauthenticated=True, # called server-side by streaming_api; see README "Authentication" + cpu=1, + memory=2048, + min_containers=1, + max_containers=1, + startup_timeout=180, + exit_grace_period=30, + secrets=[license_secret], + env={ + "HTTP_PORT": "8080", + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "LICENSE_FILE_PATH": LICENSE_PATH, + }, +) +class LicenseProxy: + @modal.enter() + def start(self) -> None: + token = os.environ.get("AAI_LICENSE_JWT") or os.environ["LICENSE_JWT"] + with open(LICENSE_PATH, "w") as fh: + fh.write(token.strip()) + self.proc = _launch([PROXY_BIN], {}) + _wait_http_ok("http://localhost:8080/health", 60) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=25) + except subprocess.TimeoutExpired: + self.proc.kill() + + +@app.server( + image=asr_image, + gpu="L40S", + cpu=4, + memory=16384, + port=ASR_GRPC_PORT, + h2_enabled=True, # gRPC needs ALPN h2 across the TLS edge + # Called server-side by streaming_api over gRPC, which cannot attach Modal + # auth headers, so the endpoint is unauthenticated. See README "Security". + unauthenticated=True, + target_concurrency=32, # mirrors MAX_OPEN_STREAMS + min_containers=1, # ~5 min warm-up; never scale a realtime backend to zero + max_containers=4, + buffer_containers=1, # scale-up lead time is the warm-up, so keep a warm spare + scaledown_window=600, + startup_timeout=900, + exit_grace_period=600, # let in-flight streams drain instead of dying + env={ + "SERVER_PORT": str(ASR_GRPC_PORT), + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "MAX_OPEN_STREAMS": os.environ.get("MAX_OPEN_STREAMS", "32"), + "VLLM_USE_FLASHINFER_SAMPLER": "0", + }, +) +class Asr: + @modal.enter() + def start(self) -> None: + self.proc = _launch([ASR_BIN], {}) + # Gate readiness on the same probe compose uses, so Modal never routes a + # session to a cold engine. + import time + + deadline = time.monotonic() + 840 + while time.monotonic() < deadline: + if subprocess.run( + ["grpc_health_probe", f"-addr=:{ASR_GRPC_PORT}"], + capture_output=True, + ).returncode == 0: + return + time.sleep(5) + raise RuntimeError("ASR engine not serving after warm-up window") + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=570) + except subprocess.TimeoutExpired: + self.proc.kill() + + +@app.server( + image=api_image, + port=8080, + unauthenticated=not REQUIRE_MODAL_AUTH, + cpu=1, + memory=2048, + target_concurrency=32, # ~32 sessions per CPU container + min_containers=1, + max_containers=4, + nonpreemptible=True, # holds live WebSocket sessions + scaledown_window=600, + startup_timeout=600, + exit_grace_period=600, + secrets=[license_secret], +) +class StreamingApi: + @modal.enter() + def start(self) -> None: + asr_host = ( + os.environ.get("ASR_ENDPOINT") + or modal.Server.from_name(APP_NAME, "Asr").get_url().split("://", 1)[1].rstrip("/") + ) + proxy_url = os.environ.get("PROXY_ENDPOINT") or modal.Server.from_name( + APP_NAME, "LicenseProxy" + ).get_url().rstrip("/") + print(f"[startup] ASR={asr_host}:443 proxy={proxy_url} require_auth={REQUIRE_MODAL_AUTH}", flush=True) + + self.proc = _launch( + [API_BIN], + { + "AAI_WSS_PORT": "8080", + "AAI_LOG_LEVEL": "INFO", + "AAI_USE_STRUCTURED_LOGGING": "False", + "AAI_ASR_ENDPOINT": f"{asr_host}:443", + "AAI_USE_SECURE_CHANNEL_TO_ASR_SERVICE": "True", + "AAI_LICENSE_AND_USAGE_PROXY_ENDPOINT": proxy_url, + }, + ) + _wait_http_ok("http://localhost:8080/v3/ws/health", 120) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=570) + except subprocess.TimeoutExpired: + self.proc.kill() diff --git a/streaming/modal/sample_streaming.py b/streaming/modal/sample_streaming.py new file mode 100644 index 0000000..a61d9ec --- /dev/null +++ b/streaming/modal/sample_streaming.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Stream a sample WAV to a deployed **streaming** stack and print turns live. + + pip install websockets + # Universal-3.5 Pro stack: + python sample_streaming.py \ + --endpoint wss://--aai-streaming-u3pro-streamingapi..modal.direct \ + --audio ../docker/example/example_audio_file.wav \ + --speech-model universal-3-5-pro + + # English + Multilingual stack (pick the model): + python sample_streaming.py --endpoint wss://--aai-streaming-english-multilang-streamingapi..modal.direct \ + --audio ../docker/example/example_audio_file.wav --speech-model universal-streaming-english + # ... or --speech-model universal-streaming-multilingual + +Audio is sent at real time by default so you watch partial turns update and +finalize, exactly as a live microphone would. Use --speed 2 to send twice as +fast, or --load N to open N sessions at once and print each one's summary. + +If the API was deployed with Modal proxy auth (the shipped default), pass +--modal-key / --modal-secret (or set MODAL_KEY / MODAL_SECRET). For a test +endpoint deployed with AAI_REQUIRE_MODAL_AUTH=0 they are not needed. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +import wave +from concurrent.futures import ThreadPoolExecutor, as_completed +from urllib.parse import urlencode + + +def load_pcm16_mono(path: str) -> tuple[bytes, int]: + with wave.open(path, "rb") as wav: + if wav.getsampwidth() != 2 or wav.getcomptype() != "NONE": + raise SystemExit(f"{path}: must be uncompressed 16-bit PCM WAV") + if wav.getnchannels() != 1: + raise SystemExit(f"{path}: must be mono (1 channel)") + rate = wav.getframerate() + return wav.readframes(wav.getnframes()), rate + + +def stream_once(args, pcm: bytes, rate: int, live: bool) -> dict: + from websockets.sync.client import connect + + params = {"sample_rate": rate, "format_turns": "true"} + if args.speech_model: + params["speech_model"] = args.speech_model + url = f"{args.endpoint.rstrip('/')}?{urlencode(params)}" + + headers = {"Authorization": "sample"} + key = args.modal_key or os.environ.get("MODAL_KEY") + secret = args.modal_secret or os.environ.get("MODAL_SECRET") + if key and secret: + headers["Modal-Key"], headers["Modal-Secret"] = key, secret + + frame = int(rate * 0.05) * 2 # 50 ms of 16-bit mono + chunks = [pcm[i : i + frame] for i in range(0, len(pcm), frame)] + + start = time.perf_counter() + first_turn: float | None = None + finals: list[str] = [] + + with connect(url, additional_headers=headers, open_timeout=args.open_timeout, max_size=None) as ws: + def writer(): + for chunk in chunks: + time.sleep(0.05 / args.speed) + ws.send(chunk) + ws.send('{"type": "Terminate"}') + + with ThreadPoolExecutor(max_workers=1) as pool: + wf = pool.submit(writer) + for message in ws: + data = json.loads(message) + if data.get("type") == "Turn": + words = data.get("words") or [] + if not words: + continue + if first_turn is None: + first_turn = time.perf_counter() - start + text = " ".join(w["text"] for w in words) + if data.get("end_of_turn"): + finals.append(text) + if live: + print(f"\r ✓ {text}", flush=True) + elif live: + # Update the current line in place while the turn forms. + print(f"\r … {text[:110]}", end="", flush=True) + elif data.get("type") == "Termination": + break + wf.result() + + return { + "elapsed_s": time.perf_counter() - start, + "first_turn_s": first_turn, + "turns": len(finals), + "text": " ".join(finals), + } + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--endpoint", required=True, help="wss://...streamingapi... URL") + ap.add_argument("--audio", required=True, help="16-bit PCM mono WAV") + ap.add_argument("--speech-model", help="universal-3-5-pro | universal-streaming-english | universal-streaming-multilingual") + ap.add_argument("--speed", type=float, default=1.0, help="send rate vs realtime (2 = twice as fast)") + ap.add_argument("--load", type=int, default=1, help="open N concurrent sessions") + ap.add_argument("--open-timeout", type=float, default=300.0, help="WS handshake wait (cold starts are slow)") + ap.add_argument("--modal-key") + ap.add_argument("--modal-secret") + args = ap.parse_args() + + pcm, rate = load_pcm16_mono(args.audio) + dur = len(pcm) / 2 / rate + print(f"{args.endpoint}\n audio {dur:.1f}s @ {rate} Hz | model={args.speech_model or '(default)'} | " + f"speed={args.speed}x | sessions={args.load}\n") + + def _first_turn(r: dict) -> str: + # None when the session produced no word-bearing turns. + return f"{r['first_turn_s']:.2f}s" if r["first_turn_s"] is not None else "n/a" + + if args.load == 1: + r = stream_once(args, pcm, rate, live=True) + print(f"\n{r['turns']} final turns | first turn {_first_turn(r)} | wall {r['elapsed_s']:.1f}s") + return 0 + + # Load mode: run N sessions at once, print a summary line per session. + started = time.perf_counter() + ok = 0 + with ThreadPoolExecutor(max_workers=args.load) as pool: + futs = {pool.submit(stream_once, args, pcm, rate, False): i for i in range(args.load)} + for fut in as_completed(futs): + i = futs[fut] + try: + r = fut.result() + ok += 1 + print(f" session {i:>2}: ok | {r['turns']} turns | first turn {_first_turn(r)}") + except Exception as exc: # noqa: BLE001 + print(f" session {i:>2}: FAIL {exc}") + print(f"\n{ok}/{args.load} sessions ok | wall {time.perf_counter() - started:.1f}s") + return 0 if ok == args.load else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/sync/.env.example b/sync/docker/.env.example similarity index 100% rename from sync/.env.example rename to sync/docker/.env.example diff --git a/sync/README.md b/sync/docker/README.md similarity index 94% rename from sync/README.md rename to sync/docker/README.md index e5d9fca..e2e3817 100644 --- a/sync/README.md +++ b/sync/docker/README.md @@ -6,7 +6,7 @@ directory. > Prerequisites (license, Docker, GPU runtime, ECR auth) and the shared > license-and-usage-proxy (usage reporting, license status endpoint, proxy -> production recommendations) are documented in the [top-level README](../README.md). +> production recommendations) are documented in the [top-level README](../../README.md). The stack (`docker-compose.universal-3-5-pro.yml`) runs two containers — `sync-api` (GPU) and `license-and-usage-proxy` — with no nginx load balancer and no @@ -22,7 +22,7 @@ header returns `401`, so make sure your proxy doesn't strip it. ## Setup -Complete the [shared prerequisites](../README.md#prerequisites-all-services) +Complete the [shared prerequisites](../../README.md#prerequisites-all-services) (GPU runtime, ECR authentication, license file) first, then configure images: ```bash @@ -121,9 +121,14 @@ python transcribe_file.py # uses the bundled example_audio_fi python transcribe_file.py path/to/audio.wav # or your own 16-bit PCM WAV ``` +## Deploying on Modal (serverless GPU) + +This stack also runs on Modal's serverless GPUs as a self-contained, +single-`modal deploy` Modal App. See [`../modal/`](../modal/). + ## Production deployment recommendations -See the [top-level README](../README.md#production-recommendations-license-and-usage-proxy) +See the [top-level README](../../README.md#production-recommendations-license-and-usage-proxy) for the license-and-usage-proxy. ### sync-api service diff --git a/sync/docker-compose.universal-3-5-pro.yml b/sync/docker/docker-compose.universal-3-5-pro.yml similarity index 100% rename from sync/docker-compose.universal-3-5-pro.yml rename to sync/docker/docker-compose.universal-3-5-pro.yml diff --git a/sync/example/example_audio_file.wav b/sync/docker/example/example_audio_file.wav similarity index 100% rename from sync/example/example_audio_file.wav rename to sync/docker/example/example_audio_file.wav diff --git a/sync/example/requirements.txt b/sync/docker/example/requirements.txt similarity index 100% rename from sync/example/requirements.txt rename to sync/docker/example/requirements.txt diff --git a/sync/example/transcribe_file.py b/sync/docker/example/transcribe_file.py similarity index 100% rename from sync/example/transcribe_file.py rename to sync/docker/example/transcribe_file.py diff --git a/sync/modal/README.md b/sync/modal/README.md new file mode 100644 index 0000000..f1931c3 --- /dev/null +++ b/sync/modal/README.md @@ -0,0 +1,130 @@ +# Sync stack on Modal (serverless GPU) + +`modal_app.py` runs the self-hosted **sync** (full-file HTTP) stack on +[Modal](https://modal.com) instead of a GPU box you manage. It is a +self-contained Modal App: one `modal deploy` brings up both services and wires +them together, and nothing depends on another deployment. Compose's two services +(see [`../docker/`](../docker/)) become two Modal Servers: + +| Compose service | Modal Server | Hardware | +|---|---|---| +| `sync-api` | `SyncApi` | L40S GPU | +| `license-and-usage-proxy` | `LicenseProxy` | CPU | + +`SyncApi` resolves `LicenseProxy`'s URL from the same App at startup, so there +is no manual wiring or two-phase deploy. + +## Prerequisites + +```bash +pip install modal && modal setup # authenticate the Modal CLI +``` + +## Store credentials as Modal secrets + +Modal has no bind mounts, so the license travels as a secret and is written to +disk at container startup. + +```bash +# ECR pull credentials, used only when Modal builds (pulls) the image. +# Prefer a dedicated pull-only IAM principal over long-lived root/admin keys +# (ecr:GetAuthorizationToken + ecr:BatchGetImage / GetDownloadUrlForLayer / +# BatchCheckLayerAvailability on the AssemblyAI repositories). If you use SSO or +# assume-role session credentials, include AWS_SESSION_TOKEN; note they expire, +# so an image *rebuild* after expiry needs fresh values (redeploys of an +# already-built image do not). +modal secret create aai-ecr-credentials \ + AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_SESSION_TOKEN=... \ + AWS_REGION=us-west-2 + +# The license itself. Usage-billed licenses: add USAGE_TRACKING_API_KEY here +# too; it reaches the proxy automatically, no code change. +modal secret create aai-license LICENSE_JWT="$(cat license.jwt)" +``` + +Both streaming stacks share these same two secrets. + +## Deploy + +```bash +modal deploy modal_app.py +``` + +The first deploy pulls and converts the ~13.5 GB sync image (several minutes); +later deploys reuse the cached image and take seconds. Two endpoint URLs are +printed, of the form `https://--aai-sync-u3pro-..modal.direct`. + +## Verify + +`syncapi` is behind Modal proxy auth by default, so its probes need a +`Modal-Key` / `Modal-Secret` header pair (a proxy-auth token from the Modal +dashboard); `licenseproxy` `/v1/status` needs none. The `syncapi` examples below +show those headers — omit them only against an endpoint deployed with +`AAI_REQUIRE_MODAL_AUTH=0`, where any non-empty `Authorization` connects. + +```bash +curl -fsS https://--aai-sync-u3pro-licenseproxy..modal.direct/v1/status +# {"state":"Connected", ...} + +curl -sS -o /dev/null -w '%{http_code}\n' \ + -H "Modal-Key: $MODAL_KEY" -H "Modal-Secret: $MODAL_SECRET" \ + https://--aai-sync-u3pro-syncapi..modal.direct/readyz +# 503 while the model is cold, 200 once warm (Modal's edge may answer 303 first) + +curl -F 'audio=@../docker/example/example_audio_file.wav;type=audio/wav' \ + -F 'config={"language_code":"en"};type=application/json' \ + -H "Modal-Key: $MODAL_KEY" -H "Modal-Secret: $MODAL_SECRET" \ + -H 'Authorization: any-non-empty-value' \ + https://--aai-sync-u3pro-syncapi..modal.direct/transcribe +``` + +Or use the [sample script](#sample-requests) (pass `--modal-key` / `--modal-secret`). + +## Authentication + +`SyncApi` requires a Modal proxy-auth token by default (`unauthenticated=False`), +so a guessed URL alone cannot reach it. Mint a proxy-auth token in the Modal +dashboard and send it on every request as `Modal-Key` / `Modal-Secret` headers +(or `Authorization: Bearer .`). For a throwaway public test +endpoint, deploy with `AAI_REQUIRE_MODAL_AUTH=0` — it then accepts any non-empty +`Authorization` header, exactly like the compose stack behind your own gateway. +`LicenseProxy` is always `unauthenticated=True` because `SyncApi` calls it +server-side and cannot attach Modal headers; its URL is unguessable but public, +so treat it as such. + +## Configuration + +The audio limits (`MAX_AUDIO_DURATION_MS`, `MIN_AUDIO_DURATION_MS`, +`MAX_REQUEST_BYTES`, `INFERENCE_TIMEOUT_SECONDS`) are read from the environment +with the compose defaults as fallback, so you can override them by adding the +variable to the `aai-license` secret (or any Server env) — your value wins. + +## Sample requests + +`sample_sync.py` sends a request and prints the transcript, server-side time, +and word count; `--concurrency N` fires N at once for a quick load check. + +```bash +pip install requests +python sample_sync.py \ + --endpoint https://--aai-sync-u3pro-syncapi..modal.direct \ + --audio ../docker/example/example_audio_file.wav +``` + +If the stack was deployed with the default proxy auth, pass `--modal-key` / +`--modal-secret` (or set `MODAL_KEY` / `MODAL_SECRET`). + +## Cost and teardown + +`SyncApi` keeps one L40S warm (`min_containers=1`) so requests do not eat a cold +start; it autoscales up to `max_containers` under load and back down after +`scaledown_window`. Modal bills the GPU while it is up, so tear the app down when +you are done: + +```bash +modal app stop aai-sync-u3pro +``` + +Audio is processed on Modal's multi-tenant cloud in `routing_region`/`compute_region` +(default `us-east`); pin them near your callers, and note this is a different +data-residency posture than a stack you host yourself. diff --git a/sync/modal/modal_app.py b/sync/modal/modal_app.py new file mode 100644 index 0000000..f8f172a --- /dev/null +++ b/sync/modal/modal_app.py @@ -0,0 +1,217 @@ +"""Run the self-hosted sync (full-file HTTP) stack on Modal as a standalone app. + +`modal deploy modal_app.py` brings up the whole stack in one command. Compose's +two services become two Modal Servers in one App, so nothing here depends on any +other deployment: + + license_proxy (CPU) -> license-and-usage-proxy + sync_api (L40S) -> sync-api, which resolves the proxy's URL at startup + +Deploy: modal deploy modal_app.py +Tear down: modal app stop aai-sync-u3pro + +Prerequisites (see README "Deploying on Modal"): + modal secret create aai-ecr-credentials AWS_ACCESS_KEY_ID=... \ + AWS_SECRET_ACCESS_KEY=... AWS_SESSION_TOKEN=... AWS_REGION=us-west-2 + modal secret create aai-license LICENSE_JWT="$(cat license.jwt)" # +USAGE_TRACKING_API_KEY if usage-billed +""" + +import os +import signal +import subprocess +import threading + +import modal + +APP_NAME = "aai-sync-u3pro" +REGISTRY = "344839248844.dkr.ecr.us-west-2.amazonaws.com" +TAG = "release-v1.0.0" + +# The API's public endpoint requires a Modal proxy-auth token by default, so a +# guessed URL alone cannot reach it. Set to False for a throwaway test endpoint +# that accepts any non-empty Authorization header (see README "Authentication"). +REQUIRE_MODAL_AUTH = os.environ.get("AAI_REQUIRE_MODAL_AUTH", "1") != "0" + +# Vendor image ENTRYPOINTs, launched explicitly in each Server's @modal.enter. +# Modal prepends an image's ENTRYPOINT to its own runtime command, so both +# images clear it with .entrypoint([]); otherwise the vendor binary consumes +# Modal's arguments, starts with default env, and this code never runs. +SYNC_BIN = "/opt/assemblyai/engineering/projects/realtime/asr_sync_u3pro/self_hosted_bin" +PROXY_BIN = "/opt/assemblyai/engineering/projects/realtime/license_and_usage_proxy/bin" + +LICENSE_PATH = "/var/aai_license.jwt" + +ecr_secret = modal.Secret.from_name("aai-ecr-credentials") +license_secret = modal.Secret.from_name("aai-license") + +app = modal.App(APP_NAME) + + +def _vendor_image(repo: str) -> modal.Image: + """A Modal-runnable image from an AssemblyAI ECR image. + + Every vendor image needs the same three adjustments: clear the ENTRYPOINT, + inject an interpreter Modal can find (the images keep theirs inside Bazel + runfiles, invisible to Modal), and install the Modal client into it (the + runtime-mounted client deps do not land on these images' sys.path). + """ + return ( + modal.Image.from_aws_ecr( + f"{REGISTRY}/{repo}:{TAG}", secret=ecr_secret, add_python="3.12" + ) + .entrypoint([]) + .pip_install(f"modal=={modal.__version__}") + ) + + +proxy_image = _vendor_image("self-hosted-streaming-license-and-usage-proxy") +sync_image = _vendor_image("self-hosted-sync-asr-u3-pro") + + +# Set by each @modal.exit stop() so the fate-share reaper can tell an intentional +# teardown from an unexpected vendor exit (one server per container). +_stopping = threading.Event() + + +def _launch(argv: list[str], env: dict[str, str]) -> subprocess.Popen: + """Start a vendor binary and fate-share it with the container. + + A bare Popen leaves the container 'up' if the binary later exits, so Modal + keeps routing to a process that is gone. The watcher exits the container on + the binary's death, turning a silent black hole into a normal replacement. + """ + proc = subprocess.Popen(argv, env={**os.environ, **env}) + + def _reap() -> None: + proc.wait() + # An exit while we are not intentionally stopping means the vendor + # process died on its own; fail so Modal replaces the container. A clean + # @modal.exit teardown sets _stopping first, so stay quiet and let the + # exit handler finish (the container then exits 0). + if not _stopping.is_set(): + os._exit(proc.returncode if (proc.returncode or 0) > 0 else 1) + + threading.Thread(target=_reap, daemon=True).start() + return proc + + +def _wait_http_ok(url: str, timeout_s: int) -> None: + import time + import urllib.request + + deadline = time.monotonic() + timeout_s + last = "" + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(url, timeout=2) as resp: + if resp.status == 200: + return + last = f"HTTP {resp.status}" + except Exception as exc: # noqa: BLE001 + last = repr(exc) + time.sleep(3) + raise RuntimeError(f"{url} not ready after {timeout_s}s (last: {last})") + + +@app.server( + image=proxy_image, + port=8080, + # Called server-side by sync_api, which cannot attach Modal auth headers to + # its request, so this endpoint must accept unauthenticated traffic. Its URL + # is unguessable but public; see README "Authentication". + unauthenticated=True, + cpu=1, + memory=2048, + min_containers=1, + max_containers=1, + startup_timeout=180, + exit_grace_period=30, + secrets=[license_secret], + env={ + "HTTP_PORT": "8080", + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "LICENSE_FILE_PATH": LICENSE_PATH, + }, +) +class LicenseProxy: + @modal.enter() + def start(self) -> None: + # Compose bind-mounts license.jwt; Modal has no bind mounts, so the JWT + # arrives as a secret and is written to disk at startup. Accept either + # key name so the same secret works across tooling. + token = os.environ.get("AAI_LICENSE_JWT") or os.environ["LICENSE_JWT"] + with open(LICENSE_PATH, "w") as fh: + fh.write(token.strip()) + # Usage-billed licenses: add USAGE_TRACKING_API_KEY to the aai-license + # secret and it reaches the proxy here through the environment. Nothing + # else to change. + self.proc = _launch([PROXY_BIN], {}) + _wait_http_ok("http://localhost:8080/health", 60) + + @modal.exit() + def stop(self) -> None: + # Graceful stop lets the proxy flush queued usage before exit. + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=25) + except subprocess.TimeoutExpired: + self.proc.kill() + + +@app.server( + image=sync_image, + gpu="L40S", + cpu=4, + memory=16384, + port=8080, + unauthenticated=not REQUIRE_MODAL_AUTH, + # Scale-out signal: concurrent in-flight /transcribe requests (GPU-bound). + # Start conservative and tune against bench/harness.py on your hardware. + target_concurrency=8, + min_containers=1, # ~2-4 min cold start; keep one warm (503 while cold) + max_containers=4, + scaledown_window=300, + startup_timeout=900, # weights load + CUDA-graph capture + exit_grace_period=60, # requests are short (INFERENCE_TIMEOUT_SECONDS below) + secrets=[license_secret], +) +class SyncApi: + @modal.enter() + def start(self) -> None: + proxy_url = os.environ.get("PROXY_ENDPOINT") or modal.Server.from_name( + APP_NAME, "LicenseProxy" + ).get_url().rstrip("/") + print(f"[startup] proxy={proxy_url} require_auth={REQUIRE_MODAL_AUTH}", flush=True) + + self.proc = _launch( + [SYNC_BIN], + { + "HTTP_PORT": "8080", + "AAI_ENV": "production", + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "GPU_MONITORING_ENABLED": "False", + "LICENSE_AND_USAGE_PROXY_ENDPOINT": proxy_url, + # Audio limits: customer-overridable via the aai-license secret + # (or any Server env). User value wins; the compose defaults are + # only the fallback. Raising MAX_AUDIO_DURATION_MS usually means + # raising MAX_REQUEST_BYTES and INFERENCE_TIMEOUT_SECONDS too. + "MAX_AUDIO_DURATION_MS": os.environ.get("MAX_AUDIO_DURATION_MS", "120000"), + "MIN_AUDIO_DURATION_MS": os.environ.get("MIN_AUDIO_DURATION_MS", "80"), + "MAX_REQUEST_BYTES": os.environ.get("MAX_REQUEST_BYTES", "41943040"), + "INFERENCE_TIMEOUT_SECONDS": os.environ.get("INFERENCE_TIMEOUT_SECONDS", "30"), + "VLLM_USE_FLASHINFER_SAMPLER": "0", + }, + ) + _wait_http_ok("http://localhost:8080/readyz", 840) + + @modal.exit() + def stop(self) -> None: + _stopping.set() + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=50) + except subprocess.TimeoutExpired: + self.proc.kill() diff --git a/sync/modal/sample_sync.py b/sync/modal/sample_sync.py new file mode 100644 index 0000000..a56bedd --- /dev/null +++ b/sync/modal/sample_sync.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Send a sample request to a deployed **sync** stack and print the transcript. + + pip install requests + python sample_sync.py --endpoint https://--aai-sync-u3pro-syncapi..modal.direct \ + --audio ../docker/example/example_audio_file.wav + +Load test (fire N in parallel, watch throughput): + python sample_sync.py --endpoint https://... --audio a.wav --concurrency 8 + +If the API was deployed with Modal proxy auth (the shipped default), pass +--modal-key / --modal-secret (or set MODAL_KEY / MODAL_SECRET); for a test +endpoint deployed with AAI_REQUIRE_MODAL_AUTH=0 they are not needed. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + +import requests + + +def _auth_headers(args: argparse.Namespace) -> dict[str, str]: + # Any non-empty Authorization satisfies the self-hosted API; Modal proxy + # auth, if enabled, rides its own headers. + headers = {"Authorization": "sample"} + key = args.modal_key or os.environ.get("MODAL_KEY") + secret = args.modal_secret or os.environ.get("MODAL_SECRET") + if key and secret: + headers["Modal-Key"] = key + headers["Modal-Secret"] = secret + return headers + + +def one_request(args: argparse.Namespace, audio: bytes, headers: dict[str, str]) -> dict: + start = time.perf_counter() + resp = requests.post( + f"{args.endpoint.rstrip('/')}/transcribe", + files={"audio": ("audio.wav", audio, "audio/wav")}, + data={"config": json.dumps({"language_code": args.language})}, + headers=headers, + timeout=args.timeout, + ) + elapsed = time.perf_counter() - start + resp.raise_for_status() + body = resp.json() + return { + "elapsed_s": elapsed, + "server_ms": body.get("request_time_ms"), + "audio_ms": body.get("audio_duration_ms"), + "words": len(body.get("words", [])), + "text": body.get("text", ""), + } + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--endpoint", required=True, help="https://...syncapi... URL") + ap.add_argument("--audio", required=True, help="16-bit PCM WAV file") + ap.add_argument("--language", default="en") + ap.add_argument("--concurrency", type=int, default=1, help="fire N requests at once") + ap.add_argument("--timeout", type=float, default=300.0) + ap.add_argument("--modal-key") + ap.add_argument("--modal-secret") + args = ap.parse_args() + + with open(args.audio, "rb") as fh: + audio = fh.read() + headers = _auth_headers(args) + print(f"POST {args.endpoint.rstrip('/')}/transcribe x{args.concurrency}\n") + + started = time.perf_counter() + results, failures = [], 0 + with ThreadPoolExecutor(max_workers=args.concurrency) as pool: + futs = [pool.submit(one_request, args, audio, headers) for _ in range(args.concurrency)] + for i, fut in enumerate(as_completed(futs), 1): + try: + r = fut.result() + results.append(r) + print(f"[{i}/{args.concurrency}] ok {r['elapsed_s']:.2f}s wall | " + f"server {r['server_ms']}ms | {r['words']} words") + except Exception as exc: # noqa: BLE001 + failures += 1 + print(f"[{i}/{args.concurrency}] FAIL {exc}") + wall = time.perf_counter() - started + + if results: + print("\n--- sample transcript ---") + print(results[0]["text"][:600]) + audio_s = (results[0]["audio_ms"] or 0) / 1000 + xrt = (len(results) * audio_s / wall) if wall else 0 + print(f"\n{len(results)} ok, {failures} failed | wall {wall:.2f}s | ~{xrt:.1f}x realtime aggregate") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main())