feat: add voice prompt extraction for cached voice cloning

willgriffin · willgriffin · commit 9cd19f1bb7d1 · 2026-01-26T07:45:26.000-07:00
- Add POST /v1/voice/extract endpoint to extract reusable voice prompts
- Update POST /v1/audio/speech to accept voice_prompt parameter
- Voice prompts are serialized as base64-encoded numpy arrays
- Allows caching voice embeddings to avoid re-processing ref_audio
- Add tests for voice prompt extraction and synthesis

Workflow:
1. Extract: POST /v1/voice/extract with ref_audio → get voice_prompt
2. Reuse: POST /v1/audio/speech with voice_prompt → faster synthesis
diff --git a/README.md b/README.md
@@ -6,14 +6,15 @@ Multi-model text-to-speech API with voice cloning support.
 
 - **Multi-model architecture** - Pluggable backend system for different TTS models
 - **Voice cloning** - Clone voices with reference audio + transcript
+- **Voice prompt caching** - Extract and reuse voice embeddings for faster synthesis
 - **ref_text support** - Provide transcript for better voice cloning quality
 - **GPU accelerated** - CUDA support for fast inference
 
 ## Supported Backends
 
-| Backend | Voice Cloning | ref_text Support |
-|---------|--------------|------------------|
-| `qwen3-tts` | ✅ | ✅ |
+| Backend | Voice Cloning | ref_text | Voice Prompt |
+|---------|--------------|----------|--------------|
+| `qwen3-tts` | ✅ | ✅ | ✅ |
 
 ## API Endpoints
 
@@ -25,8 +26,9 @@ Synthesize speech from text.
 - `text` (required): Text to synthesize
 - `language`: Target language (default: "English")
 - `speaker`: Preset speaker for basic TTS (e.g., "Vivian", "Ryan")
-- `ref_audio`: Reference audio file for voice cloning
+- `ref_audio`: Reference audio file for voice cloning (on-the-fly)
 - `ref_text`: Transcript of reference audio (improves cloning quality)
+- `voice_prompt`: Pre-extracted voice prompt from `/v1/voice/extract` (cached)
 - `speed`: Speech speed multiplier (default: 1.0)
 
 **Example:**
@@ -42,12 +44,51 @@ curl -X POST http://localhost:8000/v1/audio/speech \
   -F "speaker=Ryan" \
   -o output.wav
 
-# Voice cloning with ref_text
+# Voice cloning (on-the-fly) - processes ref_audio each time
 curl -X POST http://localhost:8000/v1/audio/speech \
   -F "text=Hello, this is my cloned voice." \
   -F "ref_audio=@reference.wav" \
   -F "ref_text=This is the transcript of my reference audio." \
   -o cloned.wav
+
+# Voice cloning (cached) - faster, uses pre-extracted prompt
+curl -X POST http://localhost:8000/v1/audio/speech \
+  -F "text=Hello, this is my cloned voice." \
+  -F "voice_prompt=$VOICE_PROMPT" \
+  -o cloned.wav
+```
+
+### `POST /v1/voice/extract`
+
+Extract a reusable voice prompt from reference audio. The returned prompt can be cached and reused with `/v1/audio/speech` to avoid re-processing the reference audio on every request.
+
+**Parameters:**
+- `ref_audio` (required): Reference audio file
+- `ref_text`: Transcript of reference audio (improves quality)
+- `language`: Language of the reference audio (default: "English")
+
+**Returns:**
+- `voice_prompt`: Base64-encoded voice embedding (store this)
+- `format`: Encoding format (e.g., "base64-numpy")
+
+**Example:**
+```bash
+# Extract voice prompt
+VOICE_PROMPT=$(curl -X POST http://localhost:8000/v1/voice/extract \
+  -F "ref_audio=@reference.wav" \
+  -F "ref_text=This is the transcript of my reference audio." \
+  | jq -r '.voice_prompt')
+
+# Use the cached prompt for multiple synthesis requests
+curl -X POST http://localhost:8000/v1/audio/speech \
+  -F "text=First sentence with cloned voice." \
+  -F "voice_prompt=$VOICE_PROMPT" \
+  -o output1.wav
+
+curl -X POST http://localhost:8000/v1/audio/speech \
+  -F "text=Second sentence with same voice." \
+  -F "voice_prompt=$VOICE_PROMPT" \
+  -o output2.wav
 ```
 
 ### `GET /health`
diff --git a/requirements.txt b/requirements.txt
@@ -5,6 +5,7 @@ python-multipart>=0.0.12
 
 # Audio processing
 soundfile>=0.12.1
+numpy>=1.24.0
 
 # Qwen3-TTS backend
 qwen-tts>=0.0.5
diff --git a/server.py b/server.py
@@ -3,15 +3,21 @@
 
 Supported models:
 - qwen3-tts: Qwen3-TTS with voice cloning via ref_audio + ref_text
+
+Voice cloning can be done two ways:
+1. On-the-fly: Pass ref_audio + ref_text with each synthesis request
+2. Cached: Extract a voice_prompt once, then reuse it for multiple requests
 """
 
 import io
 import os
+import base64
 from abc import ABC, abstractmethod
-from typing import Optional, Tuple, List
+from typing import Optional, Tuple, List, Any
 from contextlib import asynccontextmanager
 
 import torch
+import numpy as np
 import soundfile as sf
 from fastapi import FastAPI, UploadFile, File, Form, HTTPException
 from fastapi.responses import StreamingResponse, JSONResponse
@@ -65,6 +71,46 @@ def get_speakers(self) -> List[str]:
         """Return available preset speakers."""
         pass
 
+    def extract_voice_prompt(
+        self,
+        ref_audio: Tuple[Any, int],
+        ref_text: Optional[str] = None,
+        language: str = "English",
+    ) -> str:
+        """
+        Extract a reusable voice prompt from reference audio.
+
+        Args:
+            ref_audio: Tuple of (audio_data, sample_rate)
+            ref_text: Optional transcript of reference audio
+            language: Language of the reference audio
+
+        Returns:
+            Base64-encoded voice prompt that can be reused
+        """
+        raise NotImplementedError("This backend does not support voice prompt extraction")
+
+    def synthesize_with_prompt(
+        self,
+        text: str,
+        voice_prompt: str,
+        language: str = "English",
+        speed: float = 1.0,
+    ) -> Tuple[bytes, int]:
+        """
+        Synthesize speech using a pre-extracted voice prompt.
+
+        Args:
+            text: Text to synthesize
+            voice_prompt: Base64-encoded voice prompt from extract_voice_prompt
+            language: Target language
+            speed: Speech speed multiplier
+
+        Returns:
+            Tuple of (wav_bytes, sample_rate)
+        """
+        raise NotImplementedError("This backend does not support voice prompt synthesis")
+
 
 class Qwen3TTSBackend(TTSBackend):
     """
@@ -149,6 +195,7 @@ def get_info(self) -> dict:
             "base_model": self.base_model_name,
             "supports_voice_cloning": True,
             "supports_ref_text": True,
+            "supports_voice_prompt": True,
             "device": "cuda" if torch.cuda.is_available() else "cpu",
         }
 
@@ -162,6 +209,65 @@ def get_speakers(self) -> List[str]:
             # Fallback to known speakers
             return ["Vivian", "Ryan", "Sophia", "Isabella", "Evan", "Lily"]
 
+    def extract_voice_prompt(
+        self,
+        ref_audio: Tuple[Any, int],
+        ref_text: Optional[str] = None,
+        language: str = "English",
+    ) -> str:
+        """Extract a reusable voice prompt from reference audio."""
+        if self.base_model is None:
+            raise RuntimeError("Base model not loaded")
+
+        # Use the Base model's create_voice_clone_prompt method
+        voice_prompt = self.base_model.create_voice_clone_prompt(
+            ref_audio=ref_audio,
+            ref_text=ref_text,
+            language=language,
+        )
+
+        # Serialize to base64 - voice_prompt is typically tensor data
+        # Convert to numpy, then to bytes, then base64
+        if hasattr(voice_prompt, 'cpu'):
+            # It's a torch tensor
+            prompt_np = voice_prompt.cpu().numpy()
+        else:
+            prompt_np = np.array(voice_prompt)
+
+        buffer = io.BytesIO()
+        np.save(buffer, prompt_np, allow_pickle=False)
+        return base64.b64encode(buffer.getvalue()).decode('utf-8')
+
+    def synthesize_with_prompt(
+        self,
+        text: str,
+        voice_prompt: str,
+        language: str = "English",
+        speed: float = 1.0,
+    ) -> Tuple[bytes, int]:
+        """Synthesize speech using a pre-extracted voice prompt."""
+        if self.base_model is None:
+            raise RuntimeError("Base model not loaded")
+
+        # Decode the voice prompt
+        buffer = io.BytesIO(base64.b64decode(voice_prompt))
+        prompt_np = np.load(buffer, allow_pickle=False)
+        prompt_tensor = torch.from_numpy(prompt_np).to(self.device)
+
+        # Generate using the cached voice prompt
+        wavs, sr = self.base_model.generate_voice_clone(
+            text=text,
+            language=language,
+            voice_clone_prompt=prompt_tensor,
+        )
+
+        # Convert to WAV bytes
+        wav_buffer = io.BytesIO()
+        sf.write(wav_buffer, wavs[0], sr, format='WAV')
+        wav_buffer.seek(0)
+
+        return wav_buffer.read(), sr
+
 
 # =============================================================================
 # Backend Registry
@@ -203,7 +309,7 @@ async def lifespan(app: FastAPI):
 app = FastAPI(
     title="TTS Server",
     description="Multi-model text-to-speech API with voice cloning support",
-    version="0.2.0",
+    version="0.3.0",
     lifespan=lifespan,
 )
 
@@ -233,45 +339,111 @@ async def list_speakers():
     return {"speakers": backend.get_speakers()}
 
 
+@app.post("/v1/voice/extract")
+async def extract_voice_prompt(
+    ref_audio: UploadFile = File(..., description="Reference audio for voice extraction"),
+    ref_text: Optional[str] = Form(None, description="Transcript of reference audio"),
+    language: str = Form("English", description="Language of the reference audio"),
+):
+    """
+    Extract a reusable voice prompt from reference audio.
+
+    The returned `voice_prompt` can be cached and reused with `/v1/audio/speech`
+    to avoid re-processing the reference audio on every request.
+
+    **Returns:**
+    - `voice_prompt`: Base64-encoded voice embedding (store this)
+    - `format`: Always "base64-numpy" for this backend
+    """
+    if backend is None:
+        raise HTTPException(status_code=503, detail="Model not loaded")
+
+    try:
+        # Read reference audio
+        audio_bytes = await ref_audio.read()
+        audio_buffer = io.BytesIO(audio_bytes)
+        audio_data, sample_rate = sf.read(audio_buffer)
+        ref_audio_data = (audio_data, sample_rate)
+
+        # Extract voice prompt
+        voice_prompt = backend.extract_voice_prompt(
+            ref_audio=ref_audio_data,
+            ref_text=ref_text,
+            language=language,
+        )
+
+        return JSONResponse({
+            "voice_prompt": voice_prompt,
+            "format": "base64-numpy",
+            "ref_text": ref_text,
+            "language": language,
+        })
+
+    except NotImplementedError as e:
+        raise HTTPException(status_code=501, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 @app.post("/v1/audio/speech")
 async def synthesize_speech(
     text: str = Form(..., description="Text to synthesize"),
     language: str = Form("English", description="Target language"),
     speaker: Optional[str] = Form(None, description="Preset speaker for basic TTS (e.g., Vivian, Ryan)"),
     ref_audio: Optional[UploadFile] = File(None, description="Reference audio for voice cloning"),
     ref_text: Optional[str] = Form(None, description="Transcript of reference audio"),
+    voice_prompt: Optional[str] = Form(None, description="Pre-extracted voice prompt from /v1/voice/extract"),
     speed: float = Form(1.0, description="Speech speed multiplier"),
 ):
     """
     Synthesize speech from text.
 
     **Basic TTS:** Just provide `text` and optionally `speaker`.
 
-    **Voice Cloning:** Provide `ref_audio` and `ref_text` for best quality.
+    **Voice Cloning (on-the-fly):** Provide `ref_audio` and `ref_text`.
     The `ref_text` should be the exact transcript of the reference audio.
+
+    **Voice Cloning (cached):** Provide `voice_prompt` from `/v1/voice/extract`.
+    This is faster as it skips re-processing the reference audio.
     """
     if backend is None:
         raise HTTPException(status_code=503, detail="Model not loaded")
 
     try:
-        # Read reference audio if provided
-        ref_audio_data = None
-        if ref_audio:
+        # Priority: voice_prompt > ref_audio > speaker
+        if voice_prompt:
+            # Use pre-extracted voice prompt
+            wav_bytes, sample_rate = backend.synthesize_with_prompt(
+                text=text,
+                voice_prompt=voice_prompt,
+                language=language,
+                speed=speed,
+            )
+        elif ref_audio:
+            # On-the-fly voice cloning
             audio_bytes = await ref_audio.read()
-            # Load audio to get actual sample rate
             audio_buffer = io.BytesIO(audio_bytes)
             audio_data, sample_rate = sf.read(audio_buffer)
             ref_audio_data = (audio_data, sample_rate)
 
-        # Synthesize
-        wav_bytes, sample_rate = backend.synthesize(
-            text=text,
-            language=language,
-            speaker=speaker,
-            ref_audio=ref_audio_data,
-            ref_text=ref_text,
-            speed=speed,
-        )
+            wav_bytes, sample_rate = backend.synthesize(
+                text=text,
+                language=language,
+                speaker=speaker,
+                ref_audio=ref_audio_data,
+                ref_text=ref_text,
+                speed=speed,
+            )
+        else:
+            # Basic TTS with preset speaker
+            wav_bytes, sample_rate = backend.synthesize(
+                text=text,
+                language=language,
+                speaker=speaker,
+                ref_audio=None,
+                ref_text=None,
+                speed=speed,
+            )
 
         return StreamingResponse(
             io.BytesIO(wav_bytes),
@@ -282,6 +454,8 @@ async def synthesize_speech(
             }
         )
 
+    except NotImplementedError as e:
+        raise HTTPException(status_code=501, detail=str(e))
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
diff --git a/tests/test_server.py b/tests/test_server.py
