- English -
-
-
-
-
-
# DashScope Python SDK
The DashScope Python SDK provides a comprehensive interface to [Alibaba Cloud Model Studio (Bailian)](https://www.alibabacloud.com/help/en/model-studio/) APIs, covering text generation, multi-modal understanding, embeddings, reranking, image/video generation, speech synthesis & recognition, and more.
diff --git a/dashscope/__init__.py b/dashscope/__init__.py
index 637d5de..672f24c 100644
--- a/dashscope/__init__.py
+++ b/dashscope/__init__.py
@@ -21,6 +21,7 @@
HttpSpeechSynthesizer,
)
from dashscope.audio.tts.speech_synthesizer import SpeechSynthesizer
+from dashscope.api_entities.aio_session import close_shared_aio_session
from dashscope.common.api_key import save_api_key
from dashscope.common.env import (
api_key,
@@ -46,7 +47,7 @@
from dashscope.files import Files
from dashscope.models import Models
from dashscope.nlp.understanding import Understanding
-from dashscope.rerank.text_rerank import TextReRank
+from dashscope.rerank import AioTextReRank, TextReRank
from dashscope.threads import (
MessageFile,
Messages,
@@ -75,6 +76,7 @@
"api_key",
"api_key_file_path",
"save_api_key",
+ "close_shared_aio_session",
"AioGeneration",
"Conversation",
"Generation",
@@ -106,6 +108,7 @@
"list_tokenizers",
"Application",
"TextReRank",
+ "AioTextReRank",
"Assistants",
"Threads",
"Messages",
diff --git a/dashscope/aigc/generation.py b/dashscope/aigc/generation.py
index 6ec2a39..054a4f4 100644
--- a/dashscope/aigc/generation.py
+++ b/dashscope/aigc/generation.py
@@ -179,11 +179,13 @@ def call( # pylint: disable=arguments-renamed,too-many-branches,too-many-statem
to_merge_incremental_output = True
parameters["incremental_output"] = True
- # Pass incremental_to_full flag via headers user-agent
- if "headers" not in parameters:
- parameters["headers"] = {}
+ # Pass incremental_to_full flag via user_agent parameter
flag = "1" if to_merge_incremental_output else "0"
- parameters["headers"]["user-agent"] = f"incremental_to_full/{flag}"
+ existing_ua = parameters.get("user_agent", "")
+ new_ua = f"incremental_to_full/{flag}"
+ parameters["user_agent"] = (
+ f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua
+ )
response = super().call(
model=model,
@@ -434,11 +436,13 @@ async def call( # type: ignore[override] # pylint: disable=arguments-renamed,to
to_merge_incremental_output = True
parameters["incremental_output"] = True
- # Pass incremental_to_full flag via headers user-agent
- if "headers" not in parameters:
- parameters["headers"] = {}
+ # Pass incremental_to_full flag via user_agent parameter
flag = "1" if to_merge_incremental_output else "0"
- parameters["headers"]["user-agent"] = f"incremental_to_full/{flag}"
+ existing_ua = parameters.get("user_agent", "")
+ new_ua = f"incremental_to_full/{flag}"
+ parameters["user_agent"] = (
+ f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua
+ )
response = await super().call(
model=model,
diff --git a/dashscope/aigc/image_generation.py b/dashscope/aigc/image_generation.py
index 7dbfbf0..b830495 100644
--- a/dashscope/aigc/image_generation.py
+++ b/dashscope/aigc/image_generation.py
@@ -99,14 +99,15 @@ def call( # type: ignore[override]
to_merge_incremental_output = True
kwargs["incremental_output"] = True
- # Pass incremental_to_full flag via headers user-agent
- if "headers" not in kwargs:
- kwargs["headers"] = {}
-
+ # Pass incremental_to_full flag via user_agent parameter
flag = "1" if to_merge_incremental_output else "0"
- kwargs["headers"]["user-agent"] = f"incremental_to_full/{flag}"
+ existing_ua = kwargs.get("user_agent", "")
+ new_ua = f"incremental_to_full/{flag}"
+ kwargs["user_agent"] = (
+ f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua
+ )
if kwargs.get("is_async", False):
- kwargs["headers"]["X-DashScope-Async"] = "enable"
+ kwargs.setdefault("headers", {})["X-DashScope-Async"] = "enable"
task = cls.async_task
else:
task = cls.sync_task
@@ -177,6 +178,7 @@ def wait(
task: Union[str, ImageGenerationResponse], # type: ignore[override]
api_key: str = None,
workspace: str = None,
+ wait_timeout: int = -1,
**kwargs,
) -> DashScopeAPIResponse:
"""Wait for image(s) synthesis task to complete, and return the result.
@@ -186,11 +188,18 @@ def wait(
ImageGenerationResponse return by async_call().
api_key (str, optional): The api api_key. Defaults to None.
workspace (str): The dashscope workspace id.
+ wait_timeout (int, optional): The maximum seconds to wait.
+ Default is -1 (no timeout).
Returns:
DashScopeAPIResponse: The task result.
"""
- response = super().wait(task, api_key, workspace=workspace)
+ response = super().wait(
+ task,
+ api_key,
+ workspace=workspace,
+ wait_timeout=wait_timeout,
+ )
return ImageGenerationResponse.from_api_response(response)
@classmethod
@@ -408,14 +417,15 @@ async def call( # type: ignore[override]
to_merge_incremental_output = True
kwargs["incremental_output"] = True
- # Pass incremental_to_full flag via headers user-agent
- if "headers" not in kwargs:
- kwargs["headers"] = {}
-
+ # Pass incremental_to_full flag via user_agent parameter
flag = "1" if to_merge_incremental_output else "0"
- kwargs["headers"]["user-agent"] = f"incremental_to_full/{flag}"
+ existing_ua = kwargs.get("user_agent", "")
+ new_ua = f"incremental_to_full/{flag}"
+ kwargs["user_agent"] = (
+ f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua
+ )
if kwargs.get("is_async", False):
- kwargs["headers"]["X-DashScope-Async"] = "enable"
+ kwargs.setdefault("headers", {})["X-DashScope-Async"] = "enable"
task = cls.async_task
else:
task = cls.sync_task
@@ -487,6 +497,7 @@ async def wait(
task: Union[str, ImageGenerationResponse], # type: ignore[override]
api_key: str = None,
workspace: str = None,
+ wait_timeout: int = -1,
**kwargs,
) -> DashScopeAPIResponse:
"""Wait for image(s) synthesis task to complete, and return the result.
@@ -496,11 +507,18 @@ async def wait(
ImageGenerationResponse return by async_call().
api_key (str, optional): The api api_key. Defaults to None.
workspace (str): The dashscope workspace id.
+ wait_timeout (int, optional): The maximum seconds to wait.
+ Default is -1 (no timeout).
Returns:
DashScopeAPIResponse: The task result.
"""
- response = await super().wait(task, api_key, workspace=workspace)
+ response = await super().wait(
+ task,
+ api_key,
+ workspace=workspace,
+ wait_timeout=wait_timeout,
+ )
return ImageGenerationResponse.from_api_response(response)
@classmethod
diff --git a/dashscope/aigc/image_synthesis.py b/dashscope/aigc/image_synthesis.py
index 41bd060..b9843b1 100644
--- a/dashscope/aigc/image_synthesis.py
+++ b/dashscope/aigc/image_synthesis.py
@@ -261,16 +261,16 @@ def _get_input( # pylint: disable=too-many-branches
kwargs["headers"] = headers
def __get_i2i_task(task, model) -> str:
- # 处理task参数：优先使用有效的task值
+ # Handle task parameter: prefer valid task value
if task is not None and task != "":
return task
- # 根据model确定任务类型
+ # Determine task type based on model
if model is not None and model != "":
if "imageedit" in model or "wan2.5-i2i" in model:
return "image2image"
- # 默认返回文本到图像任务
+ # Default to text-to-image task
return ImageSynthesis.task
task = __get_i2i_task(task, model)
@@ -387,6 +387,7 @@ def wait( # type: ignore[override]
task: Union[str, ImageSynthesisResponse],
api_key: str = None,
workspace: str = None,
+ **kwargs,
) -> ImageSynthesisResponse:
"""Wait for image(s) synthesis task to complete, and return the result.
@@ -399,7 +400,12 @@ def wait( # type: ignore[override]
Returns:
ImageSynthesisResponse: The task result.
"""
- response = super().wait(task, api_key, workspace=workspace)
+ response = super().wait(
+ task,
+ api_key,
+ workspace=workspace,
+ **kwargs,
+ )
return ImageSynthesisResponse.from_api_response(response)
@classmethod
@@ -733,6 +739,7 @@ async def wait(
task: Union[str, ImageSynthesisResponse], # type: ignore[override]
api_key: str = None,
workspace: str = None,
+ wait_timeout: int = -1,
**kwargs,
) -> ImageSynthesisResponse:
"""Wait for image(s) synthesis task to complete, and return the result.
@@ -742,11 +749,18 @@ async def wait(
ImageSynthesisResponse return by async_call().
api_key (str, optional): The api api_key. Defaults to None.
workspace (str): The dashscope workspace id.
+ wait_timeout (int, optional): The maximum seconds to wait.
+ Default is -1 (no timeout).
Returns:
ImageSynthesisResponse: The task result.
"""
- response = await super().wait(task, api_key, workspace=workspace)
+ response = await super().wait(
+ task,
+ api_key,
+ workspace=workspace,
+ wait_timeout=wait_timeout,
+ )
return ImageSynthesisResponse.from_api_response(response)
@classmethod
diff --git a/dashscope/aigc/multimodal_conversation.py b/dashscope/aigc/multimodal_conversation.py
index cac2141..8243e77 100644
--- a/dashscope/aigc/multimodal_conversation.py
+++ b/dashscope/aigc/multimodal_conversation.py
@@ -163,11 +163,13 @@ def call( # pylint: disable=arguments-renamed,too-many-branches,too-many-statem
to_merge_incremental_output = True
kwargs["incremental_output"] = True
- # Pass incremental_to_full flag via headers user-agent
- if "headers" not in kwargs:
- kwargs["headers"] = {}
+ # Pass incremental_to_full flag via user_agent parameter
flag = "1" if to_merge_incremental_output else "0"
- kwargs["headers"]["user-agent"] = f"incremental_to_full/{flag}"
+ existing_ua = kwargs.get("user_agent", "")
+ new_ua = f"incremental_to_full/{flag}"
+ kwargs["user_agent"] = (
+ f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua
+ )
response = super().call(
model=model,
diff --git a/dashscope/aigc/video_synthesis.py b/dashscope/aigc/video_synthesis.py
index 0bc5032..f13c155 100644
--- a/dashscope/aigc/video_synthesis.py
+++ b/dashscope/aigc/video_synthesis.py
@@ -509,6 +509,7 @@ def wait( # type: ignore[override]
task: Union[str, VideoSynthesisResponse],
api_key: str = None,
workspace: str = None,
+ **kwargs,
) -> VideoSynthesisResponse:
"""Wait for video synthesis task to complete, and return the result.
@@ -521,7 +522,12 @@ def wait( # type: ignore[override]
Returns:
VideoSynthesisResponse: The task result.
"""
- response = super().wait(task, api_key, workspace=workspace)
+ response = super().wait(
+ task,
+ api_key,
+ workspace=workspace,
+ **kwargs,
+ )
return VideoSynthesisResponse.from_api_response(response)
@classmethod
@@ -888,6 +894,7 @@ async def wait(
task: Union[str, VideoSynthesisResponse], # type: ignore[override]
api_key: str = None,
workspace: str = None,
+ wait_timeout: int = -1,
**kwargs,
) -> VideoSynthesisResponse:
"""Wait for video synthesis task to complete, and return the result.
@@ -897,11 +904,18 @@ async def wait(
VideoSynthesisResponse return by async_call().
api_key (str, optional): The api api_key. Defaults to None.
workspace (str): The dashscope workspace id.
+ wait_timeout (int, optional): The maximum seconds to wait.
+ Default is -1 (no timeout).
Returns:
VideoSynthesisResponse: The task result.
"""
- response = await super().wait(task, api_key, workspace=workspace)
+ response = await super().wait(
+ task,
+ api_key,
+ workspace=workspace,
+ wait_timeout=wait_timeout,
+ )
return VideoSynthesisResponse.from_api_response(response)
@classmethod
diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py
new file mode 100644
index 0000000..2757f3c
--- /dev/null
+++ b/dashscope/api_entities/aio_session.py
@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) Alibaba, Inc. and its affiliates.
+"""Shared aiohttp session pool with cached SSL context.
+
+Provides connection reuse across async API calls. Each event loop gets
+its own ClientSession (aiohttp sessions are loop-bound). The SSL context
+is created once and shared across all sessions.
+"""
+import asyncio
+import ssl
+import threading
+import weakref
+from typing import Optional
+
+import aiohttp
+import certifi
+
+_shared_ssl_context: Optional[ssl.SSLContext] = None
+_aio_sessions: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary()
+_lock = threading.RLock()
+
+
+def get_ssl_context() -> ssl.SSLContext:
+ global _shared_ssl_context
+ with _lock:
+ if _shared_ssl_context is None:
+ _shared_ssl_context = ssl.create_default_context(
+ cafile=certifi.where(),
+ )
+ return _shared_ssl_context
+
+
+async def get_shared_aio_session() -> aiohttp.ClientSession:
+ """Return a shared aiohttp.ClientSession bound to the running event loop.
+
+ The session is lazily created on first use and reused for all
+ subsequent calls on the same event loop. Connection pooling (keep-alive)
+ is handled by the underlying TCPConnector.
+ """
+ loop = asyncio.get_running_loop()
+
+ with _lock:
+ session = _aio_sessions.get(loop)
+ if session is not None and not session.closed:
+ return session
+
+ connector = aiohttp.TCPConnector(ssl=get_ssl_context())
+ session = aiohttp.ClientSession(connector=connector, trust_env=True)
+ _aio_sessions[loop] = session
+ return session
+
+
+async def close_shared_aio_session() -> None:
+ """Close the shared session for the current event loop."""
+ loop = asyncio.get_running_loop()
+ with _lock:
+ session = _aio_sessions.pop(loop, None)
+ if session is not None and not session.closed:
+ await session.close()
diff --git a/dashscope/api_entities/aiohttp_request.py b/dashscope/api_entities/aiohttp_request.py
index 75b3965..e5487b5 100644
--- a/dashscope/api_entities/aiohttp_request.py
+++ b/dashscope/api_entities/aiohttp_request.py
@@ -3,9 +3,11 @@
import json
from http import HTTPStatus
+from typing import Optional
import aiohttp
+from dashscope.api_entities.aio_session import get_shared_aio_session
from dashscope.api_entities.base_request import AioBaseRequest
from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
from dashscope.common.constants import (
@@ -30,6 +32,7 @@ def __init__(
timeout: int = DEFAULT_REQUEST_TIMEOUT_SECONDS,
task_id: str = None,
user_agent: str = "",
+ session: Optional[aiohttp.ClientSession] = None,
) -> None:
"""HttpSSERequest, processing http server sent event stream.
@@ -38,20 +41,26 @@ def __init__(
api_key (str): The api key.
method (str): The http method(GET|POST).
stream (bool, optional): Is stream request. Defaults to True.
- timeout (int, optional): Total request timeout.
+ timeout (int, optional): Request timeout in seconds. For streaming
+ requests, this is the idle timeout between chunks (sock_read);
+ for non-streaming requests, this is the total request timeout.
Defaults to DEFAULT_REQUEST_TIMEOUT_SECONDS.
user_agent (str, optional): Additional user agent string to
append. Defaults to ''.
+ session (aiohttp.ClientSession, optional): External aiohttp
+ session to use instead of the shared session. The caller is
+ responsible for closing it. Defaults to None.
"""
super().__init__(user_agent=user_agent)
self.url = url
self.async_request = async_request
+ self._external_aio_session = session
self.headers = {
"Accept": "application/json",
"Authorization": f"Bearer {api_key}",
"Cache-Control": "no-cache",
- **self.headers,
+ **self.headers, # type: ignore[has-type]
}
self.query = query
if self.async_request and self.query is False:
@@ -244,21 +253,37 @@ async def _handle_response( # pylint: disable=too-many-branches
message=msg.decode("utf-8"),
)
+ # pylint: disable=too-many-branches
async def _handle_request(self):
try:
- async with aiohttp.ClientSession(
- timeout=aiohttp.ClientTimeout(total=self.timeout),
- headers=self.headers,
- ) as session:
+ if self._external_aio_session is not None:
+ session = self._external_aio_session
+ should_close = False
+ else:
+ session = await get_shared_aio_session()
+ should_close = False
+
+ if self.stream:
+ request_timeout = aiohttp.ClientTimeout(
+ total=None,
+ sock_read=self.timeout,
+ )
+ else:
+ request_timeout = aiohttp.ClientTimeout(total=self.timeout)
+
+ try:
logger.debug("Starting request: %s", self.url)
if self.method == HTTPMethod.POST:
- is_form, obj = self.data.get_aiohttp_payload()
+ is_form, obj = False, {}
+ if hasattr(self, "data") and self.data is not None:
+ is_form, obj = self.data.get_aiohttp_payload()
if is_form:
headers = {**self.headers, **obj.headers}
response = await session.post(
url=self.url,
data=obj,
headers=headers,
+ timeout=request_timeout,
)
else:
response = await session.request(
@@ -266,12 +291,17 @@ async def _handle_request(self):
url=self.url,
json=obj,
headers=self.headers,
+ timeout=request_timeout,
)
elif self.method == HTTPMethod.GET:
+ params = {}
+ if hasattr(self, "data") and self.data is not None:
+ params = getattr(self.data, "parameters", {})
response = await session.get(
url=self.url,
- params=self.data.parameters,
+ params=params,
headers=self.headers,
+ timeout=request_timeout,
)
else:
raise UnsupportedHTTPMethod(
@@ -281,9 +311,9 @@ async def _handle_request(self):
async with response:
async for rsp in self._handle_response(response):
yield rsp
- except aiohttp.ClientConnectorError as e:
- logger.error(e)
- raise e
+ finally:
+ if should_close:
+ await session.close()
except Exception as e:
- logger.error(e)
+ logger.debug(e)
raise e
diff --git a/dashscope/api_entities/api_request_data.py b/dashscope/api_entities/api_request_data.py
index 37a8593..aee5664 100644
--- a/dashscope/api_entities/api_request_data.py
+++ b/dashscope/api_entities/api_request_data.py
@@ -157,7 +157,12 @@ def get_batch_binary_data(self) -> bytes: # type: ignore[return]
return content
def _only_parameters(self) -> str:
+ temp_input = None
+ if "raw_input" in self.parameters:
+ temp_input = self.parameters.pop("raw_input")
obj = {"model": self.model, "parameters": self.parameters, "input": {}}
+ if temp_input is not None:
+ obj["input"] = temp_input
if self.task is not None:
obj["task"] = self.task
if self.task_group is not None:
diff --git a/dashscope/api_entities/api_request_factory.py b/dashscope/api_entities/api_request_factory.py
index cc58479..c231287 100644
--- a/dashscope/api_entities/api_request_factory.py
+++ b/dashscope/api_entities/api_request_factory.py
@@ -38,10 +38,16 @@ def _get_protocol_params(kwargs):
extra_url_parameters = kwargs.pop("extra_url_parameters", None)
session = kwargs.pop("session", None)
- # Extract user-agent from headers if present
- user_agent = ""
+ # Extract user_agent from kwargs (preferred) or from headers["user-agent"]
+ user_agent = kwargs.pop("user_agent", "")
if headers and "user-agent" in headers:
- user_agent = headers.pop("user-agent")
+ header_ua = headers.pop("user-agent")
+ if user_agent:
+ user_agent = (
+ f"{header_ua}; {user_agent}" if header_ua else user_agent
+ )
+ else:
+ user_agent = header_ua
return (
api_protocol,
diff --git a/dashscope/api_entities/dashscope_response.py b/dashscope/api_entities/dashscope_response.py
index 71f16b8..58ade6b 100644
--- a/dashscope/api_entities/dashscope_response.py
+++ b/dashscope/api_entities/dashscope_response.py
@@ -59,7 +59,12 @@ def setattr(self, attr, value):
return super().__setitem__(attr, value)
def __getattr__(self, attr):
- return self[attr]
+ try:
+ return self[attr]
+ except KeyError:
+ raise AttributeError(
+ f"{type(self).__name__!r} object has no attribute {attr!r}",
+ ) from None
def __setattr__(self, attr, value):
self[attr] = value
diff --git a/dashscope/api_entities/encryption.py b/dashscope/api_entities/encryption.py
index 266a8c4..59698e4 100644
--- a/dashscope/api_entities/encryption.py
+++ b/dashscope/api_entities/encryption.py
@@ -115,69 +115,69 @@ def _generate_iv():
@staticmethod
def _encrypt_text_with_aes(plaintext, key, iv):
- """使用AES-GCM加密数据"""
+ """Encrypt data with AES-GCM"""
- # 创建AES-GCM加密器
+ # Create AES-GCM encryptor
aes_gcm = Cipher(
algorithms.AES(key),
modes.GCM(iv, tag=None),
backend=default_backend(),
).encryptor()
- # 关联数据设为空（根据需求可调整）
+ # Set associated data to empty (adjustable as needed)
aes_gcm.authenticate_additional_data(b"")
- # 加密数据
+ # Encrypt data
ciphertext = (
aes_gcm.update(plaintext.encode("utf-8")) + aes_gcm.finalize()
)
- # 获取认证标签
+ # Get authentication tag
tag = aes_gcm.tag
- # 组合密文和标签
+ # Combine ciphertext and tag
encrypted_data = ciphertext + tag
- # 返回Base64编码结果
+ # Return Base64 encoded result
return base64.b64encode(encrypted_data).decode("utf-8")
@staticmethod
def _decrypt_text_with_aes(base64_ciphertext, aes_key, iv):
- """使用AES-GCM解密响应"""
+ """Decrypt response with AES-GCM"""
- # 解码Base64数据
+ # Decode Base64 data
encrypted_data = base64.b64decode(base64_ciphertext)
- # 分离密文和标签（标签长度16字节）
+ # Separate ciphertext and tag (tag length is 16 bytes)
ciphertext = encrypted_data[:-16]
tag = encrypted_data[-16:]
- # 创建AES-GCM解密器
+ # Create AES-GCM decryptor
aes_gcm = Cipher(
algorithms.AES(aes_key),
modes.GCM(iv, tag),
backend=default_backend(),
).decryptor()
- # 验证关联数据（与加密时一致）
+ # Verify associated data (same as during encryption)
aes_gcm.authenticate_additional_data(b"")
- # 解密数据
+ # Decrypt data
decrypted_bytes = aes_gcm.update(ciphertext) + aes_gcm.finalize()
- # 明文
+ # Plaintext
plaintext = decrypted_bytes.decode("utf-8")
return json.loads(plaintext)
@staticmethod
def _encrypt_aes_key_with_rsa(aes_key, public_key_str):
- """使用RSA公钥加密AES密钥"""
+ """Encrypt AES key with RSA public key"""
- # 解码Base64格式的公钥
+ # Decode Base64 formatted public key
public_key_bytes = base64.b64decode(public_key_str)
- # 加载公钥
+ # Load public key
public_key = serialization.load_der_public_key(
public_key_bytes,
backend=default_backend(),
@@ -185,7 +185,7 @@ def _encrypt_aes_key_with_rsa(aes_key, public_key_str):
base64_aes_key = base64.b64encode(aes_key).decode("utf-8")
- # 使用RSA加密
+ # Encrypt with RSA
encrypted_bytes = public_key.encrypt(
base64_aes_key.encode("utf-8"),
padding.PKCS1v15(),
diff --git a/dashscope/api_entities/http_request.py b/dashscope/api_entities/http_request.py
index 85ee959..1208bbc 100644
--- a/dashscope/api_entities/http_request.py
+++ b/dashscope/api_entities/http_request.py
@@ -2,14 +2,13 @@
# Copyright (c) Alibaba, Inc. and its affiliates.
import datetime
import json
-import ssl
from http import HTTPStatus
from typing import Optional, Dict, Union
import aiohttp
-import certifi
import requests
+from dashscope.api_entities.aio_session import get_shared_aio_session
from dashscope.api_entities.base_request import AioBaseRequest
from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
from dashscope.common.constants import (
@@ -53,7 +52,9 @@ def __init__(
api_key (str): The api key.
method (str): The http method(GET|POST).
stream (bool, optional): Is stream request. Defaults to True.
- timeout (int, optional): Total request timeout.
+ timeout (int, optional): Request timeout in seconds. For streaming
+ requests, this is the idle timeout between chunks (sock_read);
+ for non-streaming requests, this is the total request timeout.
Defaults to DEFAULT_REQUEST_TIMEOUT_SECONDS.
user_agent (str, optional): Additional user agent string to
append. Defaults to ''.
@@ -162,24 +163,23 @@ async def aio_call(self):
async def _handle_aio_request(self): # pylint: disable=too-many-branches
try:
# Use external aio_session if provided,
- # otherwise create temporary session
+ # otherwise use shared session with connection pooling
if self._external_aio_session is not None:
session = self._external_aio_session
should_close = False
else:
- connector = aiohttp.TCPConnector(
- ssl=ssl.create_default_context(
- cafile=certifi.where(),
- ),
- )
- session = aiohttp.ClientSession(
- connector=connector,
- timeout=aiohttp.ClientTimeout(total=self.timeout),
- headers=self.headers,
- )
- should_close = True
+ session = await get_shared_aio_session()
+ should_close = False
try:
+ if self.stream:
+ request_timeout = aiohttp.ClientTimeout(
+ total=None,
+ sock_read=self.timeout,
+ )
+ else:
+ request_timeout = aiohttp.ClientTimeout(total=self.timeout)
+
logger.debug("Starting request: %s", self.url)
if self.method == HTTPMethod.POST:
is_form, obj = False, {}
@@ -191,6 +191,7 @@ async def _handle_aio_request(self): # pylint: disable=too-many-branches
url=self.url,
data=obj,
headers=headers,
+ timeout=request_timeout,
)
else:
response = await session.request(
@@ -198,9 +199,9 @@ async def _handle_aio_request(self): # pylint: disable=too-many-branches
url=self.url,
json=obj,
headers=self.headers,
+ timeout=request_timeout,
)
elif self.method == HTTPMethod.GET:
- # 添加条件判断
params = {}
if hasattr(self, "data") and self.data is not None:
params = getattr(self.data, "parameters", {})
@@ -210,6 +211,7 @@ async def _handle_aio_request(self): # pylint: disable=too-many-branches
url=self.url,
params=params,
headers=self.headers,
+ timeout=request_timeout,
)
else:
raise UnsupportedHTTPMethod(
@@ -220,14 +222,10 @@ async def _handle_aio_request(self): # pylint: disable=too-many-branches
async for rsp in self._handle_aio_response(response):
yield rsp
finally:
- # Only close if we created the session
if should_close:
await session.close()
- except aiohttp.ClientConnectorError as e:
- logger.error(e)
- raise e
- except BaseException as e:
- logger.error(e)
+ except Exception as e:
+ logger.debug(e)
raise e
@staticmethod
@@ -457,7 +455,7 @@ def _handle_response( # pylint: disable=too-many-branches
else:
yield _handle_http_failed_response(response)
- def _handle_request(self):
+ def _handle_request(self): # pylint: disable=too-many-branches
try:
# Use external session if provided,
# otherwise create temporary session
@@ -470,7 +468,9 @@ def _handle_request(self):
try:
if self.method == HTTPMethod.POST:
- is_form, form, obj = self.data.get_http_payload()
+ is_form, form, obj = False, None, {}
+ if hasattr(self, "data") and self.data is not None:
+ is_form, form, obj = self.data.get_http_payload()
if is_form:
headers = {**self.headers}
headers.pop("Content-Type")
@@ -491,9 +491,12 @@ def _handle_request(self):
timeout=self.timeout,
)
elif self.method == HTTPMethod.GET:
+ params = {}
+ if hasattr(self, "data") and self.data is not None:
+ params = getattr(self.data, "parameters", {})
response = session.get(
url=self.url,
- params=self.data.parameters,
+ params=params,
headers=self.headers,
timeout=self.timeout,
)
@@ -507,6 +510,6 @@ def _handle_request(self):
# Only close if we created the session
if should_close:
session.close()
- except BaseException as e:
- logger.error(e)
+ except Exception as e:
+ logger.debug(e)
raise e
diff --git a/dashscope/api_entities/websocket_request.py b/dashscope/api_entities/websocket_request.py
index 0473709..249b8d8 100644
--- a/dashscope/api_entities/websocket_request.py
+++ b/dashscope/api_entities/websocket_request.py
@@ -119,6 +119,7 @@ async def connection_handler(self): # pylint: disable=too-many-branches
timeout=aiohttp.ClientTimeout(
total=self.timeout,
),
+ trust_env=True,
) as session:
async with session.ws_connect(
self.url,
diff --git a/dashscope/app/application.py b/dashscope/app/application.py
index 78a77a0..0926cbd 100644
--- a/dashscope/app/application.py
+++ b/dashscope/app/application.py
@@ -21,6 +21,7 @@
)
from dashscope.common.error import InputRequired, InvalidInput
from dashscope.common.logging import logger
+from dashscope.utils.message_utils import merge_single_response
class Application(BaseApi):
@@ -149,6 +150,25 @@ def call( # type: ignore[override]
headers["X-DashScope-WorkSpace"] = workspace
kwargs["headers"] = headers
+ # Check if we need to merge incremental output (compute once)
+ is_stream = kwargs.get("stream", False)
+ is_incremental_output = kwargs.get("incremental_output", None)
+ to_merge_incremental_output = (
+ is_stream and is_incremental_output is False
+ )
+
+ if to_merge_incremental_output:
+ kwargs["incremental_output"] = True
+
+ # Pass incremental_to_full flag via user_agent parameter to avoid
+ # overwriting the default SDK user-agent
+ flag = "1" if to_merge_incremental_output else "0"
+ existing_ua = kwargs.get("user_agent", "")
+ new_ua = f"incremental_to_full/{flag}"
+ kwargs["user_agent"] = (
+ f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua
+ )
+
(
input, # pylint: disable=redefined-builtin
parameters,
@@ -171,12 +191,15 @@ def call( # type: ignore[override]
)
# call request service.
response = request.call()
- is_stream = kwargs.get("stream", False)
if is_stream:
- return (
- ApplicationResponse.from_api_response(rsp) for rsp in response
- )
+ if to_merge_incremental_output:
+ return cls._merge_application_response(response)
+ else:
+ return (
+ ApplicationResponse.from_api_response(rsp)
+ for rsp in response
+ )
else:
return ApplicationResponse.from_api_response(response)
@@ -238,3 +261,33 @@ def _build_input_parameters( # pylint: disable=too-many-branches
input_param["file_list"] = file_list
return input_param, {**parameters, **kwargs}
+
+ @classmethod
+ def _merge_application_response(cls, response):
+ """Merge incremental application response chunks.
+
+ Simulate non-incremental output by accumulating text.
+ """
+ accumulated_data = {}
+
+ for rsp in response:
+ parsed_response = ApplicationResponse.from_api_response(rsp)
+ if parsed_response.output and not hasattr(
+ parsed_response.output,
+ "choices",
+ ):
+ parsed_response.output.choices = None
+ result = merge_single_response(
+ parsed_response,
+ accumulated_data,
+ )
+ if result is True:
+ yield parsed_response
+ elif isinstance(result, list):
+ for resp in result:
+ yield resp
+ else:
+ logger.warning(
+ "Unexpected merge result type: %s, skipping",
+ type(result).__name__,
+ )
diff --git a/dashscope/audio/asr/recognition.py b/dashscope/audio/asr/recognition.py
index f243d44..7dc7bc4 100644
--- a/dashscope/audio/asr/recognition.py
+++ b/dashscope/audio/asr/recognition.py
@@ -444,8 +444,8 @@ def call( # type: ignore[override] # noqa: E501
f.close()
self._stop_stream_timestamp = time.time() * 1000
except Exception as e:
- logger.error(e)
- raise e
+ logger.debug(e)
+ raise
if not self._stream_data.empty():
self._running = True
@@ -600,6 +600,6 @@ def get_last_package_delay(self):
"""Last Package Delay is the time between stop sending audio and receive last words package""" # noqa: E501 # pylint: disable=line-too-long
return self._on_complete_timestamp - self._stop_stream_timestamp
- # 获取上一个任务的taskId
+ # Get the requestId of the last task
def get_last_request_id(self):
return self.last_request_id
diff --git a/dashscope/audio/asr/transcription.py b/dashscope/audio/asr/transcription.py
index 480b328..88631ac 100644
--- a/dashscope/audio/asr/transcription.py
+++ b/dashscope/audio/asr/transcription.py
@@ -1,11 +1,10 @@
# -*- coding: utf-8 -*-
# Copyright (c) Alibaba, Inc. and its affiliates.
-import asyncio
import time
from typing import List, Union
-import aiohttp
+import requests
from dashscope.api_entities.dashscope_response import (
DashScopeAPIResponse,
@@ -148,8 +147,8 @@ def fetch(
workspace=workspace,
**kwargs,
)
- except (asyncio.TimeoutError, aiohttp.ClientConnectorError) as e:
- logger.error(e)
+ except (requests.Timeout, requests.ConnectionError) as e:
+ logger.debug(e)
try_count += 1
if try_count <= Transcription.MAX_QUERY_TRY_COUNT:
time.sleep(2)
@@ -167,6 +166,7 @@ def wait(
task: Union[str, TranscriptionResponse], # type: ignore[override]
api_key: str = None,
workspace: str = None,
+ wait_timeout: int = -1,
**kwargs,
) -> TranscriptionResponse:
"""Poll task until the final results of transcription is obtained.
@@ -174,7 +174,12 @@ def wait(
Args:
task (Union[str, TranscriptionResponse]): The task_id or
response including task_id returned from async_call().
+ api_key (str, optional): The api_key. Defaults to None.
workspace (str): The dashscope workspace id.
+ wait_timeout (int, optional): The timeout for waiting.
+ Defaults to -1.That means no timeout.
+ If set to a value > 0, the task does not complete
+ within this time, a timeout error response will be returned.
Returns:
TranscriptionResponse: The result of batch transcription.
@@ -183,6 +188,7 @@ def wait(
task,
api_key=api_key,
workspace=workspace,
+ wait_timeout=wait_timeout,
**kwargs,
)
return TranscriptionResponse.from_api_response(response)
@@ -223,8 +229,8 @@ def _launch_request(
workspace=workspace,
**kwargs,
)
- except (asyncio.TimeoutError, aiohttp.ClientConnectorError) as e:
- logger.error(e)
+ except (requests.Timeout, requests.ConnectionError) as e:
+ logger.debug(e)
try_count += 1
if try_count <= Transcription.MAX_QUERY_TRY_COUNT:
time.sleep(2)
diff --git a/dashscope/audio/asr/translation_recognizer.py b/dashscope/audio/asr/translation_recognizer.py
index 83dcaa5..2f3405c 100644
--- a/dashscope/audio/asr/translation_recognizer.py
+++ b/dashscope/audio/asr/translation_recognizer.py
@@ -582,8 +582,8 @@ def call( # type: ignore[override]
f.close()
self._stop_stream_timestamp = time.time() * 1000
except Exception as e:
- logger.error(e)
- raise e
+ logger.debug(e)
+ raise
if not self._stream_data.empty():
self._running = True
@@ -744,7 +744,7 @@ def get_last_package_delay(self):
"""Last Package Delay is the time between stop sending audio and receive last words package""" # noqa: E501 # pylint: disable=line-too-long
return self._on_complete_timestamp - self._stop_stream_timestamp
- # 获取上一个任务的taskId
+ # Get the taskId of the last task
def get_last_request_id(self):
return self.last_request_id
@@ -1087,6 +1087,6 @@ def get_last_package_delay(self):
"""Last Package Delay is the time between stop sending audio and receive last words package""" # noqa: E501 # pylint: disable=line-too-long
return self._on_complete_timestamp - self._stop_stream_timestamp
- # 获取上一个任务的taskId
+ # Get the taskId of the last task
def get_last_request_id(self):
return self.last_request_id
diff --git a/dashscope/audio/asr/vocabulary.py b/dashscope/audio/asr/vocabulary.py
index 1e96e28..9e8510b 100644
--- a/dashscope/audio/asr/vocabulary.py
+++ b/dashscope/audio/asr/vocabulary.py
@@ -1,11 +1,10 @@
# -*- coding: utf-8 -*-
# Copyright (c) Alibaba, Inc. and its affiliates.
-import asyncio
import time
from typing import List
-import aiohttp
+import requests
from dashscope.client.base_api import BaseApi
from dashscope.common.constants import ApiProtocol, HTTPMethod
@@ -68,8 +67,8 @@ def __call_with_input(self, input): # pylint: disable=redefined-builtin
workspace=self._workspace,
**self._kwargs,
)
- except (asyncio.TimeoutError, aiohttp.ClientConnectorError) as e:
- logger.error(e)
+ except (requests.Timeout, requests.ConnectionError) as e:
+ logger.debug(e)
try_count += 1
if try_count <= VocabularyService.MAX_QUERY_TRY_COUNT:
time.sleep(2)
@@ -86,11 +85,12 @@ def create_vocabulary(
vocabulary: List[dict],
) -> str:
"""
- 创建热词表
- param: target_model 热词表对应的语音识别模型版本
- param: prefix 热词表自定义前缀，仅允许数字和小写字母，小于十个字符。
- param: vocabulary 热词表字典
- return: 热词表标识符 vocabulary_id
+ Create a hot word table.
+ param: target_model ASR model version for the hot word table
+ param: prefix Custom hot word table prefix, only digits and
+ lowercase letters allowed, less than 10 characters.
+ param: vocabulary Hot word table dictionary
+ return: Hot word table identifier vocabulary_id
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
@@ -119,11 +119,12 @@ def list_vocabularies(
page_size: int = 10,
) -> List[dict]:
"""
- 查询已创建的所有热词表
- param: prefix 自定义前缀，如果设定则只返回指定前缀的热词表标识符列表。
- param: page_index 查询的页索引
- param: page_size 查询页大小
- return: 热词表标识符列表
+ List all created hot word tables.
+ param: prefix Custom prefix, if set only returns hot word table
+ identifiers with the specified prefix.
+ param: page_index Page index for query
+ param: page_size Page size
+ return: List of hot word table identifiers
"""
if prefix:
# pylint: disable=no-value-for-parameter
@@ -157,9 +158,9 @@ def list_vocabularies(
def query_vocabulary(self, vocabulary_id: str) -> List[dict]:
"""
- 获取热词表内容
- param: vocabulary_id 热词表标识符
- return: 热词表
+ Get hot word table contents.
+ param: vocabulary_id Hot word table identifier
+ return: Hot word table
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
@@ -185,9 +186,9 @@ def update_vocabulary(
vocabulary: List[dict],
) -> None:
"""
- 用新的热词表替换已有热词表
- param: vocabulary_id 需要替换的热词表标识符
- param: vocabulary 热词表
+ Replace existing hot word table with a new one.
+ param: vocabulary_id Hot word table identifier to replace
+ param: vocabulary Hot word table
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
@@ -210,8 +211,8 @@ def update_vocabulary(
def delete_vocabulary(self, vocabulary_id: str) -> None:
"""
- 删除热词表
- param: vocabulary_id 需要删除的热词表标识符
+ Delete hot word table.
+ param: vocabulary_id Hot word table identifier to delete
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
diff --git a/dashscope/audio/qwen_asr/qwen_transcription.py b/dashscope/audio/qwen_asr/qwen_transcription.py
index 9ebed8d..f731ff9 100644
--- a/dashscope/audio/qwen_asr/qwen_transcription.py
+++ b/dashscope/audio/qwen_asr/qwen_transcription.py
@@ -1,11 +1,10 @@
# -*- coding: utf-8 -*-
# Copyright (c) Alibaba, Inc. and its affiliates.
-import asyncio
import time
from typing import Union
-import aiohttp
+import requests
from dashscope.api_entities.dashscope_response import (
DashScopeAPIResponse,
@@ -108,8 +107,8 @@ def fetch(
workspace=workspace,
**kwargs,
)
- except (asyncio.TimeoutError, aiohttp.ClientConnectorError) as e:
- logger.error(e)
+ except (requests.Timeout, requests.ConnectionError) as e:
+ logger.debug(e)
try_count += 1
if try_count <= QwenTranscription.MAX_QUERY_TRY_COUNT:
time.sleep(2)
@@ -127,6 +126,7 @@ def wait(
task: Union[str, TranscriptionResponse], # type: ignore[override]
api_key: str = None,
workspace: str = None,
+ wait_timeout: int = -1,
**kwargs,
) -> TranscriptionResponse:
"""Poll task until the final results of transcription is obtained.
@@ -135,6 +135,8 @@ def wait(
task (Union[str, TranscriptionResponse]): The task_id or
response including task_id returned from async_call().
workspace (str): The dashscope workspace id.
+ wait_timeout (int, optional): The maximum seconds to wait.
+ Default is -1 (no timeout).
Returns:
TranscriptionResponse: The result of batch transcription.
@@ -143,6 +145,7 @@ def wait(
task,
api_key=api_key,
workspace=workspace,
+ wait_timeout=wait_timeout,
**kwargs,
)
return TranscriptionResponse.from_api_response(response)
@@ -182,8 +185,8 @@ def _launch_request(
workspace=workspace,
**kwargs,
)
- except (asyncio.TimeoutError, aiohttp.ClientConnectorError) as e:
- logger.error(e)
+ except (requests.Timeout, requests.ConnectionError) as e:
+ logger.debug(e)
try_count += 1
if try_count <= QwenTranscription.MAX_QUERY_TRY_COUNT:
time.sleep(2)
diff --git a/dashscope/audio/qwen_omni/omni_realtime.py b/dashscope/audio/qwen_omni/omni_realtime.py
index dc7cda6..4f15735 100644
--- a/dashscope/audio/qwen_omni/omni_realtime.py
+++ b/dashscope/audio/qwen_omni/omni_realtime.py
@@ -151,8 +151,9 @@ def __init__(
self.last_first_text_delay = None
self.last_first_audio_delay = None
self.metrics = []
- # 添加用于同步等待连接关闭的事件
+ # Add event for synchronously waiting on connection close
self.disconnect_event = None
+ self._disconnect_error = None
def _generate_event_id(self):
"""
@@ -189,13 +190,13 @@ def connect(self) -> None:
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
- timeout = 5 # 最长等待时间（秒）
+ timeout = 5 # max wait time in seconds
start_time = time.time()
while (
not (self.ws.sock and self.ws.sock.connected)
and (time.time() - start_time) < timeout
):
- time.sleep(0.1) # 短暂休眠，避免密集轮询
+ time.sleep(0.1) # Brief sleep to avoid busy polling
if not (self.ws.sock and self.ws.sock.connected):
raise TimeoutError(
"websocket connection could not established within 5s. "
@@ -350,6 +351,7 @@ def end_session(self, timeout: int = 20) -> None:
# create the event
self.disconnect_event = threading.Event()
+ self._disconnect_error = None
self.__send_str(
json.dumps(
@@ -362,10 +364,14 @@ def end_session(self, timeout: int = 20) -> None:
# wait for the event to be set
finish_success = self.disconnect_event.wait(timeout)
- # clear the event
+ error = self._disconnect_error
self.disconnect_event = None
+ self._disconnect_error = None
- # if the event is not set, close the connection
+ # if the server returned an error or timed out, close the connection
+ if error is not None:
+ self.close()
+ raise RuntimeError(f"Session ended with error: {error}")
if not finish_success:
self.close()
raise TimeoutError(
@@ -376,7 +382,7 @@ def end_session_async(self) -> None:
"""
end session asynchronously. you need close the connection manually
"""
- # 发送结束会话消息
+ # Send end session message
self.__send_str(
json.dumps(
{
@@ -511,7 +517,7 @@ def close(self) -> None:
"""
self.ws.close()
- # 监听消息的回调函数
+ # Callback for listening to messages
def _on_message( # pylint: disable=unused-argument,too-many-branches
self,
ws,
@@ -523,7 +529,7 @@ def _on_message( # pylint: disable=unused-argument,too-many-branches
message[:1024],
)
try:
- # 尝试将消息解析为JSON
+ # Attempt to parse message as JSON
json_data = json.loads(message)
self.last_message = json_data
self.callback.on_event(json_data)
@@ -536,6 +542,14 @@ def _on_message( # pylint: disable=unused-argument,too-many-branches
logger.info("[omni realtime] session finished")
if self.disconnect_event is not None:
self.disconnect_event.set()
+ elif "error" == json_data.get("type"):
+ if self.disconnect_event is not None:
+ self._disconnect_error = json_data.get("error")
+ logger.warning(
+ "[omni realtime] error during end_session: %s",
+ self._disconnect_error,
+ )
+ self.disconnect_event.set()
if "response.created" == json_data["type"]:
self.last_response_id = json_data["response"]["id"]
self.last_response_create_time = time.time() * 1000
@@ -574,7 +588,7 @@ def _on_message( # pylint: disable=unused-argument,too-many-branches
# pylint: disable=broad-exception-raised,raise-missing-from
raise Exception("Failed to parse message as JSON.")
elif isinstance(message, (bytes, bytearray)):
- # 如果失败，认为是二进制消息
+ # If parsing fails, treat as binary message
logger.error(
"should not receive binary message in omni realtime api",
)
@@ -591,13 +605,13 @@ def _on_close( # pylint: disable=unused-argument
):
self.callback.on_close(close_status_code, close_msg)
- # WebSocket发生错误的回调函数
+ # Callback for WebSocket error
def _on_error(self, ws, error): # pylint: disable=unused-argument
# pylint: disable=broad-exception-raised
logger.error("websocket closed due to %s", error)
raise Exception(f"websocket closed due to {error}")
- # 获取上一个任务的taskId
+ # Get the taskId of the last task
def get_session_id(self) -> str:
return self.session_id # type: ignore[return-value]
diff --git a/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py b/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py
index 7142a66..415c098 100644
--- a/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py
+++ b/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py
@@ -141,13 +141,13 @@ def connect(self) -> None:
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
- timeout = 5 # 最长等待时间（秒）
+ timeout = 5 # max wait time in seconds
start_time = time.time()
while (
not (self.ws.sock and self.ws.sock.connected)
and (time.time() - start_time) < timeout
):
- time.sleep(0.1) # 短暂休眠，避免密集轮询
+ time.sleep(0.1) # Brief sleep to avoid busy polling
if not (self.ws.sock and self.ws.sock.connected):
raise TimeoutError(
"websocket connection could not established within 5s. "
@@ -215,14 +215,16 @@ def update_session(
"response_format": response_format.format,
"sample_rate": response_format.sample_rate,
}
- if sample_rate is not None: # 如果配置，则更新
+ if sample_rate is not None: # update if configured
self.config["sample_rate"] = sample_rate
if volume is not None:
self.config["volume"] = volume
if speech_rate is not None:
self.config["speech_rate"] = speech_rate
if audio_format is not None:
- self.config["response_format"] = audio_format # 如果配置，则更新
+ self.config[
+ "response_format"
+ ] = audio_format # update if configured
if pitch_rate is not None:
self.config["pitch_rate"] = pitch_rate
if bit_rate is not None:
@@ -332,7 +334,7 @@ def close(self) -> None:
"""
self.ws.close()
- # 监听消息的回调函数
+ # Callback for listening to messages
def on_message( # pylint: disable=unused-argument
self,
ws,
@@ -344,7 +346,7 @@ def on_message( # pylint: disable=unused-argument
message[:1024],
)
try:
- # 尝试将消息解析为JSON
+ # Attempt to parse message as JSON
json_data = json.loads(message)
self.last_message = json_data
self.callback.on_event(json_data)
@@ -372,7 +374,7 @@ def on_message( # pylint: disable=unused-argument
# pylint: disable=broad-exception-raised,raise-missing-from
raise Exception("Failed to parse message as JSON.")
elif isinstance(message, (bytes, bytearray)):
- # 如果失败，认为是二进制消息
+ # If parsing fails, treat as binary message
logger.error(
"should not receive binary message in omni realtime api",
)
@@ -394,13 +396,13 @@ def on_close( # pylint: disable=unused-argument
)
self.callback.on_close(close_status_code, close_msg)
- # WebSocket发生错误的回调函数
+ # Callback for WebSocket error
def on_error(self, ws, error): # pylint: disable=unused-argument
print(f"websocket closed due to {error}")
# pylint: disable=broad-exception-raised
raise Exception(f"websocket closed due to {error}")
- # 获取上一个任务的taskId
+ # Get the taskId of the last task
def get_session_id(self):
return self.session_id
diff --git a/dashscope/audio/tts_v2/enrollment.py b/dashscope/audio/tts_v2/enrollment.py
index 2483862..239606a 100644
--- a/dashscope/audio/tts_v2/enrollment.py
+++ b/dashscope/audio/tts_v2/enrollment.py
@@ -1,11 +1,10 @@
# -*- coding: utf-8 -*-
# Copyright (c) Alibaba, Inc. and its affiliates.
-import asyncio
import time
from typing import List
-import aiohttp
+import requests
from dashscope.client.base_api import BaseApi
from dashscope.common.constants import ApiProtocol, HTTPMethod
@@ -71,8 +70,8 @@ def __call_with_input( # pylint: disable=redefined-builtin
workspace=self._workspace,
**self._kwargs,
)
- except (asyncio.TimeoutError, aiohttp.ClientConnectorError) as e:
- logger.error(e)
+ except (requests.Timeout, requests.ConnectionError) as e:
+ logger.debug(e)
try_count += 1
if try_count <= VoiceEnrollmentService.MAX_QUERY_TRY_COUNT:
time.sleep(2)
@@ -92,13 +91,15 @@ def create_voice(
**kwargs,
) -> str:
"""
- 创建新克隆音色
- param: target_model 克隆音色对应的语音合成模型版本
- param: prefix 音色自定义前缀，仅允许数字和小写字母，小于十个字符。
- param: url 用于克隆的音频文件url
- param: language_hints 克隆音色目标语言
- param: max_prompt_audio_length 音频预处理输出的prompt audio最长长度。单位为秒。默认为10s。
- param: kwargs 额外参数
+ Create a new cloned voice.
+ param: target_model TTS model version for the cloned voice
+ param: prefix Custom voice prefix, only digits and lowercase
+ letters allowed, less than 10 characters.
+ param: url Audio file URL for voice cloning
+ param: language_hints Target language for the cloned voice
+ param: max_prompt_audio_length Max length of prompt audio output
+ from audio preprocessing, in seconds. Default is 10s.
+ param: kwargs Additional parameters
return: voice_id
"""
@@ -133,10 +134,11 @@ def list_voices(
page_size: int = 10,
) -> List[dict]:
"""
- 查询已创建的所有音色
- param: page_index 查询的页索引
- param: page_size 查询页大小
- return: List[dict] 音色列表，包含每个音色的id，创建时间，修改时间，状态。
+ List all created voices.
+ param: page_index Page index for query
+ param: page_size Page size
+ return: List[dict] Voice list, including id, creation time,
+ modification time, and status for each voice.
"""
if prefix:
# pylint: disable=no-value-for-parameter
@@ -170,9 +172,9 @@ def list_voices(
def query_voice(self, voice_id: str) -> List[str]:
"""
- 查询已创建的所有音色
- param: voice_id 需要查询的音色
- return: bytes 注册音色使用的音频
+ Query voice details.
+ param: voice_id Voice ID to query
+ return: bytes Audio used for voice registration
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
@@ -194,9 +196,9 @@ def query_voice(self, voice_id: str) -> List[str]:
def update_voice(self, voice_id: str, url: str) -> None:
"""
- 更新音色
- param: voice_id 音色id
- param: url 用于克隆的音频文件url
+ Update voice.
+ param: voice_id Voice ID
+ param: url Audio file URL for cloning
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
@@ -219,8 +221,8 @@ def update_voice(self, voice_id: str, url: str) -> None:
def delete_voice(self, voice_id: str) -> None:
"""
- 删除音色
- param: voice_id 需要删除的音色
+ Delete voice.
+ param: voice_id Voice ID to delete
"""
# pylint: disable=no-value-for-parameter
response = self.__call_with_input(
diff --git a/dashscope/audio/tts_v2/speech_synthesizer.py b/dashscope/audio/tts_v2/speech_synthesizer.py
index 5f5af14..4420456 100644
--- a/dashscope/audio/tts_v2/speech_synthesizer.py
+++ b/dashscope/audio/tts_v2/speech_synthesizer.py
@@ -167,7 +167,7 @@ def __init__( # pylint: disable=redefined-builtin
self.language_hints = language_hints
def gen_uid(self):
- # 生成随机UUID
+ # Generate random UUID
return uuid.uuid4().hex
def get_websocket_headers(self, headers, workspace):
@@ -403,13 +403,13 @@ def __connect(self, timeout_seconds=5) -> None:
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
- # 等待连接建立
+ # Wait for connection to be established
start_time = time.time()
while (
not (self.ws.sock and self.ws.sock.connected)
and (time.time() - start_time) < timeout_seconds
):
- time.sleep(0.1) # 短暂休眠，避免密集轮询
+ time.sleep(0.1) # Brief sleep to avoid busy polling
if not (self.ws.sock and self.ws.sock.connected):
raise TimeoutError(
"websocket connection could not established within 5s. "
@@ -540,10 +540,10 @@ def __start_stream(self):
if self._is_started:
raise InvalidTask("task has already started.")
- # 建立ws连接
+ # Establish WebSocket connection
if self.ws is None:
self.__connect(5)
- # 发送run-task指令
+ # Send run-task command
request = self.request.get_start_request(self.additional_params)
self.__send_str(request)
if not self.start_event.wait(10):
@@ -680,7 +680,7 @@ def streaming_cancel(self):
self.start_event.set()
self.complete_event.set()
- # 监听消息的回调函数
+ # Callback for listening to messages
def on_message( # pylint: disable=unused-argument,too-many-branches
self,
ws,
@@ -689,11 +689,11 @@ def on_message( # pylint: disable=unused-argument,too-many-branches
if isinstance(message, str):
logger.debug("<<