feat: add qwen live api

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)

dashscope/resources/audio/_ws_session.py

@@ -0,0 +1,292 @@
+# -*- 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."""
+        closing_ws = None
+        try:
+            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)
+                    closing_ws = self._detach_socket()
+                    self._reset_state()
+                    raise
+        finally:
+            self._close_socket(closing_ws)
+        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:
+        """Wait up to one second for sends, then one for the Close handshake."""
+        self._close(abort=False)
+
+    def _close(self, *, abort: bool) -> None:
+        with self._lock:
+            ws = self._detach_socket()
+        self._close_socket(ws, abort=abort)
+
+    def _detach_socket(self) -> Any:
+        """Detach under _lock; the caller must close the socket outside it."""
+        ws, self._ws = self._ws, None
+        if ws is not None:
+            self._closed = self._closing = True
+        return ws
+
+    def _close_socket(self, ws: Any, *, abort: bool = False) -> None:
+        """Release only the detached socket, keeping reconnect blocked until done."""
+        from websocket import (
+            WebSocket,
+        )  # pylint: disable=import-outside-toplevel
+
+        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
+                # 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
+                # Waiting for the sender must not consume the handshake budget.
+                deadline = threading.Timer(_CLOSE_TIMEOUT, interrupt)
+                deadline.daemon = True
+                try:
+                    deadline.start()
+                except Exception:
+                    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