diff --git a/.github/workflows/test-realtime-benchmark.yml b/.github/workflows/test-realtime-benchmark.yml new file mode 100644 index 000000000..9ef63ed79 --- /dev/null +++ b/.github/workflows/test-realtime-benchmark.yml @@ -0,0 +1,35 @@ +name: Validate realtime benchmark + +on: + pull_request: + paths: + - "examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py" + - "tests/test_realtime_ws_benchmark.py" + - "docs/benchmark/realtime_ws_benchmark.md" + - ".github/workflows/test-realtime-benchmark.yml" + push: + branches: + - main + paths: + - "examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py" + - "tests/test_realtime_ws_benchmark.py" + - "docs/benchmark/realtime_ws_benchmark.md" + - ".github/workflows/test-realtime-benchmark.yml" + +permissions: + contents: read + +jobs: + benchmark: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + - name: Install benchmark test dependencies + run: python -m pip install pytest websockets + - name: Run CPU-only benchmark tests + run: python -m pytest -q tests/test_realtime_ws_benchmark.py diff --git a/docs/benchmark/realtime_ws_benchmark.md b/docs/benchmark/realtime_ws_benchmark.md index 6cf8f57f7..71eaf77bc 100644 --- a/docs/benchmark/realtime_ws_benchmark.md +++ b/docs/benchmark/realtime_ws_benchmark.md @@ -75,12 +75,23 @@ stress signal, not as user-facing realtime latency. |--------|---------| | `aggregate_audio_per_wall` | Total input audio seconds across all clients divided by benchmark wall time | | `first_update_ms_p50/p95` | Time from first audio frame to first result message with `sentences`, `partial`, or `is_final` | +| `first_text_ms_p50/p95` | Time from first audio frame to first result containing non-blank `partial` or sentence `text`, including final-only text | +| `clients_with_text` | Number of clients that received non-blank text; compare with `clients` when interpreting first-text percentiles | | `final_after_stop_ms_p50/p95` | Time from sending `STOP` to receiving the final result | | `client_response_lag_ms_p95_max` | Largest per-client p95 of non-final `(client receive time - audio start) - server duration_ms`; useful mainly in paced mode for preview/partial lag | | `partial_messages` | Count of non-final result messages with a non-empty `partial` | | `final_messages` | Count of final result messages | | `errors` | Connection, timeout, protocol, or client-side validation errors | +`first_update_ms` is unchanged: an empty result message can set it before any +transcript arrives. Use `first_text_ms` for the first client-visible transcript, +not model token-generation latency or a guarantee that the text is correct or +stable. Each client JSONL row includes both values. A client with no non-blank +text has `first_text_ms: null` and is excluded from the first-text percentiles; +if no client receives text, both percentiles are `null`. Always report +`clients_with_text`, total `clients`, and `errors` together so missing transcripts +cannot look like a latency improvement. + The script can observe only client-side timing and fields returned by the server. For a performance investigation, add `--log-decode-profile` to record one structured line per engine call with the request and sample counts, audio diff --git a/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py b/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py index 0bd2c37e1..1068d9c72 100644 --- a/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py +++ b/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py @@ -103,6 +103,13 @@ async def recv_results(ws, metrics, audio_started_at, stop_sent_at_ref, timeout) metrics["result_messages"] += 1 if metrics["first_update_ms"] is None: metrics["first_update_ms"] = (now - audio_started_at) * 1000.0 + texts = [data.get("partial")] + [ + sentence.get("text") for sentence in data.get("sentences", []) + ] + if metrics["first_text_ms"] is None and any( + isinstance(text, str) and text.strip() for text in texts + ): + metrics["first_text_ms"] = (now - audio_started_at) * 1000.0 if data.get("partial"): metrics["partial_messages"] += 1 duration_ms = data.get("duration_ms") @@ -135,6 +142,7 @@ async def run_client(client_id, args, audio_bytes, audio_seconds): "final_messages": 0, "events": {}, "first_update_ms": None, + "first_text_ms": None, "final_update_ms": None, "final_after_stop_ms": None, "response_lag_ms": [], @@ -194,6 +202,7 @@ async def run_client(client_id, args, audio_bytes, audio_seconds): metrics.update( { "first_update_ms": round_or_none(metrics["first_update_ms"], 1), + "first_text_ms": round_or_none(metrics["first_text_ms"], 1), "final_update_ms": round_or_none(metrics["final_update_ms"], 1), "final_after_stop_ms": round_or_none(metrics["final_after_stop_ms"], 1), "wall_seconds": round(wall_seconds, 3), @@ -210,6 +219,7 @@ async def run_client(client_id, args, audio_bytes, audio_seconds): def summarize(results, elapsed_seconds): total_audio = sum(item["audio_seconds"] for item in results) first_updates = [item["first_update_ms"] for item in results if item["first_update_ms"] is not None] + first_texts = [item["first_text_ms"] for item in results if item["first_text_ms"] is not None] final_after_stop = [ item["final_after_stop_ms"] for item in results if item["final_after_stop_ms"] is not None ] @@ -224,6 +234,11 @@ def summarize(results, elapsed_seconds): if first_updates else None, "first_update_ms_p95": round_or_none(percentile(first_updates, 95), 1), + "first_text_ms_p50": round_or_none(statistics.median(first_texts), 1) + if first_texts + else None, + "first_text_ms_p95": round_or_none(percentile(first_texts, 95), 1), + "clients_with_text": len(first_texts), "final_after_stop_ms_p50": round_or_none(statistics.median(final_after_stop), 1) if final_after_stop else None, @@ -242,6 +257,8 @@ def print_summary(summary, results): print(f"elapsed seconds: {summary['elapsed_seconds']}") print(f"aggregate audio/wall: {summary['aggregate_audio_per_wall']}x") print(f"first update p50/p95 ms: {summary['first_update_ms_p50']} / {summary['first_update_ms_p95']}") + print(f"first text p50/p95 ms: {summary['first_text_ms_p50']} / {summary['first_text_ms_p95']}") + print(f"clients with text: {summary['clients_with_text']}/{summary['clients']}") print( "final after STOP p50/p95 ms: " f"{summary['final_after_stop_ms_p50']} / {summary['final_after_stop_ms_p95']}" diff --git a/tests/test_realtime_ws_benchmark.py b/tests/test_realtime_ws_benchmark.py index f1224ea42..cae0ee30d 100644 --- a/tests/test_realtime_ws_benchmark.py +++ b/tests/test_realtime_ws_benchmark.py @@ -3,6 +3,10 @@ import importlib.util import json import sys from pathlib import Path +from types import SimpleNamespace +import wave + +import pytest REPO_ROOT = Path(__file__).resolve().parents[1] @@ -48,6 +52,7 @@ def test_final_message_does_not_contribute_to_response_lag(monkeypatch): "final_messages": 0, "events": {}, "first_update_ms": None, + "first_text_ms": None, "final_update_ms": None, "final_after_stop_ms": None, "response_lag_ms": [], @@ -123,3 +128,145 @@ def test_client_ping_timeout_zero_disables_timeout(): args = module.parse_args(["audio.wav", "--client-ping-timeout", "0"]) assert args.client_ping_timeout is None + + +@pytest.mark.parametrize( + "messages,expected", + [ + ([{"partial": ""}, {"partial": "hello"}, {"partial": "later"}], 1000.0), + ([{"sentences": []}, {"sentences": [{"text": "hello"}]}], 1000.0), + ( + [{"partial": ""}, {"is_final": True, "sentences": [{"text": "hello"}]}], + 1000.0, + ), + ( + [{"partial": " \t", "sentences": [{"text": "\n"}]}, {"partial": "hello"}], + 1000.0, + ), + ([{"partial": "hello"}, {"partial": "later"}], 0.0), + ([{"sentences": []}, {"is_final": True, "sentences": []}], None), + ([{"partial": None, "sentences": [{"text": None}]}], None), + ], +) +def test_first_text_waits_for_nonblank_transcript(monkeypatch, messages, expected): + module = load_benchmark_module() + responses = iter([*messages, {"event": "stopped"}]) + timestamps = iter(float(i) for i in range(len(messages) + 1)) + + async def receive_message(_ws, _timeout): + return next(responses) + + monkeypatch.setattr(module, "receive_message", receive_message) + monkeypatch.setattr( + module, "time", SimpleNamespace(perf_counter=lambda: next(timestamps)) + ) + metrics = { + "messages": 0, + "result_messages": 0, + "partial_messages": 0, + "final_messages": 0, + "events": {}, + "first_update_ms": None, + "first_text_ms": None, + "final_update_ms": None, + "final_after_stop_ms": None, + "response_lag_ms": [], + "stopped": False, + "errors": [], + } + asyncio.run(module.recv_results(object(), metrics, 0.0, {"value": None}, 1.0)) + + assert metrics["first_update_ms"] == 0.0 + assert metrics["first_text_ms"] == expected + assert metrics["stopped"] is True + assert metrics["errors"] == [] + + +@pytest.mark.parametrize( + "latencies,expected_p50,expected_p95,count", + [ + ([None, 0.0, 1000.0, 2000.0], 1000.0, 1900.0, 3), + ([None, None], None, None, 0), + ], +) +def test_first_text_summary_excludes_missing_text_but_counts_coverage( + latencies, expected_p50, expected_p95, count +): + module = load_benchmark_module() + results = [ + { + "audio_seconds": 1.0, + "first_update_ms": 0.0, + "first_text_ms": latency, + "final_after_stop_ms": 0.0, + "response_lag_ms_p95": 0.0, + "partial_messages": 0, + "final_messages": 1, + "errors": [], + } + for latency in latencies + ] + summary = module.summarize(results, 3.0) + + assert summary["first_text_ms_p50"] == expected_p50 + assert summary["first_text_ms_p95"] == expected_p95 + assert summary["clients_with_text"] == count + assert summary["clients"] == len(latencies) + assert summary["first_update_ms_p50"] == 0.0 + + +def test_first_text_round_trip_exports_client_and_summary_jsonl(tmp_path, capsys): + module = load_benchmark_module() + wav_path = tmp_path / "silence.wav" + output_path = tmp_path / "metrics.jsonl" + with wave.open(str(wav_path), "wb") as audio: + audio.setnchannels(1) + audio.setsampwidth(2) + audio.setframerate(16000) + audio.writeframes(b"\0\0" * 3200) + + async def replay_server(ws, *_args): + assert await ws.recv() == "START" + await ws.send(json.dumps({"event": "started"})) + assert isinstance(await ws.recv(), bytes) + await ws.send(json.dumps({"sentences": [], "partial": ""})) + assert isinstance(await ws.recv(), bytes) + await ws.send(json.dumps({"partial": "hello"})) + assert await ws.recv() == "STOP" + await ws.send(json.dumps({"is_final": True, "sentences": [{"text": "hello"}]})) + await ws.send(json.dumps({"event": "stopped"})) + + async def run(): + async with module.websockets.serve(replay_server, "127.0.0.1", 0) as server: + port = server.sockets[0].getsockname()[1] + args = module.parse_args( + [ + str(wav_path), + "--server", + f"ws://127.0.0.1:{port}", + "--no-pace", + "--clients", + "2", + "--output-jsonl", + str(output_path), + ] + ) + return await asyncio.wait_for(module.async_main(args), timeout=5) + + assert asyncio.run(run()) == 0 + records = [json.loads(line) for line in output_path.read_text().splitlines()] + clients = [row for row in records if row["type"] == "client"] + summary = records[-1] + assert len(clients) == 2 + for client in clients: + assert client["errors"] == [] + assert 0 <= client["first_update_ms"] <= client["first_text_ms"] + assert client["first_text_ms"] == round(client["first_text_ms"], 1) + assert client["stopped"] is True + assert summary["type"] == "summary" + assert summary["clients_with_text"] == summary["clients"] == 2 + assert summary["first_text_ms_p50"] is not None + assert summary["first_text_ms_p95"] is not None + printed = capsys.readouterr().out + assert "first text p50/p95 ms:" in printed + assert "clients with text: 2/2" in printed