827 lines
31 KiB
Python
827 lines
31 KiB
Python
"""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"]
|