from __future__ import annotations from pathlib import Path from typing import Any, Dict, Iterator, List, Optional import requests from agent.tts_provider import TTSProvider from tools.tool_backend_helpers import resolve_openai_audio_api_key from tools.tts_streaming import StreamingTTSProvider, register as register_streamer from tools.tts_tool import _load_tts_config NAME = "qwen3-stream" SAMPLE_RATE = 24000 def _settings() -> Dict[str, Any]: config = _load_tts_config() own = dict(config.get(NAME) or {}) fallback = dict(config.get("openai") or {}) own.setdefault("base_url", fallback.get("base_url", "")) own.setdefault("api_key", fallback.get("api_key", "")) own.setdefault("model", fallback.get("model", "tts-1")) own.setdefault("voice", fallback.get("voice", "alloy")) own.setdefault("language", fallback.get("language", "German")) own.setdefault("chunk_size", 4) return own def _url(path: str, section: Optional[Dict[str, Any]] = None) -> str: cfg = section or _settings() base = str(cfg.get("base_url") or "").rstrip("/") if not base: raise RuntimeError("tts.qwen3-stream.base_url is not configured") if not base.endswith("/v1"): base += "/v1" return base + path def _headers(section: Optional[Dict[str, Any]] = None) -> Dict[str, str]: cfg = section or _settings() key = str(cfg.get("api_key") or resolve_openai_audio_api_key() or "").strip() headers = {"Accept": "application/octet-stream"} if key: headers["Authorization"] = f"Bearer {key}" return headers def _payload(text: str, section: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: cfg = section or _settings() payload: Dict[str, Any] = { "input": text, "model": cfg.get("model") or "tts-1", "voice": cfg.get("voice") or "alloy", "language": cfg.get("language") or "German", } instruct = str(cfg.get("instruct") or "").strip() if instruct: payload["instruct"] = instruct return payload class Qwen3PCMStreamer(StreamingTTSProvider): sample_rate = SAMPLE_RATE channels = 1 sample_width = 2 @staticmethod def available() -> bool: try: return bool(_settings().get("base_url")) except Exception: return False def stream(self, text: str) -> Iterator[bytes]: cfg = dict(_settings()) cfg.update(self.section or {}) payload = _payload(text, cfg) payload["chunk_size"] = max(1, int(cfg.get("chunk_size", 4))) with requests.post( _url("/audio/speech/pcm-stream", cfg), json=payload, headers=_headers(cfg), stream=True, timeout=(5, 120), ) as response: response.raise_for_status() pending = b"" # An explicit read size prevents urllib3 from buffering the # unknown-length response until connection close. for chunk in response.iter_content(chunk_size=4096): if not chunk: continue data = pending + chunk even = len(data) & ~1 if even: yield data[:even] pending = data[even:] class Qwen3TTSProvider(TTSProvider): @property def name(self) -> str: return NAME @property def display_name(self) -> str: return "Athena Qwen3-TTS Streaming" def is_available(self) -> bool: return Qwen3PCMStreamer.available() def list_voices(self) -> List[Dict[str, Any]]: voice = str(_settings().get("voice") or "alloy") return [{"id": voice, "display": voice, "language": "de"}] def synthesize( self, text: str, output_path: str, *, voice: Optional[str] = None, model: Optional[str] = None, speed: Optional[float] = None, format: str = "mp3", **extra: Any, ) -> str: cfg = _settings() payload = _payload(text, cfg) payload["response_format"] = format if voice: payload["voice"] = voice if model: payload["model"] = model if speed is not None: payload["speed"] = speed response = requests.post( _url("/audio/speech", cfg), json=payload, headers=_headers(cfg), timeout=(5, 120), ) response.raise_for_status() Path(output_path).write_bytes(response.content) return output_path register_streamer(NAME)(Qwen3PCMStreamer) def register(ctx) -> None: ctx.register_tts_provider(Qwen3TTSProvider())