parakeet opts

docs/models/nemotron-speech-streaming-en-0.6b.md

@@ -50,6 +50,32 @@ NVIDIA's self-reported number on the same split at
 `att_context_size=[70, 13]` (1.12s chunk, w/o PnC) is 2.32% (from the
 [HF model card](https://huggingface.co/nvidia/nemotron-speech-streaming-en-0.6b)).
 
+## Streaming WER
+
+In cache-aware streaming mode the encoder runs incrementally over fixed
+chunks while carrying constant-memory caches. WER on the full LibriSpeech
+test-clean split (2620 utterances) at the default `att_context_right=13`
+setting.
+
+| Quantization | WER (streaming, R=13) |
+| --- | ---: |
+| F16  | 2.29% |
+| Q8_0 | 2.31% |
+
+Streaming matches offline (2.31%) and NVIDIA's NeMo cache-aware streaming
+reference on the same split (2.31%)
+
+Reproduction:
+
+```bash
+uv run scripts/wer/run.py \
+  --cli build/bin/transcribe-cli \
+  --model models/nemotron-speech-streaming-en-0.6b/nemotron-speech-streaming-en-0.6b-Q8_0.gguf \
+  --manifest <librispeech test-clean manifest> --out hyps.jsonl \
+  --stream-chunk-ms 1040 --stream-att-right 13
+uv run scripts/wer/score.py hyps.jsonl
+```
+
 ## Quick Start
 
 ```bash

scripts/wer/remote/cache_paths.py

@@ -15,6 +15,7 @@ def hyp_cache_paths(
     timestamps: str = "none",
     language: str = "",
     stream_chunk_ms: int = 0,
+    stream_att_right: int = -1,
 ) -> tuple[str, str]:
     """Deterministic Volume paths for the (model, dataset, subset, batch,
     sort) tuple.
@@ -40,8 +41,12 @@ def hyp_cache_paths(
     # Streaming (--stream-chunk-ms) produces different hyps than the offline
     # path, so it gets its own cache slot. 0 = offline (no tag, compatible).
     stream_tag = "" if stream_chunk_ms <= 0 else f".stream{stream_chunk_ms}ms"
+    # Cache-aware latency preset (--stream-att-right). -1 = unset (model
+    # default R), no tag so it stays compatible with already-cached entries;
+    # any explicit R gets its own slot so e.g. R=13 and R=0 never collide.
+    r_tag = "" if stream_att_right < 0 else f".r{stream_att_right}"
     base = (f"/data/wer/hyps/{hyp_fp}/{slug}."
-            f"{dataset_id(dataset_spec)}.{subset_tag}{bs_tag}{sort_tag}{ts_tag}{lang_tag}{stream_tag}")
+            f"{dataset_id(dataset_spec)}.{subset_tag}{bs_tag}{sort_tag}{ts_tag}{lang_tag}{stream_tag}{r_tag}")
     return f"{base}.jsonl", f"{base}.summary.json"
 
 

scripts/wer/remote/modal_sweep.py

@@ -461,6 +461,7 @@ def _run_wer_impl(
     timestamps: str = "none",
     language: str = "",
     stream_chunk_ms: int = 0,
+    stream_att_right: int = -1,
     dataset_status: dict | None = None,
 ) -> dict:
     """Run scripts/wer/run.py on `n_utts` (or full manifest) and return the
@@ -480,7 +481,7 @@ def _run_wer_impl(
     # when a hyp for this (fingerprint, model, dataset, subset) already exists.
     cache_hyp, cache_sum = hyp_cache_paths(
         HYP_FP, model_file, dataset_spec, n_utts, batch_size, sort_by_length,
-        timestamps, language, stream_chunk_ms)
+        timestamps, language, stream_chunk_ms, stream_att_right)
     if os.path.exists(cache_hyp) and os.path.exists(cache_sum) \
        and os.path.getsize(cache_hyp) > 0:
         _log_prepared_dataset("wer", dataset_status)
@@ -537,6 +538,8 @@ def _run_wer_impl(
         cmd += ["--language", language]
     if stream_chunk_ms and stream_chunk_ms > 0:
         cmd += ["--stream-chunk-ms", str(stream_chunk_ms)]
+        if stream_att_right >= 0:
+            cmd += ["--stream-att-right", str(stream_att_right)]
     print(f"[wer] $ {' '.join(cmd)}")
     t0 = time.time()
     rc, stderr_tail = run_subprocess_capturing_stderr(cmd, cwd="/work", env=env)
@@ -584,6 +587,7 @@ def _run_wer_impl(
         "utt_per_s": subset_count / wall_s if wall_s > 0 else 0.0,
         "batch_size": batch_size,
         "stream_chunk_ms": stream_chunk_ms,
+        "stream_att_right": stream_att_right,
         "mel_s": mel_ms / 1000.0,
         "encode_s": enc_ms / 1000.0,
         "decode_s": dec_ms / 1000.0,
@@ -633,6 +637,7 @@ def runner(
         timestamps: str = "none",
         language: str = "",
         stream_chunk_ms: int = 0,
+        stream_att_right: int = -1,
         dataset_status: dict | None = None,
     ) -> dict:
         # Prefer the build_dir the local entrypoint computed and built into:
@@ -645,6 +650,7 @@ def runner(
             batch_size=batch_size, sort_by_length=sort_by_length,
             timestamps=timestamps, language=language,
             stream_chunk_ms=stream_chunk_ms,
+            stream_att_right=stream_att_right,
             dataset_status=dataset_status,
         )
 
@@ -885,6 +891,7 @@ def sweep(
     timestamps: str = "none",
     language: str = "",
     stream_chunk_ms: int = 0,
+    stream_att_right: int = -1,
 ) -> None:
     """Fan WER across one or more models on one GPU class.
 
@@ -968,7 +975,8 @@ def sweep(
     n = None if n_utts < 0 else n_utts
     futs = [(c, runner.spawn(c["repo"], c["file"], c["dataset"], n,
                              c["bs"], sort_by_length, build_dir, timestamps,
-                             language, stream_chunk_ms, dataset_status))
+                             language, stream_chunk_ms, stream_att_right,
+                             dataset_status))
             for c in cells]
 
     rows, failures = [], []
@@ -982,7 +990,8 @@ def sweep(
             p = write_hyp(repo_root, res["hyp_jsonl"], c["file"], c["dataset"],
                           batch_size=(bs if bs != 1 else None),
                           timestamps=timestamps,
-                          stream_chunk_ms=stream_chunk_ms)
+                          stream_chunk_ms=stream_chunk_ms,
+                          stream_att_right=stream_att_right)
             s = res["summary"]
             rows.append((slug, c["dataset"], s["n_utts"], s["audio_s"],
                          s["wall_s"], s["rtf_wall"], str(p)))

scripts/wer/remote/output_paths.py

@@ -30,6 +30,7 @@ def write_hyp(
     batch_size: int | None = None,
     timestamps: str = "none",
     stream_chunk_ms: int = 0,
+    stream_att_right: int = -1,
 ) -> Path:
     out_dir = root / "reports" / "wer"
     out_dir.mkdir(parents=True, exist_ok=True)
@@ -38,6 +39,7 @@ def write_hyp(
     bs_tag = "" if batch_size is None else f".b{batch_size}"
     ts_tag = "" if timestamps == "none" else f".ts-{timestamps}"
     stream_tag = "" if stream_chunk_ms <= 0 else f".stream{stream_chunk_ms}ms"
-    out_path = out_dir / f"{slug}.{ds}{bs_tag}{ts_tag}{stream_tag}.jsonl"
+    r_tag = "" if stream_att_right < 0 else f".r{stream_att_right}"
+    out_path = out_dir / f"{slug}.{ds}{bs_tag}{ts_tag}{stream_tag}{r_tag}.jsonl"
     out_path.write_text(hyp_jsonl)
     return out_path

scripts/wer/run_reference_parakeet_streaming_nemo.py

@@ -55,6 +55,16 @@ def main() -> int:
     p.add_argument("--pad-and-drop-preencoded", action="store_true",
                    help="Treat the first chunk like subsequent (ONNX-export "
                         "first-chunk semantics).")
+    # Uniform reference contract (modal_sweep _run_one_reference always passes
+    # these). Streaming is inherently per-utterance, so --batch-size >1 and
+    # --mode are accepted and ignored; --device selects the torch device.
+    p.add_argument("--device", default="cuda",
+                   help="Torch device (cuda|cpu); falls back to cpu if cuda "
+                        "is unavailable.")
+    p.add_argument("--batch-size", type=int, default=1,
+                   help="Uniform-contract arg; ignored (streaming is bs=1).")
+    p.add_argument("--mode", default="streaming",
+                   help="Uniform-contract arg; this runner is always streaming.")
     args = p.parse_args()
 
     if not args.manifest.exists():
@@ -95,6 +105,9 @@ def main() -> int:
         else:
             raise last
     model.eval()
+    dev = "cuda" if (args.device == "cuda" and torch.cuda.is_available()) else "cpu"
+    model = model.to(dev)
+    print(f"device: {dev}")
     load_ms = (time.monotonic() - t0) * 1000
 
     if model.encoder.att_context_style != "chunked_limited":
@@ -167,8 +180,8 @@ def transcribe_one_streaming(audio_path: str) -> str:
                     channel_len,
                     previous_hypotheses,
                 ) = model.conformer_stream_step(
-                    processed_signal=chunk_audio.to(torch.float32),
-                    processed_signal_length=chunk_lengths,
+                    processed_signal=chunk_audio.to(dev, torch.float32),
+                    processed_signal_length=chunk_lengths.to(dev),
                     cache_last_channel=cache_lc,
                     cache_last_time=cache_lt,
                     cache_last_channel_len=channel_len,

src/arch/parakeet/decoder.cpp

@@ -546,8 +546,23 @@ inline void linear(const float * W,
 #endif
     for (int r = 0; r < out_dim; ++r) {
         const float * row = W + static_cast<size_t>(r) * static_cast<size_t>(in_dim);
-        float acc = 0.0f;
-        for (int c = 0; c < in_dim; ++c) {
+        // Eight independent accumulators break the single-acc reduction
+        // dependency chain so the compiler can auto-vectorize the dot (the
+        // serial version pins it to one FMA's latency — ~7x off on MSVC, the
+        // dominant decode cost). Pure fp32, DL-safe, portable: no intrinsics,
+        // no ggml-cpu symbols, no arch flags. The tree-shaped final sum changes
+        // float reassociation only (argmax-robust, transcript-identical).
+        float a0 = 0, a1 = 0, a2 = 0, a3 = 0, a4 = 0, a5 = 0, a6 = 0, a7 = 0;
+        int c = 0;
+        const int in8 = in_dim & ~7;
+        for (; c < in8; c += 8) {
+            a0 += row[c + 0] * x[c + 0]; a1 += row[c + 1] * x[c + 1];
+            a2 += row[c + 2] * x[c + 2]; a3 += row[c + 3] * x[c + 3];
+            a4 += row[c + 4] * x[c + 4]; a5 += row[c + 5] * x[c + 5];
+            a6 += row[c + 6] * x[c + 6]; a7 += row[c + 7] * x[c + 7];
+        }
+        float acc = ((a0 + a1) + (a2 + a3)) + ((a4 + a5) + (a6 + a7));
+        for (; c < in_dim; ++c) {
             acc += row[c] * x[c];
         }
         y[r] = acc + (b != nullptr ? b[r] : 0.0f);
@@ -774,7 +789,15 @@ void joint_step(const HostJoint &     j,
     // Cost: ~joint_n adds + 1 logsumexp. For joint_n=1030 that's ~2K
     // ops per decode iter, negligible vs the joint_h × pred_hidden +
     // joint_h × joint_n matmuls (~1.3M ops).
-    {
+    //
+    // GREEDY FAST PATH: the log_softmax is a uniform per-row shift, so it leaves
+    // BOTH the token/duration argmax AND token_confidence (which re-softmaxes the
+    // token sub-range, absorbing the shift) invariant. It is therefore only
+    // needed to make the `dec.joint.0` dump comparable to NeMo's normalized
+    // reference. Skip it entirely unless dumping — saves a joint_n-wide
+    // max+exp+log+sub pass (joint_n≈1030, with a double exp) on every decode
+    // iteration. Bit-identical decode output either way.
+    if (transcribe::debug::enabled()) {
         float max_v = out_logits[0];
         for (int i = 1; i < j.joint_n; ++i) {
             if (out_logits[i] > max_v) max_v = out_logits[i];
@@ -813,20 +836,39 @@ void precompute_enc_proj(const HostJoint &    j,
                 1.0f, enc_out, d_enc,
                 j.enc_w.data(), d_enc,
                 0.0f, out.data(), joint_h);
-#else
+    // Add bias to every row.
     for (int t = 0; t < T; ++t) {
-        const float * frame = enc_out + static_cast<size_t>(t) * static_cast<size_t>(d_enc);
         float * proj = out.data() + static_cast<size_t>(t) * static_cast<size_t>(joint_h);
-        linear(j.enc_w.data(), frame, nullptr, joint_h, d_enc, proj, n_threads);
+        for (int k = 0; k < joint_h; ++k) {
+            proj[k] += j.enc_b[static_cast<size_t>(k)];
+        }
     }
+#else
+    // No BLAS: this is a [T, d_enc] x [d_enc, joint_h]^T GEMM. The previous code
+    // ran it as T serial sgemv calls via linear(), and linear() only threads when
+    // out_dim >= 2048 — joint_h is below that, so the whole projection ran
+    // single-threaded (~90 ms for T=138 on this model, the dominant decode cost,
+    // paid on CPU even under the Vulkan backend since the decoder is host code).
+    // The rows are fully independent, so parallelize over T and fold the bias in.
+    // The inner dot auto-vectorizes (FMA/AVX-512) under the project's arch flags.
+    const float * enc_w = j.enc_w.data();
+    const float * enc_b = j.enc_b.data();
+#ifdef _OPENMP
+    #pragma omp parallel for schedule(static) num_threads(n_threads > 0 ? n_threads : 1)
 #endif
-    // Add bias to every row.
     for (int t = 0; t < T; ++t) {
+        const float * frame = enc_out + static_cast<size_t>(t) * static_cast<size_t>(d_enc);
         float * proj = out.data() + static_cast<size_t>(t) * static_cast<size_t>(joint_h);
-        for (int k = 0; k < joint_h; ++k) {
-            proj[k] += j.enc_b[static_cast<size_t>(k)];
+        for (int r = 0; r < joint_h; ++r) {
+            const float * row = enc_w + static_cast<size_t>(r) * static_cast<size_t>(d_enc);
+            float acc = 0.0f;
+            for (int c = 0; c < d_enc; ++c) {
+                acc += row[c] * frame[c];
+            }
+            proj[r] = acc + enc_b[static_cast<size_t>(r)];
         }
     }
+#endif
 }
 
 // Argmax over a contiguous fp32 range. Returns the index of the