hermes(stt): accurate large-v3-turbo final decode, fast tiny partials

Proper nouns (Córdoba, Cancún) and dropped words came from decoding the
committed transcript with the small model. The image already ships
large-v3-turbo, so the final decode now uses it with beam_size=5, a
temperature fallback ladder, and a proper-noun/accents initial_prompt
that fixes first-pass capitalization and diacritics across EN/ES/RU;
the rolling previews stay on tiny at greedy so the on-the-fly feel is
unchanged. The accurate decode runs in the speculative predecode during
the end-of-speech silence and is cache-reused at commit, so perceived
latency stays low. All decode knobs are env-overridable for on-device
tuning (beam/temperature/prompt), with small as the guaranteed-present
rollback if turbo underperforms on the Jetson. 206 STT tests pass.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf

dockerfiles/Dockerfile.hermes-jetson-stt

@@ -35,7 +35,7 @@ RUN python3 -c "import stat; from pathlib import Path; import torch, whisper; p=
 
 ENV HERMES_STT_HOST=0.0.0.0 \
     HERMES_STT_PORT=9000 \
-    HERMES_STT_MODEL=small \
+    HERMES_STT_MODEL=large-v3-turbo \
     HERMES_STT_ROLLING_MODEL=tiny \
     HERMES_STT_CACHE=/opt/models/whisper \
     PYTHONDONTWRITEBYTECODE=1 \

dockerfiles/hermes-jetson-stt-server.py

@@ -24,9 +24,43 @@ import whisper
 
 HOST = os.getenv("HERMES_STT_HOST", "0.0.0.0")
 PORT = int(os.getenv("HERMES_STT_PORT", "9000"))
-MODEL_NAME = os.getenv("HERMES_STT_MODEL", "small")
+MODEL_NAME = os.getenv("HERMES_STT_MODEL", "large-v3-turbo")
 ROLLING_MODEL_NAME = os.getenv("HERMES_STT_ROLLING_MODEL", "tiny")
 CACHE_DIR = Path(os.getenv("HERMES_STT_CACHE", "/cache/whisper"))
+
+# A short, neutral vocabulary/style prime for the accurate final decode only.
+# Whisper treats ``initial_prompt`` as soft bias, not a transcript, so a handful
+# of well-known accented place names across EN/ES/RU teaches correct
+# capitalization and diacritics on the FIRST pass (no "say it twice") without
+# memorizing any single user's phrases. It is never applied to the fast rolling
+# previews, which must stay clean and cheap.
+DEFAULT_INITIAL_PROMPT = (
+    "Proper nouns keep their capitalization and accents, for example "
+    "Córdoba, Cancún, Málaga, Moscú, Москва, and New York."
+)
+INITIAL_PROMPT = os.getenv("HERMES_STT_INITIAL_PROMPT", DEFAULT_INITIAL_PROMPT).strip()
+
+
+def _parse_temperature_fallback(raw: str) -> tuple[float, ...]:
+    """Parse a comma-separated Whisper temperature fallback ladder."""
+    values: list[float] = []
+    for token in raw.split(","):
+        token = token.strip()
+        if not token:
+            continue
+        try:
+            values.append(float(token))
+        except ValueError:
+            continue
+    return tuple(values) if values else (0.0,)
+
+
+# The accurate committed transcript retries a low-confidence greedy pass at a
+# few higher temperatures; the fast rolling preview never does.
+FINAL_BEAM_SIZE = max(1, int(os.getenv("HERMES_STT_FINAL_BEAM_SIZE", "5")))
+FINAL_TEMPERATURE = _parse_temperature_fallback(
+    os.getenv("HERMES_STT_FINAL_TEMPERATURE", "0.0,0.2,0.4")
+)
 MAX_AUDIO_BYTES = 30 * 1024 * 1024
 STREAM_SAMPLE_RATE = 16_000
 STREAM_SAMPLE_WIDTH = 2
@@ -261,26 +295,61 @@ def _transcription_payload(result: dict) -> dict:
     }
 
 
-def _decode_path(model: object, path: str, language: str) -> dict:
-    """Decode one path with the quality settings shared by partial and final work."""
+# Shared hallucination guards; the accuracy/latency split lives in the two
+# wrappers below, not here.
+_COMMON_DECODE_PARAMS = {
+    "task": "transcribe",
+    "condition_on_previous_text": False,
+    "compression_ratio_threshold": 2.0,
+    "logprob_threshold": -0.8,
+    "no_speech_threshold": 0.5,
+    "verbose": False,
+}
+
+
+def _decode(model: object, path: str, language: str, *, params: dict) -> dict:
+    """Decode one path with an explicit quality profile (final vs. rolling)."""
     result = model.transcribe(
         path,
         language=None if language in {"", "auto"} else language,
-        task="transcribe",
         fp16=torch.cuda.is_available(),
-        condition_on_previous_text=False,
-        temperature=0,
-        compression_ratio_threshold=2.0,
-        logprob_threshold=-0.8,
-        no_speech_threshold=0.5,
-        verbose=False,
+        **_COMMON_DECODE_PARAMS,
+        **params,
     )
     return _transcription_payload(result)
 
 
+def _final_decode_params() -> dict:
+    """Accurate profile: beam search, a small temperature fallback, and priming.
+
+    Used for the authoritative POST decode, the committed streaming turn, and the
+    speculative EOS predecode that the commit reuses verbatim (so the frozen
+    cache stays byte-identical to a direct final decode).
+    """
+    params = {"beam_size": FINAL_BEAM_SIZE, "temperature": FINAL_TEMPERATURE}
+    if INITIAL_PROMPT:
+        params["initial_prompt"] = INITIAL_PROMPT
+    return params
+
+
+# Fast, clean profile for disposable rolling previews: greedy, no fallback ladder
+# and deliberately no ``initial_prompt`` so live partials stay cheap and snappy.
+_ROLLING_DECODE_PARAMS = {"beam_size": 1, "temperature": 0}
+
+
+def _decode_final(model: object, path: str, language: str) -> dict:
+    """Decode with the accurate, primed profile for the committed transcript."""
+    return _decode(model, path, language, params=_final_decode_params())
+
+
+def _decode_rolling(model: object, path: str, language: str) -> dict:
+    """Decode with the fast greedy profile for a disposable live preview."""
+    return _decode(model, path, language, params=_ROLLING_DECODE_PARAMS)
+
+
 def _transcribe_path(model: object, path: str, language: str) -> dict:
-    """Run the one canonical full-utterance Whisper decode configuration."""
-    return INFERENCE_GATE.run_final(lambda: _decode_path(model, path, language))
+    """Run the accurate full-utterance decode behind the authoritative gate."""
+    return INFERENCE_GATE.run_final(lambda: _decode_final(model, path, language))
 
 
 def _write_pcm_wav(pcm: bytes) -> str:
@@ -311,12 +380,16 @@ def _transcribe_pcm(model: object, pcm: bytes, language: str) -> dict:
                 pass
 
 
-def _warm_model(model: object) -> float:
-    """Pay Whisper's lazy CUDA initialization cost before readiness."""
+def _warm_model(model: object, decode=_decode_rolling) -> float:
+    """Pay Whisper's lazy CUDA initialization cost before readiness.
+
+    Each model is warmed with the exact profile it will serve so the first real
+    request never inherits beam-search or fallback CUDA kernel compilation.
+    """
     temp_path = _write_pcm_wav(bytes(_pcm_bytes_for_ms(1_000)))
     started = time.monotonic()
     try:
-        _decode_path(model, temp_path, "auto")
+        decode(model, temp_path, "auto")
         torch.cuda.synchronize()
     finally:
         try:
@@ -337,7 +410,7 @@ def _transcribe_pcm_rolling(model: object, pcm: bytes, language: str) -> dict |
     try:
         temp_path = _write_pcm_wav(pcm)
         return INFERENCE_GATE.try_background(
-            lambda: _decode_path(model, temp_path, language),
+            lambda: _decode_rolling(model, temp_path, language),
             kind="rolling",
         )
     finally:
@@ -351,12 +424,17 @@ def _transcribe_pcm_rolling(model: object, pcm: bytes, language: str) -> dict |
 def _transcribe_pcm_speculative(
     model: object, pcm: bytes, language: str
 ) -> dict | None:
-    """Predecode one EOS snapshot only when the model is currently idle."""
+    """Predecode one EOS snapshot only when the model is currently idle.
+
+    This runs on the accurate final model with the final profile, so a commit
+    that reuses the frozen EOS cache gets the exact same high-accuracy result a
+    direct final decode would have produced.
+    """
     temp_path = ""
     try:
         temp_path = _write_pcm_wav(pcm)
         return INFERENCE_GATE.try_background(
-            lambda: _decode_path(model, temp_path, language),
+            lambda: _decode_final(model, temp_path, language),
             kind="speculative",
         )
     finally:
@@ -1152,9 +1230,10 @@ def main() -> None:
         "[stt] rolling CUDA warm-up completed in " f"{STARTUP_ROLLING_WARMUP_MS:.1f}ms",
         flush=True,
     )
-    # Warm the authoritative model last so the first real final never inherits
-    # lazy CUDA work from either model's startup path.
-    STARTUP_WARMUP_MS = _warm_model(model)
+    # Warm the authoritative model last, with its own beam-search profile, so
+    # the first real final never inherits lazy CUDA work from either model's
+    # startup path.
+    STARTUP_WARMUP_MS = _warm_model(model, _decode_final)
     print(
         f"[stt] final CUDA warm-up completed in {STARTUP_WARMUP_MS:.1f}ms", flush=True
     )

services/hermes/voice-deployment.yaml

@@ -19,7 +19,7 @@ spec:
         app: hermes-stt
       annotations:
         ai.bstein.dev/role: private-chat-speech-to-text
-        ai.bstein.dev/model: whisper-small-multilingual
+        ai.bstein.dev/model: whisper-large-v3-turbo-multilingual (tiny rolling previews)
         ai.bstein.dev/gpu: titan-21 dedicated speech time-slice
     spec:
       runtimeClassName: nvidia
@@ -39,7 +39,7 @@ spec:
             - {name: HOME, value: /tmp}
             - {name: XDG_CACHE_HOME, value: /tmp/cache}
             - {name: HERMES_STT_PORT, value: "9000"}
-            - {name: HERMES_STT_MODEL, value: small}
+            - {name: HERMES_STT_MODEL, value: large-v3-turbo}
             - {name: HERMES_STT_CACHE, value: /opt/models/whisper}
             - {name: NVIDIA_VISIBLE_DEVICES, value: all}
             - {name: NVIDIA_DRIVER_CAPABILITIES, value: "compute,utility"}

testing/tests/test_hermes_stt_rolling_model.py

@@ -281,7 +281,7 @@ def test_dual_warmup_and_separate_timing_are_declared(monkeypatch):
     assert not Path(model.calls[0]["path"]).exists()
     source = SERVER.read_text()
     rolling_warm = "STARTUP_ROLLING_WARMUP_MS = _warm_model(rolling_model)"
-    final_warm = "STARTUP_WARMUP_MS = _warm_model(model)"
+    final_warm = "STARTUP_WARMUP_MS = _warm_model(model, _decode_final)"
     assert source.index(rolling_warm) < source.index(final_warm)
     assert source.index(final_warm) < source.index("server = ThreadingHTTPServer")
     assert 'kind="rolling"' in source

testing/tests/test_hermes_stt_streaming.py

@@ -465,3 +465,54 @@ def test_stream_contract_rejects_wrong_format_and_oversize(monkeypatch):
     monkeypatch.setattr(module, "MAX_STREAM_AUDIO_BYTES", 4)
     with pytest.raises(module.WebSocketError, match="too long"):
         session.append(bytes(6))
+
+
+class _RecordingModel:
+    """Record the exact decode options each path passes to Whisper."""
+
+    def transcribe(self, path: str, **options: object) -> dict:
+        self.options = options
+        return {
+            "text": " ok",
+            "language": "en",
+            "segments": [{"text": " ok", "no_speech_prob": 0.0, "avg_logprob": -0.1}],
+        }
+
+
+def test_final_decode_is_accurate_and_primed_while_rolling_stays_fast_and_clean(
+    monkeypatch,
+):
+    """The committed/speculative path uses beam search + a temperature ladder +
+    an initial_prompt; the disposable rolling preview stays greedy and clean."""
+    module = _load_server(monkeypatch)
+    wav = module._write_pcm_wav(_speech(samples=1_600))
+
+    final_model = _RecordingModel()
+    module._decode_final(final_model, wav, "auto")
+    assert final_model.options["beam_size"] == module.FINAL_BEAM_SIZE >= 5
+    assert final_model.options["temperature"] == module.FINAL_TEMPERATURE
+    assert final_model.options["temperature"][0] == 0.0
+    assert len(final_model.options["temperature"]) > 1
+    assert final_model.options["initial_prompt"] == module.INITIAL_PROMPT
+    assert final_model.options["condition_on_previous_text"] is False
+
+    rolling_model = _RecordingModel()
+    module._decode_rolling(rolling_model, wav, "auto")
+    assert rolling_model.options["beam_size"] == 1
+    assert rolling_model.options["temperature"] == 0
+    assert "initial_prompt" not in rolling_model.options
+    assert rolling_model.options["condition_on_previous_text"] is False
+
+    Path(wav).unlink()
+
+
+def test_initial_prompt_is_env_overridable_and_default_primes_places(monkeypatch):
+    module = _load_server(monkeypatch)
+    # The default primes accented, capitalized place names across EN/ES/RU.
+    assert "Córdoba" in module.DEFAULT_INITIAL_PROMPT
+    assert "Москва" in module.DEFAULT_INITIAL_PROMPT
+    assert module.INITIAL_PROMPT == module.DEFAULT_INITIAL_PROMPT
+
+    monkeypatch.setenv("HERMES_STT_INITIAL_PROMPT", "Custom prime.")
+    reloaded = _load_server(monkeypatch)
+    assert reloaded.INITIAL_PROMPT == "Custom prime."