feat(morpheus,gateway-chat): streaming decode — TTFA ~4.5s -> ~0.8s

Wrapper gains POST /tts/stream: reads the vLLM token stream, decodes SNAC in WINDOWED
CHUNKS (every 6 frames, decode [2 ctx | 6 | 2 ctx] and emit only the middle 6 — context
both sides => seamless), and streams raw PCM16 (24kHz mono) as it generates. Windowed
(not per-frame) because per-frame CPU decode's per-call overhead x ~60 frames serialized
to ~7s (RTF 2.2); windowed keeps up (RTF ~0.97). Whole-clip /tts kept for non-browser use.

gateway-chat plays the stream via the Web Audio API (fetch reader -> int16->float32 ->
scheduled AudioBufferSourceNodes on a running clock; a new reply supersedes the prior
stream via a generation counter; 🔊 replays). Measured: TTFA 0.80s (was ~4.5s whole-clip),
RTF 0.97, full-duration match. CORS already covers the new route.

Deployed: tts rebuilt on irv-ml1, page pushed to ana-docker.

stacks/gateway-chat/conf/index.html

@@ -144,33 +144,54 @@ function addMsg(who, cls){
 }
 function showErr(text){ const m = addMsg('error','err'); m.body.className = 'err'; m.body.textContent = text; }
 
-// --- mOrpheus TTS: auto-voice quoted text from a completed assistant reply ---
+// --- mOrpheus TTS: streaming-decode auto-voice of quoted text (Web Audio, low TTFA) ---
-let ttsAudio = null;   // current clip, so a new reply interrupts the previous one
+let audioCtx = null;   // shared AudioContext
+let ttsGen = 0;        // generation counter — a newer reply's stream supersedes older ones
 function extractQuotes(text){
   const re = /“([^”]+)”|"([^"]+)"/g;   // typographic "…" or straight "…"
   const out = []; let m;
   while ((m = re.exec(text)) !== null){ const q = (m[1] || m[2] || '').trim(); if (q) out.push(q); }
   return out;
 }
+async function streamSpeak(speech, tag){
+  const mine = ++ttsGen;                                // cancels any in-flight stream
+  if (!audioCtx) audioCtx = new (window.AudioContext || window.webkitAudioContext)();
+  if (audioCtx.state === 'suspended') await audioCtx.resume();
+  const url = $('ttsUrl').value.trim().replace(/\/+$/,'').replace(/\/tts$/,'') + '/tts/stream';
+  const res = await fetch(url, {
+    method:'POST', headers:{ 'Content-Type':'application/json' },
+    body: JSON.stringify({ text: speech, voice: ($('ttsVoice').value.trim() || 'baddy'), max_tokens: 1200 })
+  });
+  if (!res.ok || !res.body){ tag.textContent = ' 🔇' + res.status; return; }
+  const reader = res.body.getReader();
+  let playhead = audioCtx.currentTime + 0.12, leftover = new Uint8Array(0);
+  for(;;){
+    const { done, value } = await reader.read();
+    if (done || mine !== ttsGen) break;                 // stop if a newer reply took over
+    let bytes = new Uint8Array(leftover.length + value.length);
+    bytes.set(leftover); bytes.set(value, leftover.length);
+    const n = bytes.length - (bytes.length % 2);        // whole int16 samples only
+    leftover = bytes.slice(n);
+    if (!n) continue;
+    const dv = new DataView(bytes.buffer, 0, n), N = n / 2, f32 = new Float32Array(N);
+    for (let i = 0; i < N; i++) f32[i] = dv.getInt16(i * 2, true) / 32768;
+    const abuf = audioCtx.createBuffer(1, N, 24000); abuf.getChannelData(0).set(f32);
+    const src = audioCtx.createBufferSource(); src.buffer = abuf; src.connect(audioCtx.destination);
+    if (playhead < audioCtx.currentTime) playhead = audioCtx.currentTime + 0.02;  // underrun catch-up
+    src.start(playhead); playhead += abuf.duration;
+  }
+  if (mine === ttsGen) tag.textContent = ' 🔈';
+}
 async function speakQuotes(text, msg){
   if (!$('ttsOn').checked) return;
   const quotes = extractQuotes(text);
   if (!quotes.length) return;
+  const speech = quotes.join('  ');
   const tag = document.createElement('span'); tag.textContent = ' 🔊';
-  tag.style.cursor = 'pointer'; tag.title = 'mOrpheus'; msg.wrap.querySelector('.who').append(tag);
+  tag.style.cursor = 'pointer'; tag.title = 'mOrpheus — click to replay';
-  try {
+  tag.onclick = () => streamSpeak(speech, tag).catch(() => tag.textContent = ' 🔇');
-    const res = await fetch($('ttsUrl').value.trim(), {
+  msg.wrap.querySelector('.who').append(tag);
-      method:'POST', headers:{ 'Content-Type':'application/json' },
+  try { await streamSpeak(speech, tag); } catch (e){ tag.textContent = ' 🔇'; }
-      body: JSON.stringify({ text: quotes.join('  '), voice: ($('ttsVoice').value.trim() || 'baddy'), max_tokens: 1200 })
-    });
-    if (!res.ok){ tag.textContent = ' 🔇' + res.status; return; }
-    const url = URL.createObjectURL(await res.blob());
-    if (ttsAudio) ttsAudio.pause();
-    ttsAudio = new Audio(url);
-    tag.onclick = () => ttsAudio.play();          // click 🔊 to replay
-    ttsAudio.onended = () => tag.textContent = ' 🔈';
-    ttsAudio.play();
-  } catch (e){ tag.textContent = ' 🔇'; }
 }
 
 async function send(){

stacks/mOrpheus/tts/app.py

@@ -11,11 +11,11 @@ pos2/3/5/6->L3, offsets k*4096). Named-speaker prompt: [SOH] "voice: text" [EOT]
 Zero-shot cloning: [BOS][SOH] ref_text [EOT][SOA][SOS] <ref audio tokens> [EOS_sp] then
 [SOH] text [EOT][SOA]. Generation stops at end-of-speech (128258).
 """
-import os, io, base64
+import os, io, re, json, base64
 import numpy as np, torch, soundfile as sf, requests
 from scipy.signal import resample_poly
 from fastapi import FastAPI, HTTPException
-from fastapi.responses import Response
+from fastapi.responses import Response, StreamingResponse
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from transformers import AutoTokenizer
@@ -128,3 +128,92 @@ def tts(req: TTSReq):
     sf.write(buf, audio, 24000, format="WAV", subtype="PCM_16")
     return Response(content=buf.getvalue(), media_type="audio/wav",
                     headers={"X-Audio-Seconds": f"{len(audio)/24000:.2f}"})
+
+
+# --- streaming decode: emit raw PCM16 (24kHz mono) as the engine generates -----------
+# vLLM emits audio tokens as <custom_token_N> strings (token_id = 128256 + N). We decode in
+# WINDOWED CHUNKS: every CHUNK new frames, decode a small window [CTX left | CHUNK | CTX
+# right] and emit only the middle CHUNK frames (context on both sides => seamless, no
+# clicks). O(window) per call and few calls => decode keeps up with gen on CPU (per-frame
+# decode did NOT — per-call overhead x ~60 frames serialized to seconds). TTFA ~0.7s.
+CUSTOM_RE = re.compile(r"<custom_token_(\d+)>")
+FRAME = 2048  # audio samples per SNAC frame
+
+
+def _decode_frames(ids: list[int]):
+    codes = [t - AUDIO_BASE for t in ids if AUDIO_BASE <= t <= AUDIO_MAX]
+    n = len(codes) // 7
+    if n == 0:
+        return None
+    l1, l2, l3 = [], [], []
+    for i in range(n):
+        f = codes[7 * i:7 * i + 7]
+        c = [f[0], f[1] - 4096, f[2] - 8192, f[3] - 12288, f[4] - 16384, f[5] - 20480, f[6] - 24576]
+        if any(x < 0 or x > 4095 for x in c):
+            return None
+        l1.append(c[0]); l2 += [c[1], c[4]]; l3 += [c[2], c[3], c[5], c[6]]
+    ct = [torch.tensor(l1).unsqueeze(0).to(SNAC_DEVICE),
+          torch.tensor(l2).unsqueeze(0).to(SNAC_DEVICE),
+          torch.tensor(l3).unsqueeze(0).to(SNAC_DEVICE)]
+    with torch.inference_mode():
+        return snac_model.decode(ct).squeeze().cpu().numpy()
+
+
+def _pcm16(arr) -> bytes:
+    return (np.clip(arr, -1, 1) * 32767).astype(np.int16).tobytes()
+
+
+@app.post("/tts/stream")
+def tts_stream(req: TTSReq):
+    prompt_ids = _build_prompt(req)
+    payload = {"model": VLLM_MODEL, "prompt": prompt_ids, "max_tokens": req.max_tokens,
+               "temperature": req.temperature, "top_p": req.top_p, "skip_special_tokens": False,
+               "stop_token_ids": [EOS_SP], "repetition_penalty": req.repetition_penalty, "stream": True}
+    CTX, CHUNK = 2, 6  # context frames each side, frames emitted per decode
+
+    def emit(emitted, total, final):
+        # decode [emitted-CTX : end] with left/right context; return (pcm_bytes, new_emitted)
+        end = total if final else emitted + CHUNK
+        ws = max(0, emitted - CTX)
+        audio = _decode_frames(buf[7 * ws:7 * (end + (0 if final else CTX))])
+        if audio is None:
+            return b"", end
+        off = (emitted - ws) * FRAME
+        chunk = audio[off:] if final else audio[off:off + CHUNK * FRAME]
+        return (_pcm16(chunk) if len(chunk) else b""), end
+
+    buf = []
+
+    def gen():
+        acc, cursor, emitted = "", 0, 0
+        with requests.post(VLLM_URL, json=payload, stream=True, timeout=180) as r:
+            for raw in r.iter_lines():
+                if not raw:
+                    continue
+                s = raw.decode("utf-8", "ignore")
+                if not s.startswith("data:"):
+                    continue
+                d = s[5:].strip()
+                if d == "[DONE]":
+                    break
+                try:
+                    j = json.loads(d)
+                except Exception:
+                    continue
+                acc += j["choices"][0].get("text", "")
+                for m in CUSTOM_RE.finditer(acc, cursor):
+                    cursor = m.end()
+                    tid = 128256 + int(m.group(1))
+                    if AUDIO_BASE <= tid <= AUDIO_MAX:
+                        buf.append(tid)
+                while (len(buf) // 7) - emitted >= CHUNK + CTX:   # enough new frames + right ctx
+                    b, emitted = emit(emitted, len(buf) // 7, final=False)
+                    if b:
+                        yield b
+        if (len(buf) // 7) > emitted:                             # flush the tail
+            b, emitted = emit(emitted, len(buf) // 7, final=True)
+            if b:
+                yield b
+
+    return StreamingResponse(gen(), media_type="application/octet-stream",
+                             headers={"Cache-Control": "no-cache", "X-Sample-Rate": "24000"})

tools/gateway-chat.html

@@ -144,33 +144,54 @@ function addMsg(who, cls){
 }
 function showErr(text){ const m = addMsg('error','err'); m.body.className = 'err'; m.body.textContent = text; }
 
-// --- mOrpheus TTS: auto-voice quoted text from a completed assistant reply ---
+// --- mOrpheus TTS: streaming-decode auto-voice of quoted text (Web Audio, low TTFA) ---
-let ttsAudio = null;   // current clip, so a new reply interrupts the previous one
+let audioCtx = null;   // shared AudioContext
+let ttsGen = 0;        // generation counter — a newer reply's stream supersedes older ones
 function extractQuotes(text){
   const re = /“([^”]+)”|"([^"]+)"/g;   // typographic "…" or straight "…"
   const out = []; let m;
   while ((m = re.exec(text)) !== null){ const q = (m[1] || m[2] || '').trim(); if (q) out.push(q); }
   return out;
 }
+async function streamSpeak(speech, tag){
+  const mine = ++ttsGen;                                // cancels any in-flight stream
+  if (!audioCtx) audioCtx = new (window.AudioContext || window.webkitAudioContext)();
+  if (audioCtx.state === 'suspended') await audioCtx.resume();
+  const url = $('ttsUrl').value.trim().replace(/\/+$/,'').replace(/\/tts$/,'') + '/tts/stream';
+  const res = await fetch(url, {
+    method:'POST', headers:{ 'Content-Type':'application/json' },
+    body: JSON.stringify({ text: speech, voice: ($('ttsVoice').value.trim() || 'baddy'), max_tokens: 1200 })
+  });
+  if (!res.ok || !res.body){ tag.textContent = ' 🔇' + res.status; return; }
+  const reader = res.body.getReader();
+  let playhead = audioCtx.currentTime + 0.12, leftover = new Uint8Array(0);
+  for(;;){
+    const { done, value } = await reader.read();
+    if (done || mine !== ttsGen) break;                 // stop if a newer reply took over
+    let bytes = new Uint8Array(leftover.length + value.length);
+    bytes.set(leftover); bytes.set(value, leftover.length);
+    const n = bytes.length - (bytes.length % 2);        // whole int16 samples only
+    leftover = bytes.slice(n);
+    if (!n) continue;
+    const dv = new DataView(bytes.buffer, 0, n), N = n / 2, f32 = new Float32Array(N);
+    for (let i = 0; i < N; i++) f32[i] = dv.getInt16(i * 2, true) / 32768;
+    const abuf = audioCtx.createBuffer(1, N, 24000); abuf.getChannelData(0).set(f32);
+    const src = audioCtx.createBufferSource(); src.buffer = abuf; src.connect(audioCtx.destination);
+    if (playhead < audioCtx.currentTime) playhead = audioCtx.currentTime + 0.02;  // underrun catch-up
+    src.start(playhead); playhead += abuf.duration;
+  }
+  if (mine === ttsGen) tag.textContent = ' 🔈';
+}
 async function speakQuotes(text, msg){
   if (!$('ttsOn').checked) return;
   const quotes = extractQuotes(text);
   if (!quotes.length) return;
+  const speech = quotes.join('  ');
   const tag = document.createElement('span'); tag.textContent = ' 🔊';
-  tag.style.cursor = 'pointer'; tag.title = 'mOrpheus'; msg.wrap.querySelector('.who').append(tag);
+  tag.style.cursor = 'pointer'; tag.title = 'mOrpheus — click to replay';
-  try {
+  tag.onclick = () => streamSpeak(speech, tag).catch(() => tag.textContent = ' 🔇');
-    const res = await fetch($('ttsUrl').value.trim(), {
+  msg.wrap.querySelector('.who').append(tag);
-      method:'POST', headers:{ 'Content-Type':'application/json' },
+  try { await streamSpeak(speech, tag); } catch (e){ tag.textContent = ' 🔇'; }
-      body: JSON.stringify({ text: quotes.join('  '), voice: ($('ttsVoice').value.trim() || 'baddy'), max_tokens: 1200 })
-    });
-    if (!res.ok){ tag.textContent = ' 🔇' + res.status; return; }
-    const url = URL.createObjectURL(await res.blob());
-    if (ttsAudio) ttsAudio.pause();
-    ttsAudio = new Audio(url);
-    tag.onclick = () => ttsAudio.play();          // click 🔊 to replay
-    ttsAudio.onended = () => tag.textContent = ' 🔈';
-    ttsAudio.play();
-  } catch (e){ tag.textContent = ' 🔇'; }
 }
 
 async function send(){