From eabd1304cce0e053cd32ec910d2f0ea429e8af14 Mon Sep 17 00:00:00 2001 From: soryu Date: Wed, 28 Jan 2026 02:54:17 +0000 Subject: Add Qwen3-TTS streaming endpoint for voice synthesis (#40) * Task completion checkpoint * Task completion checkpoint * Task completion checkpoint * Add Qwen3-TTS research document for live TTS replacement Research findings for replacing Chatterbox TTS with Qwen3-TTS-12Hz-0.6B-Base: - Current TTS: Chatterbox-Turbo-ONNX with batch-only generation, no streaming - Qwen3-TTS: 97ms end-to-end latency, streaming support, 3-second voice cloning - Voice cloning: Requires 3s reference audio + transcript (Makima voice planned) - Integration: Python service with WebSocket bridge (no ONNX export available) - Languages: 10 supported including English and Japanese Document includes: - Current architecture analysis (makima/src/tts.rs) - Qwen3-TTS capabilities and requirements - Feasibility assessment for live/streaming TTS - Audio clip requirements for voice cloning - Preliminary technical approach with architecture diagrams Co-Authored-By: Claude Opus 4.5 * [WIP] Heartbeat checkpoint - 2026-01-27 03:11:15 UTC * Add Qwen3-TTS research documentation Comprehensive research on replacing Chatterbox TTS with Qwen3-TTS-12Hz-0.6B-Base: - Current TTS implementation analysis (Chatterbox-Turbo-ONNX in makima/src/tts.rs) - Qwen3-TTS capabilities: 97ms streaming latency, voice cloning with 3s reference - Cross-lingual support: Japanese voice (Makima/Tomori Kusunoki) speaking English - Python microservice architecture recommendation (FastAPI + WebSocket) - Implementation phases and technical approach - Hardware requirements and dependencies Key findings: - Live/streaming TTS is highly feasible with 97ms latency - Voice cloning fully supported with 0.95 speaker similarity - Recommended: Python microservice with WebSocket streaming Co-Authored-By: Claude Opus 4.5 * Add comprehensive Qwen3-TTS integration specification This specification document defines the complete integration of Qwen3-TTS-12Hz-0.6B-Base as a replacement for the existing Chatterbox-Turbo TTS implementation. The document covers: ## Functional Requirements - WebSocket endpoint /api/v1/speak for streaming TTS - Voice cloning with default Makima voice (Japanese VA speaking English) - Support for custom voice references - Detailed client-to-server and server-to-client message protocols - Integration with Listen page for bidirectional speech ## Non-Functional Requirements - Latency targets: < 200ms first audio byte - Audio quality: 24kHz, mono, PCM16/PCM32f - Hardware requirements: CUDA GPU with 4-8GB VRAM - Scalability: 10 concurrent sessions per GPU ## Architecture Specification - Python TTS microservice with FastAPI/WebSocket - Rust proxy endpoint in makima server - Voice prompt caching mechanism (LRU cache) - Error handling and recovery strategies ## API Contract - Complete WebSocket message format definitions (TypeScript) - Error codes and responses (TTS_UNAVAILABLE, SYNTHESIS_ERROR, etc.) - Session state machine and lifecycle management ## Voice Asset Requirements - Makima voice clip specifications (5-10s WAV, transcript required) - Storage location: models/voices/makima/ - Metadata format for voice management ## Testing Strategy - Unit tests for Python TTS service and Rust proxy - Integration tests for WebSocket flow - Latency benchmarks with performance targets - Test data fixtures for various text lengths Co-Authored-By: Claude Opus 4.5 * Add Qwen3-TTS implementation plan Comprehensive implementation plan for replacing Chatterbox-TTS with Qwen3-TTS streaming TTS service, including: - Task breakdown with estimated hours for each phase - Phase 1: Python TTS microservice (FastAPI, WebSocket) - Phase 2: Rust proxy integration (speak.rs, tts_client.rs) - Detailed file changes and new module structure - Testing plan with unit, integration, and latency benchmarks - Risk assessment with mitigation strategies - Success criteria for each phase Based on specification in docs/specs/qwen3-tts-spec.md Co-Authored-By: Claude Opus 4.5 * Add author and research references to TTS implementation plan Add links to research documentation and author attribution. Co-Authored-By: Claude Opus 4.5 * [WIP] Heartbeat checkpoint - 2026-01-27 03:25:06 UTC * Add Python TTS service project structure (Phase 1.1-1.3) Create the initial makima-tts Python service directory structure with: - pyproject.toml with FastAPI, Qwen-TTS, and torch dependencies - config.py with pydantic-settings TTSConfig class - models.py with Pydantic message models (Start, Speak, Stop, Ready, etc.) This implements tasks P1.1, P1.2, and P1.3 from the Qwen3-TTS implementation plan. Co-Authored-By: Claude Opus 4.5 * Add TTS engine and voice manager for Qwen3-TTS (Phase 1.4-1.5) Implement core TTS functionality: - tts_engine.py: Qwen3-TTS wrapper with streaming audio chunk generation - voice_manager.py: Voice prompt caching with LRU eviction and TTL support Co-Authored-By: Claude Opus 4.5 * [WIP] Heartbeat checkpoint - 2026-01-27 03:30:06 UTC * Add TTS proxy client and message types (Phase 2.1, 2.2, 2.4) - Add tts_client.rs with TtsConfig, TtsCircuitBreaker, TtsError, TtsProxyClient, and TtsConnection structs for WebSocket proxying - Add TTS message types to messages.rs (TtsAudioEncoding, TtsPriority, TtsStartMessage, TtsSpeakMessage, TtsStopMessage, TtsClientMessage, TtsReadyMessage, TtsAudioChunkMessage, TtsCompleteMessage, TtsErrorMessage, TtsStoppedMessage, TtsServerMessage) - Export tts_client module from server mod.rs - tokio-tungstenite already present in Cargo.toml Co-Authored-By: Claude Opus 4.5 * Add TTS WebSocket handler and route (Phase 2.3, 2.5, 2.6) - Create speak.rs WebSocket handler that proxies to Python TTS service - Add TtsState fields (tts_client, tts_config) to AppState - Add with_tts() builder and is_tts_healthy() methods to AppState - Register /api/v1/speak route in the router - Add speak module export in handlers/mod.rs The handler forwards WebSocket messages bidirectionally between the client and the Python TTS microservice with proper error handling. Co-Authored-By: Claude Opus 4.5 * Add Makima voice profile assets for TTS voice cloning Creates the voice assets directory structure with: - manifest.json containing voice configuration (voice_id, speaker, language, reference audio path, and Japanese transcript placeholder) - README.md with instructions for obtaining voice reference audio Co-Authored-By: Claude Opus 4.5 * Add Rust-native Qwen3-TTS integration research document Research findings for integrating Qwen3-TTS-12Hz-0.6B-Base directly into the makima Rust codebase without Python. Key conclusions: - ONNX export is not viable (unsupported architecture) - Candle (HF Rust ML framework) is the recommended approach - Model weights available in safetensors format (2.52GB total) - Three components needed: LM backbone, code predictor, speech tokenizer - Crane project has Qwen3-TTS as highest priority (potential upstream) Co-Authored-By: Claude Opus 4.5 * [WIP] Heartbeat checkpoint - 2026-01-27 11:21:43 UTC * [WIP] Heartbeat checkpoint - 2026-01-27 11:24:19 UTC * [WIP] Heartbeat checkpoint - 2026-01-27 11:26:43 UTC * feat: implement Rust-native Qwen3-TTS using candle framework Replace monolithic tts.rs with modular tts/ directory structure: - tts/mod.rs: TtsEngine trait, TtsEngineFactory, shared types (AudioChunk, TtsError), and utility functions (save_wav, resample, argmax) - tts/chatterbox.rs: existing ONNX-based ChatterboxTTS adapted to implement TtsEngine trait with Mutex-wrapped sessions for Send+Sync - tts/qwen3/mod.rs: Qwen3Tts entry point with HuggingFace model loading - tts/qwen3/config.rs: Qwen3TtsConfig parsing from HF config.json - tts/qwen3/model.rs: 28-layer Qwen3 transformer with RoPE, GQA (16 heads, 8 KV heads), SiLU MLP, RMS norm, and KV cache - tts/qwen3/code_predictor.rs: 5-layer MTP module predicting 16 codebooks - tts/qwen3/speech_tokenizer.rs: ConvNet encoder/decoder with 16-layer RVQ - tts/qwen3/generate.rs: autoregressive generation loop with streaming support Add candle-core, candle-nn, candle-transformers, safetensors to Cargo.toml. Co-Authored-By: Claude Opus 4.5 * feat: integrate TTS engine into speak WebSocket handler - Update speak.rs handler to use TTS engine directly from SharedState instead of returning a stub "not implemented" error - Add TtsEngine (OnceCell lazy-loaded) to AppState in state.rs with get_tts_engine() method for lazy initialization on first connection - Implement full WebSocket protocol: client sends JSON speak/cancel/stop messages, server streams binary PCM audio chunks and audio_end signals - Create voices/makima/manifest.json for Makima voice profile configuration - All files compile successfully with zero errors Co-Authored-By: Claude Opus 4.5 * feat: add /speak TTS page with WebSocket audio playback Add a new /speak frontend page for text-to-speech via WebSocket. The page accepts text input and streams synthesized PCM audio through the Web Audio API. Includes model loading indicator, cancel support, and connection status. Also adds a loading bar to the listen page ControlPanel during WebSocket connection. Co-Authored-By: Claude Opus 4.5 --------- Co-authored-by: Claude Opus 4.5 --- makima/src/server/handlers/speak.rs | 274 ++++++++++++++++++++++++++++++++++++ 1 file changed, 274 insertions(+) create mode 100644 makima/src/server/handlers/speak.rs (limited to 'makima/src/server/handlers/speak.rs') diff --git a/makima/src/server/handlers/speak.rs b/makima/src/server/handlers/speak.rs new file mode 100644 index 0000000..75e7780 --- /dev/null +++ b/makima/src/server/handlers/speak.rs @@ -0,0 +1,274 @@ +//! WebSocket handler for TTS streaming (direct in-process inference). +//! +//! This module implements the `/api/v1/speak` endpoint which performs +//! text-to-speech synthesis directly using the candle-based TTS engine. +//! No external Python service or proxy — the model runs in-process. +//! +//! ## Architecture +//! +//! The speak handler will: +//! 1. Accept a WebSocket connection from the client +//! 2. Lazily load the TTS model (candle) on first request +//! 3. Parse JSON control messages (start, speak, stop, cancel) +//! 4. Run inference directly and stream audio chunks back +//! +//! See `makima/src/tts/` for the TTS engine implementation. +//! See `docs/specs/qwen3-tts-spec.md` for the full protocol specification. + +use axum::{ + extract::{ws::Message, ws::WebSocket, State, WebSocketUpgrade}, + response::Response, +}; +use futures::{SinkExt, StreamExt}; +use serde::Deserialize; +use uuid::Uuid; + +use crate::server::state::SharedState; + +/// Client-to-server control messages. +#[derive(Debug, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum ClientMessage { + /// Request speech synthesis for the given text. + Speak { + text: String, + /// Optional voice ID (e.g., "makima"). Not yet used — reserved for future voice selection. + #[serde(default)] + #[allow(dead_code)] + voice: Option, + }, + /// Cancel any in-progress synthesis. + Cancel, + /// Graceful close. + Stop, +} + +/// WebSocket upgrade handler for TTS streaming. +/// +/// This endpoint accepts WebSocket connections for text-to-speech synthesis. +/// The TTS model runs directly in-process using candle — no external service. +#[utoipa::path( + get, + path = "/api/v1/speak", + responses( + (status = 101, description = "WebSocket connection established"), + (status = 503, description = "TTS engine not available"), + ), + tag = "Speak" +)] +pub async fn websocket_handler( + ws: WebSocketUpgrade, + State(state): State, +) -> Response { + ws.on_upgrade(|socket| handle_speak_socket(socket, state)) +} + +/// Handle TTS WebSocket session with direct in-process inference. +/// +/// Protocol: +/// - Client sends JSON `{ "type": "speak", "text": "..." }` messages +/// - Server responds with binary audio chunks (16-bit PCM @ 24kHz) +/// - Server sends JSON `{ "type": "audio_end" }` when synthesis is complete +/// - Server sends JSON `{ "type": "error", ... }` on failures +async fn handle_speak_socket(socket: WebSocket, state: SharedState) { + let session_id = Uuid::new_v4().to_string(); + tracing::info!(session_id = %session_id, "New TTS WebSocket connection"); + + let (mut sender, mut receiver) = socket.split(); + + // Process incoming messages + while let Some(msg) = receiver.next().await { + let msg = match msg { + Ok(m) => m, + Err(e) => { + tracing::warn!(session_id = %session_id, error = %e, "WebSocket receive error"); + break; + } + }; + + match msg { + Message::Text(text) => { + let client_msg: ClientMessage = match serde_json::from_str(&text) { + Ok(m) => m, + Err(e) => { + let _ = send_error( + &mut sender, + "INVALID_MESSAGE", + &format!("Failed to parse message: {e}"), + ) + .await; + continue; + } + }; + + match client_msg { + ClientMessage::Speak { text, .. } => { + tracing::info!( + session_id = %session_id, + text_len = text.len(), + "TTS speak request" + ); + + // Get or lazily load the TTS engine + let engine = match state.get_tts_engine().await { + Ok(e) => e, + Err(e) => { + tracing::error!( + session_id = %session_id, + error = %e, + "Failed to load TTS engine" + ); + let _ = send_error( + &mut sender, + "TTS_LOAD_FAILED", + &format!("Failed to load TTS engine: {e}"), + ) + .await; + continue; + } + }; + + if !engine.is_ready() { + let _ = send_error( + &mut sender, + "TTS_NOT_READY", + "TTS engine is not ready yet", + ) + .await; + continue; + } + + // Run TTS inference (no voice reference for now — uses default) + match engine.generate(&text, None, None).await { + Ok(chunks) => { + for chunk in &chunks { + // Send binary PCM audio data + let pcm_bytes = chunk.to_pcm16_bytes(); + if sender + .send(Message::Binary(pcm_bytes.into())) + .await + .is_err() + { + tracing::warn!( + session_id = %session_id, + "Failed to send audio chunk — client disconnected" + ); + return; + } + } + + // Signal end of audio + let end_msg = serde_json::json!({ + "type": "audio_end", + "sample_rate": engine.sample_rate(), + "format": "pcm_s16le", + "channels": 1, + }); + let _ = sender + .send(Message::Text(end_msg.to_string().into())) + .await; + } + Err(e) => { + tracing::error!( + session_id = %session_id, + error = %e, + "TTS inference failed" + ); + let _ = send_error( + &mut sender, + "TTS_INFERENCE_FAILED", + &format!("TTS inference failed: {e}"), + ) + .await; + } + } + } + ClientMessage::Cancel => { + tracing::info!(session_id = %session_id, "TTS cancel requested"); + // TODO: support cancellation of in-progress inference + } + ClientMessage::Stop => { + tracing::info!(session_id = %session_id, "TTS stop requested, closing"); + break; + } + } + } + Message::Close(_) => { + tracing::info!(session_id = %session_id, "TTS WebSocket closed by client"); + break; + } + _ => { + // Ignore ping/pong/binary from client + } + } + } + + tracing::info!(session_id = %session_id, "TTS WebSocket connection closed"); +} + +/// Send an error message to the client. +async fn send_error(sender: &mut S, code: &str, message: &str) -> Result<(), axum::Error> +where + S: SinkExt + Unpin, + >::Error: std::error::Error, +{ + let error_msg = serde_json::json!({ + "type": "error", + "code": code, + "message": message, + "recoverable": false + }); + + sender + .send(Message::Text(error_msg.to_string().into())) + .await + .ok(); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_error_message_format() { + let error = serde_json::json!({ + "type": "error", + "code": "TEST_ERROR", + "message": "Test message", + "recoverable": false + }); + + assert_eq!(error["type"], "error"); + assert_eq!(error["code"], "TEST_ERROR"); + assert_eq!(error["message"], "Test message"); + assert_eq!(error["recoverable"], false); + } + + #[test] + fn test_client_message_parse_speak() { + let json = r#"{"type": "speak", "text": "Hello world"}"#; + let msg: ClientMessage = serde_json::from_str(json).unwrap(); + match msg { + ClientMessage::Speak { text, voice } => { + assert_eq!(text, "Hello world"); + assert!(voice.is_none()); + } + _ => panic!("Expected Speak message"), + } + } + + #[test] + fn test_client_message_parse_cancel() { + let json = r#"{"type": "cancel"}"#; + let msg: ClientMessage = serde_json::from_str(json).unwrap(); + assert!(matches!(msg, ClientMessage::Cancel)); + } + + #[test] + fn test_client_message_parse_stop() { + let json = r#"{"type": "stop"}"#; + let msg: ClientMessage = serde_json::from_str(json).unwrap(); + assert!(matches!(msg, ClientMessage::Stop)); + } +} -- cgit v1.2.3