986 lines
38 KiB
Python
986 lines
38 KiB
Python
"""
|
|
Qwen3 TTS Handler
|
|
|
|
- On Apple Silicon: Uses mlx-audio with MLX-converted Qwen3-TTS models.
|
|
- On CUDA/CPU: Uses faster-qwen3-tts for low-latency streaming.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import re
|
|
import tempfile
|
|
import unicodedata
|
|
|
|
# Sanitize ApiPilot markers directly before TTS as an additional safeguard
|
|
# alongside the S2S relay's audio gating and sanitization.
|
|
# Observed formats:
|
|
# (a) <<<APIPILOT_CALL:NAME=tool,ARGUMENTS=json>>> standard triple brackets
|
|
# (b) <APIPILOT_CALL:NAME=tool,ARGUMENTS={"format":"voice"}
|
|
# single brackets
|
|
# (c) <<<APIPILOT_CALL:...}> \n>> split across lines
|
|
# (d) >>>APIPILOT_CALL:...<<< inverted brackets
|
|
# Replace CALL with a natural Spanish utterance; strip RESULT markers.
|
|
_APIPILOT_CALL_NORMAL_RE = re.compile(
|
|
r"[<]{1,3}\s*APIPILOT_CALL:NAME=([A-Za-z0-9_.-]+),ARGUMENTS=(.*?)[>]{1,3}",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_APIPILOT_CALL_INVERTED_RE = re.compile(
|
|
r"[>]{1,3}\s*APIPILOT_CALL:NAME=([A-Za-z0-9_.-]+),ARGUMENTS=(.*?)[<]{1,3}",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_APIPILOT_RESULT_RE = re.compile(
|
|
r"[<]{1,3}\s*APIPILOT_RESULT(?::PASS2)?[^<>\n]{0,2000}[>]{1,3}",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_APIPILOT_SPANISH_PLACEHOLDER = "Voy a consultar tus datos, un momento por favor."
|
|
|
|
|
|
def _tts_sanitize_apipilot_markers(text: str) -> tuple[str, bool, str]:
|
|
"""Remove ApiPilot markers from TTS text.
|
|
|
|
Returns (sanitized_text, found_any, log_tag).
|
|
"""
|
|
if not isinstance(text, str) or not text.strip():
|
|
return text, False, "empty"
|
|
found_call = False
|
|
found_result = False
|
|
tool_name = ""
|
|
# 1) Replace standard/inverted CALL markers with the Spanish placeholder.
|
|
def _repl_call(m: re.Match) -> str:
|
|
nonlocal found_call, tool_name
|
|
found_call = True
|
|
try:
|
|
nm = m.group(1)
|
|
if isinstance(nm, str) and nm.strip():
|
|
tool_name = nm.strip()
|
|
except Exception:
|
|
pass
|
|
return f" {_APIPILOT_SPANISH_PLACEHOLDER} "
|
|
text2 = _APIPILOT_CALL_NORMAL_RE.sub(_repl_call, text)
|
|
if not found_call:
|
|
text2 = _APIPILOT_CALL_INVERTED_RE.sub(_repl_call, text2)
|
|
# 2) Strip RESULT markers silently; they must not be spoken.
|
|
if _APIPILOT_RESULT_RE.search(text2):
|
|
found_result = True
|
|
text2 = _APIPILOT_RESULT_RE.sub(" ", text2)
|
|
# 3) Catch marker fragments without a complete wrapper.
|
|
low = text2.lower()
|
|
if "apipilot" in low and ("call" in low or "result" in low):
|
|
import re as _re_sf
|
|
text2 = _re_sf.sub(r"(?i)\bAPIPILOT(?:_CALL|_RESULT)?\b", " ", text2)
|
|
text2 = _re_sf.sub(r"(?i)\b(?:NAME|ARGUMENTS|STATUS|PASS2)\s*[:=]", " ", text2)
|
|
found_call = True
|
|
if found_call or found_result:
|
|
text2 = re.sub(r"\s{2,}", " ", text2).strip()
|
|
tag = "-"
|
|
if found_call and found_result:
|
|
tag = f"CALL({tool_name or '?'})+RESULT"
|
|
elif found_call:
|
|
tag = f"CALL({tool_name or '?'})"
|
|
elif found_result:
|
|
tag = "RESULT"
|
|
return text2, (found_call or found_result), tag
|
|
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from sys import platform
|
|
from threading import Event
|
|
from time import perf_counter
|
|
from typing import Any, Iterator, Optional
|
|
|
|
import numpy as np
|
|
import torch
|
|
from openai.types.realtime.realtime_response_create_params import RealtimeResponseCreateParams
|
|
from rich.console import Console
|
|
|
|
from speech_to_speech.api.openai_realtime.runtime_config import RuntimeConfig
|
|
from speech_to_speech.baseHandler import BaseHandler
|
|
from speech_to_speech.pipeline.cancel_scope import CancelScope
|
|
from speech_to_speech.pipeline.control import SESSION_END, is_control_message
|
|
from speech_to_speech.pipeline.handler_types import TTSIn, TTSOut
|
|
from speech_to_speech.pipeline.messages import AUDIO_RESPONSE_DONE, PIPELINE_END, EndOfResponse, TTSInput
|
|
from speech_to_speech.pipeline.speculative_turns import SpeculativeTurnTracker
|
|
from speech_to_speech.utils.mlx_lock import MLXLockContext
|
|
|
|
logger = logging.getLogger(__name__)
|
|
console = Console()
|
|
|
|
DEFAULT_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"
|
|
DEFAULT_MLX_MODEL = "mlx-community/Qwen3-TTS-12Hz-1.7B-CustomVoice-6bit"
|
|
DEFAULT_REF_TEXT = "I'm confused why some people have super short timelines, yet at the same time are bullish on scaling up reinforcement learning atop LLMs. If we're actually close to a human-like learner, then this whole approach of training on verifiable outcomes."
|
|
DEFAULT_FASTER_STREAMING_CHUNK_SIZE = 8
|
|
DEFAULT_MLX_STREAMING_CHUNK_SIZE = 4
|
|
DEFAULT_QWEN3_TTS_MAX_NEW_TOKENS = 1536
|
|
MIN_QWEN3_TTS_UTTERANCE_TOKENS = 360
|
|
VALID_MLX_QUANTIZATION_SUFFIXES = ("bf16", "4bit", "6bit", "8bit")
|
|
VALID_FASTER_BACKENDS = ("ggml", "torch")
|
|
MLX_STREAMING_TOKENS_PER_SECOND = 12.5
|
|
PIPELINE_SR = 16000
|
|
ESTIMATED_QWEN3_WORDS_PER_SECOND = 2.6
|
|
ESTIMATED_QWEN3_CHARS_PER_SECOND = 14.0
|
|
QWEN3_TOKEN_SAFETY_MARGIN = 1.35
|
|
QWEN3_BASE_PROMPT_SECONDS = 1.0
|
|
QWEN3_PUNCTUATION_PAUSE_SECONDS = 0.5
|
|
QWEN3_LANGUAGE_ALIASES = {
|
|
"zh": "chinese",
|
|
"zh-cn": "chinese",
|
|
"zh-hans": "chinese",
|
|
"zh-hant": "chinese",
|
|
"cmn": "chinese",
|
|
"en": "english",
|
|
"en-us": "english",
|
|
"en-gb": "english",
|
|
"ja": "japanese",
|
|
"jp": "japanese",
|
|
"ko": "korean",
|
|
"kr": "korean",
|
|
"de": "german",
|
|
"fr": "french",
|
|
"ru": "russian",
|
|
"pt": "portuguese",
|
|
"pt-br": "portuguese",
|
|
"pt-pt": "portuguese",
|
|
"es": "spanish",
|
|
"it": "italian",
|
|
}
|
|
|
|
|
|
class Qwen3TTSHandler(BaseHandler[TTSIn, TTSOut]):
|
|
"""
|
|
Handles Text-to-Speech using Qwen3-TTS.
|
|
|
|
Backend selection:
|
|
- Apple Silicon (Darwin): mlx-audio
|
|
- Other platforms: faster-qwen3-tts
|
|
|
|
Supports three generation modes depending on the loaded model:
|
|
- Voice cloning (ref_audio + ref_text)
|
|
- Custom voice (preset speakers)
|
|
- Voice design (instruct prompt)
|
|
"""
|
|
|
|
def setup(
|
|
self,
|
|
should_listen: Event,
|
|
model_name: str = DEFAULT_MODEL,
|
|
device: str = "cuda",
|
|
dtype: str | torch.dtype = "auto",
|
|
attn_implementation: str = "eager",
|
|
backend: str = "ggml",
|
|
ref_audio: str | Path | None = None,
|
|
ref_text: str = DEFAULT_REF_TEXT,
|
|
language: str = "auto",
|
|
speaker: Optional[str] = "Aiden",
|
|
instruct: Optional[str] = None,
|
|
xvec_only: bool = False,
|
|
parity_mode: bool = False,
|
|
non_streaming_mode: bool | None = True,
|
|
mlx_quantization: Optional[str] = None,
|
|
streaming_chunk_size: int | None = None,
|
|
max_new_tokens: int = DEFAULT_QWEN3_TTS_MAX_NEW_TOKENS,
|
|
blocksize: int = 512,
|
|
gen_kwargs: dict[str, Any] | None = None,
|
|
cancel_scope: CancelScope | None = None,
|
|
speculative_turns: SpeculativeTurnTracker | None = None,
|
|
) -> None:
|
|
self.cancel_scope = cancel_scope
|
|
self.speculative_turns = speculative_turns
|
|
self.should_listen = should_listen
|
|
self.requested_device = device
|
|
self.ref_audio = ref_audio
|
|
self.ref_text = ref_text
|
|
self.language = self._normalize_language(language)
|
|
self.speaker = speaker
|
|
self.instruct = instruct
|
|
self.xvec_only = xvec_only
|
|
self.parity_mode = parity_mode
|
|
self.non_streaming_mode = non_streaming_mode
|
|
self.faster_backend = self._normalize_faster_backend(backend)
|
|
self.mlx_quantization = self._normalize_mlx_quantization(mlx_quantization)
|
|
self.max_new_tokens = max_new_tokens
|
|
self.blocksize = blocksize
|
|
self.dtype: torch.dtype | None | str = None
|
|
self.gen_kwargs = gen_kwargs or {}
|
|
self._mlx_ref_audio_cache: dict[str, Any] = {}
|
|
self._mlx_temp_ref_audio_files: set[str] = set()
|
|
|
|
self.backend = "mlx" if platform == "darwin" else "faster_qwen3_tts"
|
|
self.streaming_chunk_size = self._resolve_streaming_chunk_size(streaming_chunk_size)
|
|
|
|
if self.backend == "mlx":
|
|
self.device = "mps"
|
|
self.model_name = self._resolve_mlx_model_name(model_name)
|
|
logger.info(f"Loading Qwen3-TTS model: {self.model_name} via mlx-audio on Apple Silicon")
|
|
if self.non_streaming_mode is not None:
|
|
logger.debug(
|
|
"qwen3_tts_non_streaming_mode=%s is ignored on Apple Silicon because "
|
|
"mlx-audio does not expose non_streaming_mode yet.",
|
|
self.non_streaming_mode,
|
|
)
|
|
model_quantization = self._model_name_quantization_suffix(self.model_name)
|
|
if model_quantization and model_quantization != "bf16":
|
|
logger.info(
|
|
"Using MLX quantized Qwen3-TTS variant: %s",
|
|
model_quantization,
|
|
)
|
|
self._setup_mlx(self.model_name)
|
|
else:
|
|
self.device = device
|
|
self.model_name = model_name
|
|
logger.info(
|
|
"Loading Qwen3-TTS model: %s via faster-qwen3-tts (%s backend)",
|
|
self.model_name,
|
|
self.faster_backend,
|
|
)
|
|
self._setup_faster(
|
|
model_name=self.model_name,
|
|
dtype=dtype,
|
|
attn_implementation=attn_implementation,
|
|
backend=self.faster_backend,
|
|
)
|
|
|
|
logger.info(
|
|
"Using Qwen3-TTS streaming chunk size %d (~%.0fms audio per chunk) on %s",
|
|
self.streaming_chunk_size,
|
|
self.streaming_chunk_size / MLX_STREAMING_TOKENS_PER_SECOND * 1000,
|
|
self.backend,
|
|
)
|
|
|
|
self._initial_speaker = self.speaker
|
|
self._initial_ref_audio = self.ref_audio
|
|
|
|
self.warmup()
|
|
|
|
def _setup_faster(self, model_name: str, dtype: Any, attn_implementation: str, backend: str) -> None:
|
|
try:
|
|
import torch
|
|
except ImportError as e:
|
|
raise ImportError("torch is required. Install with: pip install torch") from e
|
|
|
|
if dtype == "auto":
|
|
self.dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16
|
|
elif isinstance(dtype, str):
|
|
self.dtype = getattr(torch, dtype)
|
|
else:
|
|
self.dtype = dtype
|
|
|
|
try:
|
|
from faster_qwen3_tts import FasterQwen3TTS
|
|
except ImportError as e:
|
|
raise ImportError(
|
|
"faster-qwen3-tts is required for Qwen3 TTS on non-macOS platforms. "
|
|
"Install with: pip install 'faster-qwen3-tts[ggml]'"
|
|
) from e
|
|
|
|
self.model = FasterQwen3TTS.from_pretrained(
|
|
model_name,
|
|
device=self.device,
|
|
dtype=self.dtype,
|
|
attn_implementation=attn_implementation,
|
|
backend=backend,
|
|
)
|
|
logger.info("Qwen3-TTS model loaded")
|
|
|
|
def _setup_mlx(self, model_name: str) -> None:
|
|
try:
|
|
from mlx_audio.tts.utils import load_model
|
|
|
|
self.model = load_model(model_name)
|
|
except ImportError as e:
|
|
message = str(e)
|
|
if any(
|
|
dep in message
|
|
for dep in (
|
|
"misaki",
|
|
"spacy",
|
|
"phonemizer",
|
|
"espeakng_loader",
|
|
)
|
|
):
|
|
raise ImportError(
|
|
"Qwen3-TTS on Apple Silicon requires mlx-audio and its TTS dependencies. "
|
|
f"Missing dependency: {message}. "
|
|
"Install with: pip install mlx-audio misaki spacy phonemizer-fork espeakng-loader"
|
|
) from e
|
|
raise ImportError(
|
|
"mlx-audio is required for Qwen3 TTS on Apple Silicon. Install with: pip install mlx-audio"
|
|
) from e
|
|
|
|
logger.info("MLX Audio Qwen3-TTS model loaded")
|
|
|
|
def _normalize_faster_backend(self, backend: Any) -> str:
|
|
value = str(backend or "ggml").strip().lower()
|
|
if value not in VALID_FASTER_BACKENDS:
|
|
raise ValueError(
|
|
f"Unsupported qwen3_tts_backend value {backend!r}. Supported values: {', '.join(VALID_FASTER_BACKENDS)}"
|
|
)
|
|
return value
|
|
|
|
def _normalize_mlx_quantization(self, mlx_quantization: Any) -> Optional[str]:
|
|
if mlx_quantization is None:
|
|
return None
|
|
|
|
value = str(mlx_quantization).strip().lower()
|
|
if value in ("", "none", "default"):
|
|
return None
|
|
if value not in VALID_MLX_QUANTIZATION_SUFFIXES:
|
|
raise ValueError(
|
|
"Unsupported qwen3_tts_mlx_quantization value "
|
|
f"{mlx_quantization!r}. Supported values: {', '.join(VALID_MLX_QUANTIZATION_SUFFIXES)}"
|
|
)
|
|
return value
|
|
|
|
def _apply_mlx_quantization_suffix(self, model_name: str) -> str:
|
|
if self.mlx_quantization is None:
|
|
return model_name
|
|
|
|
desired_suffix = f"-{self.mlx_quantization}"
|
|
for suffix in VALID_MLX_QUANTIZATION_SUFFIXES:
|
|
current_suffix = f"-{suffix}"
|
|
if model_name.endswith(current_suffix):
|
|
return model_name[: -len(current_suffix)] + desired_suffix
|
|
|
|
return f"{model_name}{desired_suffix}"
|
|
|
|
def _model_name_quantization_suffix(self, model_name: str) -> Optional[str]:
|
|
if not model_name:
|
|
return None
|
|
|
|
for suffix in VALID_MLX_QUANTIZATION_SUFFIXES:
|
|
if model_name.endswith(f"-{suffix}"):
|
|
return suffix
|
|
|
|
return None
|
|
|
|
def _resolve_mlx_model_name(self, model_name: str) -> str:
|
|
if not model_name:
|
|
return self._apply_mlx_quantization_suffix(DEFAULT_MLX_MODEL)
|
|
if model_name.startswith("mlx-community/"):
|
|
if self.mlx_quantization is None and self._model_name_quantization_suffix(model_name) is None:
|
|
return f"{model_name}-6bit"
|
|
return self._apply_mlx_quantization_suffix(model_name)
|
|
if model_name.startswith("Qwen/"):
|
|
mapped = model_name.replace("Qwen/", "mlx-community/", 1)
|
|
if self._model_name_quantization_suffix(mapped) is None:
|
|
if self.mlx_quantization is None:
|
|
return f"{mapped}-6bit"
|
|
mapped = f"{mapped}-bf16"
|
|
return self._apply_mlx_quantization_suffix(mapped)
|
|
return model_name
|
|
|
|
def _resolve_streaming_chunk_size(self, streaming_chunk_size: int | None) -> int:
|
|
if streaming_chunk_size is not None:
|
|
return max(1, int(streaming_chunk_size))
|
|
if self.backend == "mlx":
|
|
return DEFAULT_MLX_STREAMING_CHUNK_SIZE
|
|
return DEFAULT_FASTER_STREAMING_CHUNK_SIZE
|
|
|
|
def _normalize_language(self, language: str | None) -> str:
|
|
if language is None:
|
|
return "auto"
|
|
normalized = str(language).strip().replace("_", "-").lower()
|
|
if not normalized:
|
|
return "auto"
|
|
return QWEN3_LANGUAGE_ALIASES.get(normalized, normalized)
|
|
|
|
def _infer_model_type_from_name(self) -> str:
|
|
name = (self.model_name or "").lower()
|
|
if "voicedesign" in name:
|
|
return "voice_design"
|
|
if "customvoice" in name:
|
|
return "custom_voice"
|
|
return "base"
|
|
|
|
def _resolve_audio_path(self, audio: Any) -> Path | None:
|
|
if not isinstance(audio, (str, Path)) or not audio:
|
|
return None
|
|
|
|
candidate = Path(audio).expanduser()
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
search_paths = []
|
|
|
|
if candidate.is_absolute():
|
|
search_paths.append(candidate)
|
|
else:
|
|
search_paths.append(Path.cwd() / candidate)
|
|
search_paths.append(repo_root / candidate)
|
|
|
|
seen = set()
|
|
for path in search_paths:
|
|
normalized = str(path)
|
|
if normalized in seen:
|
|
continue
|
|
seen.add(normalized)
|
|
if path.exists():
|
|
return path.resolve()
|
|
|
|
return None
|
|
|
|
def _prepare_mlx_ref_audio(self, ref_audio: Any) -> Any:
|
|
if self.backend != "mlx" or ref_audio is None:
|
|
return ref_audio
|
|
|
|
if not isinstance(ref_audio, (str, Path)):
|
|
return ref_audio
|
|
|
|
resolved_path = self._resolve_audio_path(ref_audio)
|
|
if resolved_path is None:
|
|
raise FileNotFoundError(
|
|
"Qwen3-TTS on Apple Silicon requires qwen3_tts_ref_audio to point to "
|
|
f"a readable audio file. Got: {ref_audio!r}"
|
|
)
|
|
|
|
cache_key = str(resolved_path)
|
|
cached_path = self._mlx_ref_audio_cache.get(cache_key)
|
|
if cached_path and Path(cached_path).exists():
|
|
return cached_path
|
|
|
|
try:
|
|
import soundfile as sf
|
|
from scipy.signal import resample_poly
|
|
|
|
waveform, sample_rate = sf.read(str(resolved_path), always_2d=False, dtype="float32")
|
|
waveform = np.asarray(waveform, dtype=np.float32)
|
|
if waveform.ndim > 1:
|
|
waveform = waveform.mean(axis=1)
|
|
target_sample_rate = getattr(self.model, "sample_rate", 24000)
|
|
if sample_rate != target_sample_rate:
|
|
gcd = np.gcd(int(sample_rate), int(target_sample_rate))
|
|
waveform = resample_poly(
|
|
waveform,
|
|
up=int(target_sample_rate) // gcd,
|
|
down=int(sample_rate) // gcd,
|
|
)
|
|
sample_rate = target_sample_rate
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
prefix="qwen3_ref_",
|
|
suffix=".wav",
|
|
delete=False,
|
|
) as temp_file:
|
|
normalized_path = temp_file.name
|
|
|
|
sf.write(
|
|
normalized_path,
|
|
waveform,
|
|
sample_rate,
|
|
format="WAV",
|
|
subtype="PCM_16",
|
|
)
|
|
except Exception as e:
|
|
raise RuntimeError(f"Failed to normalize Qwen3-TTS reference audio {resolved_path}: {e}") from e
|
|
|
|
self._mlx_ref_audio_cache[cache_key] = normalized_path
|
|
self._mlx_temp_ref_audio_files.add(normalized_path)
|
|
return normalized_path
|
|
|
|
def _apply_session_voice_override(
|
|
self,
|
|
model_type: str,
|
|
runtime_config: RuntimeConfig | None = None,
|
|
response: RealtimeResponseCreateParams | None = None,
|
|
) -> None:
|
|
session_voice: Optional[str] = None
|
|
if response and response.audio and response.audio.output:
|
|
resp_voice = response.audio.output.voice
|
|
session_voice = str(resp_voice) if resp_voice else None
|
|
if not session_voice and runtime_config is not None:
|
|
audio = runtime_config.session.audio
|
|
output = audio.output if audio is not None else None
|
|
sess_voice = output.voice if output is not None else None
|
|
session_voice = str(sess_voice) if sess_voice else None
|
|
if not session_voice:
|
|
return
|
|
|
|
if model_type == "custom_voice":
|
|
supported_speakers = self._supported_speakers()
|
|
if supported_speakers is not None:
|
|
speakers_by_lower = {speaker.lower(): speaker for speaker in supported_speakers}
|
|
speaker = speakers_by_lower.get(session_voice.lower())
|
|
if speaker is None:
|
|
supported_text = ", ".join(sorted(supported_speakers, key=str.lower)) or "unknown"
|
|
logger.warning(
|
|
"Ignoring Qwen3-TTS session voice override %r because it is not a supported "
|
|
"CustomVoice speaker. Supported speakers: %s",
|
|
session_voice,
|
|
supported_text,
|
|
)
|
|
return
|
|
session_voice = speaker
|
|
|
|
self.speaker = session_voice
|
|
self.ref_audio = None
|
|
return
|
|
|
|
if self._resolve_audio_path(session_voice) is not None:
|
|
self.ref_audio = session_voice
|
|
return
|
|
|
|
logger.warning(
|
|
"Ignoring Qwen3-TTS session voice override because it is not an audio file path: %r",
|
|
session_voice,
|
|
)
|
|
|
|
def warmup(self) -> None:
|
|
logger.info(f"Warming up {self.__class__.__name__}")
|
|
|
|
if self.backend == "faster_qwen3_tts":
|
|
if self.parity_mode:
|
|
logger.info("Qwen3-TTS parity mode enabled: skipping backend warmup")
|
|
else:
|
|
try:
|
|
self.model.warmup(prefill_len=100)
|
|
except Exception as e:
|
|
logger.warning("Qwen3-TTS backend warmup failed: %s", e)
|
|
|
|
try:
|
|
for _ in self._warmup_process("Hello, this is a warmup."):
|
|
pass
|
|
logger.info(f"{self.__class__.__name__} warmed up")
|
|
except Exception as e:
|
|
logger.warning(f"Warmup generation failed: {e}")
|
|
|
|
def _model_type(self) -> str:
|
|
if self.backend == "mlx":
|
|
config = getattr(self.model, "config", None)
|
|
return getattr(config, "tts_model_type", None) or self._infer_model_type_from_name()
|
|
|
|
inner = getattr(getattr(self.model, "model", None), "model", None)
|
|
return getattr(inner, "tts_model_type", None) or self._infer_model_type_from_name()
|
|
|
|
def _supported_speakers(self) -> list[str] | None:
|
|
for candidate in (
|
|
self.model,
|
|
getattr(self.model, "model", None),
|
|
getattr(getattr(self.model, "model", None), "model", None),
|
|
):
|
|
get_speakers = getattr(candidate, "get_supported_speakers", None)
|
|
if callable(get_speakers):
|
|
speakers = get_speakers()
|
|
if speakers is None:
|
|
return None
|
|
return [str(speaker) for speaker in speakers if speaker]
|
|
return None
|
|
|
|
def _resolve_speaker(self) -> Optional[str]:
|
|
if self.speaker:
|
|
return self.speaker
|
|
|
|
for candidate in (
|
|
self.model,
|
|
getattr(getattr(self.model, "model", None), "model", None),
|
|
):
|
|
get_speakers = getattr(candidate, "get_supported_speakers", None)
|
|
if callable(get_speakers):
|
|
speakers = list(get_speakers() or [])
|
|
if speakers:
|
|
return speakers[0]
|
|
|
|
return None
|
|
|
|
def _to_int16(self, audio: np.ndarray) -> np.ndarray:
|
|
return np.clip(audio * 32768, -32768, 32767).astype(np.int16)
|
|
|
|
def _estimate_max_new_tokens(self, text: Optional[str]) -> int:
|
|
text = (text or "").strip()
|
|
chunk_size = max(1, int(getattr(self, "streaming_chunk_size", 1)))
|
|
configured_cap = max(1, int(getattr(self, "max_new_tokens", DEFAULT_QWEN3_TTS_MAX_NEW_TOKENS)))
|
|
|
|
if not text:
|
|
return min(configured_cap, MIN_QWEN3_TTS_UTTERANCE_TOKENS)
|
|
|
|
word_count = len(re.findall(r"\w+", text, flags=re.UNICODE))
|
|
char_count = len(re.sub(r"\s+", "", text))
|
|
word_seconds = word_count / ESTIMATED_QWEN3_WORDS_PER_SECOND if word_count else 0.0
|
|
char_seconds = char_count / ESTIMATED_QWEN3_CHARS_PER_SECOND if char_count else 0.0
|
|
punctuation_count = sum(unicodedata.category(ch).startswith("P") for ch in text)
|
|
punctuation_seconds = punctuation_count * QWEN3_PUNCTUATION_PAUSE_SECONDS
|
|
estimated_seconds = max(word_seconds, char_seconds) + punctuation_seconds + QWEN3_BASE_PROMPT_SECONDS
|
|
estimated_tokens = math.ceil(estimated_seconds * MLX_STREAMING_TOKENS_PER_SECOND * QWEN3_TOKEN_SAFETY_MARGIN)
|
|
aligned_tokens = max(
|
|
chunk_size,
|
|
math.ceil(estimated_tokens / chunk_size) * chunk_size,
|
|
)
|
|
requested_tokens = max(MIN_QWEN3_TTS_UTTERANCE_TOKENS, aligned_tokens)
|
|
resolved_tokens = min(configured_cap, requested_tokens)
|
|
|
|
if resolved_tokens < requested_tokens:
|
|
logger.warning(
|
|
"Qwen3-TTS estimated %d codec tokens for a %d-character utterance, "
|
|
"but max_new_tokens is capped at %d; output may still truncate.",
|
|
requested_tokens,
|
|
len(text),
|
|
configured_cap,
|
|
)
|
|
|
|
logger.debug(
|
|
"Qwen3-TTS using max_new_tokens=%d for utterance with %d words and %d chars",
|
|
resolved_tokens,
|
|
word_count,
|
|
char_count,
|
|
)
|
|
return resolved_tokens
|
|
|
|
def _warmup_process(self, llm_sentence: str) -> Iterator[bytes | np.ndarray]:
|
|
model_type = self._model_type()
|
|
if self.ref_audio:
|
|
yield from self._process_voice_clone(llm_sentence)
|
|
elif model_type == "custom_voice":
|
|
yield from self._process_custom_voice(llm_sentence)
|
|
elif model_type == "voice_design":
|
|
yield from self._process_voice_design(llm_sentence)
|
|
else:
|
|
raise ValueError(
|
|
"Qwen3-TTS Base model requires ref_audio for voice cloning. "
|
|
"Provide qwen3_tts_ref_audio or use a CustomVoice/VoiceDesign model."
|
|
)
|
|
|
|
def _resample_to_pipeline_sr(self, audio: np.ndarray, sr: int) -> np.ndarray:
|
|
if sr == PIPELINE_SR:
|
|
return audio
|
|
from scipy.signal import resample_poly
|
|
|
|
gcd = np.gcd(PIPELINE_SR, sr)
|
|
return resample_poly(audio, up=PIPELINE_SR // gcd, down=sr // gcd)
|
|
|
|
def _prepare_audio_chunk(self, item: Any) -> tuple[np.ndarray | None, int | None]:
|
|
if isinstance(item, tuple):
|
|
audio_chunk, sr, _timing = item
|
|
return np.asarray(audio_chunk, dtype=np.float32), sr
|
|
|
|
audio = getattr(item, "audio", None)
|
|
if audio is None:
|
|
return None, None
|
|
|
|
audio_chunk = np.asarray(audio, dtype=np.float32).squeeze()
|
|
sr = getattr(item, "sample_rate", None) or PIPELINE_SR
|
|
return audio_chunk, sr
|
|
|
|
def _stream(self, gen: Any, label: str) -> Iterator[bytes | np.ndarray]:
|
|
"""Common streaming loop: log TTFA and RTF, yield int16 chunks."""
|
|
cancel_gen = self.cancel_scope.generation if self.cancel_scope else None
|
|
start = perf_counter()
|
|
total_samples = 0
|
|
first_chunk = True
|
|
found_speech = False
|
|
leftover = np.array([], dtype=np.int16)
|
|
|
|
for item in gen:
|
|
if cancel_gen is not None and self.cancel_scope is not None and self.cancel_scope.is_stale(cancel_gen):
|
|
logger.info("TTS generation cancelled (interruption)")
|
|
return
|
|
|
|
audio_chunk, sr = self._prepare_audio_chunk(item)
|
|
if audio_chunk is None or sr is None or audio_chunk.size == 0:
|
|
continue
|
|
|
|
if first_chunk:
|
|
logger.info(f"Qwen3-TTS TTFA: {perf_counter() - start:.2f}s ({label})")
|
|
first_chunk = False
|
|
|
|
audio_chunk = self._resample_to_pipeline_sr(audio_chunk, sr)
|
|
audio_chunk = self._to_int16(audio_chunk)
|
|
|
|
# Trim the initial silent ramp-up, but keep enough preroll to avoid
|
|
# shaving soft initial phonemes at the start of the utterance.
|
|
if not found_speech:
|
|
threshold = int(32768 * 0.01)
|
|
above = np.abs(audio_chunk) > threshold
|
|
if not np.any(above):
|
|
continue
|
|
start_idx = max(0, int(np.argmax(above)) - int(PIPELINE_SR * 0.040))
|
|
audio_chunk = audio_chunk[start_idx:]
|
|
found_speech = True
|
|
|
|
audio_chunk = np.concatenate([leftover, audio_chunk])
|
|
|
|
n = (len(audio_chunk) // self.blocksize) * self.blocksize
|
|
for i in range(0, n, self.blocksize):
|
|
yield audio_chunk[i : i + self.blocksize]
|
|
total_samples += self.blocksize
|
|
leftover = audio_chunk[n:]
|
|
|
|
if len(leftover) > 0:
|
|
chunk = np.pad(leftover, (0, self.blocksize - len(leftover)))
|
|
yield chunk
|
|
total_samples += len(leftover)
|
|
|
|
generation_time = perf_counter() - start
|
|
audio_duration = total_samples / PIPELINE_SR
|
|
rtf = audio_duration / generation_time if generation_time > 0 else 0
|
|
logger.info(
|
|
f"Qwen3-TTS generated {audio_duration:.2f}s audio in {generation_time:.2f}s (RTF: {rtf:.2f}, {label})"
|
|
)
|
|
|
|
def _coalesce_pending_tts_input(self, current_input: TTSInput) -> tuple[str, Optional[str], bool]:
|
|
"""Combine already-queued text chunks before the next TTS synthesis call."""
|
|
if not hasattr(self.queue_in, "mutex") or not hasattr(self.queue_in, "queue"):
|
|
return current_input.text, current_input.language_code, False
|
|
|
|
text = current_input.text
|
|
language_code = current_input.language_code
|
|
|
|
parts = [text.strip()] if text and text.strip() else []
|
|
saw_end_of_response = False
|
|
|
|
with self.queue_in.mutex:
|
|
while self.queue_in.queue:
|
|
next_item = self.queue_in.queue[0]
|
|
if is_control_message(next_item, SESSION_END.kind):
|
|
break
|
|
if isinstance(next_item, bytes) and next_item == PIPELINE_END:
|
|
break
|
|
if isinstance(next_item, EndOfResponse):
|
|
saw_end_of_response = True
|
|
break
|
|
if not isinstance(next_item, TTSInput):
|
|
break
|
|
if current_input.turn_id != next_item.turn_id or current_input.turn_revision != next_item.turn_revision:
|
|
break
|
|
if (
|
|
language_code is not None
|
|
and next_item.language_code is not None
|
|
and next_item.language_code != language_code
|
|
):
|
|
break
|
|
|
|
self.queue_in.queue.popleft()
|
|
if next_item.text.strip():
|
|
parts.append(next_item.text.strip())
|
|
if language_code is None:
|
|
language_code = next_item.language_code
|
|
|
|
combined_text = " ".join(parts).strip()
|
|
return combined_text, language_code, saw_end_of_response
|
|
|
|
def process(self, tts_input: TTSIn) -> Iterator[TTSOut]:
|
|
speculative_turns = getattr(self, "speculative_turns", None)
|
|
if isinstance(tts_input, EndOfResponse):
|
|
if speculative_turns and not speculative_turns.is_latest_after_reopen_grace(
|
|
tts_input.turn_id,
|
|
tts_input.turn_revision,
|
|
):
|
|
return
|
|
yield AUDIO_RESPONSE_DONE
|
|
return
|
|
|
|
if speculative_turns and not speculative_turns.is_latest_after_reopen_grace(
|
|
tts_input.turn_id,
|
|
tts_input.turn_revision,
|
|
):
|
|
logger.debug("Dropping stale TTS input for turn=%s rev=%s", tts_input.turn_id, tts_input.turn_revision)
|
|
return
|
|
if speculative_turns:
|
|
speculative_turns.commit(tts_input.turn_id, tts_input.turn_revision)
|
|
|
|
runtime_config = tts_input.runtime_config
|
|
response = tts_input.response
|
|
|
|
coalesced_text, _language_code, _saw_end_of_response = self._coalesce_pending_tts_input(tts_input)
|
|
|
|
raw_text = coalesced_text or "Hello."
|
|
# A race or a new marker format can bypass relay sanitization. Strip or
|
|
# replace markers again before passing text to the TTS model.
|
|
text, found_marker, sanitize_tag = _tts_sanitize_apipilot_markers(raw_text)
|
|
if found_marker:
|
|
logger.warning(
|
|
"[TTS-APIPILOT-SANITIZE] marker=%s rawLen=%d sanitizedLen=%d → replaced with the Spanish placeholder or stripped.",
|
|
sanitize_tag, len(raw_text or ""), len(text or ""),
|
|
)
|
|
if not text.strip():
|
|
# Use the Spanish placeholder when the marker was the entire utterance.
|
|
text = _APIPILOT_SPANISH_PLACEHOLDER
|
|
|
|
model_type = self._model_type()
|
|
self._apply_session_voice_override(model_type, runtime_config, response)
|
|
|
|
console.print(f"[green]ASSISTANT: {text}")
|
|
|
|
try:
|
|
if self.ref_audio:
|
|
audio_iter = self._process_voice_clone(text)
|
|
elif model_type == "custom_voice":
|
|
audio_iter = self._process_custom_voice(text)
|
|
elif model_type == "voice_design":
|
|
audio_iter = self._process_voice_design(text)
|
|
else:
|
|
raise ValueError(
|
|
"Qwen3-TTS Base model requires ref_audio for voice cloning. "
|
|
"Provide qwen3_tts_ref_audio or use a CustomVoice/VoiceDesign model."
|
|
)
|
|
first_audio = True
|
|
for audio_chunk in audio_iter:
|
|
if first_audio:
|
|
self._log_first_audio_latency(tts_input)
|
|
first_audio = False
|
|
yield audio_chunk
|
|
except Exception as e:
|
|
logger.error(f"Error during Qwen3-TTS generation: {e}", exc_info=True)
|
|
|
|
def _log_first_audio_latency(self, tts_input: TTSInput) -> None:
|
|
if tts_input.speech_stopped_at_s is None:
|
|
return
|
|
latency_s = perf_counter() - tts_input.speech_stopped_at_s
|
|
if latency_s < 0:
|
|
return
|
|
logger.info(
|
|
"Last speech detected to first speech out: %.3fs (turn=%s rev=%s)",
|
|
latency_s,
|
|
tts_input.turn_id,
|
|
tts_input.turn_revision,
|
|
)
|
|
|
|
def _mlx_streaming_interval(self) -> float:
|
|
return max(1, self.streaming_chunk_size) / MLX_STREAMING_TOKENS_PER_SECOND
|
|
|
|
def _mlx_stream_kwargs(self, max_tokens: int) -> dict[str, Any]:
|
|
return {
|
|
"max_tokens": max_tokens,
|
|
"verbose": False,
|
|
"stream": True,
|
|
"streaming_interval": self._mlx_streaming_interval(),
|
|
**self.gen_kwargs,
|
|
}
|
|
|
|
def _stream_mlx_generation(
|
|
self,
|
|
generation_fn: Callable,
|
|
label: str,
|
|
max_tokens: int,
|
|
**generation_kwargs: Any,
|
|
) -> Iterator[bytes | np.ndarray]:
|
|
with MLXLockContext(handler_name="Qwen3TTS", timeout=10.0) as acquired:
|
|
if not acquired:
|
|
raise TimeoutError("Timed out waiting for MLX lock")
|
|
yield from self._stream(
|
|
generation_fn(
|
|
**self._mlx_stream_kwargs(max_tokens=max_tokens),
|
|
**generation_kwargs,
|
|
),
|
|
label=label,
|
|
)
|
|
|
|
def _process_voice_clone(self, text: str) -> Iterator[bytes | np.ndarray]:
|
|
utterance_max_new_tokens = self._estimate_max_new_tokens(text)
|
|
if self.backend == "mlx":
|
|
if self.xvec_only:
|
|
logger.warning("mlx-audio Qwen3-TTS does not support xvec_only; ignoring it")
|
|
if self.parity_mode:
|
|
logger.info("Qwen3-TTS parity mode is CUDA-specific and is ignored on mlx-audio")
|
|
|
|
yield from self._stream_mlx_generation(
|
|
self.model.generate,
|
|
label="voice_clone_mlx",
|
|
max_tokens=utterance_max_new_tokens,
|
|
text=text,
|
|
ref_audio=self._prepare_mlx_ref_audio(self.ref_audio),
|
|
ref_text=self.ref_text,
|
|
lang_code=self.language,
|
|
)
|
|
return
|
|
|
|
yield from self._stream(
|
|
self.model.generate_voice_clone_streaming(
|
|
text=text,
|
|
language=self.language,
|
|
ref_audio=self.ref_audio,
|
|
ref_text=self.ref_text,
|
|
xvec_only=self.xvec_only,
|
|
chunk_size=self.streaming_chunk_size,
|
|
max_new_tokens=utterance_max_new_tokens,
|
|
parity_mode=self.parity_mode,
|
|
non_streaming_mode=self.non_streaming_mode,
|
|
),
|
|
label="voice_clone_parity" if self.parity_mode else "voice_clone",
|
|
)
|
|
|
|
def _process_custom_voice(self, text: str) -> Iterator[bytes | np.ndarray]:
|
|
utterance_max_new_tokens = self._estimate_max_new_tokens(text)
|
|
speaker = self._resolve_speaker()
|
|
if not speaker:
|
|
raise ValueError(
|
|
"CustomVoice generation requires a speaker. "
|
|
"Set qwen3_tts_speaker or use a voice-clone model with ref_audio."
|
|
)
|
|
|
|
if self.backend == "mlx":
|
|
yield from self._stream_mlx_generation(
|
|
self.model.generate_custom_voice,
|
|
label="custom_voice_mlx",
|
|
max_tokens=utterance_max_new_tokens,
|
|
text=text,
|
|
speaker=speaker,
|
|
language=self.language,
|
|
instruct=self.instruct,
|
|
)
|
|
return
|
|
|
|
yield from self._stream(
|
|
self.model.generate_custom_voice_streaming(
|
|
text=text,
|
|
speaker=speaker,
|
|
language=self.language,
|
|
instruct=self.instruct,
|
|
chunk_size=self.streaming_chunk_size,
|
|
max_new_tokens=utterance_max_new_tokens,
|
|
non_streaming_mode=self.non_streaming_mode,
|
|
),
|
|
label="custom_voice",
|
|
)
|
|
|
|
def _process_voice_design(self, text: str) -> Iterator[bytes | np.ndarray]:
|
|
utterance_max_new_tokens = self._estimate_max_new_tokens(text)
|
|
if self.backend == "mlx":
|
|
yield from self._stream_mlx_generation(
|
|
self.model.generate_voice_design,
|
|
label="voice_design_mlx",
|
|
max_tokens=utterance_max_new_tokens,
|
|
text=text,
|
|
instruct=self.instruct,
|
|
language=self.language,
|
|
)
|
|
return
|
|
|
|
yield from self._stream(
|
|
self.model.generate_voice_design_streaming(
|
|
text=text,
|
|
instruct=self.instruct,
|
|
language=self.language,
|
|
chunk_size=self.streaming_chunk_size,
|
|
max_new_tokens=utterance_max_new_tokens,
|
|
non_streaming_mode=self.non_streaming_mode,
|
|
),
|
|
label="voice_design",
|
|
)
|
|
|
|
def on_session_end(self) -> None:
|
|
self.speaker = self._initial_speaker
|
|
self.ref_audio = self._initial_ref_audio
|
|
logger.debug("Qwen3-TTS session state reset")
|
|
|
|
def cleanup(self) -> None:
|
|
try:
|
|
del self.model
|
|
for path in list(getattr(self, "_mlx_temp_ref_audio_files", set())):
|
|
try:
|
|
Path(path).unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
if self.backend == "mlx":
|
|
try:
|
|
import mlx.core as mx
|
|
|
|
mx.clear_cache()
|
|
except Exception:
|
|
pass
|
|
else:
|
|
import torch
|
|
|
|
if torch.cuda.is_available():
|
|
torch.cuda.empty_cache()
|
|
logger.info("Qwen3-TTS handler cleaned up")
|
|
except Exception as e:
|
|
logger.warning(f"Cleanup error: {e}")
|