tts_providers/qwen3_provider.py
Rama main · réplica de lectura del repositorio original.
"""
Qwen3-TTS Provider — fal.ai hosted Qwen3-TTS models.
Supports:
- Named speaker TTS (0.6B and 1.7B)
- Voice cloning from audio samples via clone-voice endpoint
- Emotion/style control via prompt (1.7B)
- Cloned voice embeddings stored locally for reuse
API key: FAL_KEY env var
"""
import json
import os
import re
import subprocess
import time
import logging
from pathlib import Path
from typing import Optional
import httpx
from .base_provider import (
CloneCapableTTSProvider, CloneInfo, CloneRequest, CloneResult,
soft_retire_clone_dir,
)
logger = logging.getLogger(__name__)
# fal.ai endpoints
FAL_TTS_1_7B = "https://fal.run/fal-ai/qwen-3-tts/text-to-speech/1.7b"
FAL_TTS_0_6B = "https://fal.run/fal-ai/qwen-3-tts/text-to-speech/0.6b"
FAL_CLONE_1_7B = "https://fal.run/fal-ai/qwen-3-tts/clone-voice/1.7b"
FAL_CLONE_0_6B = "https://fal.run/fal-ai/qwen-3-tts/clone-voice/0.6b"
BUILTIN_VOICES = [
"Vivian", # Female, warm
"Serena", # Female, clear
"Dylan", # Male, casual
"Eric", # Male, professional
"Ryan", # Male, energetic
"Aiden", # Male, deep
"Uncle_Fu", # Male, character
"Ono_Anna", # Female, Japanese accent
"Sohee", # Female, Korean accent
]
def _get_clones_dir() -> Path:
"""Resolve voice clones directory from paths module or fallback."""
try:
from services.paths import VOICE_CLONES_DIR
return VOICE_CLONES_DIR
except ImportError:
return Path(os.getenv("VOICE_CLONES_DIR", "./runtime/voice-clones"))
def _fal_request(api_key: str, endpoint: str, payload: dict,
timeout: float = 90.0) -> dict:
"""Make a JSON request to fal.ai and return the parsed response."""
headers = {
'Authorization': f'Key {api_key}',
'Content-Type': 'application/json',
}
with httpx.Client(timeout=httpx.Timeout(timeout, connect=10.0)) as client:
resp = client.post(endpoint, json=payload, headers=headers)
# JamBot Books: record the fal.ai call (guarded, fire-and-forget).
try:
from services.jambot_books_hook import record_provider_call
record_provider_call('fal', endpoint=endpoint, status=resp.status_code)
except Exception:
pass
resp.raise_for_status()
return resp.json()
def _fal_download(url: str, timeout: float = 30.0) -> bytes:
"""Download binary content from a fal.ai result URL."""
with httpx.Client(timeout=httpx.Timeout(timeout)) as client:
resp = client.get(url)
resp.raise_for_status()
return resp.content
class Qwen3Provider(CloneCapableTTSProvider):
"""
TTS Provider using Qwen3-TTS via fal.ai.
Built-in voices: Vivian, Serena, Dylan, Eric, Ryan, Aiden, Uncle_Fu, Ono_Anna, Sohee
Cloned voices: stored locally as .safetensors embeddings, referenced by voice_id
Output: MP3 audio bytes
"""
# Clone seam (base_provider.CloneCapableTTSProvider)
clone_label = 'Qwen3'
clone_input = ('url',) # fal.ai fetches the reference from a URL
clone_languages = ('en', 'zh', 'es', 'fr', 'de', 'it', 'ja', 'ko', 'pt', 'ru')
def __init__(self):
super().__init__()
self.api_key = os.getenv('FAL_KEY', '')
self._status = 'active' if self.api_key else 'error'
self._init_error = None if self.api_key else 'FAL_KEY not set in environment'
# ------------------------------------------------------------------
# Voice cloning
# ------------------------------------------------------------------
def clone_voice(self, audio_url: str, name: str,
reference_text: Optional[str] = None) -> dict:
"""
Clone a voice from a reference audio sample.
Args:
audio_url: Public URL to reference audio (WAV/MP3, 3+ seconds).
name: Human-readable name for this cloned voice.
reference_text: Optional transcript of what's said in the audio
(improves quality).
Returns:
dict with: voice_id, name, embedding_url, created_at, metadata
"""
if not self.api_key:
raise RuntimeError("FAL_KEY not set — cannot clone voice")
t = time.time()
logger.info(f"[Qwen3] Cloning voice '{name}' from {audio_url[:80]}")
payload = {"audio_url": audio_url}
if reference_text:
payload["reference_text"] = reference_text
try:
result = _fal_request(self.api_key, FAL_CLONE_1_7B, payload,
timeout=120.0)
except httpx.HTTPStatusError as e:
raise RuntimeError(
f"fal.ai clone error {e.response.status_code}: {e.response.text}"
)
# Extract embedding URL from response
embedding_url = result.get('speaker_embedding', {}).get('url')
if not embedding_url:
# Try alternate response shapes
embedding_url = result.get('audio', {}).get('url')
if not embedding_url:
raise RuntimeError(f"No embedding URL in fal.ai response: {result}")
elapsed_ms = int((time.time() - t) * 1000)
# Download and persist the embedding locally
embedding_bytes = _fal_download(embedding_url)
clones_dir = _get_clones_dir()
# voice_id = sanitized name
voice_id = "clone_" + "".join(
c for c in name.lower().replace(" ", "_")
if c.isalnum() or c == "_"
)[:40]
voice_dir = clones_dir / voice_id
voice_dir.mkdir(parents=True, exist_ok=True)
embedding_path = voice_dir / "embedding.safetensors"
with open(embedding_path, 'wb') as f:
f.write(embedding_bytes)
metadata = {
"voice_id": voice_id,
"name": name,
"embedding_url": embedding_url,
"embedding_size": len(embedding_bytes),
"reference_text": reference_text,
"source_audio_url": audio_url,
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"clone_time_ms": elapsed_ms,
"provider": "qwen3",
"fal_response": result,
}
with open(voice_dir / "metadata.json", 'w') as f:
json.dump(metadata, f, indent=2)
logger.info(
f"[Qwen3] Voice cloned: {voice_id} ({len(embedding_bytes)} bytes) "
f"in {elapsed_ms}ms"
)
return metadata
def list_cloned_voices(self) -> list:
"""List all locally stored cloned voice embeddings."""
clones_dir = _get_clones_dir()
voices = []
if not clones_dir.exists():
return voices
for voice_dir in sorted(clones_dir.iterdir()):
# '_'-prefixed entries are seam state (VOICE_CLONE_SEAM_DIR =
# VOICE_CLONES_DIR/_seam), never a voice; voice ids are clone_*.
if voice_dir.name.startswith('_'):
continue
# Retired clones keep their metadata.json (rename, never delete)
# but are no longer voices.
if voice_dir.name.endswith('.retired'):
continue
meta_path = voice_dir / "metadata.json"
if meta_path.exists():
try:
with open(meta_path) as f:
meta = json.load(f)
meta["has_embedding"] = (voice_dir / "embedding.safetensors").exists()
voices.append(meta)
except Exception as e:
logger.warning(f"Bad voice metadata in {voice_dir}: {e}")
return voices
# ------------------------------------------------------------------
# Clone seam shims — wrap clone_voice exactly as the pre-seam route did
# ------------------------------------------------------------------
def create_clone(self, req: CloneRequest) -> CloneResult:
record = self.clone_voice(
audio_url=req.audio_url,
name=req.name,
reference_text=req.reference_text,
)
voice_ref = ''
if record.get('voice_id'):
voice_ref = str(_get_clones_dir() / record['voice_id']
/ 'embedding.safetensors')
return CloneResult.from_legacy('qwen3', req.name, record, voice_ref)
def list_clones(self, tenant_id: str) -> list:
# Clones live in this container's own volume, so every one is this
# tenant's (jail-cell rule: one OVU container per tenant).
return [
CloneInfo(voice_id=c['voice_id'], name=c.get('name', ''),
provider='qwen3', created_at=c.get('created_at', ''),
tenant_id=tenant_id, metadata=c)
for c in self.list_cloned_voices() if c.get('voice_id')
]
def disable_clone(self, tenant_id: str, voice_id: str, reason: str = '') -> None:
soft_retire_clone_dir(_get_clones_dir(), voice_id, reason)
def get_clone_embedding_url(self, voice_id: str) -> Optional[str]:
"""Get the fal.ai embedding URL for a cloned voice.
Returns the cached remote URL from metadata. The embedding is also
stored locally as a fallback, but fal.ai needs the URL for generation.
"""
clones_dir = _get_clones_dir()
meta_path = clones_dir / voice_id / "metadata.json"
if not meta_path.exists():
return None
try:
with open(meta_path) as f:
meta = json.load(f)
return meta.get("embedding_url")
except Exception:
return None
# ------------------------------------------------------------------
# Speech generation
# ------------------------------------------------------------------
def generate_speech(self, text: str, voice: str = 'Vivian', **kwargs) -> bytes:
"""
Generate speech via fal.ai Qwen3-TTS.
Args:
text: Text to synthesize.
voice: Built-in voice name OR cloned voice_id (clone_xxx).
**kwargs:
language: Language name (default 'English').
prompt: Style/emotion instruction for 1.7B model.
speaker_embedding_url: Direct embedding URL override.
reference_text: Reference text for cloned voice quality.
model: '0.6b' or '1.7b' (default '1.7b').
Returns:
MP3 audio bytes.
"""
if not self.api_key:
raise RuntimeError("FAL_KEY not set — cannot call fal.ai API")
self.validate_text(text)
language = kwargs.get('language', 'English')
prompt = kwargs.get('prompt', '')
embedding_url = kwargs.get('speaker_embedding_url')
reference_text = kwargs.get('reference_text', '')
model = kwargs.get('model', '1.7b')
endpoint = FAL_TTS_1_7B if model == '1.7b' else FAL_TTS_0_6B
# Resolve cloned voice → embedding URL
is_cloned = voice.startswith("clone_") if voice else False
if is_cloned and not embedding_url:
embedding_url = self.get_clone_embedding_url(voice)
if not embedding_url:
raise RuntimeError(
f"Cloned voice '{voice}' not found or missing embedding"
)
# Load reference_text from metadata if not provided
if not reference_text:
clones_dir = _get_clones_dir()
meta_path = clones_dir / voice / "metadata.json"
if meta_path.exists():
try:
with open(meta_path) as f:
meta = json.load(f)
reference_text = meta.get("reference_text", "")
except Exception:
pass
# ── Comic timing: add pauses at sentence boundaries ──
# Ellipsis after sentence-ending punctuation creates natural pauses
# in qwen3 TTS output without needing SSML.
comic_timing = kwargs.get('comic_timing', True) if is_cloned else False
if comic_timing:
text = re.sub(r'([.!?])\s+', r'\1 ... ', text)
# Scale max_new_tokens based on text length — default 200 truncates long responses.
# ~4 tokens per character is a safe ratio for speech codec tokens.
estimated_tokens = max(2000, len(text) * 4)
max_tokens = min(estimated_tokens, 8192)
payload = {
"text": text,
"language": language,
"max_new_tokens": max_tokens,
}
if embedding_url:
# Cloned voice — use embedding, skip built-in voice
payload["speaker_voice_embedding_file_url"] = embedding_url
if reference_text:
payload["reference_text"] = reference_text
if prompt:
payload["prompt"] = prompt
else:
# Built-in voice
if voice not in BUILTIN_VOICES:
logger.warning(f"Unknown voice '{voice}', falling back to Vivian")
voice = 'Vivian'
payload["voice"] = voice
if prompt:
payload["prompt"] = prompt
t = time.time()
voice_label = voice if not is_cloned else f"{voice} (cloned)"
logger.info(f"[Qwen3] TTS: '{text[:60]}...' voice={voice_label}")
try:
result = _fal_request(self.api_key, endpoint, payload)
except httpx.HTTPStatusError as e:
raise RuntimeError(
f"fal.ai API error {e.response.status_code}: {e.response.text}"
)
except Exception as e:
raise RuntimeError(f"fal.ai request failed: {e}")
audio_url = result.get('audio', {}).get('url')
if not audio_url:
raise RuntimeError(f"No audio URL in fal.ai response: {result}")
audio_bytes = _fal_download(audio_url)
# ── Post-processing: speed adjustment via ffmpeg ──
# Read speed from active profile if not passed directly
speed = kwargs.get('speed', None)
if speed is None:
speed = self._get_profile_speed()
if speed and speed != 1.0 and 0.5 <= speed <= 2.0:
audio_bytes = self._ffmpeg_speed(audio_bytes, speed)
elapsed = int((time.time() - t) * 1000)
logger.info(f"[Qwen3] Generated {len(audio_bytes)} bytes in {elapsed}ms speed={speed}")
return audio_bytes
@staticmethod
def _get_profile_speed():
"""Read speed from the active profile's voice settings."""
try:
from profiles.manager import get_profile_manager
from routes.profiles import _active_profile_id
mgr = get_profile_manager()
prof = mgr.get_profile(_active_profile_id)
if prof and hasattr(prof, 'voice') and prof.voice:
return prof.voice.speed
except Exception:
pass
return 1.0
@staticmethod
def _ffmpeg_speed(audio_bytes: bytes, speed: float) -> bytes:
"""Adjust audio speed using ffmpeg atempo filter."""
try:
cmd = [
'ffmpeg', '-y', '-i', 'pipe:0',
'-filter:a', f'atempo={speed}',
'-f', 'mp3', 'pipe:1'
]
proc = subprocess.run(
cmd, input=audio_bytes, capture_output=True, timeout=30
)
if proc.returncode == 0 and proc.stdout:
logger.info(
f"[Qwen3] ffmpeg speed={speed}: "
f"{len(audio_bytes)}→{len(proc.stdout)} bytes"
)
return proc.stdout
else:
logger.warning(f"[Qwen3] ffmpeg failed: {proc.stderr[:200]}")
except Exception as e:
logger.warning(f"[Qwen3] ffmpeg speed adjust error: {e}")
return audio_bytes
# ------------------------------------------------------------------
# Provider interface
# ------------------------------------------------------------------
def health_check(self) -> dict:
if not self.api_key:
return {"ok": False, "latency_ms": 0, "detail": "FAL_KEY not set"}
t = time.time()
try:
with httpx.Client(timeout=httpx.Timeout(8.0)) as client:
resp = client.get(
"https://fal.run/",
headers={"Authorization": f"Key {self.api_key}"},
)
latency_ms = int((time.time() - t) * 1000)
return {
"ok": True, "latency_ms": latency_ms,
"detail": "fal.ai reachable — Qwen3-TTS ready",
}
except Exception as e:
latency_ms = int((time.time() - t) * 1000)
return {"ok": False, "latency_ms": latency_ms, "detail": str(e)}
def list_voices(self) -> list:
voices = BUILTIN_VOICES.copy()
for clone in self.list_cloned_voices():
voices.append(clone["voice_id"])
return voices
def get_default_voice(self) -> str:
return 'Vivian'
def is_available(self) -> bool:
return bool(self.api_key)
def get_info(self) -> dict:
cloned = self.list_cloned_voices()
return {
'name': 'Qwen3-TTS (fal.ai)',
'provider_id': 'qwen3',
'status': self._status,
'description': (
'Qwen3-TTS via fal.ai — expressive, multilingual, '
'voice cloning, emotion control'
),
'quality': 'very-high',
'latency': 'fast',
'cost_per_minute': 0.003,
'voices': BUILTIN_VOICES.copy(),
'cloned_voices': [
{"voice_id": c["voice_id"], "name": c["name"]}
for c in cloned
],
'features': [
'multilingual', 'expressive', 'voice-cloning',
'emotion-control', 'cloud', 'mp3-output',
],
'requires_api_key': True,
'languages': ['en', 'zh', 'es', 'fr', 'de', 'it', 'ja', 'ko', 'pt', 'ru'],
'max_characters': 5000,
'notes': 'Qwen3-TTS 1.7B + 0.6B. Voice cloning via clone-voice endpoint. FAL_KEY required.',
'default_voice': 'Vivian',
'audio_format': 'mp3',
'sample_rate': 24000,
'capabilities': self.clone_capabilities(),
'error': self._init_error,
}