"""Provider-specific Bailian adapters. Credentials and endpoints are resolved only from the execution environment. They never enter agent versions, MQ messages, reports, or source control. """ from __future__ import annotations import base64 import contextlib import http.client import io import json import os import queue import threading import time import uuid import wave from collections.abc import Iterator, Mapping from typing import Any, cast from urllib.parse import urlsplit, urlunsplit from .ai_runtime import ( AIConfigError, AIProviderError, StreamChunk, normalize_agent_config, pcm16_to_alaw, resample_pcm16, validate_wav, ) class BailianLLM: """DashScope OpenAI-compatible streaming Chat Completions adapter.""" def __init__(self, base_url: str, api_key: str, timeout_s: float = 45.0) -> None: parsed = urlsplit(base_url) if parsed.scheme != "https" or not parsed.netloc: raise AIConfigError( "BAILIAN_BASE_URL_INVALID", "BAILIAN_BASE_URL must be an HTTPS URL" ) if not api_key: raise AIConfigError( "BAILIAN_API_KEY_MISSING", "BAILIAN_API_KEY is not configured" ) self.base_url = base_url.rstrip("/") self.api_key = api_key self.timeout_s = max(1.0, min(120.0, timeout_s)) @classmethod def from_env(cls) -> BailianLLM: return cls( os.environ.get("BAILIAN_BASE_URL", ""), os.environ.get("BAILIAN_API_KEY", ""), ) def stream( self, messages: list[dict[str, str]], config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: llm = config["llm"] payload: dict[str, Any] = { "model": llm["model"], "messages": messages, "stream": True, "stream_options": {"include_usage": True}, } for key in ("temperature", "max_tokens"): if key in llm: payload[key] = llm[key] parsed = urlsplit(self.base_url) if parsed.hostname is None: raise AIConfigError( "BAILIAN_BASE_URL_INVALID", "BAILIAN_BASE_URL host is invalid" ) path = parsed.path.rstrip("/") + "/chat/completions" if parsed.query: path += "?" + parsed.query connection = http.client.HTTPSConnection( parsed.hostname, parsed.port or 443, timeout=self.timeout_s, ) model: str | None = None sequence = 0 try: connection.request( "POST", path, body=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "Accept": "text/event-stream", }, ) response = connection.getresponse() if response.status >= 400: raise AIProviderError( f"LLM_HTTP_{response.status}", f"Bailian Chat API returned HTTP {response.status}", retryable=response.status == 429 or response.status >= 500, ) for raw_line in response: if cancelled.is_set(): return line = raw_line.decode("utf-8", "replace").strip() if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": break try: event = json.loads(data) except json.JSONDecodeError as exc: raise AIProviderError( "LLM_STREAM_INVALID", "Bailian returned invalid SSE JSON" ) from exc if isinstance(event.get("model"), str): model = event["model"] if isinstance(event.get("error"), Mapping): error = event["error"] raise AIProviderError( "LLM_PROVIDER_ERROR", str(error.get("message", "Bailian returned an error")), retryable=False, ) for choice in event.get("choices", []): delta = choice.get("delta") or {} content = delta.get("content") if isinstance(content, str) and content: yield StreamChunk( sequence=sequence, text=content, provider_model=model, ) sequence += 1 yield StreamChunk(sequence=sequence, done=True, provider_model=model) except AIProviderError: raise except TimeoutError as exc: raise AIProviderError( "LLM_TIMEOUT", "Bailian Chat API timed out", True ) from exc except (http.client.HTTPException, OSError) as exc: raise AIProviderError( "LLM_CONNECT_FAILED", "Bailian Chat API connection failed", True ) from exc finally: connection.close() class BailianASR: """Bailian realtime ASR adapter using the existing run-task protocol.""" def __init__( self, websocket_base_url: str, api_key: str, timeout_s: float = 45.0 ) -> None: parsed = urlsplit(websocket_base_url) if parsed.scheme != "wss" or not parsed.netloc: raise AIConfigError( "BAILIAN_WSS_BASE_URL_INVALID", "BAILIAN_WSS_BASE_URL must be a WSS URL" ) if not api_key: raise AIConfigError( "BAILIAN_API_KEY_MISSING", "BAILIAN_API_KEY is not configured" ) self.websocket_base_url = websocket_base_url.rstrip("/") self.api_key = api_key self.timeout_s = max(1.0, min(120.0, timeout_s)) @classmethod def from_env(cls) -> BailianASR: return cls( os.environ.get("BAILIAN_WSS_BASE_URL", ""), os.environ.get("BAILIAN_API_KEY", ""), ) def transcribe( self, audio: bytes, config: Mapping[str, Any], turn_id: str ) -> Iterator[dict[str, Any]]: try: metadata = validate_wav(audio) with wave.open(io.BytesIO(audio), "rb") as wav: raw_audio = wav.readframes(wav.getnframes()) input_rate = metadata["sample_rate_hz"] raw_audio = resample_pcm16(raw_audio, input_rate, 16000) timeout_s = min( self.timeout_s, max(1.0, int(config["asr"].get("timeout_ms", 45000)) / 1000), ) model = str(config["asr"].get("model", "fun-asr-realtime")) except (AIConfigError, KeyError, TypeError, ValueError, OverflowError) as exc: if isinstance(exc, AIConfigError): raise raise AIProviderError( "ASR_CONFIG_INVALID", "Bailian ASR configuration is invalid" ) from exc try: import websocket except ImportError as exc: raise AIProviderError( "BAILIAN_ASR_SDK_MISSING", "websocket-client is required for Bailian ASR", ) from exc connection: Any | None = None task_id = f"task_{uuid.uuid4().hex}" sequence = 0 try: connection = websocket.create_connection( self.websocket_base_url, header=["Authorization: Bearer " + self.api_key], timeout=timeout_s, ) connection.send( json.dumps( { "header": { "action": "run-task", "task_id": task_id, "streaming": "duplex", }, "payload": { "task_group": "audio", "task": "asr", "function": "recognition", "model": model, "parameters": {"format": "pcm", "sample_rate": 16000}, "input": {}, }, }, ensure_ascii=False, ) ) started = self._recv_json(connection) header = started.get("header") or {} if header.get("event") != "task-started": raise AIProviderError( "ASR_START_FAILED", str( header.get("error_message", "Bailian ASR task failed to start") ), ) for offset in range(0, len(raw_audio), 3200): connection.send( raw_audio[offset : offset + 3200], opcode=websocket.ABNF.OPCODE_BINARY, ) connection.send( json.dumps( { "header": { "action": "finish-task", "task_id": task_id, "streaming": "duplex", }, "payload": {"input": {}}, } ) ) while True: event = self._recv_json(connection) header = event.get("header") or {} event_type = header.get("event") if event_type == "result-generated": sentence = ((event.get("payload") or {}).get("output") or {}).get( "sentence" ) or {} text = sentence.get("text") if isinstance(text, str) and text: is_final = bool(sentence.get("sentence_end")) yield { "turn_id": turn_id, "sequence": sequence, "text": text, "is_final": is_final, } sequence += 1 elif event_type == "task-failed": raise AIProviderError( "ASR_PROVIDER_ERROR", str(header.get("error_message", "Bailian ASR task failed")), ) elif event_type == "task-finished": break except AIProviderError: raise except TimeoutError as exc: raise AIProviderError("ASR_TIMEOUT", "Bailian ASR timed out", True) from exc except Exception as exc: raise AIProviderError( "ASR_CONNECTION_FAILED", "Bailian ASR connection failed", True ) from exc finally: if connection is not None: with contextlib.suppress(Exception): connection.close() @staticmethod def _recv_json(connection: Any) -> dict[str, Any]: try: message = connection.recv() except Exception as exc: raise AIProviderError( "ASR_CONNECTION_FAILED", "Bailian ASR connection failed", True ) from exc if isinstance(message, bytes): raise AIProviderError( "ASR_PROTOCOL_INVALID", "Bailian ASR returned unexpected binary data" ) try: event = json.loads(message) except (TypeError, json.JSONDecodeError) as exc: raise AIProviderError( "ASR_PROTOCOL_INVALID", "Bailian ASR returned invalid JSON" ) from exc if not isinstance(event, dict): raise AIProviderError( "ASR_PROTOCOL_INVALID", "Bailian ASR returned an invalid event" ) return event class _BailianTTSCallback: def __init__(self) -> None: self.events: queue.Queue[dict[str, Any]] = queue.Queue() def on_open(self) -> None: self.events.put({"type": "__open__"}) def on_close(self, close_status_code: Any, close_msg: Any) -> None: self.events.put( { "type": "__closed__", "code": close_status_code, "message": str(close_msg or ""), } ) def on_event(self, message: dict[str, Any]) -> None: if isinstance(message, dict): self.events.put(message) class _BailianCosyVoiceCallback: def __init__(self, max_events: int) -> None: self.events: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=max_events) self.provider_model: str | None = None self.overflowed = False def _put(self, event: dict[str, Any]) -> None: try: self.events.put_nowait(event) except queue.Full: self.overflowed = True def on_open(self) -> None: self._put({"type": "__open__"}) def on_complete(self) -> None: self._put({"type": "__complete__"}) def on_error(self, message: Any) -> None: self._put({"type": "__error__", "message": str(message)}) def on_close(self) -> None: self._put({"type": "__closed__"}) def on_event(self, message: str) -> None: try: event = json.loads(message) except (TypeError, json.JSONDecodeError): self._put({"type": "__event_invalid__"}) return if isinstance(event, dict): header = event.get("header") or {} model = header.get("model") or event.get("model") if isinstance(model, str) and model: self.provider_model = model self._put({"type": "__event__", "event": event}) def on_data(self, data: bytes) -> None: self._put({"type": "__audio__", "audio": bytes(data)}) class BailianTTS: """Bailian streaming TTS for Qwen realtime and CosyVoice models.""" def __init__( self, websocket_base_url: str, api_key: str, timeout_s: float = 45.0, ) -> None: parsed = urlsplit(websocket_base_url) if parsed.scheme != "wss" or not parsed.netloc: raise AIConfigError( "BAILIAN_WSS_BASE_URL_INVALID", "BAILIAN_WSS_BASE_URL must be a WSS URL", ) if not api_key: raise AIConfigError( "BAILIAN_API_KEY_MISSING", "BAILIAN_API_KEY is not configured" ) self.websocket_base_url = websocket_base_url.rstrip("/") self.api_key = api_key self.timeout_s = max(1.0, min(120.0, timeout_s)) @classmethod def from_env(cls) -> BailianTTS: return cls( os.environ.get("BAILIAN_TTS_WSS_BASE_URL") or os.environ.get("BAILIAN_WSS_BASE_URL", ""), os.environ.get("BAILIAN_API_KEY", ""), ) def stream( self, text: str, config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: tts = config["tts"] model = str(tts["model"]) if model.startswith("cosyvoice-"): yield from self._stream_cosyvoice(text, config, turn_id, cancelled) return try: import dashscope from dashscope.audio.qwen_tts_realtime import AudioFormat, QwenTtsRealtime except ImportError as exc: raise AIProviderError( "BAILIAN_TTS_SDK_MISSING", "dashscope is required for Bailian Qwen-TTS-Realtime", ) from exc dashscope.api_key = self.api_key callback = _BailianTTSCallback() connection: Any | None = None response_done = False finish_sent = False sequence = 0 returned_model: str | None = None try: configured_timeout_s = int(tts.get("timeout_ms", 45000)) / 1000 except (TypeError, ValueError, OverflowError) as exc: raise AIProviderError( "TTS_CONFIG_INVALID", "TTS timeout is invalid" ) from exc timeout_s = min(self.timeout_s, max(1.0, configured_timeout_s)) try: connection = QwenTtsRealtime( model=model, callback=cast(Any, callback), url=self._url_for_qwen_realtime(), headers={"X-DashScope-DataInspection": "disable"}, ) connection.connect() connection.update_session( voice=str(tts["voice"]), response_format=AudioFormat.PCM_24000HZ_MONO_16BIT, mode="commit", sample_rate=24000, language_type="Chinese" if config["asr"].get("language", "").startswith("zh") else "Auto", ) if not self._wait_for_event(callback.events, "session.updated", timeout_s): raise AIProviderError( "TTS_SESSION_TIMEOUT", "Bailian TTS session was not acknowledged", True, ) if cancelled.is_set(): connection.finish() return connection.append_text(text) connection.commit() while True: try: event = callback.events.get(timeout=timeout_s) except queue.Empty as exc: raise AIProviderError( "TTS_TIMEOUT", "Bailian TTS response timed out", True ) from exc event_type = event.get("type") if event_type == "session.created" or event_type == "session.updated": returned_model = (event.get("session") or {}).get( "model" ) or returned_model elif event_type == "response.created": returned_model = (event.get("response") or {}).get( "model" ) or returned_model elif event_type == "error": error = event.get("error") or {} raise AIProviderError( "TTS_PROVIDER_ERROR", str(error.get("message", "Bailian TTS returned an error")), ) elif event_type == "response.audio.delta": try: audio = base64.b64decode(event.get("delta", ""), validate=True) except (ValueError, TypeError) as exc: raise AIProviderError( "TTS_AUDIO_INVALID", "Bailian returned invalid audio data" ) from exc yield StreamChunk( sequence=sequence, audio=audio, provider_model=returned_model, ) sequence += 1 elif event_type == "response.done": response_done = True if not finish_sent: connection.finish() finish_sent = True elif event_type == "session.finished": break elif event_type == "__closed__": if response_done: break raise AIProviderError( "TTS_CONNECTION_CLOSED", "Bailian TTS connection closed early", True, ) if cancelled.is_set(): with contextlib.suppress(Exception): connection.cancel_response() if not finish_sent: connection.finish() finish_sent = True return yield StreamChunk( sequence=sequence, done=True, provider_model=returned_model ) except AIProviderError: raise except TimeoutError as exc: raise AIProviderError( "TTS_CONNECT_TIMEOUT", "Bailian TTS connection timed out", True ) from exc except Exception as exc: raise AIProviderError( "TTS_CONNECTION_FAILED", "Bailian TTS connection failed", True ) from exc finally: if connection is not None: with contextlib.suppress(Exception): connection.close() def _url_for_qwen_realtime(self) -> str: parsed = urlsplit(self.websocket_base_url) path = parsed.path.rstrip("/") if path.endswith("/inference"): path = path[: -len("/inference")] + "/realtime" elif not path: path = "/api-ws/v1/realtime" return urlunsplit((parsed.scheme, parsed.netloc, path, "", "")) def _stream_cosyvoice( self, text: str, config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: del turn_id try: import dashscope from dashscope.audio.tts_v2 import AudioFormat, SpeechSynthesizer except ImportError as exc: raise AIProviderError( "BAILIAN_TTS_SDK_MISSING", "dashscope is required for Bailian CosyVoice", ) from exc tts = config["tts"] try: timeout_s = min( self.timeout_s, max(1.0, int(tts.get("timeout_ms", 45000)) / 1000), ) sample_rate = int(tts["format"]["sample_rate_hz"]) encoding = str(tts["format"]["encoding"]) speed = float(tts.get("speed", 1.0)) except (KeyError, TypeError, ValueError, OverflowError) as exc: raise AIProviderError( "TTS_CONFIG_INVALID", "TTS configuration is invalid" ) from exc if encoding not in {"pcm_s16le", "pcma"}: raise AIProviderError( "TTS_FORMAT_INVALID", "TTS output encoding is unsupported" ) format_name = f"PCM_{sample_rate}HZ_MONO_16BIT" output_format = getattr(AudioFormat, format_name, None) if output_format is None: raise AIProviderError( "TTS_FORMAT_INVALID", "CosyVoice does not support the requested sample rate", ) dashscope.api_key = self.api_key try: max_events = int(config["conversation"]["max_pending_audio_chunks"]) except (KeyError, TypeError, ValueError, OverflowError) as exc: raise AIProviderError( "TTS_CONFIG_INVALID", "audio queue limit is invalid" ) from exc callback = _BailianCosyVoiceCallback(max(16, min(256, max_events * 4))) synthesizer: Any | None = None sequence = 0 response_done = False deadline = time.monotonic() + timeout_s try: synthesizer = SpeechSynthesizer( model=str(tts["model"]), voice=str(tts["voice"]), format=output_format, speech_rate=speed, url=self.websocket_base_url, headers={"X-DashScope-DataInspection": "disable"}, callback=cast(Any, callback), ) synthesizer.streaming_call(text) synthesizer.async_streaming_complete(int(timeout_s * 1000)) while True: if cancelled.is_set(): with contextlib.suppress(Exception): synthesizer.streaming_cancel() return if callback.overflowed: raise AIProviderError( "TTS_BACKPRESSURE", "Bailian TTS callback queue exceeded its bound", ) remaining = deadline - time.monotonic() if remaining <= 0: raise AIProviderError( "TTS_TIMEOUT", "Bailian TTS response timed out", True ) try: event = callback.events.get(timeout=min(0.25, remaining)) except queue.Empty: continue event_type = event.get("type") if event_type == "__audio__": raw_audio = event.get("audio", b"") if ( not isinstance(raw_audio, bytes) or not raw_audio or len(raw_audio) % 2 ): raise AIProviderError( "TTS_AUDIO_INVALID", "Bailian returned invalid PCM audio" ) audio = ( pcm16_to_alaw(raw_audio, sample_rate, sample_rate) if encoding == "pcma" else raw_audio ) yield StreamChunk( sequence=sequence, audio=audio, late=cancelled.is_set(), provider_model=callback.provider_model, ) sequence += 1 elif event_type == "__error__": raise AIProviderError( "TTS_PROVIDER_ERROR", event.get("message", "Bailian TTS failed") ) elif event_type == "__event_invalid__": raise AIProviderError( "TTS_EVENT_INVALID", "Bailian returned invalid TTS metadata" ) elif event_type == "__event__": provider_event = event.get("event") or {} header = provider_event.get("header") or {} if header.get("event") == "task-failed": raise AIProviderError( "TTS_PROVIDER_ERROR", str(header.get("error_message", "Bailian TTS task failed")), ) if header.get("event") == "task-finished": response_done = True elif event_type == "__complete__": response_done = True elif event_type == "__closed__": if response_done: break raise AIProviderError( "TTS_CONNECTION_CLOSED", "Bailian TTS connection closed early", True, ) if response_done: break yield StreamChunk( sequence=sequence, done=True, provider_model=callback.provider_model, ) except AIProviderError: raise except TimeoutError as exc: raise AIProviderError( "TTS_CONNECT_TIMEOUT", "Bailian TTS connection timed out", True ) from exc except Exception as exc: raise AIProviderError( "TTS_CONNECTION_FAILED", "Bailian TTS connection failed", True ) from exc finally: if synthesizer is not None: with contextlib.suppress(Exception): synthesizer.close() @staticmethod def _wait_for_event( events: queue.Queue[dict[str, Any]], expected: str, timeout_s: float ) -> bool: try: while True: event = events.get(timeout=timeout_s) event_type = event.get("type") if event_type == expected: return True if event_type == "error": error = event.get("error") or {} raise AIProviderError( "TTS_SESSION_ERROR", str(error.get("message", "Bailian TTS session failed")), ) if event_type == "__closed__": return False except queue.Empty: return False def build_bailian_config( version_id: str, prompt: str, model: str = "qwen-plus", tts_model: str | None = None, voice: str | None = None, language: str = "zh-CN", asr_model: str | None = None, opening: str = "", max_turns: int = 20, max_duration_ms: int = 120000, ) -> dict[str, Any]: resolved_voice = voice or os.environ.get("BAILIAN_TTS_VOICE", "") if not resolved_voice: raise AIConfigError( "BAILIAN_TTS_VOICE_MISSING", "BAILIAN_TTS_VOICE is not configured" ) resolved_model = tts_model or os.environ.get("BAILIAN_TTS_MODEL", "") resolved_asr_model = asr_model or os.environ.get( "BAILIAN_ASR_MODEL", "fun-asr-realtime" ) if not resolved_model: resolved_model = ( "cosyvoice-v3.5-plus" if resolved_voice.startswith("cosyvoice-") else "qwen3-tts-flash-realtime" ) return normalize_agent_config( { "agent_version_id": version_id, "immutable": True, "llm": { "provider_ref": "bailian", "credential_ref": "env:BAILIAN_API_KEY", "model": model, "temperature": 0.2, "max_tokens": 256, "timeout_ms": 45000, }, "prompt": {"text": prompt, "allowed_variables": []}, "tts": { "provider_ref": "bailian", "credential_ref": "env:BAILIAN_API_KEY", "model": resolved_model, "voice": resolved_voice, "speed": 1.0, "format": { "encoding": "pcm_s16le", "sample_rate_hz": 24000, "channels": 1, }, "timeout_ms": 45000, }, "asr": { "provider_ref": "bailian", "credential_ref": "env:BAILIAN_API_KEY", "model": resolved_asr_model, "language": language, "input": { "encoding": "pcm_s16le", "sample_rate_hz": 16000, "channels": 1, "sample_width_bytes": 2, }, "interim": True, "timeout_ms": 5000, }, "conversation": { "opening": opening, "allow_interrupt": True, "silence_timeout_ms": 3000, "max_duration_ms": max_duration_ms, "max_turns": max_turns, "sentence_max_chars": 80, "max_pending_audio_chunks": 32, }, } ) __all__ = ["BailianASR", "BailianLLM", "BailianTTS", "build_bailian_config"]