From e667d2597f2baabf4a5ca4f59b23e43b595224b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Tue, 10 Nov 2026 16:23:47 +0800 Subject: [PATCH 1/2] feat(audio): add Qwen Live sessions on shared realtime transport Preserve Live and Omni error contracts while providing bounded standard WebSocket closure and explicit cancellation, with examples and regression coverage. --- dashscope/resources/audio/__init__.py | 71 +- dashscope/resources/audio/_ws_session.py | 278 +++ dashscope/resources/audio/live.py | 719 +++++++ dashscope/resources/audio/realtime.py | 255 +-- .../audio/realtime_live.py | 120 ++ tests/test_audio_live_session.py | 1880 +++++++++++++++++ tests/test_audio_samples_direct_call.py | 5 +- tests/test_audio_ws_integration.py | 62 + tests/test_audio_ws_session.py | 554 +++++ 9 files changed, 3762 insertions(+), 182 deletions(-) create mode 100644 dashscope/resources/audio/_ws_session.py create mode 100644 dashscope/resources/audio/live.py create mode 100644 examples/01_atomic_capabilities/audio/realtime_live.py create mode 100644 tests/test_audio_live_session.py create mode 100644 tests/test_audio_ws_session.py diff --git a/dashscope/resources/audio/__init__.py b/dashscope/resources/audio/__init__.py index 71a972a2..55a2fb05 100644 --- a/dashscope/resources/audio/__init__.py +++ b/dashscope/resources/audio/__init__.py @@ -6,8 +6,12 @@ here is two levels deep at most: ``client.audio.asr.transcribe``, ``client.audio.asr.recognize``, ``client.audio.tts.synthesize``, ``client.audio.tts.http.synthesize``, ``client.audio.realtime.connect``, -and the management plane ``client.audio.vocabularies`` / ``phrases`` / -``voices`` / ``voice_design``. +``client.audio.live.connect``, and the management plane +``client.audio.vocabularies`` / ``phrases`` / ``voices`` / +``voice_design``. + +The Qwen-Live wire event names (``EVENT_*``) are re-exported here too, so +callers can import them from the same package as ``LiveSession``. """ from dashscope.resources._blocking import run_blocking @@ -34,6 +38,38 @@ HttpSynthesisStream, HttpTts, ) +from dashscope.resources.audio.live import ( + EVENT_ERROR, + EVENT_RESPONSE_CREATE, + EVENT_RESPONSE_EVENT, + EVENT_RESPONSE_ITEM_CREATE, + EVENT_SESSION_CLOSE, + EVENT_SESSION_CLOSED, + EVENT_SESSION_COMMENTARY_APPEND, + EVENT_SESSION_COMMENTARY_APPENDED, + EVENT_SESSION_CREATED, + EVENT_SESSION_DELEGATION_CREATED, + EVENT_SESSION_EVENT, + EVENT_SESSION_INPUT_AUDIO_APPEND, + EVENT_SESSION_INPUT_AUDIO_MUTE, + EVENT_SESSION_INPUT_AUDIO_MUTED, + EVENT_SESSION_INPUT_AUDIO_UNMUTE, + EVENT_SESSION_INPUT_AUDIO_UNMUTED, + EVENT_SESSION_INPUT_TRANSCRIPT_DELTA, + EVENT_SESSION_INSTRUCTION_APPEND, + EVENT_SESSION_INSTRUCTIONS_APPENDED, + EVENT_SESSION_ITEM_CREATE, + EVENT_SESSION_OUTPUT_AUDIO_DELTA, + EVENT_SESSION_OUTPUT_AUDIO_INTERRUPTED, + EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA, + EVENT_SESSION_THINKING_APPEND, + EVENT_SESSION_THINKING_APPENDED, + EVENT_SESSION_UPDATE, + EVENT_SESSION_UPDATED, + EVENT_SESSION_USAGE_UPDATED, + Live, + LiveSession, +) from dashscope.resources.audio.phrases import ( Phrase, PhraseList, @@ -50,10 +86,40 @@ "Audio", "DesignedVoice", "DesignedVoiceList", + "EVENT_ERROR", + "EVENT_RESPONSE_CREATE", + "EVENT_RESPONSE_EVENT", + "EVENT_RESPONSE_ITEM_CREATE", + "EVENT_SESSION_CLOSE", + "EVENT_SESSION_CLOSED", + "EVENT_SESSION_COMMENTARY_APPEND", + "EVENT_SESSION_COMMENTARY_APPENDED", + "EVENT_SESSION_CREATED", + "EVENT_SESSION_DELEGATION_CREATED", + "EVENT_SESSION_EVENT", + "EVENT_SESSION_INPUT_AUDIO_APPEND", + "EVENT_SESSION_INPUT_AUDIO_MUTE", + "EVENT_SESSION_INPUT_AUDIO_MUTED", + "EVENT_SESSION_INPUT_AUDIO_UNMUTE", + "EVENT_SESSION_INPUT_AUDIO_UNMUTED", + "EVENT_SESSION_INPUT_TRANSCRIPT_DELTA", + "EVENT_SESSION_INSTRUCTION_APPEND", + "EVENT_SESSION_INSTRUCTIONS_APPENDED", + "EVENT_SESSION_ITEM_CREATE", + "EVENT_SESSION_OUTPUT_AUDIO_DELTA", + "EVENT_SESSION_OUTPUT_AUDIO_INTERRUPTED", + "EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA", + "EVENT_SESSION_THINKING_APPEND", + "EVENT_SESSION_THINKING_APPENDED", + "EVENT_SESSION_UPDATE", + "EVENT_SESSION_UPDATED", + "EVENT_SESSION_USAGE_UPDATED", "HttpSynthesis", "HttpSynthesisChunk", "HttpSynthesisStream", "HttpTts", + "Live", + "LiveSession", "Phrase", "PhraseList", "Phrases", @@ -110,6 +176,7 @@ def __init__(self, client) -> None: self.asr = Asr(client) self.tts = Tts(client) self.realtime = Realtime(client) + self.live = Live(client) self.vocabularies = Vocabularies(client) self.phrases = Phrases(client) self.voices = Voices(client) diff --git a/dashscope/resources/audio/_ws_session.py b/dashscope/resources/audio/_ws_session.py new file mode 100644 index 00000000..6776d53c --- /dev/null +++ b/dashscope/resources/audio/_ws_session.py @@ -0,0 +1,278 @@ +# -*- coding: utf-8 -*- +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Raw WebSocket transport shared by Live and Omni realtime sessions. + +Session payloads, terminal events and error policy belong to the callers. +This layer only assembles connections, exchanges raw frames and releases I/O. +""" + +from __future__ import annotations + +import base64 +import socket +import threading +from typing import Any, Mapping, Optional, TypeVar, Union +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from dashscope._config import WS_REALTIME_PATH, workspace_ws_url +from dashscope._logging import request_id_from_headers +from dashscope.transport import default_user_agent +from dashscope.transport._ident import sdk_headers +from dashscope.transport.ws import ( + Connector, + _close_websocket, + _default_connector, +) + +__all__ = ["WsSessionBase"] + +MODULE = "audio" + +_UNSET = object() +_CLOSE_TIMEOUT = 1.0 + +_SelfT = TypeVar("_SelfT", bound="WsSessionBase") + + +def _as_base64(data: Union[str, bytes]) -> str: + if isinstance(data, str): + return data + if isinstance(data, bytes): + return base64.b64encode(data).decode("ascii") + raise TypeError("expected a base64 string or bytes") + + +class WsSessionBase: + """One socket per session; subclasses own all protocol state.""" + + _LABEL = "realtime" + + def __init__( + self, + client, + *, + model: str, + ws_url: Optional[str] = None, + connector: Optional[Connector] = None, + timeout: Optional[float] = None, + headers: Optional[Mapping[str, str]] = None, + api_key: Optional[str] = None, + workspace: Optional[str] = None, + ) -> None: + config = client.config + if timeout is not None and timeout <= 0: + raise ValueError("timeout must be positive") + # Keep the derived gateway and workspace header consistent. + space = workspace or config.workspace + url = ws_url or workspace_ws_url( + space, + WS_REALTIME_PATH, + config.region, + ) + parts = urlsplit(url) + query = parse_qsl(parts.query, keep_blank_values=True) + if not any(key == "model" for key, _ in query): + query.append(("model", model)) + url = urlunsplit(parts._replace(query=urlencode(query))) + # Inherit the client's agent (and any surface suffix on it) so the + # realtime socket reports the same identity as its HTTP calls; the + # dedup below still lets an explicit per-call header win. The + # lookup is defensive for the same reason as + # ``transport.ws.with_client_user_agent``. + all_headers = { + "user-agent": ( + getattr( + getattr(client, "transport", None), + "user_agent", + None, + ) + or default_user_agent() + ), + **sdk_headers(MODULE), + } + key = api_key or config.api_key + if key: + all_headers["Authorization"] = f"Bearer {key}" + if space: + all_headers["X-DashScope-WorkSpace"] = space + user_headers = dict(headers or {}) + user_keys = {key.lower() for key in user_headers} + all_headers = { + key: value + for key, value in all_headers.items() + if key.lower() not in user_keys + } + all_headers.update(user_headers) + self._model = model + self._wire_model = next( + value for key, value in query if key == "model" + ) + self._url = url + self._headers = all_headers + self._connector = connector + self._timeout = config.timeout if timeout is None else timeout + self._lock = threading.RLock() + self._send_lock = threading.RLock() + self._recv_lock = threading.Lock() + self._generation = 0 + self._ws = None + self._closing = False + self._reset_state() + + def _reset_state(self) -> None: + self._closed = False + self._header_request_id: Optional[str] = None + + @property + def model(self) -> str: + return self._model + + @property + def url(self) -> str: + return self._url + + def connect(self: _SelfT) -> _SelfT: + """Open the connection, either explicitly or on context entry.""" + with self._lock: + if self._closing: + raise RuntimeError(f"{self._LABEL} session is closing") + if self._ws is not None: + raise RuntimeError( + f"{self._LABEL} session is already open", + ) + self._generation += 1 + # Reset first so handshake context survives into failure logs. + self._reset_state() + try: + if self._connector is None: + self._ws = _default_connector( + self._url, + self._headers, + timeout=self._timeout, + ) + else: + self._ws = self._connector(self._url, self._headers) + self._header_request_id = request_id_from_headers( + getattr(self._ws, "headers", None), + ) + if self._timeout is not None: + self._ws.settimeout(self._timeout) + # pylint: disable=import-outside-toplevel + from websocket import WebSocket + + if isinstance(self._ws, WebSocket): + # Raw writes share the native lock with automatic Close + # replies; send_frame() acquires this same lock again. + self._ws.lock = threading.RLock() + except BaseException as exc: + self._on_connect_error(exc) + self.close() + self._reset_state() + raise + return self + + def _on_connect_error(self, exc: BaseException) -> None: + """Hook for subclass diagnostics before the failed socket is released.""" + + def __enter__(self: _SelfT) -> _SelfT: + return self.connect() + + def __exit__(self, *exc: Any) -> None: + self.close() + + def _send_raw(self, ws: Any, raw: Union[str, bytes]) -> None: + """Write without retries or holding the session state lock.""" + from websocket import ( # pylint: disable=import-outside-toplevel + WebSocket, + WebSocketConnectionClosedException, + ) + + with self._lock: + if ws is not self._ws: + raise WebSocketConnectionClosedException("socket is closed") + if isinstance(ws, WebSocket): + with ws.lock: + if not ws.connected: + raise WebSocketConnectionClosedException( + "socket is closed", + ) + ws.send(raw) + else: + ws.send(raw) + + def _recv_raw(self, ws: Any, generation: Optional[int] = None) -> Any: + """Skip detached sockets; read errors are left to session policy.""" + with self._lock: + if ws is not self._ws or ( + generation is not None and generation != self._generation + ): + return None + return ws.recv() + + def close(self) -> None: + """Send a normal WebSocket Close, allowing one second for the handshake.""" + self._close(abort=False) + + def _close(self, *, abort: bool) -> None: + from websocket import ( + WebSocket, + ) # pylint: disable=import-outside-toplevel + + with self._lock: + ws, self._ws = self._ws, None + if ws is not None: + self._closed = self._closing = True + if ws is None: + return + + deadline = None + sending = False + try: + if isinstance(ws, WebSocket) and ws.sock is not None: + sock = ws.sock + + def interrupt() -> None: + # close() sets connected=False before sending, so abort() + # cannot wake its blocked writer or reader at the deadline. + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + def release() -> None: + interrupt() + # Retain the descriptor until the native writer wakes; + # closing it too early can strand socket.send on macOS. + writing = ws.lock.acquire(timeout=0.1) + try: + ws.shutdown() + finally: + if writing: + ws.lock.release() + + if abort: + release() + return + deadline = threading.Timer(_CLOSE_TIMEOUT, interrupt) + deadline.daemon = True + try: + deadline.start() + except Exception: + release() + return + # Complete an in-flight data frame before sending Close. The + # reentrant lock also permits cleanup from a failed send. + # A context manager cannot bound lock-acquisition time. + # pylint: disable-next=consider-using-with + sending = self._send_lock.acquire(timeout=_CLOSE_TIMEOUT) + if not sending: + release() + return + _close_websocket(ws) + finally: + if deadline is not None: + deadline.cancel() + if sending: + self._send_lock.release() + with self._lock: + self._closing = False diff --git a/dashscope/resources/audio/live.py b/dashscope/resources/audio/live.py new file mode 100644 index 00000000..ffb1c2c3 --- /dev/null +++ b/dashscope/resources/audio/live.py @@ -0,0 +1,719 @@ +# -*- coding: utf-8 -*- +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Qwen-Live protocol over a single business WebSocket.""" + +from __future__ import annotations + +import base64 +import json +import math +import socket +import uuid +from typing import Any, Dict, Iterator, List, Mapping, Optional, Union, cast + +from dashscope.errors import ( + APIConnectionError, + APITimeoutError, + DashScopeError, + error_from_business_code, + error_from_status, +) +from dashscope.resources.audio._ws_session import ( + _UNSET, + WsSessionBase, + _as_base64, +) +from dashscope.transport.ws import Connector, WsEventError + +__all__ = [ + "EVENT_ERROR", + "EVENT_RESPONSE_CREATE", + "EVENT_RESPONSE_EVENT", + "EVENT_RESPONSE_ITEM_CREATE", + "EVENT_SESSION_CLOSE", + "EVENT_SESSION_CLOSED", + "EVENT_SESSION_COMMENTARY_APPEND", + "EVENT_SESSION_COMMENTARY_APPENDED", + "EVENT_SESSION_CREATED", + "EVENT_SESSION_DELEGATION_CREATED", + "EVENT_SESSION_EVENT", + "EVENT_SESSION_INPUT_AUDIO_APPEND", + "EVENT_SESSION_INPUT_AUDIO_MUTE", + "EVENT_SESSION_INPUT_AUDIO_MUTED", + "EVENT_SESSION_INPUT_AUDIO_UNMUTE", + "EVENT_SESSION_INPUT_AUDIO_UNMUTED", + "EVENT_SESSION_INPUT_TRANSCRIPT_DELTA", + "EVENT_SESSION_INSTRUCTION_APPEND", + "EVENT_SESSION_INSTRUCTIONS_APPENDED", + "EVENT_SESSION_ITEM_CREATE", + "EVENT_SESSION_OUTPUT_AUDIO_DELTA", + "EVENT_SESSION_OUTPUT_AUDIO_INTERRUPTED", + "EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA", + "EVENT_SESSION_THINKING_APPEND", + "EVENT_SESSION_THINKING_APPENDED", + "EVENT_SESSION_UPDATE", + "EVENT_SESSION_UPDATED", + "EVENT_SESSION_USAGE_UPDATED", + "Live", + "LiveSession", +] + +# Wire names of the client events this session sends. +EVENT_SESSION_UPDATE = "session.update" +EVENT_SESSION_INPUT_AUDIO_APPEND = "session.input_audio.append" +EVENT_SESSION_INPUT_AUDIO_MUTE = "session.input_audio.mute" +EVENT_SESSION_INPUT_AUDIO_UNMUTE = "session.input_audio.unmute" +EVENT_SESSION_INSTRUCTION_APPEND = "session.instruction.append" +EVENT_SESSION_THINKING_APPEND = "session.thinking.append" +EVENT_SESSION_COMMENTARY_APPEND = "session.commentary.append" +EVENT_SESSION_ITEM_CREATE = "session.item.create" +EVENT_RESPONSE_ITEM_CREATE = "response.item.create" +EVENT_RESPONSE_CREATE = "response.create" +EVENT_SESSION_CLOSE = "session.close" + +# Wire names of the server events this session receives. +EVENT_ERROR = "error" +EVENT_SESSION_CREATED = "session.created" +EVENT_SESSION_UPDATED = "session.updated" +EVENT_SESSION_INPUT_AUDIO_MUTED = "session.input_audio.muted" +EVENT_SESSION_INPUT_AUDIO_UNMUTED = "session.input_audio.unmuted" +EVENT_SESSION_OUTPUT_AUDIO_DELTA = "session.output_audio.delta" +EVENT_SESSION_OUTPUT_AUDIO_INTERRUPTED = "session.output_audio.interrupted" +EVENT_SESSION_INPUT_TRANSCRIPT_DELTA = "session.input_transcript.delta" +EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA = "session.output_transcript.delta" +EVENT_SESSION_EVENT = "session.event" +EVENT_RESPONSE_EVENT = "response.event" +EVENT_SESSION_USAGE_UPDATED = "session.usage.updated" +EVENT_SESSION_CLOSED = "session.closed" +EVENT_SESSION_DELEGATION_CREATED = "session.delegation.created" +EVENT_SESSION_INSTRUCTIONS_APPENDED = "session.instructions.appended" +EVENT_SESSION_THINKING_APPENDED = "session.thinking.appended" +EVENT_SESSION_COMMENTARY_APPENDED = "session.commentary.appended" + +# Typed update_session arguments; the parameters passthrough may not shadow them. +_RESERVED_SESSION_FIELDS = frozenset( + { + "model", + "instructions", + "voice", + "tts", + "input_audio_format", + "output_audio_format", + "tools", + "delegation", + "transport", + "modalities", + "context", + "turn_detection", + "input_audio_transcription", + }, +) + + +def _transport_error(exc: BaseException, operation: str) -> BaseException: + """Normalize Live failures without changing Omni's native error contract.""" + from websocket import ( # pylint: disable=import-outside-toplevel + WebSocketBadStatusException, + WebSocketException, + WebSocketTimeoutException, + ) + + if isinstance(exc, DashScopeError): + return exc + if ( + isinstance(exc, WebSocketBadStatusException) + and exc.status_code is not None + ): + body = exc.resp_body + try: + body = json.loads(body) + except (ValueError, TypeError): + pass + return error_from_status( + exc.status_code, + message=str(exc), + body=body, + headers=exc.resp_headers, + ) + if isinstance( + exc, + (TimeoutError, socket.timeout, WebSocketTimeoutException), + ): + return APITimeoutError(f"websocket {operation} timed out: {exc}") + if isinstance(exc, (OSError, WebSocketException)): + return APIConnectionError(f"websocket {operation} failed: {exc}") + return exc + + +def _ms_window( + start_ms: Optional[float], + end_ms: Optional[float], +) -> Dict[str, Any]: + if (start_ms is None) != (end_ms is None): + raise ValueError("start_ms and end_ms must be given together") + if start_ms is None or end_ms is None: + return {} + for value in (start_ms, end_ms): + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + ): + raise ValueError("start_ms and end_ms must be finite numbers") + if value < 0: + raise ValueError("start_ms and end_ms must be non-negative") + if end_ms < start_ms: + raise ValueError("end_ms must be >= start_ms") + return {"start_ms": start_ms, "end_ms": end_ms} + + +class LiveSession(WsSessionBase): + """Live session; send helpers return request IDs, not acknowledgements. + + events() yields recoverable error events; recv_event() raises them while + leaving the connection open. Fatal errors and unexpected disconnects raise + SDK exceptions in both forms. close() cancels locally, without draining. + """ + + _FINISH_EVENT = EVENT_SESSION_CLOSE + _TERMINAL_EVENT = EVENT_SESSION_CLOSED + _LABEL = "live" + + def _reset_state(self) -> None: + super()._reset_state() + self._finish_requested = False + self._finish_event_id: Optional[str] = None + self._finished = False + self._failure: Optional[DashScopeError] = None + self.session_id: Optional[str] = None + self.last_message: Optional[Dict[str, Any]] = None + self.close_reason: Optional[str] = None + + def connect(self) -> LiveSession: + """Open the shared transport, mapping only Live's connection errors.""" + try: + return super().connect() + except BaseException as exc: + error = _transport_error(exc, "connect") + if error is exc: + raise + raise error from exc + + def send_event(self, event: Dict[str, Any]) -> str: + """Send a raw event and return its supplied or generated event_id.""" + with self._send_lock: + with self._lock: + ws, generation = self._ws, self._generation + if ws is None: + raise RuntimeError("live session is not open") + if self._finish_requested: + if event.get("type") == EVENT_SESSION_CLOSE: + assert self._finish_event_id is not None + return self._finish_event_id + raise RuntimeError("live session is finishing") + payload = dict(event) + kind = payload.get("type") + if not isinstance(kind, str) or not kind.strip(): + raise ValueError("event type must be a nonempty string") + payload.setdefault("event_id", "event_" + uuid.uuid4().hex) + event_id = payload["event_id"] + if ( + not isinstance(event_id, str) + or not 1 <= len(event_id) <= 128 + ): + raise ValueError( + "event_id must be a string of 1 to 128 characters", + ) + raw = json.dumps(payload) + if payload.get("type") == EVENT_SESSION_CLOSE: + self._finish_requested = True + self._finish_event_id = event_id + try: + self._send_raw(ws, raw) + except BaseException as exc: + error = _transport_error(exc, "send") + with self._lock: + if ws is self._ws and generation == self._generation: + if isinstance(error, DashScopeError): + self._failure = error + self.close() + if error is exc: + raise + raise error from exc + return event_id + + def recv_event(self) -> Optional[Dict[str, Any]]: + """Receive an event; recoverable errors raise but leave the socket open.""" + return self._recv_event(raise_request_errors=True) + + # Keep ownership checks and protocol state in one receive critical section. + # pylint: disable-next=too-many-branches,too-many-statements + def _recv_event( # pylint: disable=too-many-return-statements + self, + *, + raise_request_errors: bool, + expected_generation: Optional[int] = None, + ) -> Optional[Dict[str, Any]]: + # Socket I/O never holds the state lock needed by send or close. + with self._recv_lock: + with self._lock: + if ( + expected_generation is not None + and expected_generation != self._generation + ): + return None + ws, generation = self._ws, self._generation + if ws is None: + self._check_unfinished() + if self._closed: + return None + raise RuntimeError("live session is not open") + failure = None + recoverable = False + try: + try: + raw = self._recv_raw(ws, generation) + except Exception as exc: + with self._lock: + if generation != self._generation: + return None + if ws is not self._ws: + self._check_unfinished() + return None + error = _transport_error(exc, "receive") + if error is exc: + raise + raise error from exc + with self._lock: + if generation != self._generation: + return None + if ws is not self._ws: + self._check_unfinished() + return None + if raw is None or raw == "" or raw == b"": + raise APIConnectionError( + "live websocket disconnected before session.closed", + ) + try: + if isinstance(raw, (bytes, bytearray)): + raw = raw.decode("utf-8") + event = json.loads(raw) + except (ValueError, TypeError) as exc: + raise WsEventError( + "expected a UTF-8 JSON event object", + body=raw, + ) from exc + if not isinstance(event, dict): + raise WsEventError( + "expected a JSON event object", + body=event, + ) + if ( + not isinstance(event.get("type"), str) + or not event["type"] + ): + raise WsEventError( + "expected a non-empty string event type", + body=event, + ) + self.last_message = event + self._track_event(event) + if event["type"] == EVENT_SESSION_CLOSED: + self._finished = True + self.close() + self._check_unfinished() + elif event["type"] == EVENT_ERROR: + error = event.get("error") + if error is None: + error = {} + if not isinstance(error, dict): + raise WsEventError( + "expected a JSON error object", + body=event, + ) + code = error.get("code") + if code is None or code == "": + code = event.get("code") + if code is not None and not isinstance(code, str): + raise WsEventError( + "expected a string error code", + body=event, + ) + failure = error_from_business_code( + message=error.get("message") or "live error", + code=code, + body=event, + ) + failure.status_code = None + recoverable = ( + error.get("type") == "invalid_request_error" + ) + if recoverable: + if self._finish_event_id is not None and ( + error.get("event_id") == self._finish_event_id + ): + self._finish_requested = False + self._finish_event_id = None + else: + self._failure = failure + self.close() + except BaseException as exc: + with self._lock: + if ws is self._ws and generation == self._generation: + if isinstance(exc, DashScopeError): + self._failure = exc + self.close() + raise + if failure is not None and ( + raise_request_errors or not recoverable + ): + raise failure + return event + + def _check_unfinished(self) -> None: + if self._failure is not None: + raise self._failure + + def _track_event(self, event: Dict[str, Any]) -> None: + if event["type"] in {EVENT_SESSION_CREATED, EVENT_SESSION_UPDATED}: + session = event.get("session") + if session is None: + session = {} + if not isinstance(session, dict): + raise WsEventError( + "expected a JSON session object", + body=event, + ) + self.session_id = session.get("id", self.session_id) + elif event["type"] == EVENT_SESSION_CLOSED: + self.close_reason = event.get("reason") + + def events(self) -> Iterator[Dict[str, Any]]: + """Yield recoverable errors; fatal failures raise, never becoming EOF. + + Once started, an iterator belongs to that connection. Reconnecting + requires a new iterator; audio and tasks are never replayed. + """ + with self._lock: + generation = self._generation + while True: + event = self._recv_event( + raise_request_errors=False, + expected_generation=generation, + ) + if event is None: + return + yield event + if event["type"] == EVENT_SESSION_CLOSED: + return + + def cancel(self) -> None: + """Immediately abort the socket without a WebSocket close handshake.""" + self._close(abort=True) + + def _send(self, event: Dict[str, Any], event_id: Optional[str]) -> str: + if event_id is not None: + event["event_id"] = event_id + return self.send_event(event) + + def update_session( + self, + *, + model: str, + delegation: Dict[str, Any], + instructions: str = cast(str, _UNSET), + voice: str = cast(str, _UNSET), + tts: Dict[str, Any] = cast(Dict[str, Any], _UNSET), + input_audio_format: str = cast(str, _UNSET), + output_audio_format: str = cast(str, _UNSET), + tools: List[Dict[str, Any]] = cast(List[Dict[str, Any]], _UNSET), + transport: Dict[str, Any] = cast(Dict[str, Any], _UNSET), + modalities: List[str] = cast(List[str], _UNSET), + context: Dict[str, Any] = cast(Dict[str, Any], _UNSET), + turn_detection: Optional[Dict[str, Any]] = cast( + Optional[Dict[str, Any]], + _UNSET, + ), + input_audio_transcription: Optional[Dict[str, Any]] = cast( + Optional[Dict[str, Any]], + _UNSET, + ), + parameters: Optional[Dict[str, Any]] = None, + event_id: Optional[str] = None, + ) -> str: + """Configure the front end and delegation; confirmed by session.updated. + + ``parameters`` merges extra keys into the ``session`` node for server + fields this SDK does not type yet; it may not shadow a typed argument. + """ + if not isinstance(delegation, dict): + raise ValueError("delegation must be an object") + kind = delegation.get("type") + if kind not in ("none", "client", "responses"): + raise ValueError( + "delegation.type must be none, client or responses", + ) + if kind == "responses": + responses = delegation.get("responses") + if not isinstance(responses, dict): + raise ValueError("delegation.responses must be an object") + backend_model = responses.get("model") + if not isinstance(backend_model, str) or not backend_model.strip(): + raise ValueError( + "delegation.responses.model must be a nonempty string", + ) + elif "responses" in delegation: + raise ValueError("delegation.responses requires responses mode") + for value in (input_audio_format, output_audio_format): + if value is not _UNSET and value != "pcm": + raise ValueError("Live audio formats must be pcm") + if parameters is not None: + if not isinstance(parameters, dict): + raise ValueError("parameters must be an object") + shadowed = sorted(set(parameters) & _RESERVED_SESSION_FIELDS) + if shadowed: + raise ValueError( + "parameters cannot override typed session fields: " + + ", ".join(shadowed), + ) + session = { + key: value + for key, value in ( + ("model", model), + ("instructions", instructions), + ("voice", voice), + ("tts", tts), + ("input_audio_format", input_audio_format), + ("output_audio_format", output_audio_format), + ("tools", tools), + ("delegation", delegation), + ("transport", transport), + ("modalities", modalities), + ("context", context), + ("turn_detection", turn_detection), + ("input_audio_transcription", input_audio_transcription), + ) + if value is not _UNSET + } + if parameters: + session.update(parameters) + return self._send( + {"type": EVENT_SESSION_UPDATE, "session": session}, + event_id, + ) + + def append_audio( + self, + audio_b64: Union[str, bytes], + *, + event_id: Optional[str] = None, + ) -> str: + """Send raw or base64 16 kHz mono PCM16LE; AOQ input uses its own SDK.""" + audio = _as_base64(audio_b64) + pcm = ( + audio_b64 + if isinstance(audio_b64, bytes) + else base64.b64decode(audio, validate=True) + ) + if not pcm or len(pcm) % 2: + raise ValueError( + "audio must contain nonempty, even-length PCM16LE", + ) + return self._send( + {"type": EVENT_SESSION_INPUT_AUDIO_APPEND, "audio": audio}, + event_id, + ) + + def mute_input_audio(self, *, event_id: Optional[str] = None) -> str: + """Replace subsequent input with silence without stopping playback.""" + return self._send({"type": EVENT_SESSION_INPUT_AUDIO_MUTE}, event_id) + + def unmute_input_audio(self, *, event_id: Optional[str] = None) -> str: + """Resume normal input without replaying muted audio.""" + return self._send({"type": EVENT_SESSION_INPUT_AUDIO_UNMUTE}, event_id) + + def _append_task_event( + self, + kind: str, + content: str, + *, + delegation_id: Optional[str], + task_id: Optional[str], + start_ms: Optional[float], + end_ms: Optional[float], + event_id: Optional[str], + error: Optional[Dict[str, Any]] = None, + ) -> str: + if kind != "commentary" or delegation_id is not None: + if not isinstance(delegation_id, str) or not delegation_id: + raise ValueError("delegation_id must be a nonempty string") + if delegation_id is None and ( + task_id is not None or error is not None + ): + raise ValueError("task_id and error require a delegation_id") + event: Dict[str, Any] = {"type": f"session.{kind}.append"} + if content is not _UNSET: + event["content"] = content + for key, value in ( + ("delegation_id", delegation_id), + ("task_id", task_id), + ("error", error), + ): + if value is not None: + event[key] = value + event.update(_ms_window(start_ms, end_ms)) + return self._send(event, event_id) + + def append_instruction( + self, + content: str, + *, + delegation_id: str, + task_id: Optional[str] = None, + start_ms: Optional[float] = None, + end_ms: Optional[float] = None, + event_id: Optional[str] = None, + ) -> str: + """Yield for user input; acknowledged by session.instructions.appended.""" + return self._append_task_event( + "instruction", + content, + delegation_id=delegation_id, + task_id=task_id, + start_ms=start_ms, + end_ms=end_ms, + event_id=event_id, + ) + + def append_thinking( + self, + content: str, + *, + delegation_id: str, + start_ms: Optional[float] = None, + end_ms: Optional[float] = None, + event_id: Optional[str] = None, + ) -> str: + """Report silent client-task progress without ending the execution.""" + return self._append_task_event( + "thinking", + content, + delegation_id=delegation_id, + task_id=None, + start_ms=start_ms, + end_ms=end_ms, + event_id=event_id, + ) + + def append_commentary( + self, + content: str = cast(str, _UNSET), + *, + delegation_id: Optional[str] = None, + start_ms: Optional[float] = None, + end_ms: Optional[float] = None, + error: Optional[Dict[str, Any]] = None, + event_id: Optional[str] = None, + ) -> str: + """Send user text or a client-task result; content may be omitted.""" + return self._append_task_event( + "commentary", + content, + delegation_id=delegation_id, + task_id=None, + start_ms=start_ms, + end_ms=end_ms, + error=error, + event_id=event_id, + ) + + def create_session_item( + self, + item: Dict[str, Any] = cast(Dict[str, Any], _UNSET), + *, + event_id: Optional[str] = None, + ) -> str: + """Send session.item.create with an optional front-end item.""" + event: Dict[str, Any] = {"type": EVENT_SESSION_ITEM_CREATE} + if item is not _UNSET: + event["item"] = item + return self._send(event, event_id) + + def submit_function_output( + self, + call_id: str, + output: Any, + *, + event_id: Optional[str] = None, + ) -> str: + """Reply to a front-end function from session.event, never a backend tool.""" + return self.create_session_item( + { + "type": "function_call_output", + "call_id": call_id, + "output": ( + output + if isinstance(output, str) + else json.dumps(output, ensure_ascii=False) + ), + }, + event_id=event_id, + ) + + def create_response_item( + self, + item: Dict[str, Any], + *, + event_id: Optional[str] = None, + ) -> str: + """Buffer next-task input in Responses mode, not an active task's result.""" + return self._send( + {"type": EVENT_RESPONSE_ITEM_CREATE, "item": item}, + event_id, + ) + + def create_response( + self, + response: Optional[Dict[str, Any]] = None, + *, + event_id: Optional[str] = None, + ) -> str: + """Start a new Responses task, or request a reply from a capable front end.""" + event: Dict[str, Any] = {"type": EVENT_RESPONSE_CREATE} + if response is not None: + event["response"] = response + return self._send(event, event_id) + + def end_session(self, *, event_id: Optional[str] = None) -> str: + """Send session.close; keep receiving until session.closed or disconnect.""" + return self._send({"type": EVENT_SESSION_CLOSE}, event_id) + + +class Live: + """client.audio.live: Qwen-Live delegated voice sessions.""" + + def __init__(self, client) -> None: + self._client = client + + def connect( + self, + *, + model: str, + ws_url: Optional[str] = None, + connector: Optional[Connector] = None, + timeout: Optional[float] = None, + headers: Optional[Mapping[str, str]] = None, + api_key: Optional[str] = None, + workspace: Optional[str] = None, + ) -> LiveSession: + """Create an unopened session; model routes the URL, not the front-end model.""" + return LiveSession( + self._client, + model=model, + ws_url=ws_url, + connector=connector, + timeout=timeout, + headers=headers, + api_key=api_key, + workspace=workspace, + ) diff --git a/dashscope/resources/audio/realtime.py b/dashscope/resources/audio/realtime.py index e38664f4..f7b9c413 100644 --- a/dashscope/resources/audio/realtime.py +++ b/dashscope/resources/audio/realtime.py @@ -4,33 +4,30 @@ from __future__ import annotations -import base64 import json -import threading import time import uuid from typing import Any, Dict, Iterator, List, Mapping, Optional, Union, cast -from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit -from dashscope._config import WS_REALTIME_PATH, workspace_ws_url from dashscope._logging import get_logger, request_id_from_headers from dashscope.errors import ( APIConnectionError, APITimeoutError, error_from_business_code, ) -from dashscope.transport import default_user_agent -from dashscope.transport._ident import sdk_headers -from dashscope.transport.ws import Connector, _close_websocket +from dashscope.resources.audio._ws_session import ( + _UNSET, + WsSessionBase, + _as_base64, +) +from dashscope.transport.ws import Connector __all__ = [ "Realtime", "RealtimeSession", ] -MODULE = "audio" -_UNSET = object() _logger = get_logger("audio.realtime") @@ -54,25 +51,7 @@ def _event_request_id(event: Any) -> Optional[str]: return None -def _default_connector( - url: str, - headers: Dict[str, str], - timeout: Optional[float], -) -> Any: - import websocket # pylint: disable=import-outside-toplevel - - return websocket.create_connection(url, header=headers, timeout=timeout) - - -def _as_base64(data: Union[str, bytes]) -> str: - if isinstance(data, str): - return data - if isinstance(data, bytes): - return base64.b64encode(data).decode("ascii") - raise TypeError("expected a base64 string or bytes") - - -class RealtimeSession: +class RealtimeSession(WsSessionBase): """One realtime websocket session (context manager). Usage:: @@ -105,59 +84,17 @@ def __init__( api_key: Optional[str] = None, workspace: Optional[str] = None, ) -> None: - config = client.config - if timeout is not None and timeout <= 0: - raise ValueError("timeout must be positive") - # A per-call workspace override picks that workspace's gateway - # for the URL as well, so the socket and the - # X-DashScope-WorkSpace header never name different workspaces. - space = workspace or config.workspace - # URL precedence: explicit ``ws_url`` > the workspace-scoped MaaS - # gateway realtime endpoint; without a workspace the legacy shared - # gateway is the fallback, with a warning. (The inference - # endpoint's DASHSCOPE_WEBSOCKET_BASE_URL does not apply here: the - # realtime path differs.) - url = ws_url or workspace_ws_url( - space, - WS_REALTIME_PATH, - config.region, + super().__init__( + client, + model=model, + ws_url=ws_url, + connector=connector, + timeout=timeout, + headers=headers, + api_key=api_key, + workspace=workspace, ) - parts = urlsplit(url) - query = parse_qsl(parts.query, keep_blank_values=True) - if not any(key == "model" for key, _ in query): - query.append(("model", model)) - url = urlunsplit(parts._replace(query=urlencode(query))) - # Inherit the client's agent (and any surface suffix on it) so the - # realtime socket reports the same identity as its HTTP calls; the - # dedup below still lets an explicit per-call header win. The - # lookup is defensive for the same reason as - # ``transport.ws.with_client_user_agent``. - all_headers = { - "user-agent": ( - getattr( - getattr(client, "transport", None), - "user_agent", - None, - ) - or default_user_agent() - ), - **sdk_headers(MODULE), - } - key = api_key or config.api_key - if key: - all_headers["Authorization"] = f"Bearer {key}" - if space: - all_headers["X-DashScope-WorkSpace"] = space - user_headers = dict(headers or {}) - user_keys = {key.lower() for key in user_headers} - all_headers = { - key: value - for key, value in all_headers.items() - if key.lower() not in user_keys - } - all_headers.update(user_headers) - self._model = model - wire_model = next(value for key, value in query if key == "model") + wire_model = self._wire_model if wire_model.startswith("qwen3.8-omni-"): self._protocol = "omni_v38" elif wire_model.startswith("qwen-audio-"): @@ -170,22 +107,12 @@ def __init__( self._protocol = "livetranslate" else: self._protocol = "omni" - self._url = url - self._headers = all_headers - self._connector = connector - self._timeout = config.timeout if timeout is None else timeout - self._lock = threading.RLock() - self._recv_lock = threading.Lock() - self._generation = 0 - self._ws = None - self._reset_state() def _reset_state(self) -> None: + super()._reset_state() # Logging context is separate from the public latency/response metrics. self._request_id: Optional[str] = None self._protocol_request_id: Optional[str] = None - self._header_request_id: Optional[str] = None - self._closed = False self._finish_requested = False self._finished = False self._response_active = False @@ -197,62 +124,36 @@ def _reset_state(self) -> None: self.last_first_text_delay: Optional[float] = None self.last_first_audio_delay: Optional[float] = None - @property - def model(self) -> str: - return self._model - - @property - def url(self) -> str: - return self._url - def connect(self) -> "RealtimeSession": """Open the connection, either explicitly or on context entry.""" - with self._lock: - if self._ws is not None: - raise RuntimeError("realtime session is already open") - self._generation += 1 - self._reset_state() - try: - if self._connector is None: - self._ws = _default_connector( - self._url, - self._headers, - self._timeout, - ) - else: - self._ws = self._connector(self._url, self._headers) - self._header_request_id = request_id_from_headers( - getattr(self._ws, "headers", None), - ) - if self._timeout is not None: - self._ws.settimeout(self._timeout) - except BaseException as exc: - body = getattr(exc, "resp_body", None) - if isinstance(body, (str, bytes, bytearray)): - try: - body = json.loads(body) - except (ValueError, UnicodeError, RecursionError): - body = None - request_id = ( - _event_request_id(body) - or request_id_from_headers( - getattr(exc, "resp_headers", None), - ) - or self._header_request_id - ) - _logger.error( - "Realtime handshake failed: %s, request_id=%s", - type(exc).__name__, - request_id or "unknown", - ) - self.close() - self._reset_state() - from websocket import WebSocketTimeoutException + from websocket import ( # pylint: disable=import-outside-toplevel + WebSocketTimeoutException, + ) - if isinstance(exc, WebSocketTimeoutException): - raise APITimeoutError(detail=str(exc)) from exc - raise - return self + try: + return super().connect() + except WebSocketTimeoutException as exc: + raise APITimeoutError(detail=str(exc)) from exc + + def _on_connect_error(self, exc: BaseException) -> None: + body = getattr(exc, "resp_body", None) + if isinstance(body, (str, bytes, bytearray)): + try: + body = json.loads(body) + except (ValueError, UnicodeError, RecursionError): + body = None + request_id = ( + _event_request_id(body) + or request_id_from_headers( + getattr(exc, "resp_headers", None), + ) + or self._header_request_id + ) + _logger.error( + "Realtime handshake failed: %s, request_id=%s", + type(exc).__name__, + request_id or "unknown", + ) def _request_id_for_log(self) -> str: return ( @@ -262,37 +163,42 @@ def _request_id_for_log(self) -> str: or "unknown" ) - def __enter__(self) -> "RealtimeSession": - return self.connect() - - def __exit__(self, *exc: Any) -> None: - self.close() - def send_event(self, event: Dict[str, Any]) -> None: """Send a raw event, assigning an event_id unless one is supplied.""" - with self._lock: - if self._ws is None: - raise RuntimeError("realtime session is not open") - if self._finish_requested: - if event.get("type") == "session.finish": - return - raise RuntimeError("realtime session is finishing") - payload = dict(event) - payload.setdefault("event_id", "event_" + uuid.uuid4().hex) - raw = json.dumps(payload) - if payload.get("type") == "session.finish": - self._finish_requested = True - try: - self._ws.send(raw) + with self._send_lock: + with self._lock: + ws, generation = self._ws, self._generation + if ws is None: + raise RuntimeError("realtime session is not open") + if self._finish_requested: + if event.get("type") == "session.finish": + return + raise RuntimeError("realtime session is finishing") + payload = dict(event) + payload.setdefault("event_id", "event_" + uuid.uuid4().hex) + raw = json.dumps(payload) + if payload.get("type") == "session.finish": + self._finish_requested = True if payload.get("type") == "response.create": + # Register before I/O: a response may arrive before send returns. self._pending_response_ids.append(payload["event_id"]) + try: + self._send_raw(ws, raw) except BaseException as exc: - _logger.error( - "Realtime send failed: %s, request_id=%s", - type(exc).__name__, - self._request_id_for_log(), - ) - self.close() + with self._lock: + if ws is self._ws and generation == self._generation: + _logger.error( + "Realtime send failed: %s, request_id=%s", + type(exc).__name__, + self._request_id_for_log(), + ) + if payload.get("type") == "response.create" and ( + payload["event_id"] in self._pending_response_ids + ): + self._pending_response_ids.remove( + payload["event_id"], + ) + self.close() raise def update_session( @@ -755,7 +661,7 @@ def recv_event(self) -> Optional[Dict[str, Any]]: received_on_current_connection = False try: try: - raw = ws.recv() + raw = self._recv_raw(ws, generation) except WebSocketConnectionClosedException: raw = None except WebSocketTimeoutException as exc: @@ -952,13 +858,6 @@ def events(self) -> Iterator[Dict[str, Any]]: if event.get("type") == "session.finished": return - def close(self) -> None: - with self._lock: - ws, self._ws = self._ws, None - if ws is not None: - self._closed = True - _close_websocket(ws) - class Realtime: """``client.audio.realtime``: Omni, Qwen-Audio and Livetranslate.""" diff --git a/examples/01_atomic_capabilities/audio/realtime_live.py b/examples/01_atomic_capabilities/audio/realtime_live.py new file mode 100644 index 00000000..77fb310c --- /dev/null +++ b/examples/01_atomic_capabilities/audio/realtime_live.py @@ -0,0 +1,120 @@ +# -*- coding: utf-8 -*- +"""Live 简洁示例：发送文字或音频文件，实时打印回答转写；Ctrl+C 退出。 + +设置 LIVE_API_KEY 或 DASHSCOPE_API_KEY；密钥须有目标环境访问权限。 +默认复用已实测跑通的预发环境，可用 LIVE_WS_URL / LIVE_ROUTE_MODEL / +LIVE_FRONTEND_MODEL / LIVE_BACKEND_MODEL 覆写：路由模型决定接入地址，前端模型 +处理语音，设置后端模型才委派 responses 后台推理。服务端必填 context.compaction， +缺失会拒绝初始化。WAV / PCM 文件需为 16 kHz、单声道 PCM16LE。 +音频按 100 ms 帧发送，不追赶补发；本示例不采集或播放音频。 +回答后连接保持打开；连续 30 秒未收到事件会报 SDK 超时异常。 +运行：python -m examples.01_atomic_capabilities.audio.realtime_live [audio_file] +""" + +import argparse +import os +import sys +import wave +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Event + +from dashscope import Client + +WS_URL = os.getenv( + "LIVE_WS_URL", "wss://poc-01.cn-beijing.pre-maas.aliyuncs.com/api-ws/v1/realtime", +) +ROUTE_MODEL = os.getenv("LIVE_ROUTE_MODEL", "pre-qwen-live-test") +FRONTEND_MODEL = os.getenv("LIVE_FRONTEND_MODEL", "qwen-audio-realtime") +BACKEND_MODEL = os.getenv("LIVE_BACKEND_MODEL", "") + + +def read_audio(path): + path = Path(path) + if path.suffix.lower() == ".wav": + with wave.open(str(path), "rb") as audio: + if ( + audio.getnchannels() != 1 or audio.getsampwidth() != 2 + or audio.getframerate() != 16000 or audio.getcomptype() != "NONE" + ): + raise ValueError("WAV 必须是 16 kHz、单声道、16-bit PCM") + pcm = audio.readframes(audio.getnframes()) + if len(pcm) != audio.getnframes() * 2: + raise ValueError("WAV 音频数据被截断") + elif path.suffix.lower() == ".pcm": + pcm = path.read_bytes() # 裸 PCM 无格式信息，由调用者保证 16 kHz 单声道。 + else: + raise ValueError("仅支持 .wav 或 .pcm 音频文件") + if not pcm or len(pcm) % 2: + raise ValueError("音频必须包含非空、偶数字节的 PCM16LE 数据") + return pcm + + +def send_audio(session, pcm, stopped): + pcm += bytes(32000) # 一秒静音，让服务端 VAD 检测到本轮语音结束。 + for offset in range(0, len(pcm), 3200): + if stopped.is_set(): + return + session.append_audio(pcm[offset:offset + 3200].ljust(3200, b"\0")) + if stopped.wait(0.1): # 每帧 100 ms；延迟后不追赶补发。 + return + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("audio_file", nargs="?", help="16 kHz 单声道 PCM WAV / PCM 文件") + args = parser.parse_args(argv) + pcm = read_audio(args.audio_file) if args.audio_file else None + delegation = {"type": "none"} if not BACKEND_MODEL else { + "type": "responses", + "responses": { + "model": BACKEND_MODEL, + "instructions": "直接回答用户问题，不使用外部工具", + }, + } + stopped = Event() + try: + with Client( + api_key=os.getenv("LIVE_API_KEY"), + ) as client, client.audio.live.connect( + model=ROUTE_MODEL, ws_url=WS_URL, timeout=30, + ) as session, ThreadPoolExecutor(max_workers=1) as pool: + sender = None + try: + session.update_session( + model=FRONTEND_MODEL, + delegation=delegation, + context={"compaction": {"enabled": False, "mode": "local"}}, + instructions="请用简短的中文回答。", + input_audio_format="pcm", + output_audio_format="pcm", + ) + sent = False + for event in session.events(): + if event["type"] == "session.updated" and not sent: + sent = True + if pcm is None: + session.append_commentary("你好，请用一句话介绍自己。") + session.create_response() # 纯文本不会自动触发作答。 + else: + sender = pool.submit(send_audio, session, pcm, stopped) + elif event["type"] == "session.output_transcript.delta": + print(event["delta"], end="", flush=True) + elif event["type"] == "error": + error = event["error"] + message = ( + error.get("message") or error.get("code") or "Live error" + ) + print(message, file=sys.stderr) + finally: + stopped.set() + session.close() # 先中断 I/O，再等待发送线程退出。 + if sender is not None: + sender.result() + except KeyboardInterrupt: + pass # with 会释放连接；Ctrl+C 是本地主动退出，不是异常断连。 + print() + + +if __name__ == "__main__": + main() diff --git a/tests/test_audio_live_session.py b/tests/test_audio_live_session.py new file mode 100644 index 00000000..04f83065 --- /dev/null +++ b/tests/test_audio_live_session.py @@ -0,0 +1,1880 @@ +# -*- coding: utf-8 -*- +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Qwen-Live wire payloads and synchronous session lifecycle.""" + +from __future__ import annotations + +import base64 +import importlib +import json +import re +import runpy +import socket +import threading +import time +import wave + +import pytest +from websocket import ( + WebSocketBadStatusException, + WebSocketConnectionClosedException, + WebSocketException, + WebSocketTimeoutException, +) + +from dashscope import Client +from dashscope.errors import ( + APIConnectionError, + APITimeoutError, + AuthenticationError, + BadRequestError, + DashScopeError, + RateLimitError, +) +from dashscope.transport._ident import sdk_headers +from dashscope.transport.ws import WsEventError + + +class FakeWebSocket: + def __init__(self, inbox=()): + self.inbox = list(inbox) + self.sent = [] + self.timeouts = [] + self.close_calls = 0 + + def send(self, payload): + self.sent.append(payload) + + def recv(self): + raw = self.inbox.pop(0) if self.inbox else None + if isinstance(raw, BaseException): + raise raw # pylint: disable=raising-bad-type + return json.dumps(raw) if isinstance(raw, dict) else raw + + def settimeout(self, timeout): + self.timeouts.append(timeout) + + def close(self): + self.close_calls += 1 + + +@pytest.fixture +def client(): + with Client( + api_key="sk-test", + workspace="ws-test", + timeout=12.5, + ) as value: + yield value + + +@pytest.fixture +def opened(client): + ws = FakeWebSocket() + with client.audio.live.connect( + model="qwen-live", + connector=lambda url, headers: ws, + ) as session: + yield session, ws + + +def _payloads(ws): + frames = [json.loads(raw) for raw in ws.sent] + event_ids = [frame.pop("event_id") for frame in frames] + assert all( + re.fullmatch(r"event_[0-9a-f]{32}", value) for value in event_ids + ) + assert len(set(event_ids)) == len(event_ids) + return frames + + +def test_connect_targets_the_realtime_endpoint_with_the_model_query(client): + ws = FakeWebSocket() + seen = {} + + def connector(url, headers): + seen["url"] = url + seen["headers"] = headers + return ws + + with client.audio.live.connect( + model="qwen-live", + connector=connector, + ) as session: + assert session.model == "qwen-live" + assert session.url == ( + "wss://ws-test.cn-beijing.maas.aliyuncs.com" + "/api-ws/v1/realtime?model=qwen-live" + ) + assert seen["url"] == session.url + assert seen["headers"]["Authorization"] == "Bearer sk-test" + assert seen["headers"]["X-DashScope-WorkSpace"] == "ws-test" + for key, value in sdk_headers("audio").items(): + assert seen["headers"][key] == value + assert ws.timeouts == [12.5] + assert ws.close_calls == 1 + + +def test_update_session_matches_the_document_client_envelope(opened): + session, ws = opened + request_id = session.update_session( + model="qwen-audio-realtime", + delegation={"type": "client"}, + voice="星悦", + tts={"system_prompt": "请用清晰自然的女声说话"}, + tools=[], + event_id="evt_update_client_001", + ) + assert request_id == "evt_update_client_001" + assert json.loads(ws.sent[0]) == { + "type": "session.update", + "event_id": "evt_update_client_001", + "session": { + "model": "qwen-audio-realtime", + "delegation": {"type": "client"}, + "voice": "星悦", + "tts": {"system_prompt": "请用清晰自然的女声说话"}, + "tools": [], + }, + } + + +@pytest.mark.parametrize( + "context", + [{}, {"compaction": {"enabled": False, "mode": "local"}}], +) +def test_update_session_passes_through_context_configuration(opened, context): + session, ws = opened + returned = session.update_session( + model="audio-live", + delegation={"type": "none"}, + context=json.loads(json.dumps(context)), + event_id="configure_context_1", + ) + assert returned == "configure_context_1" + assert json.loads(ws.sent[0]) == { + "type": "session.update", + "event_id": "configure_context_1", + "session": { + "model": "audio-live", + "delegation": {"type": "none"}, + "context": context, + }, + } + + +def test_update_session_merges_parameters_passthrough(opened): + session, ws = opened + session.update_session( + model="audio-live", + delegation={"type": "none"}, + parameters={"future_field": 1, "another_new_field": {"nested": True}}, + event_id="configure_parameters_1", + ) + payload = json.loads(ws.sent[0])["session"] + assert payload["future_field"] == 1 + assert payload["another_new_field"] == {"nested": True} + assert payload["model"] == "audio-live" + assert payload["delegation"] == {"type": "none"} + + +@pytest.mark.parametrize( + "parameters", + [ + {"model": "other"}, + {"delegation": {"type": "client"}}, + {"instructions": "x", "context": {}}, + ], +) +def test_update_session_parameters_cannot_shadow_typed_fields( + opened, + parameters, +): + session, ws = opened + with pytest.raises(ValueError, match="parameters"): + session.update_session( + model="audio-live", + delegation={"type": "none"}, + parameters=parameters, + ) + assert ws.sent == [] + + +@pytest.mark.parametrize("parameters", ["duplex", 1, []]) +def test_update_session_rejects_non_mapping_parameters(opened, parameters): + session, ws = opened + with pytest.raises(ValueError, match="parameters"): + session.update_session( + model="audio-live", + delegation={"type": "none"}, + parameters=parameters, + ) + assert ws.sent == [] + + +def test_update_session_passes_through_responses_configuration(opened): + session, ws = opened + session.update_session( + model="audio-live", + delegation={ + "type": "responses", + "responses": { + "model": "qwen3.8-max", + "instructions": "be brief", + "tools": [{"type": "web_search"}], + "metadata": {"purpose": "travel"}, + }, + }, + ) + assert _payloads(ws) == [ + { + "type": "session.update", + "session": { + "model": "audio-live", + "delegation": { + "type": "responses", + "responses": { + "model": "qwen3.8-max", + "instructions": "be brief", + "tools": [{"type": "web_search"}], + "metadata": {"purpose": "travel"}, + }, + }, + }, + }, + ] + + +@pytest.mark.parametrize("updated", [False, True]) +@pytest.mark.parametrize( + "kwargs,missing", + [ + ({}, "model"), + ({"model": "audio-live"}, "delegation"), + ({"delegation": {"type": "client"}}, "model"), + ], +) +def test_update_session_requires_model_and_delegation( + opened, + updated, + kwargs, + missing, +): + session, ws = opened + if updated: + session.update_session( + model="audio-live", + delegation={"type": "client"}, + ) + sent = list(ws.sent) + with pytest.raises(TypeError, match=missing): + session.update_session(**kwargs) + assert ws.sent == sent + + +@pytest.mark.parametrize("kind", ["client", "none"]) +@pytest.mark.parametrize( + "options", + [ + {}, + {"instructions": "", "tools": []}, + {"turn_detection": None}, + {"transport": {"type": "aoq", "aoq": {"ws_audio": False}}}, + {"input_audio_format": "pcm", "output_audio_format": "pcm"}, + ], +) +def test_update_session_omits_unset_fields_and_preserves_explicit_values( + opened, + kind, + options, +): + session, ws = opened + session.update_session( + model="audio-live", + delegation={"type": kind}, + **options, + ) + assert _payloads(ws) == [ + { + "type": "session.update", + "session": { + "model": "audio-live", + "delegation": {"type": kind}, + **options, + }, + }, + ] + + +@pytest.mark.parametrize( + "delegation", + [ + None, + {}, + {"responses": {"model": "qwen3.8-max"}}, + {"type": "delegate"}, + {"type": "client", "responses": {"model": "qwen3.8-max"}}, + {"type": "none", "responses": {}}, + {"type": "responses"}, + {"type": "responses", "responses": None}, + {"type": "responses", "responses": {}}, + {"type": "responses", "responses": {"model": ""}}, + {"type": "responses", "responses": {"model": " "}}, + {"type": "responses", "responses": {"model": None}}, + {"type": "responses", "responses": {"model": 123}}, + ], +) +def test_update_session_rejects_structurally_invalid_delegation( + opened, + delegation, +): + session, ws = opened + with pytest.raises(ValueError, match="delegation"): + session.update_session(model="audio-live", delegation=delegation) + assert ws.sent == [] + + +@pytest.mark.parametrize( + "field", + ["input_audio_format", "output_audio_format"], +) +def test_update_session_rejects_native_realtime_audio_formats(opened, field): + session, ws = opened + with pytest.raises(ValueError, match="pcm"): + session.update_session( + model="audio-live", + delegation={"type": "client"}, + **{field: "pcm16"}, + ) + assert ws.sent == [] + + +def test_append_audio_encodes_raw_bytes(opened): + session, ws = opened + session.append_audio(b"\x00\x01") + session.append_audio("AAAAAA==") + assert _payloads(ws) == [ + {"type": "session.input_audio.append", "audio": "AAE="}, + {"type": "session.input_audio.append", "audio": "AAAAAA=="}, + ] + + +def test_mute_and_unmute_input_audio(opened): + session, ws = opened + session.mute_input_audio() + session.unmute_input_audio() + assert _payloads(ws) == [ + {"type": "session.input_audio.mute"}, + {"type": "session.input_audio.unmute"}, + ] + + +def test_append_commentary_carries_the_delegation_id(opened): + session, ws = opened + session.append_commentary( + "table booked", + delegation_id="task_reservation_001", + ) + assert _payloads(ws) == [ + { + "type": "session.commentary.append", + "delegation_id": "task_reservation_001", + "content": "table booked", + }, + ] + + +def test_append_instruction_carries_the_delegation_id(opened): + session, ws = opened + session.append_instruction( + "prefers outdoors", + delegation_id="task_reservation_001", + ) + assert _payloads(ws) == [ + { + "type": "session.instruction.append", + "delegation_id": "task_reservation_001", + "content": "prefers outdoors", + }, + ] + + +def test_append_thinking_carries_the_delegation_id(opened): + session, ws = opened + session.append_thinking( + "checking availability", + delegation_id="task_reservation_001", + ) + assert _payloads(ws) == [ + { + "type": "session.thinking.append", + "delegation_id": "task_reservation_001", + "content": "checking availability", + }, + ] + + +def test_append_commentary_without_a_delegation_id_is_plain_user_text(opened): + session, ws = opened + session.append_commentary("请帮我查询杭州天气。") + assert _payloads(ws) == [ + { + "type": "session.commentary.append", + "content": "请帮我查询杭州天气。", + }, + ] + + +def test_append_commentary_carries_the_task_result_extras(opened): + session, ws = opened + session.append_commentary( + "杭州今日晴。", + delegation_id="task_weather_001", + start_ms=1200, + end_ms=1400, + error={"code": "upstream_timeout", "outcome": "not_sent"}, + ) + assert _payloads(ws) == [ + { + "type": "session.commentary.append", + "delegation_id": "task_weather_001", + "content": "杭州今日晴。", + "start_ms": 1200, + "end_ms": 1400, + "error": {"code": "upstream_timeout", "outcome": "not_sent"}, + }, + ] + + +def test_append_commentary_can_send_an_error_without_content(opened): + session, ws = opened + returned = session.append_commentary( + delegation_id="task_1", + error={"code": "upstream_timeout"}, + event_id="error_result_1", + ) + assert returned == "error_result_1" + assert json.loads(ws.sent[0]) == { + "type": "session.commentary.append", + "event_id": "error_result_1", + "delegation_id": "task_1", + "error": {"code": "upstream_timeout"}, + } + + +def test_append_commentary_preserves_explicit_empty_content(opened): + session, ws = opened + session.append_commentary("") + assert _payloads(ws) == [ + {"type": "session.commentary.append", "content": ""}, + ] + + +def test_append_instruction_preserves_optional_task_id(opened): + session, ws = opened + session.append_instruction( + "which city?", + delegation_id="task_1", + task_id="task_1", + ) + assert _payloads(ws) == [ + { + "type": "session.instruction.append", + "content": "which city?", + "delegation_id": "task_1", + "task_id": "task_1", + }, + ] + + +@pytest.mark.parametrize("method", ["append_thinking", "append_commentary"]) +def test_task_id_is_not_accepted_for_thinking_or_commentary(opened, method): + session, ws = opened + with pytest.raises(TypeError, match="task_id"): + getattr(session, method)( + "result", + delegation_id="task_1", + task_id="task_1", + ) + assert ws.sent == [] + + +def test_append_thinking_accepts_a_timeline_window(opened): + session, ws = opened + session.append_thinking( + "正在查询符合条件的餐厅。", + delegation_id="task_1", + start_ms=2000, + end_ms=2000, + ) + assert _payloads(ws) == [ + { + "type": "session.thinking.append", + "delegation_id": "task_1", + "content": "正在查询符合条件的餐厅。", + "start_ms": 2000, + "end_ms": 2000, + }, + ] + + +@pytest.mark.parametrize( + "start_ms,end_ms,match", + [ + (1200, None, "together"), + (None, 1400, "together"), + (-1, 1400, "non-negative"), + (1400, 1200, "end_ms must be >="), + (float("nan"), 1400, "finite"), + (1200, float("inf"), "finite"), + (False, 1400, "finite"), + ], +) +def test_append_rejects_a_malformed_timeline_window( + opened, + start_ms, + end_ms, + match, +): + session, ws = opened + with pytest.raises(ValueError, match=match): + session.append_commentary( + "result", + delegation_id="task_1", + start_ms=start_ms, + end_ms=end_ms, + ) + assert ws.sent == [] + + +def test_session_item_create_carries_a_front_end_function_result(opened): + session, ws = opened + item = { + "type": "function_call_output", + "call_id": "f0:call_2", + "output": '{"online":true}', + } + session.create_session_item(item) + assert _payloads(ws) == [{"type": "session.item.create", "item": item}] + + +def test_create_session_item_omits_unset_item(opened): + session, ws = opened + session.create_session_item() + assert _payloads(ws) == [{"type": "session.item.create"}] + + +def test_create_session_item_preserves_an_explicit_empty_object(opened): + session, ws = opened + session.create_session_item({}) + assert _payloads(ws) == [{"type": "session.item.create", "item": {}}] + + +def test_submit_function_output_json_encodes_non_strings(opened): + session, ws = opened + session.submit_function_output("f0:call_2", {"online": True}) + assert _payloads(ws) == [ + { + "type": "session.item.create", + "item": { + "type": "function_call_output", + "call_id": "f0:call_2", + "output": '{"online": true}', + }, + }, + ] + + +def test_response_item_create_then_create_response(opened): + session, ws = opened + item = { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "a table for two"}], + } + session.create_response_item(item) + session.create_response() + assert _payloads(ws) == [ + {"type": "response.item.create", "item": item}, + {"type": "response.create"}, + ] + + +def test_create_response_item_requires_an_item(opened): + session, ws = opened + with pytest.raises(TypeError, match="item"): + session.create_response_item() + assert ws.sent == [] + + +def test_create_response_passes_through_backend_parameters(opened): + session, ws = opened + session.create_response( + { + "instructions": "be brief", + "tools": [{"type": "web_search"}], + "metadata": {"purpose": "travel"}, + }, + ) + assert _payloads(ws) == [ + { + "type": "response.create", + "response": { + "instructions": "be brief", + "tools": [{"type": "web_search"}], + "metadata": {"purpose": "travel"}, + }, + }, + ] + + +def test_output_audio_preserves_timestamps_without_an_event_id(opened): + session, ws = opened + ws.inbox.append( + { + "type": "session.output_audio.delta", + "delta": "AAAAAA==", + "start_ms": 1200, + "end_ms": 1200.083, + }, + ) + assert session.recv_event() == { + "type": "session.output_audio.delta", + "delta": "AAAAAA==", + "start_ms": 1200, + "end_ms": 1200.083, + } + + +def test_session_created_tracks_the_session_id(opened): + session, ws = opened + ws.inbox.append( + {"type": "session.created", "session": {"id": "live_abc123"}}, + ) + session.recv_event() + assert session.session_id == "live_abc123" + + +def test_session_closed_reports_the_reason_and_ends_the_stream(opened): + session, ws = opened + ws.inbox.extend( + [ + {"type": "session.output_transcript.delta", "delta": "hi"}, + { + "type": "session.closed", + "reason": "close_requested", + "session": {"id": "live_abc123"}, + "usage": {"seconds": 45.8}, + }, + {"type": "session.output_transcript.delta", "delta": "late"}, + ], + ) + kinds = [event["type"] for event in session.events()] + assert kinds == [ + "session.output_transcript.delta", + "session.closed", + ] + assert session.close_reason == "close_requested" + assert ws.close_calls == 1 + + +@pytest.mark.parametrize("finish_requested", [False, True]) +@pytest.mark.parametrize("consume", ["recv_event", "events"]) +@pytest.mark.parametrize( + "eof", + [None, "", b"", WebSocketConnectionClosedException()], +) +def test_a_connection_drop_without_session_closed_raises( + opened, + finish_requested, + consume, + eof, +): + session, ws = opened + session.append_audio(b"\x00\x00") + if finish_requested: + session.end_session() + ws.inbox.append(eof) + with pytest.raises(APIConnectionError): + if consume == "events": + list(session.events()) + else: + session.recv_event() + assert session.close_reason is None + assert ws.close_calls == 1 + # A failed connection must not subsequently look like a successful EOF. + with pytest.raises(APIConnectionError): + session.recv_event() + + +def test_explicit_close_ends_receiving_normally(opened): + session, ws = opened + session.close() + assert session.recv_event() is None + assert list(session.events()) == [] + assert ws.close_calls == 1 + + +def test_terminal_event_ends_receiving_normally(opened): + session, ws = opened + ws.inbox.append({"type": "session.closed", "reason": "close_requested"}) + assert list(session.events()) == [ + {"type": "session.closed", "reason": "close_requested"}, + ] + assert session.recv_event() is None + assert ws.close_calls == 1 + + +def test_a_client_error_keeps_the_session_usable(opened): + session, ws = opened + ws.inbox.extend( + [ + { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "InvalidParameter", + "message": "invalid voice", + }, + }, + {"type": "session.updated", "session": {"id": "live_abc123"}}, + ], + ) + with pytest.raises(BadRequestError, match="invalid voice"): + session.recv_event() + assert ws.close_calls == 0 + assert session.recv_event()["type"] == "session.updated" + + +@pytest.mark.parametrize("code", ["InvalidParameter", "future_request_error"]) +def test_events_yields_recoverable_errors_and_keeps_the_same_iterator( + opened, + code, +): + session, ws = opened + error = { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": code, + "message": "invalid voice", + "event_id": "update_1", + }, + } + ws.inbox.extend( + [ + {"type": "session.created", "session": {"id": "live_1"}}, + error, + {"type": "session.updated", "session": {"id": "live_1"}}, + {"type": "session.output_transcript.delta", "delta": "hello"}, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + events = session.events() + assert next(events)["type"] == "session.created" + assert next(events) == error + assert session.last_message == error + assert ws.close_calls == 0 + session.append_commentary("hello") + assert [event["type"] for event in events] == [ + "session.updated", + "session.output_transcript.delta", + "session.closed", + ] + + +def test_events_exposes_a_rejected_close_and_allows_retry(opened): + session, ws = opened + session.end_session(event_id="close_1") + ws.inbox.extend( + [ + { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "BUSY", + "message": "try later", + "event_id": "close_1", + }, + }, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + events = session.events() + assert next(events)["error"]["event_id"] == "close_1" + assert session.end_session(event_id="close_2") == "close_2" + assert next(events)["type"] == "session.closed" + assert list(events) == [] + + +def test_events_raises_fatal_errors_instead_of_yielding_them(opened): + session, ws = opened + ws.inbox.append( + { + "type": "error", + "error": { + "type": "server_error", + "code": "InternalError", + "message": "gone", + }, + }, + ) + with pytest.raises(DashScopeError, match="gone"): + next(session.events()) + assert ws.close_calls == 1 + with pytest.raises(DashScopeError, match="gone"): + session.recv_event() + + +def test_websocket_business_errors_have_no_invented_http_status(opened): + session, ws = opened + ws.inbox.append( + { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "InvalidParameter", + "message": "invalid voice", + "event_id": "update_1", + }, + }, + ) + with pytest.raises(BadRequestError) as caught: + session.recv_event() + assert caught.value.status_code is None + assert caught.value.code == "InvalidParameter" + assert caught.value.body["error"]["event_id"] == "update_1" + + +@pytest.mark.parametrize("stage", ["connect", "timeout", "send", "receive"]) +@pytest.mark.parametrize( + "failure,error_type", + [ + (OSError("network failed"), APIConnectionError), + (socket.timeout("timed out"), APITimeoutError), + (WebSocketTimeoutException("timed out"), APITimeoutError), + ( + WebSocketConnectionClosedException("connection closed"), + APIConnectionError, + ), + (WebSocketException("transport failed"), APIConnectionError), + ], +) +def test_transport_failures_use_sdk_errors(client, stage, failure, error_type): + calls = [] + + class BrokenWebSocket(FakeWebSocket): + def settimeout(self, timeout): + if stage == "timeout": + raise failure + super().settimeout(timeout) + + def send(self, payload): + if stage == "send": + raise failure + super().send(payload) + + ws = BrokenWebSocket([failure] if stage == "receive" else []) + + def connector(url, headers): + calls.append(url) + if stage == "connect": + raise failure + return ws + + session = client.audio.live.connect(model="qwen-live", connector=connector) + with pytest.raises(error_type) as caught: + with session: + if stage == "send": + session.append_commentary("hello") + elif stage == "receive": + session.recv_event() + assert caught.value.__cause__ is failure + assert str(failure) in str(caught.value) + assert ( + len(calls) == 1 + ) # Live never automatically retries or replays input. + assert ws.close_calls == (0 if stage == "connect" else 1) + if stage in {"send", "receive"}: + with pytest.raises(error_type): + session.recv_event() + + +@pytest.mark.parametrize( + "status,error_type", + [(401, AuthenticationError), (429, RateLimitError)], +) +def test_handshake_status_errors_are_mapped(client, status, error_type): + failure = WebSocketBadStatusException( + "handshake failed", + status_code=status, + resp_headers={"x-request-id": "handshake_1"}, + resp_body=json.dumps({"message": "handshake failed"}), + ) + + def connector(url, headers): + raise failure + + with pytest.raises(error_type) as caught: + client.audio.live.connect( + model="qwen-live", + connector=connector, + ).connect() + assert caught.value.status_code == status + assert caught.value.request_id == "handshake_1" + assert caught.value.__cause__ is failure + + +@pytest.mark.parametrize( + "raw", + [ + "not json", + "[]", + b"\xff", + '{"type":"error","error":[]}', + '{"type":"error","error":{"code":123}}', + '{"type":"session.updated","session":[]}', + "{}", + '{"type":null}', + '{"type":""}', + '{"type":123}', + '{"type":false}', + '{"type":[]}', + '{"type":{}}', + '{"type":"error","code":[],"error":{"type":"server_error"}}', + '{"type":"error","code":{},"error":{"type":"server_error"}}', + '{"type":"error","code":123,"error":{"type":"server_error"}}', + '{"type":"error","code":[],"error":{"code":""}}', + ], +) +@pytest.mark.parametrize("consume", ["recv_event", "events"]) +def test_malformed_server_events_use_sdk_protocol_errors(opened, raw, consume): + session, ws = opened + ws.inbox.append(raw) + with pytest.raises(WsEventError): + if consume == "events": + next(session.events()) + else: + session.recv_event() + assert ws.close_calls == 1 + with pytest.raises(WsEventError): + session.recv_event() + + +def test_local_serialization_errors_do_not_close_the_session(opened): + session, ws = opened + with pytest.raises(TypeError): + session.send_event({"type": "custom", "value": object()}) + session.append_commentary("valid") + assert ws.close_calls == 0 + assert json.loads(ws.sent[0])["content"] == "valid" + + +def test_a_server_error_terminates_the_connection(opened): + session, ws = opened + ws.inbox.append( + { + "type": "error", + "error": { + "type": "server_error", + "code": "InternalError", + "message": "upstream gone", + }, + }, + ) + with pytest.raises(DashScopeError, match="upstream gone"): + session.recv_event() + assert ws.close_calls == 1 + + +def test_end_session_sends_session_close(opened): + session, ws = opened + session.end_session() + assert _payloads(ws) == [{"type": "session.close"}] + + +def test_end_session_is_idempotent_and_refuses_further_sends(opened): + session, ws = opened + session.end_session() + session.end_session() + assert len(_payloads(ws)) == 1 + with pytest.raises(RuntimeError, match="live session is finishing"): + session.append_audio(b"\x00\x00") + + +def test_session_closed_after_end_session_reports_the_reason(opened): + session, ws = opened + ws.inbox.append( + { + "type": "session.closed", + "reason": "close_requested", + "session": {"id": "live_abc123", "status": "active"}, + "usage": {"seconds": 45.8}, + }, + ) + session.end_session() + kinds = [event["type"] for event in session.events()] + assert kinds == ["session.closed"] + assert session.close_reason == "close_requested" + assert ws.close_calls == 1 + + +def test_sending_after_close_reports_a_closed_session(opened): + session, _ = opened + session.close() + with pytest.raises(RuntimeError, match="live session is not open"): + session.append_audio(b"\x00\x00") + + +@pytest.mark.parametrize("audio", [b"", b"\x00", "", "AAEC", "not base64"]) +def test_append_audio_rejects_empty_odd_or_invalid_pcm(opened, audio): + session, ws = opened + with pytest.raises(ValueError): + session.append_audio(audio) + assert ws.sent == [] + + +def test_user_commentary_cannot_carry_a_task_error(opened): + session, ws = opened + with pytest.raises(ValueError, match="delegation_id"): + session.append_commentary("result", error={"code": "timeout"}) + assert ws.sent == [] + + +@pytest.mark.parametrize("method", ["append_instruction", "append_thinking"]) +@pytest.mark.parametrize("delegation_id", [None, ""]) +def test_task_reply_requires_nonempty_delegation_id( + opened, + method, + delegation_id, +): + session, ws = opened + with pytest.raises(ValueError, match="delegation_id"): + getattr(session, method)("progress", delegation_id=delegation_id) + assert ws.sent == [] + + +@pytest.mark.parametrize( + "method,args,kwargs,kind", + [ + ( + "update_session", + (), + {"model": "audio-live", "delegation": {"type": "client"}}, + "session.update", + ), + ("append_audio", (b"\x00\x00",), {}, "session.input_audio.append"), + ("mute_input_audio", (), {}, "session.input_audio.mute"), + ("unmute_input_audio", (), {}, "session.input_audio.unmute"), + ("append_commentary", (), {}, "session.commentary.append"), + ("append_commentary", ("hello",), {}, "session.commentary.append"), + ( + "append_instruction", + ("which city?",), + {"delegation_id": "task_1"}, + "session.instruction.append", + ), + ( + "append_thinking", + ("checking",), + {"delegation_id": "task_1"}, + "session.thinking.append", + ), + ("create_session_item", (), {}, "session.item.create"), + ("create_session_item", ({},), {}, "session.item.create"), + ("submit_function_output", ("call_1", {}), {}, "session.item.create"), + ("create_response_item", ({},), {}, "response.item.create"), + ("create_response", (), {}, "response.create"), + ("end_session", (), {}, "session.close"), + ], +) +@pytest.mark.parametrize("event_id", [None, "client_event_001"]) +def test_helpers_return_the_wire_event_id( + opened, + method, + args, + kwargs, + kind, + event_id, +): + session, ws = opened + returned = getattr(session, method)(*args, event_id=event_id, **kwargs) + payload = json.loads(ws.sent[0]) + assert payload["type"] == kind + assert returned == payload["event_id"] + if event_id is None: + assert re.fullmatch(r"event_[0-9a-f]{32}", returned) + else: + assert returned == event_id + if kind == "session.update": + assert "event_id" not in payload["session"] + + +@pytest.mark.parametrize("event_id", ["", "x" * 129, 1]) +def test_invalid_event_ids_are_rejected_before_send(opened, event_id): + session, ws = opened + with pytest.raises(ValueError, match="event_id"): + session.send_event( + { + "type": "session.update", + "session": {}, + "event_id": event_id, + }, + ) + assert ws.sent == [] + session.update_session(model="audio-live", delegation={"type": "client"}) + assert len(ws.sent) == 1 + + +def test_raw_send_returns_id_without_mutating_input(opened): + session, ws = opened + event = {"type": "session.update", "session": {"instructions": "hello"}} + returned = session.send_event(event) + assert returned == json.loads(ws.sent[0])["event_id"] + assert "event_id" not in event + + +@pytest.mark.parametrize( + "event", + [ + {}, + {"type": ""}, + {"type": " "}, + {"type": None}, + {"type": 1}, + {"type": []}, + ], +) +def test_send_event_rejects_missing_or_blank_type(opened, event): + session, ws = opened + with pytest.raises(ValueError, match="type"): + session.send_event(event) + assert ws.sent == [] + # 本地校验失败不关闭会话，之后仍可正常发送。 + session.update_session(model="audio-live", delegation={"type": "none"}) + assert len(ws.sent) == 1 + + +def test_cancel_releases_socket_and_is_repeatable(opened): + session, ws = opened + session.cancel() + session.cancel() + assert ws.close_calls == 1 + assert session.recv_event() is None + + +def test_event_constants_cover_the_protocol_and_stay_in_sync(): + from dashscope.resources.audio import live + + names = {name for name in live.__all__ if name.startswith("EVENT_")} + assert names == { + "EVENT_ERROR", + "EVENT_RESPONSE_CREATE", + "EVENT_RESPONSE_EVENT", + "EVENT_RESPONSE_ITEM_CREATE", + "EVENT_SESSION_CLOSE", + "EVENT_SESSION_CLOSED", + "EVENT_SESSION_COMMENTARY_APPEND", + "EVENT_SESSION_COMMENTARY_APPENDED", + "EVENT_SESSION_CREATED", + "EVENT_SESSION_DELEGATION_CREATED", + "EVENT_SESSION_EVENT", + "EVENT_SESSION_INPUT_AUDIO_APPEND", + "EVENT_SESSION_INPUT_AUDIO_MUTE", + "EVENT_SESSION_INPUT_AUDIO_MUTED", + "EVENT_SESSION_INPUT_AUDIO_UNMUTE", + "EVENT_SESSION_INPUT_AUDIO_UNMUTED", + "EVENT_SESSION_INPUT_TRANSCRIPT_DELTA", + "EVENT_SESSION_INSTRUCTION_APPEND", + "EVENT_SESSION_INSTRUCTIONS_APPENDED", + "EVENT_SESSION_ITEM_CREATE", + "EVENT_SESSION_OUTPUT_AUDIO_DELTA", + "EVENT_SESSION_OUTPUT_AUDIO_INTERRUPTED", + "EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA", + "EVENT_SESSION_THINKING_APPEND", + "EVENT_SESSION_THINKING_APPENDED", + "EVENT_SESSION_UPDATE", + "EVENT_SESSION_UPDATED", + "EVENT_SESSION_USAGE_UPDATED", + } + values = [getattr(live, name) for name in names] + assert len(set(values)) == len(values), "duplicate wire value" + # 易错的单复数与 append/appended 对子逐个钉死。 + assert ( + live.EVENT_SESSION_INSTRUCTION_APPEND == "session.instruction.append" + ) + assert ( + live.EVENT_SESSION_INSTRUCTIONS_APPENDED + == "session.instructions.appended" + ) + assert ( + live.EVENT_SESSION_INPUT_AUDIO_APPEND == "session.input_audio.append" + ) + assert live.EVENT_SESSION_INPUT_AUDIO_MUTED == "session.input_audio.muted" + assert ( + live.EVENT_SESSION_OUTPUT_AUDIO_DELTA == "session.output_audio.delta" + ) + # 常量必须与会话内部实际使用的协议名一致，避免两处漂移。 + assert live.LiveSession._FINISH_EVENT == live.EVENT_SESSION_CLOSE + assert live.LiveSession._TERMINAL_EVENT == live.EVENT_SESSION_CLOSED + + +def test_event_constants_are_reexported_from_the_audio_package(): + from dashscope.resources import audio + from dashscope.resources.audio import live + + expected = {name for name in live.__all__ if name.startswith("EVENT_")} + assert expected, "expected public EVENT_* constants" + assert expected <= set(audio.__all__), "EVENT_* missing from audio.__all__" + for name in sorted(expected): + assert getattr(audio, name) == getattr(live, name) + + +def test_repeated_end_session_returns_original_request_id(opened): + session, ws = opened + first = session.end_session(event_id="close_1") + second = session.end_session(event_id="close_2") + assert first == second == "close_1" + assert len(ws.sent) == 1 + + +def test_rejected_close_can_be_corrected_and_retried(opened): + session, ws = opened + session.end_session(event_id="close_1") + ws.inbox.append( + { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "BUSY", + "message": "busy", + "event_id": "close_1", + }, + }, + ) + with pytest.raises(DashScopeError, match="busy"): + session.recv_event() + assert ws.close_calls == 0 + session.end_session(event_id="close_2") + assert [json.loads(raw)["event_id"] for raw in ws.sent] == [ + "close_1", + "close_2", + ] + + +@pytest.mark.parametrize("error_event_id", [None, "earlier_request"]) +def test_unrelated_request_error_does_not_interrupt_close_drain( + opened, + error_event_id, +): + session, ws = opened + session.end_session() + ws.inbox.extend( + [ + { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "InvalidParameter", + "message": "bad earlier request", + "event_id": error_event_id, + }, + }, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + with pytest.raises(DashScopeError, match="bad earlier request"): + session.recv_event() + assert ws.close_calls == 0 + with pytest.raises(RuntimeError, match="finishing"): + session.mute_input_audio() + assert session.recv_event()["type"] == "session.closed" + + +@pytest.mark.parametrize("reconnect", [False, True]) +@pytest.mark.parametrize("failure", [KeyboardInterrupt(), SystemExit(7)]) +def test_close_or_reconnect_does_not_swallow_receive_interrupts( + client, + reconnect, + failure, +): + entered, release = threading.Event(), threading.Event() + results, failures = [], [] + + class InterruptedWebSocket(FakeWebSocket): + def recv(self): + entered.set() + assert release.wait(2), "receive was not released" + raise failure + + first = InterruptedWebSocket() + second = FakeWebSocket( + [{"type": "session.created", "session": {"id": "new"}}], + ) + sockets = iter([first, second]) + session = client.audio.live.connect( + model="qwen-live", + connector=lambda u, h: next(sockets), + ).connect() + + def receive(): + try: + results.append(session.recv_event()) + except BaseException as exc: + failures.append(exc) + + receiver = threading.Thread(target=receive, daemon=True) + receiver.start() + try: + assert entered.wait(2) + session.close() + if reconnect: + session.connect() + release.set() + receiver.join(2) + assert not receiver.is_alive() + assert results == [] + assert failures == [failure] + if reconnect: + assert second.close_calls == 0 + assert session.recv_event()["session"]["id"] == "new" + finally: + release.set() + receiver.join(3) + session.close() + + +def test_old_event_iterator_does_not_consume_a_reconnected_session(client): + first = FakeWebSocket( + [{"type": "session.created", "session": {"id": "old"}}], + ) + second = FakeWebSocket( + [ + {"type": "session.created", "session": {"id": "new"}}, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + sockets = iter([first, second]) + session = client.audio.live.connect( + model="qwen-live", + connector=lambda u, h: next(sockets), + ) + with session: + old_events = session.events() + assert next(old_events)["session"]["id"] == "old" + with session: + with pytest.raises(StopIteration): + next(old_events) + new_events = session.events() + assert next(new_events)["session"]["id"] == "new" + assert next(new_events)["type"] == "session.closed" + assert list(new_events) == [] + + +def test_reconnect_does_not_reuse_close_or_session_state(client): + first, second = FakeWebSocket(), FakeWebSocket() + sockets = iter([first, second]) + session = client.audio.live.connect( + model="qwen-live", + connector=lambda u, h: next(sockets), + ) + with session: + session.end_session() + first.inbox.append({"type": "session.closed", "reason": "expired"}) + session.recv_event() + with session: + assert session.close_reason is None + session.update_session( + model="audio-live", + delegation={"type": "client"}, + instructions="new connection", + ) + assert json.loads(second.sent[0])["type"] == "session.update" + + +@pytest.mark.parametrize("cancelled", [False, True]) +@pytest.mark.parametrize("missing_message", [False, True]) +def test_minimal_live_example_waits_for_configuration_and_prints_reply( + client, + monkeypatch, + capsys, + cancelled, + missing_message, +): + import dashscope + + class ExampleWebSocket(FakeWebSocket): + configured = False + + def recv(self): + raw = super().recv() + if ( + raw is not None + and json.loads(raw).get("type") == "session.updated" + ): + self.configured = True + return raw + + def send(self, payload): + if json.loads(payload)["type"] == "session.commentary.append": + assert ( + self.configured + ), "sample sent input before session.updated" + super().send(payload) + + ws = ExampleWebSocket( + [ + {"type": "session.created", "session": {"id": "live_example"}}, + {"type": "session.updated", "session": {"id": "live_example"}}, + {"type": "session.output_transcript.delta", "delta": "你好"}, + { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "InvalidParameter", + **( + {} + if missing_message + else {"message": "request rejected"} + ), + }, + }, + {"type": "session.output_transcript.delta", "delta": "，我是语音助手。"}, + KeyboardInterrupt() + if cancelled + else { + "type": "session.closed", + "reason": "close_requested", + }, + ], + ) + connect = client.audio.live.connect + monkeypatch.setattr("sys.argv", ["realtime_live.py"]) + monkeypatch.setattr(dashscope, "Client", lambda **kwargs: client) + monkeypatch.setattr( + client.audio.live, + "connect", + lambda **kwargs: connect(**kwargs, connector=lambda u, h: ws), + ) + runpy.run_module( + "examples.01_atomic_capabilities.audio.realtime_live", + run_name="__main__", + ) + captured = capsys.readouterr() + assert "你好，我是语音助手。" in captured.out + expected_error = ( + "InvalidParameter" if missing_message else "request rejected" + ) + assert expected_error in captured.err + frames = _payloads(ws) + assert [frame["type"] for frame in frames] == [ + "session.update", + "session.commentary.append", + "response.create", + ] + assert frames[0]["session"]["delegation"] == {"type": "none"} + assert isinstance(frames[1]["content"], str) and frames[1]["content"] + assert ws.close_calls == 1 + + +@pytest.mark.parametrize("override", [False, True]) +def test_minimal_live_example_uses_successful_environment_or_explicit_overrides( + client, + monkeypatch, + override, +): + import dashscope + + for name in ( + "LIVE_WS_URL", + "LIVE_ROUTE_MODEL", + "LIVE_FRONTEND_MODEL", + "LIVE_BACKEND_MODEL", + "LIVE_API_KEY", + ): + monkeypatch.delenv(name, raising=False) + if override: + monkeypatch.setenv( + "LIVE_WS_URL", + "wss://live.example.test/api-ws/v1/realtime", + ) + monkeypatch.setenv("LIVE_ROUTE_MODEL", "custom-live-route") + monkeypatch.setenv("LIVE_FRONTEND_MODEL", "custom-live-frontend") + monkeypatch.setenv("LIVE_API_KEY", "sk-test-live") + seen = {} + ws = FakeWebSocket( + [ + {"type": "session.updated", "session": {"id": "configured"}}, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + connect = client.audio.live.connect + + def make_client(**kwargs): + seen["client"] = kwargs + return client + + def open_session(**kwargs): + session = connect(**kwargs, connector=lambda u, h: ws) + seen["url"] = session.url + return session + + monkeypatch.setattr(dashscope, "Client", make_client) + monkeypatch.setattr(client.audio.live, "connect", open_session) + monkeypatch.setattr("sys.argv", ["realtime_live.py"]) + runpy.run_module( + "examples.01_atomic_capabilities.audio.realtime_live", + run_name="__main__", + ) + assert seen["url"] == ( + "wss://live.example.test/api-ws/v1/realtime?model=custom-live-route" + if override + else "wss://poc-01.cn-beijing.pre-maas.aliyuncs.com" + "/api-ws/v1/realtime?model=pre-qwen-live-test" + ) + assert seen["client"]["api_key"] == ("sk-test-live" if override else None) + session_config = json.loads(ws.sent[0])["session"] + assert session_config["model"] == ( + "custom-live-frontend" if override else "qwen-audio-realtime" + ) + assert session_config["delegation"] == {"type": "none"} + assert session_config["context"] == { + "compaction": {"enabled": False, "mode": "local"}, + } + assert session_config["input_audio_format"] == "pcm" + assert session_config["output_audio_format"] == "pcm" + + +def test_minimal_live_example_delegates_to_backend_model_when_configured( + client, + monkeypatch, +): + import dashscope + + for name in ( + "LIVE_WS_URL", + "LIVE_ROUTE_MODEL", + "LIVE_FRONTEND_MODEL", + "LIVE_API_KEY", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("LIVE_BACKEND_MODEL", "qwen3.8-max") + ws = FakeWebSocket( + [ + {"type": "session.updated", "session": {"id": "configured"}}, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + connect = client.audio.live.connect + monkeypatch.setattr(dashscope, "Client", lambda **kwargs: client) + monkeypatch.setattr( + client.audio.live, + "connect", + lambda **kwargs: connect(**kwargs, connector=lambda u, h: ws), + ) + monkeypatch.setattr("sys.argv", ["realtime_live.py"]) + runpy.run_module( + "examples.01_atomic_capabilities.audio.realtime_live", + run_name="__main__", + ) + assert json.loads(ws.sent[0])["session"]["delegation"] == { + "type": "responses", + "responses": { + "model": "qwen3.8-max", + "instructions": "直接回答用户问题，不使用外部工具", + }, + } + + +@pytest.mark.parametrize("kind", ["wav", "pcm"]) +def test_minimal_live_example_streams_audio_only_after_configuration( + client, + monkeypatch, + tmp_path, + capsys, + kind, +): + module = importlib.import_module( + "examples.01_atomic_capabilities.audio.realtime_live", + ) + pcm = b"\x01\x00" * 3337 + path = tmp_path / f"input.{kind}" + if kind == "wav": + with wave.open(str(path), "wb") as audio: + audio.setparams((1, 2, 16000, 0, "NONE", "not compressed")) + audio.writeframes(pcm) + else: + path.write_bytes(pcm) + sent = threading.Event() + frames, timestamps = [], [] + + class AudioWebSocket(FakeWebSocket): + configured = False + + def recv(self): + raw = super().recv() + event = json.loads(raw) + if event["type"] == "session.updated": + self.configured = True + elif event["type"] == "session.output_transcript.delta": + assert sent.wait(3), "audio sender did not finish" + return raw + + def send(self, payload): + event = json.loads(payload) + if event["type"] == "session.input_audio.append": + assert self.configured, "audio sent before session.updated" + frames.append(base64.b64decode(event["audio"], validate=True)) + timestamps.append(time.monotonic()) + if ( + len(frames) == 13 + ): # Three input frames + one second of silence. + sent.set() + super().send(payload) + + ws = AudioWebSocket( + [ + {"type": "session.created", "session": {"id": "audio_example"}}, + {"type": "session.updated", "session": {"id": "audio_example"}}, + {"type": "session.output_transcript.delta", "delta": "音频已收到。"}, + {"type": "session.closed", "reason": "close_requested"}, + ], + ) + connect = client.audio.live.connect + monkeypatch.setattr(module, "Client", lambda **kwargs: client) + monkeypatch.setattr( + client.audio.live, + "connect", + lambda **kwargs: connect(**kwargs, connector=lambda u, h: ws), + ) + module.main([str(path)]) + assert "音频已收到。" in capsys.readouterr().out + assert len(frames) == 13 + assert all(len(frame) == 3200 for frame in frames) + received = b"".join(frames) + assert ( + received[: len(pcm)] == pcm + ) # No WAV container bytes reach the wire. + assert received[len(pcm) :] == bytes(41600 - len(pcm)) + assert all( + later - earlier >= 0.095 + for earlier, later in zip(timestamps, timestamps[1:]) + ) # Slow scheduling must not trigger catch-up bursts. + assert all( + json.loads(raw)["type"] != "session.commentary.append" + for raw in ws.sent + ) + assert ws.close_calls == 1 + + +@pytest.mark.parametrize( + "kind,channels,width,rate,data", + [ + ("pcm", 1, 2, 16000, b""), + ("pcm", 1, 2, 16000, b"\x01"), + ("mp3", 1, 2, 16000, b"not pcm"), + ("wav", 2, 2, 16000, b"\0" * 64), + ("wav", 1, 3, 16000, b"\0" * 66), + ("wav", 1, 2, 8000, b"\0" * 64), + ("wav", 1, 2, 16000, b""), + ], +) +def test_minimal_live_example_rejects_bad_audio_before_connecting( + monkeypatch, + tmp_path, + kind, + channels, + width, + rate, + data, +): + module = importlib.import_module( + "examples.01_atomic_capabilities.audio.realtime_live", + ) + path = tmp_path / f"input.{kind}" + if kind == "wav": + with wave.open(str(path), "wb") as audio: + audio.setparams( + (channels, width, rate, 0, "NONE", "not compressed"), + ) + audio.writeframes(data) + else: + path.write_bytes(data) + + def forbidden(): + raise AssertionError("invalid audio opened a cloud client") + + monkeypatch.setattr(module, "Client", forbidden) + with pytest.raises(ValueError): + module.main([str(path)]) + + +@pytest.mark.parametrize("interrupted", [False, True]) +def test_minimal_live_audio_sender_stops_on_failure_or_cancellation( + client, + monkeypatch, + tmp_path, + interrupted, +): + module = importlib.import_module( + "examples.01_atomic_capabilities.audio.realtime_live", + ) + path = tmp_path / "input.pcm" + path.write_bytes(b"\x01\x00" * 32000) + first_send = threading.Event() + sends = [] + + class InterruptedWebSocket(FakeWebSocket): + def recv(self): + if not self.inbox: + assert first_send.wait(2) + if interrupted: + raise KeyboardInterrupt() + raise WebSocketConnectionClosedException("socket closed") + return super().recv() + + def send(self, payload): + if json.loads(payload)["type"] == "session.input_audio.append": + sends.append(payload) + first_send.set() + if not interrupted: + raise OSError("input send failed") + super().send(payload) + + ws = InterruptedWebSocket( + [ + {"type": "session.updated", "session": {"id": "configured"}}, + ], + ) + connect = client.audio.live.connect + monkeypatch.setattr(module, "Client", lambda **kwargs: client) + monkeypatch.setattr( + client.audio.live, + "connect", + lambda **kwargs: connect(**kwargs, connector=lambda u, h: ws), + ) + if interrupted: + module.main([str(path)]) + else: + # 发送失败会关闭连接，被阻塞的接收随即失败；两者并发，哪一个先写入 + # 会话失败状态并不确定，所以任一连接错误都证明失败已上抛。 + pattern = "input send failed|socket closed" + with pytest.raises(APIConnectionError, match=pattern): + module.main([str(path)]) + assert len(sends) == 1 + assert ws.close_calls == 1 + + +def test_minimal_live_example_help_does_not_open_resources( + monkeypatch, + capsys, +): + module = importlib.import_module( + "examples.01_atomic_capabilities.audio.realtime_live", + ) + + def forbidden(): + raise AssertionError("help opened a cloud client") + + monkeypatch.setattr(module, "Client", forbidden) + with pytest.raises(SystemExit) as caught: + module.main(["--help"]) + assert caught.value.code == 0 + assert "audio_file" in capsys.readouterr().out + + +def test_close_interrupts_native_websocket_receive_without_a_network_error( + client, +): + from websocket import WebSocket + + entered = threading.Event() + results, failures = [], [] + + class ObservedWebSocket(WebSocket): + def recv(self): + entered.set() + return super().recv() + + local, peer = socket.socketpair() + ws = ObservedWebSocket() + ws.sock = local + ws.connected = True + session = client.audio.live.connect( + model="qwen-live", + connector=lambda u, h: ws, + ).connect() + + def receive(): + try: + results.append(session.recv_event()) + except Exception as exc: + failures.append(exc) + + receiver = threading.Thread(target=receive, daemon=True) + receiver.start() + try: + assert entered.wait(2) + session.close() + receiver.join(2) + assert not receiver.is_alive() + assert failures == [] + assert results == [None] + assert session.recv_event() is None + finally: + session.close() + peer.close() + local.close() + receiver.join(3) + + +def test_close_interrupts_native_websocket_send_under_backpressure(client): + from websocket import WebSocket + + entered = threading.Event() + closed = threading.Event() + failures = [] + + class ObservedWebSocket(WebSocket): + def _send(self, data): + entered.set() + return super()._send(data) + + local, peer = socket.socketpair() + local.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024) + ws = ObservedWebSocket() + ws.sock = local + ws.connected = True + session = client.audio.live.connect( + model="qwen-live", + connector=lambda u, h: ws, + ).connect() + + def send(): + try: + session.append_audio(b"\0\0" * 1024 * 1024) + except Exception as exc: + failures.append(exc) + + def close(): + session.close() + closed.set() + + sender = threading.Thread(target=send, daemon=True) + closer = threading.Thread(target=close, daemon=True) + sender.start() + try: + assert entered.wait(2), "send did not reach the native socket" + closer.start() + assert closed.wait(2), "close blocked behind the pending send" + sender.join(2) + assert not sender.is_alive() + assert failures + assert local.fileno() == -1 + finally: + peer.close() + sender.join(3) + if closer.ident is not None: + closer.join(3) + session.close() + local.close() diff --git a/tests/test_audio_samples_direct_call.py b/tests/test_audio_samples_direct_call.py index f813d4fd..4fe08989 100644 --- a/tests/test_audio_samples_direct_call.py +++ b/tests/test_audio_samples_direct_call.py @@ -43,8 +43,9 @@ def _direct(name): NAMES = sorted( path.stem for path in SAMPLES.glob("*.py") - # tingwu is a standalone example without a direct_call copy. - if not path.name.startswith("_") and path.stem != "tingwu" + # Standalone tutorials do not have direct_call CLI copies. + if not path.name.startswith("_") + and path.stem not in {"tingwu", "realtime_live"} ) REALTIME = [realtime_omni, realtime_qwen_audio, realtime_translation] PCM = b"\x01\x00" * 1600 diff --git a/tests/test_audio_ws_integration.py b/tests/test_audio_ws_integration.py index 42f9bf6a..ace60f97 100644 --- a/tests/test_audio_ws_integration.py +++ b/tests/test_audio_ws_integration.py @@ -179,6 +179,68 @@ def serve(scenario): return serve +@pytest.mark.parametrize("receiving", [False, True]) +def test_live_close_sends_normal_control_frame( + local_ws_server, + client, + receiving, +): + entered = threading.Event() + acknowledged = Future() + results, failures = [], [] + + class ObservedSocket(websocket.WebSocket): + def recv(self): + entered.set() + return super().recv() + + async def scenario(ws, request): + message = await ws.receive(timeout=SERVER_TIMEOUT) + # close() is a transport handshake, not a session.close JSON event. + assert message.type == WSMsgType.CLOSE + assert message.data == 1000 + acknowledged.set_result(None) + + with local_ws_server(scenario) as server: + session = client.audio.live.connect( + model="live", + ws_url=server.url, + timeout=TIMEOUT, + connector=lambda url, headers: websocket.create_connection( + url, + header=headers, + timeout=TIMEOUT, + class_=ObservedSocket, + ), + ).connect() + + def receive(): + try: + results.append(session.recv_event()) + except BaseException as exc: + failures.append(exc) + + receiver = threading.Thread(target=receive, daemon=True) + try: + if receiving: + receiver.start() + assert entered.wait(TIMEOUT) + session.close() + assert acknowledged.result(timeout=TIMEOUT) is None + session.close() + session.cancel() + assert session.recv_event() is None + if receiving: + receiver.join(TIMEOUT) + assert not receiver.is_alive() + assert results == [None] + assert failures == [] + finally: + session.cancel() + if receiver.ident is not None: + receiver.join(SERVER_TIMEOUT) + + @pytest.fixture(autouse=True) def closed_client_sockets(monkeypatch): connect = websocket.create_connection diff --git a/tests/test_audio_ws_session.py b/tests/test_audio_ws_session.py new file mode 100644 index 00000000..f2ca756c --- /dev/null +++ b/tests/test_audio_ws_session.py @@ -0,0 +1,554 @@ +# -*- coding: utf-8 -*- +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Shared realtime transport and independent Live/Omni contracts.""" + +from __future__ import annotations + +import json +import socket +import sys +import threading +import traceback + +import pytest +from websocket import WebSocket, WebSocketException, WebSocketTimeoutException + +from dashscope import Client +from dashscope.errors import ( + APIConnectionError, + APITimeoutError, + BadRequestError, +) +from dashscope.resources.audio._ws_session import WsSessionBase + + +class FrameSocket: + def __init__(self, inbox=()): + self.inbox = list(inbox) + self.sent = [] + self.close_calls = 0 + + def settimeout(self, timeout): + self.timeout = timeout + + def send(self, raw): + self.sent.append(raw) + + def recv(self): + raw = self.inbox.pop(0) + if isinstance(raw, BaseException): + raise raw + return raw + + def close(self): + self.close_calls += 1 + + +@pytest.fixture +def client(): + with Client(api_key="sk-test", workspace="ws-test", timeout=10) as value: + yield value + + +@pytest.fixture(params=["live", "realtime"]) +def resource(request, client): + return getattr(client.audio, request.param) + + +@pytest.mark.parametrize("raw", [b"\x00\xff", "not JSON", '{"type":"error"}']) +def test_transport_delivers_raw_frames_without_protocol_decisions(client, raw): + ws = FrameSocket([raw]) + with WsSessionBase( + client, + model="route", + connector=lambda u, h: ws, + ) as transport: + transport._send_raw(ws, raw) + assert transport._recv_raw(ws) == raw + assert ws.sent == [raw] + assert ws.close_calls == 1 + + +def test_transport_leaves_native_receive_errors_for_the_session(client): + failure = WebSocketTimeoutException("read timed out") + ws = FrameSocket([failure]) + with WsSessionBase( + client, + model="route", + connector=lambda u, h: ws, + ) as transport: + with pytest.raises(WebSocketTimeoutException) as caught: + transport._recv_raw(ws) + assert caught.value is failure + assert ws.close_calls == 0 + + +def test_sessions_map_receive_timeouts_to_sdk_timeout(resource): + failure = WebSocketTimeoutException("read timed out") + ws = FrameSocket([failure]) + with resource.connect(model="route", connector=lambda u, h: ws) as session: + with pytest.raises(APITimeoutError) as caught: + session.recv_event() + assert caught.value.__cause__ is failure + assert ws.close_calls == 1 + + +def test_sessions_keep_distinct_idle_eof_contracts(resource): + ws = FrameSocket([None]) + with resource.connect(model="route", connector=lambda u, h: ws) as session: + if resource is resource._client.audio.live: + with pytest.raises(APIConnectionError): + session.recv_event() + else: + assert session.recv_event() is None + assert ws.close_calls == 1 + + +def test_recoverable_error_keeps_each_session_usable(resource): + error = { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "InvalidParameter", + "message": "correct the input", + }, + } + ws = FrameSocket( + [ + json.dumps(error), + '{"type":"session.updated","session":{"id":"s"}}', + ], + ) + with resource.connect(model="route", connector=lambda u, h: ws) as session: + with pytest.raises(BadRequestError): + session.recv_event() + assert ws.close_calls == 0 + assert session.recv_event()["type"] == "session.updated" + session.send_event({"type": "custom", "event_id": "next"}) + assert json.loads(ws.sent[0]) == {"type": "custom", "event_id": "next"} + + +def test_omni_response_acknowledged_during_send_does_not_remain_pending( + client, +): + sent, received = threading.Event(), threading.Event() + failures = [] + + class ReplySocket(FrameSocket): + def send(self, raw): + super().send(raw) + sent.set() + assert received.wait( + 2, + ), "response was not read before send returned" + + ws = ReplySocket( + [ + '{"type":"response.created","response":{"id":"r"}}', + '{"type":"response.done","response":{"id":"r"}}', + None, + ], + ) + with client.audio.realtime.connect( + model="omni", + connector=lambda u, h: ws, + ) as session: + + def receive(): + try: + assert sent.wait(2) + assert session.recv_event()["type"] == "response.created" + assert session.recv_event()["type"] == "response.done" + except BaseException as exc: + failures.append(exc) + finally: + received.set() + + receiver = threading.Thread(target=receive, daemon=True) + receiver.start() + try: + session.create_response() + receiver.join(2) + assert not receiver.is_alive() + assert failures == [] + assert session.recv_event() is None + finally: + session.close() + receiver.join(3) + + +def test_close_sends_control_frame_before_unacknowledged_peer_timeout( + resource, +): + local, peer = socket.socketpair() + ws, remote = WebSocket(), WebSocket() + ws.sock, ws.connected = local, True + remote.sock, remote.connected = peer, True + peer.settimeout(2) + session = resource.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + closed = threading.Event() + + def close(): + session.close() + closed.set() + + closer = threading.Thread(target=close, daemon=True) + try: + closer.start() + frame = remote.recv_frame() + assert frame.opcode == 8 + assert frame.data == b"\x03\xe8" # Normal closure, 1000. + # Do not acknowledge: the close deadline must release the descriptor. + assert closed.wait(2), "close waited indefinitely for a peer Close" + assert local.fileno() == -1 + finally: + remote.shutdown() + closer.join(3) + session.close() + local.close() + + +def test_close_still_releases_socket_when_deadline_cannot_start( + resource, + monkeypatch, +): + local, peer = socket.socketpair() + ws = WebSocket() + ws.sock, ws.connected = local, True + session = resource.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + + def unavailable(self): + raise RuntimeError("cannot start a new thread") + + monkeypatch.setattr(threading.Timer, "start", unavailable) + try: + session.close() + assert local.fileno() == -1 + assert session.recv_event() is None + finally: + peer.close() + ws.shutdown() + + +def test_reconnect_is_rejected_until_old_close_completes(resource): + entered, release = threading.Event(), threading.Event() + failures = [] + + class ClosingSocket(FrameSocket): + def close(self): + entered.set() + assert release.wait(2) + super().close() + + old, fresh = ClosingSocket(), FrameSocket() + sockets = iter([old, fresh]) + session = resource.connect( + model="route", + connector=lambda u, h: next(sockets), + ).connect() + + def close(): + try: + session.close() + except BaseException as exc: + failures.append(exc) + + closer = threading.Thread(target=close, daemon=True) + try: + closer.start() + assert entered.wait(2) + with pytest.raises(RuntimeError, match="closing"): + session.connect() + release.set() + closer.join(2) + assert not closer.is_alive() + assert failures == [] + session.connect() + session.send_event({"type": "custom"}) + assert len(fresh.sent) == 1 + assert fresh.close_calls == 0 + finally: + release.set() + closer.join(3) + session.close() + + +def test_close_waits_for_inflight_send_before_sending_control_frame(resource): + entered, detached, release = ( + threading.Event(), + threading.Event(), + threading.Event(), + ) + failures = [] + + class DelayedSocket(WebSocket): + def send(self, payload, opcode=1): + if opcode == 1: + entered.set() + assert release.wait(2) + return super().send(payload, opcode) + + local, peer = socket.socketpair() + ws, remote = DelayedSocket(), WebSocket() + ws.sock, ws.connected = local, True + remote.sock, remote.connected = peer, True + peer.settimeout(2) + session = resource.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + lock = session._lock + + class ObservedLock: + def __enter__(self): + return lock.__enter__() + + def __exit__(self, *args): + # Synchronize at detach without relying on thread scheduling delays. + if session._ws is None: + detached.set() + return lock.__exit__(*args) + + session._lock = ObservedLock() + + def send(): + try: + session.send_event({"type": "custom", "event_id": "last"}) + except BaseException as exc: + failures.append(exc) + + sender = threading.Thread(target=send, daemon=True) + closer = threading.Thread(target=session.close, daemon=True) + try: + sender.start() + assert entered.wait(2) + closer.start() + assert detached.wait(2) + release.set() + first = remote.recv_frame() + assert first.opcode == 1 + assert json.loads(first.data) == {"type": "custom", "event_id": "last"} + final = remote.recv_frame() + assert final.opcode == 8 + assert final.data == b"\x03\xe8" + remote.send_close() + sender.join(2) + closer.join(2) + assert not sender.is_alive() + assert not closer.is_alive() + assert failures == [] + finally: + release.set() + remote.shutdown() + sender.join(3) + if closer.ident is not None: + closer.join(3) + session.close() + local.close() + + +def test_remote_close_reply_cannot_overtake_inflight_data(resource): + entered, closing, release = ( + threading.Event(), + threading.Event(), + threading.Event(), + ) + failures = [] + + class DelayedSocket(WebSocket): + def send(self, payload, opcode=1): + if opcode == 1: + entered.set() + assert release.wait(2) + return super().send(payload, opcode) + + def send_close(self, *args, **kwargs): + closing.set() + return super().send_close(*args, **kwargs) + + local, peer = socket.socketpair() + ws, remote = DelayedSocket(), WebSocket() + ws.sock, ws.connected = local, True + remote.sock, remote.connected = peer, True + peer.settimeout(2) + session = resource.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + + def send(): + try: + session.send_event({"type": "custom", "event_id": "last"}) + except BaseException as exc: + failures.append(exc) + + def receive(): + try: + session.recv_event() + except APIConnectionError: + pass # Live still reports an unexpected remote transport close. + + sender = threading.Thread(target=send, daemon=True) + receiver = threading.Thread(target=receive, daemon=True) + try: + sender.start() + assert entered.wait(2) + receiver.start() + remote.send_close() + assert closing.wait(2) + release.set() + first = remote.recv_frame() + assert first.opcode == 1 + assert json.loads(first.data) == {"type": "custom", "event_id": "last"} + assert remote.recv_frame().opcode == 8 + sender.join(2) + receiver.join(2) + assert not sender.is_alive() + assert not receiver.is_alive() + assert failures == [] + finally: + release.set() + remote.shutdown() + sender.join(3) + if receiver.ident is not None: + receiver.join(3) + session.close() + local.close() + + +def test_live_cancel_releases_socket_without_a_close_handshake(client): + local, peer = socket.socketpair() + ws = WebSocket() + ws.sock, ws.connected = local, True + peer.settimeout(0.5) + session = client.audio.live.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + try: + session.cancel() + assert peer.recv(1024) == b"" + assert local.fileno() == -1 + assert session.recv_event() is None + session.cancel() + session.close() + finally: + peer.close() + session.cancel() + local.close() + + +def test_local_close_wakes_a_native_blocked_receive(resource): + entered, closed = threading.Event(), threading.Event() + results, failures = [], [] + + class ObservedSocket(WebSocket): + def recv(self): + entered.set() + return super().recv() + + local, peer = socket.socketpair() + ws = ObservedSocket() + ws.sock, ws.connected = local, True + session = resource.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + + def receive(): + try: + results.append(session.recv_event()) + except BaseException as exc: + failures.append(exc) + + def close(): + session.close() + closed.set() + + receiver = threading.Thread(target=receive, daemon=True) + closer = threading.Thread(target=close, daemon=True) + receiver.start() + try: + assert entered.wait(2) + closer.start() + assert closed.wait(2), "close waited for the blocked receive" + receiver.join(2) + assert not receiver.is_alive() + assert results == [None] + assert failures == [] + assert local.fileno() == -1 + finally: + peer.close() + receiver.join(3) + if closer.ident is not None: + closer.join(3) + session.close() + local.close() + + +def test_local_close_wakes_a_native_blocked_send(resource): + entered, closed = threading.Event(), threading.Event() + failures = [] + + class ObservedSocket(WebSocket): + def _send(self, raw): + entered.set() + return super()._send(raw) + + local, peer = socket.socketpair() + local.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024) + ws = ObservedSocket() + ws.sock, ws.connected = local, True + session = resource.connect( + model="route", + connector=lambda u, h: ws, + ).connect() + + def send(): + try: + session.send_event({"type": "custom", "data": "x" * 1024 * 1024}) + except BaseException as exc: + failures.append(exc) + + def close(): + session.close() + closed.set() + + sender = threading.Thread(target=send, daemon=True) + closer = threading.Thread(target=close, daemon=True) + sender.start() + try: + assert entered.wait(2) + closer.start() + assert closed.wait(2), "close waited for the blocked send" + sender.join(2) + if sender.is_alive(): + pytest.fail( + "send did not stop:\n" + + "".join( + traceback.format_stack( + sys._current_frames()[sender.ident], + ), + ), + ) + assert failures + expected = ( + APIConnectionError + if resource is resource._client.audio.live + else (OSError, WebSocketException) + ) + assert all(isinstance(error, expected) for error in failures) + assert local.fileno() == -1 + finally: + peer.close() + sender.join(3) + if closer.ident is not None: + closer.join(3) + session.close() + local.close() From c9a7f03829191455fc5999b78982f4cf68ffb966 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Tue, 10 Nov 2026 16:58:11 +0800 Subject: [PATCH 2/2] feat(audio): accept audio sources for voice enrollment Allow Base64 Data URIs and URLs through audio while preserving legacy url calls and enforcing a single audio source. --- dashscope/resources/audio/customization.py | 29 ++++- tests/test_audio_management.py | 143 +++++++++++++++++++++ 2 files changed, 165 insertions(+), 7 deletions(-) diff --git a/dashscope/resources/audio/customization.py b/dashscope/resources/audio/customization.py index 343be9ac..035d8068 100644 --- a/dashscope/resources/audio/customization.py +++ b/dashscope/resources/audio/customization.py @@ -639,7 +639,8 @@ def create( *, target_model: str, prefix: str, - url: str, + url: Optional[str] = None, + audio: Optional[str] = None, language_hints: Optional[List[str]] = None, max_prompt_audio_length: Optional[float] = None, extra_input: Optional[Dict[str, Any]] = None, @@ -663,7 +664,8 @@ def create( *, target_model: str, prefix: str, - url: str, + url: Optional[str] = None, + audio: Optional[str] = None, language_hints: Optional[List[str]] = None, max_prompt_audio_length: Optional[float] = None, extra_input: Optional[Dict[str, Any]] = None, @@ -686,7 +688,8 @@ def create( *, target_model: str, prefix: str, - url: str, + url: Optional[str] = None, + audio: Optional[str] = None, language_hints: Optional[List[str]] = None, max_prompt_audio_length: Optional[float] = None, extra_input: Optional[Dict[str, Any]] = None, @@ -702,7 +705,9 @@ def create( max_retries: Optional[int] = None, raw: bool = False, ) -> Union[Voice, "APIResponse"]: - """Clone a voice from the audio at ``url``; carries ``voice_id``. + """Clone a voice from ``audio`` (Data URI/URL) or a legacy ``url``. + + Provide exactly one source. The result carries ``voice_id``. ``raw=True`` returns the :class:`~dashscope.transport.APIResponse` before any error check, so a caller can read the status, headers and @@ -710,11 +715,21 @@ def create( keeps ``response_body``'s envelope: it preserves non-JSON failure bodies and injects ``status_code``/``request_id``/``is_error``. """ + if (audio is None) == (url is None): + raise ValueError("provide exactly one of audio or url") + if audio is not None and ( + not isinstance(audio, str) or not audio.strip() + ): + raise ValueError("audio must be a non-empty string") fields: Dict[str, Any] = dict(extra_input or {}) # Named arguments win over the ``extra_input=`` escape hatch. - fields.update( - {"target_model": target_model, "prefix": prefix, "url": url}, - ) + fields.update({"target_model": target_model, "prefix": prefix}) + if audio is not None: + fields.pop("url", None) + fields["audio"] = audio + else: + fields.pop("audio", None) + fields["url"] = url if language_hints is not None: fields["language_hints"] = language_hints if max_prompt_audio_length is not None: diff --git a/tests/test_audio_management.py b/tests/test_audio_management.py index 844ac8c2..0ab0b9be 100644 --- a/tests/test_audio_management.py +++ b/tests/test_audio_management.py @@ -228,6 +228,149 @@ def handler(request): assert all(call["model"] == "voice-enrollment" for call in calls) +@pytest.mark.parametrize( + "audio", + [ + "data:audio/wav;base64,UklGRg==", + "https://cdn.test/sample.wav", + ], +) +def test_voice_create_accepts_audio_source(audio): + bodies = [] + + def handler(request): + assert request.method == "POST" + assert request.url.path == "/api/v1/services/audio/tts/customization" + bodies.append(json.loads(request.content)) + return httpx.Response( + 200, + json={ + "request_id": "req-audio", + "output": {"voice_id": "voice-audio"}, + }, + ) + + with _make_client(handler) as client: + voice = client.audio.voices.create( + target_model="qwen-audio-3.0-tts-flash", + prefix="myvoice", + audio=audio, + ) + assert voice.voice_id == "voice-audio" + assert voice.request_id == "req-audio" + assert bodies == [ + { + "model": "voice-enrollment", + "input": { + "action": "create_voice", + "target_model": "qwen-audio-3.0-tts-flash", + "prefix": "myvoice", + "audio": audio, + }, + }, + ] + + +@pytest.mark.parametrize( + "source", + [ + {}, + {"url": None, "audio": None}, + { + "url": "https://cdn.test/sample.wav", + "audio": "data:audio/wav;base64,UklGRg==", + }, + ], +) +def test_voice_create_requires_exactly_one_audio_source(source): + bodies = [] + + def handler(request): + bodies.append(json.loads(request.content)) + return httpx.Response(200, json={"output": {}}) + + with _make_client(handler) as client: + with pytest.raises(ValueError, match="exactly one of audio or url"): + client.audio.voices.create( + target_model="qwen-audio-3.0-tts-flash", + prefix="myvoice", + **source, + ) + assert bodies == [] + + +@pytest.mark.parametrize("audio", ["", " ", b"\x00\x00", 123]) +def test_voice_create_rejects_empty_or_non_string_audio(audio): + bodies = [] + + def handler(request): + bodies.append(json.loads(request.content)) + return httpx.Response(200, json={"output": {}}) + + with _make_client(handler) as client: + with pytest.raises( + ValueError, + match="audio must be a non-empty string", + ): + client.audio.voices.create( + target_model="qwen-audio-3.0-tts-flash", + prefix="myvoice", + audio=audio, + ) + assert bodies == [] + + +@pytest.mark.parametrize( + "source, expected_source", + [ + ( + {"audio": "data:audio/wav;base64,UklGRg=="}, + {"audio": "data:audio/wav;base64,UklGRg=="}, + ), + ( + {"url": "https://cdn.test/sample.wav"}, + {"url": "https://cdn.test/sample.wav"}, + ), + ], +) +def test_voice_create_named_source_wins_over_extra_input( + source, + expected_source, +): + bodies = [] + extra_input = { + "audio": "data:audio/wav;base64,aWdub3JlZA==", + "url": "https://cdn.test/ignored.wav", + "extra": 1, + } + original = dict(extra_input) + + def handler(request): + bodies.append(json.loads(request.content)) + return httpx.Response(200, json={"output": {"voice_id": "voice-1"}}) + + with _make_client(handler) as client: + client.audio.voices.create( + target_model="qwen-audio-3.0-tts-flash", + prefix="myvoice", + extra_input=extra_input, + **source, + ) + assert bodies == [ + { + "model": "voice-enrollment", + "input": { + "action": "create_voice", + "target_model": "qwen-audio-3.0-tts-flash", + "prefix": "myvoice", + "extra": 1, + **expected_source, + }, + }, + ] + assert extra_input == original + + # ── voice design (voice-enrollment registry, prompt-created) ──────────