rtx-csm: Phase 8.10 — Moonshine as third AsrEngine in converse_server

Wires the Phase 8.5-8.9 Moonshine port as a drop-in alternative to
Kyutai STT in the conversation server. New `--moonshine` flag (mutex
with `--whisper` and `--vad`). Pure candle, no external runtime — no
ggml/protobuf conflicts unlike `--whisper`.

Architecture:
  - AsrEngine enum extended with Moonshine(MoonshineAsr) variant
  - MoonshineAsr bundles { encoder, decoder, tokenizer, cfg, device }
    plus a transcribe_24k() that resamples 24->16 kHz, encodes,
    greedy-decodes (KV-cached), detokenizes
  - Renamed Shared.whisper_mode -> Shared.batch_asr to cover both
    Whisper and Moonshine (both batch-only, skip parallel STT)
  - Receive loop's match arms now exhaustive over all three variants
  - At EOT, transcript construction branches:
      Kyutai    -> join words from incremental Word/EndWord stream
      Moonshine -> transcribe_24k() over accumulated audio
      Whisper   -> transcribe_24k() (asr feature)

End-to-end verified (mock LLM, Q8 + stream + warmup + Moonshine, real
LibriSpeech 10.43 s):

  recv_phase:          0 ms   (batch ASR; audio just buffers)
  stt_post:          406 ms   (Moonshine transcribe at EOT)
  llm_to_first_audio: 533 ms
  total_turn:      23617 ms
  *** Client TTFA:   939 ms ***   (sub-second!)

Compared to Kyutai (Phase 8.2 extended warmup baseline):
  Kyutai TTFA p50    4915 ms
  Moonshine TTFA      939 ms   ← -80%

Moonshine produces near-perfect transcript: "He hoped there would be
stew for dinner, turnips and carrots and bruised potatoes, and fat,
mutton pieces to be ladled out in thick, peppered, flour-fat and
sauce." matching the LibriSpeech ground truth.

This is the new production-recommended voice-loop config for
English-only deploys:

  converse_server \\
    --quantized-gguf <Q8> --stream-tts --moonshine \\
    [--vad-gate]   # energy VAD still useful for skipping silence
    [--llm-extra-body '{"thinking":{"type":"disabled"}}'   # for Z.AI]

For multilingual (en+fr) deploys, stick with Kyutai 1B (the default).

Phase 8 is now feature-complete on the optimization tracks the
research surfaced:
  - Tier 1 (warmup, energy VAD, ort gate): SHIPPED
  - Tier 2.2 (Moonshine candle port): SHIPPED end-to-end (8.4-8.10)
  - Tier 2.1 (VoXtream), Tier 3 (Frame-Stacked, VADUSA): deferred,
    documented in plan + perf_history.md

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
This commit is contained in:
osobh
2026-04-27 11:22:55 -07:00
co-authored by Claude Opus 4.7
parent 95015c17e1
commit 932a93c2ce
+114 -29
View File
@@ -175,6 +175,19 @@ struct Cli {
#[arg(long)] #[arg(long)]
stream_tts: bool, stream_tts: bool,
/// Use Moonshine-tiny (pure candle, English-only) instead of Kyutai
/// STT 1B for transcription. ~34× faster than realtime on M-series
/// Metal (10 s of audio → ~300 ms transcribe). Batch-only — fires
/// the transcript at EOT, not incrementally. Loses semantic VAD.
///
/// **Pure-Rust win**: unlike `--whisper`, Moonshine integrates
/// cleanly without ggml/protobuf runtime conflicts. No feature
/// flag needed (ships in the default build).
///
/// English-only — for multilingual deploys, stick with Kyutai.
#[arg(long)]
moonshine: bool,
/// Use Whisper-tiny (whisper-rs / whisper.cpp) instead of Kyutai STT /// Use Whisper-tiny (whisper-rs / whisper.cpp) instead of Kyutai STT
/// 1B for transcription. ~50× faster on Metal in isolation (10 s of /// 1B for transcription. ~50× faster on Metal in isolation (10 s of
/// audio → 0.2 s transcribe) but batch-only — transcript fires at /// audio → 0.2 s transcribe) but batch-only — transcript fires at
@@ -320,13 +333,51 @@ struct Metrics {
/// ASR backend wrapped behind a single Mutex so the receive loop can /// ASR backend wrapped behind a single Mutex so the receive loop can
/// branch on its kind. Whisper variant is feature-gated; without `asr` /// branch on its kind. Whisper variant is feature-gated; without `asr`
/// at compile time only Kyutai is available. /// at compile time only Kyutai + Moonshine are available.
///
/// Moonshine is pure candle (no external runtime), so it ships in the
/// default build alongside Kyutai. English-only.
enum AsrEngine { enum AsrEngine {
Kyutai(Stt), Kyutai(Stt),
Moonshine(MoonshineAsr),
#[cfg(feature = "asr")] #[cfg(feature = "asr")]
Whisper(rtx_csm::asr::WhisperAsr), Whisper(rtx_csm::asr::WhisperAsr),
} }
/// Bundle of Moonshine encoder + decoder + tokenizer + config so the
/// receive-loop's batch transcribe call has everything it needs.
struct MoonshineAsr {
encoder: rtx_csm::moonshine::Encoder,
decoder: rtx_csm::moonshine::Decoder,
tokenizer: tokenizers::Tokenizer,
cfg: rtx_csm::moonshine::MoonshineConfig,
device: candle_core::Device,
}
impl MoonshineAsr {
/// Resample 24 kHz user audio to 16 kHz, encode, greedy decode,
/// detokenize. Returns the transcript text. Batch-only (no
/// streaming, no per-frame VAD).
fn transcribe_24k(&self, samples_24k: &[f32]) -> anyhow::Result<String> {
let samples_16k = rtx_csm::audio_io::resample(samples_24k, 24_000, 16_000)
.map_err(|e| anyhow::anyhow!("resample 24k -> 16k: {e}"))?;
let pcm = candle_core::Tensor::from_vec(
samples_16k.clone(),
(1, 1, samples_16k.len()),
&self.device,
)?;
let enc = self.encoder.forward(&pcm)?;
let token_ids = self
.decoder
.generate_cached(&enc, &self.cfg, self.cfg.max_position_embeddings.min(180))?;
let text = self
.tokenizer
.decode(&token_ids, true)
.map_err(|e| anyhow::anyhow!("detok: {e}"))?;
Ok(text)
}
}
struct Shared { struct Shared {
generator: Mutex<Generator>, generator: Mutex<Generator>,
stt: Mutex<AsrEngine>, stt: Mutex<AsrEngine>,
@@ -348,10 +399,10 @@ struct Shared {
stream_tts: bool, stream_tts: bool,
/// Mimi frames per stream chunk (4 = 320ms default). /// Mimi frames per stream chunk (4 = 320ms default).
stream_chunk_frames: usize, stream_chunk_frames: usize,
/// True when the ASR backend is Whisper (batch-only). The receive /// True when the ASR backend is batch-only (Whisper or Moonshine).
/// loop skips parallel STT in this case and runs one transcribe /// The receive loop skips parallel STT in this case and runs one
/// call at EOT. /// transcribe call at EOT. (Kyutai uses incremental step_pcm.)
whisper_mode: bool, batch_asr: bool,
/// Optional VAD gate. When `Some`, each incoming binary chunk is /// Optional VAD gate. When `Some`, each incoming binary chunk is
/// classified speech/silence before being sent to STT; silence /// classified speech/silence before being sent to STT; silence
/// chunks skip step_pcm. Behind a Mutex because is_speech() takes /// chunks skip step_pcm. Behind a Mutex because is_speech() takes
@@ -520,7 +571,34 @@ async fn main() -> Result<()> {
); );
} }
let stt: AsrEngine = if cli.whisper { if cli.moonshine && cli.whisper {
anyhow::bail!("--moonshine and --whisper are mutually exclusive (pick one ASR backend).");
}
if cli.moonshine && cli.vad {
anyhow::bail!(
"--moonshine and --vad cannot combine: Moonshine has no per-frame probability \
output."
);
}
let stt: AsrEngine = if cli.moonshine {
tracing::info!("loading Moonshine-tiny (pure candle, English-only)...");
let api = hf_hub::api::sync::Api::new()?;
let repo = api.model("UsefulSensors/moonshine-tiny".to_string());
let weights = repo.get("model.safetensors")?;
let tok_path = repo.get("tokenizer.json")?;
let cfg = rtx_csm::moonshine::MoonshineConfig::tiny();
let (encoder, decoder) =
rtx_csm::moonshine::load_full(&weights, &device, &cfg)?;
let tokenizer = rtx_csm::moonshine::load_tokenizer(&tok_path)
.map_err(|e| anyhow::anyhow!("moonshine tokenizer: {e}"))?;
AsrEngine::Moonshine(MoonshineAsr {
encoder,
decoder,
tokenizer,
cfg,
device: device.clone(),
})
} else if cli.whisper {
if cli.vad { if cli.vad {
anyhow::bail!( anyhow::bail!(
"--whisper and --vad cannot combine: Whisper has no per-frame probability \ "--whisper and --vad cannot combine: Whisper has no per-frame probability \
@@ -629,7 +707,7 @@ async fn main() -> Result<()> {
llm_extra_body, llm_extra_body,
stream_tts: cli.stream_tts, stream_tts: cli.stream_tts,
stream_chunk_frames: cli.stream_chunk_frames.max(1), stream_chunk_frames: cli.stream_chunk_frames.max(1),
whisper_mode: cli.whisper, batch_asr: cli.whisper || cli.moonshine,
vad_gate: if cli.vad_gate { vad_gate: if cli.vad_gate {
tracing::info!( tracing::info!(
"energy VAD gate enabled (rms_threshold={:.4})", "energy VAD gate enabled (rms_threshold={:.4})",
@@ -825,7 +903,7 @@ async fn handle_connection(mut socket: WebSocket, shared: Arc<Shared>) {
// skips parallel ingest entirely. STT runs on the blocking pool // skips parallel ingest entirely. STT runs on the blocking pool
// so the async worker can poll other tasks during the ~80ms // so the async worker can poll other tasks during the ~80ms
// per-frame work (multi-tenant concurrency win). // per-frame work (multi-tenant concurrency win).
if !user_audio_24k.is_empty() && !shared.whisper_mode { if !user_audio_24k.is_empty() && !shared.batch_asr {
let carry_slice = user_audio_24k.clone(); let carry_slice = user_audio_24k.clone();
stt_streaming_offset = carry_slice.len(); stt_streaming_offset = carry_slice.len();
let shared_carry = shared.clone(); let shared_carry = shared.clone();
@@ -839,6 +917,7 @@ async fn handle_connection(mut socket: WebSocket, shared: Arc<Shared>) {
Vec::new() Vec::new()
} }
}, },
AsrEngine::Moonshine(_) => Vec::new(),
#[cfg(feature = "asr")] #[cfg(feature = "asr")]
AsrEngine::Whisper(_) => Vec::new(), AsrEngine::Whisper(_) => Vec::new(),
} }
@@ -885,7 +964,7 @@ async fn handle_connection(mut socket: WebSocket, shared: Arc<Shared>) {
// before STT; if silence, skip step_pcm entirely. // before STT; if silence, skip step_pcm entirely.
// Saves ~80-300 ms per silent chunk depending on // Saves ~80-300 ms per silent chunk depending on
// chunk size. // chunk size.
let events = if !shared.whisper_mode { let events = if !shared.batch_asr {
let new_slice = let new_slice =
user_audio_24k[stt_streaming_offset..].to_vec(); user_audio_24k[stt_streaming_offset..].to_vec();
stt_streaming_offset = user_audio_24k.len(); stt_streaming_offset = user_audio_24k.len();
@@ -911,6 +990,7 @@ async fn handle_connection(mut socket: WebSocket, shared: Arc<Shared>) {
Vec::new() Vec::new()
} }
}, },
AsrEngine::Moonshine(_) => Vec::new(),
#[cfg(feature = "asr")] #[cfg(feature = "asr")]
AsrEngine::Whisper(_) => Vec::new(), AsrEngine::Whisper(_) => Vec::new(),
} }
@@ -987,6 +1067,7 @@ async fn handle_connection(mut socket: WebSocket, shared: Arc<Shared>) {
let mut stt = shared.stt.lock().await; let mut stt = shared.stt.lock().await;
match &mut *stt { match &mut *stt {
AsrEngine::Kyutai(s) => s.finish().unwrap_or_default(), AsrEngine::Kyutai(s) => s.finish().unwrap_or_default(),
AsrEngine::Moonshine(_) => Vec::new(),
#[cfg(feature = "asr")] #[cfg(feature = "asr")]
AsrEngine::Whisper(_) => Vec::new(), AsrEngine::Whisper(_) => Vec::new(),
} }
@@ -1029,26 +1110,29 @@ async fn handle_connection(mut socket: WebSocket, shared: Arc<Shared>) {
// transcribe call here over the full accumulated audio. (With // transcribe call here over the full accumulated audio. (With
// Whisper-tiny on Metal, 10s of audio takes ~200ms — much // Whisper-tiny on Metal, 10s of audio takes ~200ms — much
// faster than Kyutai's 5s parallel-STT-during-receive.) // faster than Kyutai's 5s parallel-STT-during-receive.)
// Build user_text. Kyutai has been streaming words incrementally
// — just join them. Batch backends (Whisper / Moonshine) fire
// one transcribe call here over the full accumulated audio.
let stt_t = std::time::Instant::now(); let stt_t = std::time::Instant::now();
let user_text = if shared.whisper_mode { let user_text = if shared.batch_asr {
#[cfg(feature = "asr")] let stt = shared.stt.lock().await;
{ match &*stt {
let stt = shared.stt.lock().await; AsrEngine::Moonshine(m) => match m.transcribe_24k(&user_audio_24k) {
if let AsrEngine::Whisper(w) = &*stt { Ok(t) => t.trim().to_string(),
match w.transcribe_24k(&user_audio_24k) { Err(e) => {
Ok(t) => t.trim().to_string(), tracing::warn!("moonshine transcribe: {e}");
Err(e) => { String::new()
tracing::warn!("whisper transcribe: {e}");
String::new()
}
} }
} else { },
transcript_words.join(" ").trim().to_string() #[cfg(feature = "asr")]
} AsrEngine::Whisper(w) => match w.transcribe_24k(&user_audio_24k) {
} Ok(t) => t.trim().to_string(),
#[cfg(not(feature = "asr"))] Err(e) => {
{ tracing::warn!("whisper transcribe: {e}");
transcript_words.join(" ").trim().to_string() String::new()
}
},
AsrEngine::Kyutai(_) => transcript_words.join(" ").trim().to_string(),
} }
} else { } else {
transcript_words.join(" ").trim().to_string() transcript_words.join(" ").trim().to_string()
@@ -1404,8 +1488,9 @@ async fn collect_transcript_events(
let stt = stt_lock.lock().await; let stt = stt_lock.lock().await;
let kyutai = match &*stt { let kyutai = match &*stt {
AsrEngine::Kyutai(s) => s, AsrEngine::Kyutai(s) => s,
// Whisper path doesn't go through this — events come from // Batch backends (Moonshine / Whisper) don't go through this —
// Kyutai's incremental Word/EndWord stream only. // events come from Kyutai's incremental Word/EndWord stream only.
AsrEngine::Moonshine(_) => return,
#[cfg(feature = "asr")] #[cfg(feature = "asr")]
AsrEngine::Whisper(_) => return, AsrEngine::Whisper(_) => return,
}; };