Cross-lingual voice cloning with XTTS-v2: implementation notes, quality observations, and a memory leak you should know about

python
# core/voice_cloner.py

import os
import io
import base64
import hashlib
import tempfile
from pathlib import Path
from pydub import AudioSegment
import soundfile as sf
import numpy as np

class VoiceCloner:
    def __init__(self, model_path: str = None):
        self.model = None
        self.model_path = model_path
        self.generation_count = 0
        self.MAX_GENERATIONS_BEFORE_REINIT = 50
        self._supported_languages = [
            'en', 'es', 'fr', 'de', 'it', 'pt',
            'pl', 'tr', 'ru', 'nl', 'cs', 'ar',
            'zh-cn', 'ja', 'ko', 'hu', 'hi'
        ]

    def _load_model(self):
        """Load XTTS-v2. Deferred until first use."""
        from TTS.api import TTS
        import torch

        device = "cuda" if torch.cuda.is_available() else \
                 "mps" if torch.backends.mps.is_available() else \
                 "cpu"

        if self.model_path:
            self.model = TTS(
                model_path=self.model_path,
                config_path=os.path.join(self.model_path, "config.json")
            ).to(device)
        else:
            self.model = TTS(
                "tts_models/multilingual/multi-dataset/xtts_v2"
            ).to(device)

        self.generation_count = 0

    def _ensure_model_ready(self):
        """
        Load or reinitialise the model.
        Reinitialisation every 50 generations 
        works around the memory leak.
        """
        if self.model is None:
            self._load_model()
            return

        if self.generation_count >= self.MAX_GENERATIONS_BEFORE_REINIT:
            self._reinitialise()

    def _reinitialise(self):
        """
        Reinitialise model to clear leaked memory.
        Do NOT skip this in production.
        """
        import gc
        import torch

        del self.model
        self.model = None
        gc.collect()

        if torch.cuda.is_available():
            torch.cuda.empty_cache()
        elif hasattr(torch.backends, 'mps') and \
             torch.backends.mps.is_available():
            # MPS doesn't have explicit cache clearing
            # gc.collect() handles most of it
            pass

        self._load_model()

    def preprocess_audio(self, audio_path: str) -> str:
        """
        Convert any audio format to 22050Hz mono WAV.
        XTTS-v2 requires this exact format.
        Returns path to processed WAV file.
        """
        audio = AudioSegment.from_file(audio_path)

        # Convert to mono
        if audio.channels > 1:
            audio = audio.set_channels(1)

        # Resample to 22050Hz
        audio = audio.set_frame_rate(22050)

        # Validate minimum duration
        duration_seconds = len(audio) / 1000.0
        if duration_seconds < 3.0:
            raise ValueError(
                f"Reference audio too short: {duration_seconds:.1f}s. "
                f"Minimum 3 seconds required. "
                f"15-20 seconds recommended for best quality."
            )

        if duration_seconds < 8.0:
            # Warn but don't block — short audio works, just worse
            print(f"Warning: {duration_seconds:.1f}s reference audio. "
                  f"Quality improves significantly above 15 seconds.")

        # Export to temp WAV
        temp_wav = tempfile.mktemp(suffix='.wav')
        audio.export(temp_wav, format='wav')
        return temp_wav

    def chunk_text(self, text: str, 
                   max_chars: int = 250) -> list[str]:
        """
        Split text at sentence boundaries.
        XTTS-v2 has a ~300 character practical limit.
        We use 250 to leave headroom.
        """
        import spacy

        # Load spacy model — en_core_web_sm is sufficient
        try:
            nlp = spacy.load("en_core_web_sm")
        except OSError:
            # Fallback: split on punctuation
            return self._simple_chunk(text, max_chars)

        doc = nlp(text)
        chunks = []
        current_chunk = ""

        for sent in doc.sents:
            sent_text = sent.text.strip()

            if len(current_chunk) + len(sent_text) <= max_chars:
                current_chunk += (" " if current_chunk else "") + sent_text
            else:
                if current_chunk:
                    chunks.append(current_chunk)
                # If single sentence exceeds max_chars,
                # split it at word boundaries
                if len(sent_text) > max_chars:
                    chunks.extend(
                        self._simple_chunk(sent_text, max_chars)
                    )
                else:
                    current_chunk = sent_text

        if current_chunk:
            chunks.append(current_chunk)

        return [c for c in chunks if c.strip()]

    def _simple_chunk(self, text: str, 
                      max_chars: int) -> list[str]:
        """Fallback chunker — splits at word boundaries."""
        words = text.split()
        chunks = []
        current = ""

        for word in words:
            if len(current) + len(word) + 1 <= max_chars:
                current += (" " if current else "") + word
            else:
                if current:
                    chunks.append(current)
                current = word

        if current:
            chunks.append(current)

        return chunks

    def join_audio_chunks(self, 
                          chunks: list[np.ndarray],
                          sample_rate: int = 24000,
                          silence_ms: int = 150) -> np.ndarray:
        """
        Join audio chunks with natural silence between them.
        XTTS-v2 output is 24000Hz.
        """
        silence_samples = int(sample_rate * silence_ms / 1000)
        silence = np.zeros(silence_samples)

        parts = []
        for i, chunk in enumerate(chunks):
            parts.append(chunk)
            if i < len(chunks) - 1:
                parts.append(silence)

        return np.concatenate(parts)

    def clone(
        self,
        text: str,
        reference_audio_path: str,
        language: str = 'en',
        speed: float = 1.0
    ) -> dict:
        """
        Main voice cloning method.
        Returns dict with audio_base64 and duration_seconds.
        """
        if language not in self._supported_languages:
            raise ValueError(
                f"Language '{language}' not supported. "
                f"Supported: {self._supported_languages}"
            )

        self._ensure_model_ready()

        # Preprocess reference audio
        processed_ref = self.preprocess_audio(reference_audio_path)

        try:
            # Split text into chunks
            text_chunks = self.chunk_text(text)

            if not text_chunks:
                raise ValueError("Text is empty after processing")

            # Generate audio for each chunk
            audio_chunks = []

            for chunk in text_chunks:
                wav = self.model.tts(
                    text=chunk,
                    speaker_wav=processed_ref,
                    language=language,
                    speed=speed
                )
                audio_chunks.append(np.array(wav))

            # Join chunks with silence
            if len(audio_chunks) == 1:
                final_audio = audio_chunks[0]
            else:
                final_audio = self.join_audio_chunks(audio_chunks)

            # Convert to WAV bytes
            buffer = io.BytesIO()
            sf.write(buffer, final_audio, 24000, format='WAV')
            audio_bytes = buffer.getvalue()

            # Base64 encode for API response
            audio_b64 = base64.b64encode(audio_bytes).decode('utf-8')

            duration = len(final_audio) / 24000.0

            self.generation_count += 1

            return {
                "audio_base64": audio_b64,
                "duration_seconds": round(duration, 2),
                "chunks_processed": len(text_chunks),
                "language": language
            }

        finally:
            # Always clean up temp file
            if os.path.exists(processed_ref):
                os.remove(processed_ref)

    def get_supported_languages(self) -> list[dict]:
        """Return language list with metadata."""
        language_info = {
            'en': {'name': 'English', 'accent': 'American', 
                   'quality': 'excellent'},
            'es': {'name': 'Spanish', 'accent': 'Neutral', 
                   'quality': 'excellent'},
            'fr': {'name': 'French', 'accent': 'French', 
                   'quality': 'excellent'},
            'de': {'name': 'German', 'accent': 'German', 
                   'quality': 'very good'},
            'it': {'name': 'Italian', 'accent': 'Italian', 
                   'quality': 'very good'},
            'pt': {'name': 'Portuguese', 'accent': 'Brazilian', 
                   'quality': 'very good'},
            'pl': {'name': 'Polish', 'accent': 'Polish', 
                   'quality': 'good'},
            'tr': {'name': 'Turkish', 'accent': 'Turkish', 
                   'quality': 'good'},
            'ru': {'name': 'Russian', 'accent': 'Russian', 
                   'quality': 'good'},
            'nl': {'name': 'Dutch', 'accent': 'Dutch', 
                   'quality': 'good'},
            'cs': {'name': 'Czech', 'accent': 'Czech', 
                   'quality': 'good'},
            'ar': {'name': 'Arabic', 'accent': 'MSA', 
                   'quality': 'moderate'},
            'zh-cn': {'name': 'Chinese', 'accent': 'Mandarin', 
                      'quality': 'moderate'},
            'ja': {'name': 'Japanese', 'accent': 'Standard', 
                   'quality': 'moderate'},
            'ko': {'name': 'Korean', 'accent': 'Standard', 
                   'quality': 'moderate'},
            'hu': {'name': 'Hungarian', 'accent': 'Hungarian', 
                   'quality': 'good'},
            'hi': {'name': 'Hindi', 'accent': 'Standard', 
                   'quality': 'moderate'},
        }
        return [
            {'code': code, **info}
            for code, info in language_info.items()
            if code in self._supported_languages
        ]

python
def _reinitialise(self):
    import gc
    import torch

    # Delete the model object
    del self.model
    self.model = None

    # Force garbage collection
    gc.collect()

    # Clear GPU cache
    if torch.cuda.is_available():
        torch.cuda.empty_cache()

    # Reload
    self._load_model()

python
REQUIRED_FORMAT = {
    "sample_rate": 22050,  # Hz — must be exact
    "channels": 1,          # Mono only
    "format": "wav",        # WAV container
    "bit_depth": 16,        # 16-bit PCM
    "min_duration": 3.0,    # seconds
    "recommended_duration": 15.0,  # seconds
}

python
from pydub import AudioSegment

def prepare_reference_audio(input_path: str) -> str:
    audio = AudioSegment.from_file(input_path)
    audio = audio.set_channels(1)           # Mono
    audio = audio.set_frame_rate(22050)     # Required sample rate
    audio = audio.set_sample_width(2)       # 16-bit

    output_path = input_path.rsplit('.', 1)[0] + '_processed.wav'
    audio.export(output_path, format='wav')
    return output_path

python
# In server.py

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import JSONResponse
import tempfile
import os

voice_cloner = VoiceCloner()

@app.post("/api/clone-voice")
async def clone_voice_endpoint(
    text: str = Form(...),
    language: str = Form(default='en'),
    speed: float = Form(default=1.0),
    reference_audio: UploadFile = File(...)
):
    # Save uploaded audio to temp file
    suffix = os.path.splitext(reference_audio.filename)[1]
    with tempfile.NamedTemporaryFile(
        suffix=suffix, delete=False
    ) as tmp:
        content = await reference_audio.read()
        tmp.write(content)
        tmp_path = tmp.name

    try:
        result = voice_cloner.clone(
            text=text,
            reference_audio_path=tmp_path,
            language=language,
            speed=speed
        )
        return JSONResponse(content=result)

    except ValueError as e:
        return JSONResponse(
            status_code=400,
            content={"error": str(e)}
        )
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={"error": f"Generation failed: {str(e)}"}
        )
    finally:
        if os.path.exists(tmp_path):
            os.remove(tmp_path)

@app.get("/api/tts/voices")
async def get_voices():
    return voice_cloner.get_supported_languages()

@app.post("/api/tts/preview")
async def preview_voice(
    reference_audio: UploadFile = File(...),
    language: str = Form(default='en')
):
    """Generate a 5-second preview to evaluate voice quality."""
    preview_text = {
        'en': "Hello, this is a preview of your cloned voice.",
        'es': "Hola, esta es una vista previa de tu voz clonada.",
        'fr': "Bonjour, ceci est un aperçu de votre voix clonée.",
        'de': "Hallo, dies ist eine Vorschau Ihrer geklonten Stimme.",
        'ja': "こんにちは、これはクローンされた声のプレビューです。",
    }.get(language, "Hello, this is a preview of your cloned voice.")

    # Same logic as clone_voice_endpoint
    # [abbreviated for clarity]
    pass

python
MAX_RETRIES = 3

for attempt in range(MAX_RETRIES):
    try:
        wav = self.model.tts(
            text=chunk,
            speaker_wav=processed_ref,
            language=language
        )
        # Check for NaN/Inf
        audio_array = np.array(wav)
        if np.isnan(audio_array).any() or \
           np.isinf(audio_array).any():
            raise ValueError("NaN/Inf in output")
        audio_chunks.append(audio_array)
        break
    except Exception as e:
        if attempt == MAX_RETRIES - 1:
            raise
        time.sleep(0.5)