diff --git a/README.md b/README.md index 6190b4e..02b6ab4 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,3 @@ -

-

- 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/aigc/image_synthesis.py b/dashscope/aigc/image_synthesis.py index 0491d7e..0409bfc 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) diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py new file mode 100644 index 0000000..d2365db --- /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) + _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 c89491e..598be25 100644 --- a/dashscope/api_entities/aiohttp_request.py +++ b/dashscope/api_entities/aiohttp_request.py @@ -6,6 +6,7 @@ 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 ( @@ -38,7 +39,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 ''. @@ -245,43 +248,51 @@ async def _handle_response( # pylint: disable=too-many-branches ) async def _handle_request(self): - async with aiohttp.ClientSession( - timeout=aiohttp.ClientTimeout(total=self.timeout), - headers=self.headers, - ) as session: - logger.debug("Starting request: %s", self.url) - if self.method == HTTPMethod.POST: - 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, - ) - else: - response = await session.request( - "POST", - url=self.url, - json=obj, - headers=self.headers, - ) - 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( + session = await get_shared_aio_session() + 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, {} + 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, - params=params, - headers=self.headers, + data=obj, + headers=headers, + timeout=request_timeout, ) else: - raise UnsupportedHTTPMethod( - f"Unsupported http method: {self.method}", + response = await session.request( + "POST", + url=self.url, + json=obj, + headers=self.headers, + timeout=request_timeout, ) - logger.debug("Response returned: %s", self.url) - async with response: - async for rsp in self._handle_response(response): - yield rsp + 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=params, + headers=self.headers, + timeout=request_timeout, + ) + else: + raise UnsupportedHTTPMethod( + f"Unsupported http method: {self.method}", + ) + logger.debug("Response returned: %s", self.url) + async with response: + async for rsp in self._handle_response(response): + yield rsp 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 e361503..3142328 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 ''. @@ -161,67 +162,61 @@ async def aio_call(self): async def _handle_aio_request(self): # pylint: disable=too-many-branches # 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, + session = await get_shared_aio_session() + + if self.stream: + request_timeout = aiohttp.ClientTimeout( + total=None, + sock_read=self.timeout, ) - should_close = True + else: + request_timeout = aiohttp.ClientTimeout(total=self.timeout) - try: - logger.debug("Starting request: %s", self.url) - if self.method == HTTPMethod.POST: - 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, - ) - else: - response = await session.request( - "POST", - url=self.url, - json=obj, - headers=self.headers, - ) - elif self.method == HTTPMethod.GET: - # 添加条件判断 - params = {} - if hasattr(self, "data") and self.data is not None: - params = getattr(self.data, "parameters", {}) - if params: - params = self.__handle_parameters(params) - response = await session.get( + logger.debug("Starting request: %s", self.url) + if self.method == HTTPMethod.POST: + 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, - params=params, - headers=self.headers, + data=obj, + headers=headers, + timeout=request_timeout, ) else: - raise UnsupportedHTTPMethod( - f"Unsupported http method: {self.method}", + response = await session.request( + "POST", + url=self.url, + json=obj, + headers=self.headers, + timeout=request_timeout, ) - logger.debug("Response returned: %s", self.url) - async with response: - 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() + elif self.method == HTTPMethod.GET: + params = {} + if hasattr(self, "data") and self.data is not None: + params = getattr(self.data, "parameters", {}) + if params: + params = self.__handle_parameters(params) + response = await session.get( + url=self.url, + params=params, + headers=self.headers, + timeout=request_timeout, + ) + else: + raise UnsupportedHTTPMethod( + f"Unsupported http method: {self.method}", + ) + logger.debug("Response returned: %s", self.url) + async with response: + async for rsp in self._handle_aio_response(response): + yield rsp @staticmethod def __handle_parameters(params: dict) -> dict: diff --git a/dashscope/audio/asr/recognition.py b/dashscope/audio/asr/recognition.py index f243d44..8b988f6 100644 --- a/dashscope/audio/asr/recognition.py +++ b/dashscope/audio/asr/recognition.py @@ -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/translation_recognizer.py b/dashscope/audio/asr/translation_recognizer.py index 83dcaa5..f677dd2 100644 --- a/dashscope/audio/asr/translation_recognizer.py +++ b/dashscope/audio/asr/translation_recognizer.py @@ -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..8fd2ce8 100644 --- a/dashscope/audio/asr/vocabulary.py +++ b/dashscope/audio/asr/vocabulary.py @@ -86,11 +86,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 +120,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 +159,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 +187,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 +212,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_omni/omni_realtime.py b/dashscope/audio/qwen_omni/omni_realtime.py index dc7cda6..9aeabdd 100644 --- a/dashscope/audio/qwen_omni/omni_realtime.py +++ b/dashscope/audio/qwen_omni/omni_realtime.py @@ -151,7 +151,7 @@ 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 def _generate_event_id(self): @@ -189,13 +189,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. " @@ -376,7 +376,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 +511,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 +523,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) @@ -574,7 +574,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 +591,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..9a728c3 100644 --- a/dashscope/audio/tts_v2/enrollment.py +++ b/dashscope/audio/tts_v2/enrollment.py @@ -92,13 +92,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 +135,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 +173,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 +197,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 +222,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("<< 100: raise ValueError("max_size must be less than 100") self._pool = [] - # 如果重连中，则会将avaliable置为False，避免被使用 + # If reconnecting, set available to False to avoid being used self._avaliable = [] self._pool_size = max_size for i in range(self._pool_size): # pylint: disable=unused-variable @@ -918,7 +919,7 @@ def __auto_reconnect(self): current_time = time.time() for idx, poolObject in enumerate(self._pool): - # 如果超过固定时间没有使用对象，则重连 + # Reconnect if object has not been used for a fixed time if poolObject.connect_time == -1: objects_need_to_connect.append(poolObject) self._avaliable[idx] = False @@ -995,7 +996,8 @@ def borrow_synthesizer( # pylint: disable=unused-argument,redefined-builtin # n logger.debug("[SpeechSynthesizerObjectPool] get synthesizer") synthesizer: SpeechSynthesizer = None with self._lock: - # 遍历对象池，如果存在预建连的对象，则返回 + # Iterate over object pool, return pre-connected object + # if available for idx, poolObject in enumerate(self._pool): if ( self._avaliable[idx] @@ -1009,7 +1011,7 @@ def borrow_synthesizer( # pylint: disable=unused-argument,redefined-builtin # n self._avaliable.pop(idx) break - # 如果对象池不足，则返回未建连的新对象 + # If pool is exhausted, return a new unconnected object if synthesizer is None: synthesizer = self.__get_default_synthesizer() logger.warning( diff --git a/dashscope/client/base_api.py b/dashscope/client/base_api.py index cc52fa2..a609584 100644 --- a/dashscope/client/base_api.py +++ b/dashscope/client/base_api.py @@ -81,9 +81,10 @@ def _handle_kwargs( @classmethod async def _handle_request(cls, request): - # 如果 aio_call 返回的是异步生成器，则需要从中获取响应 + # If aio_call returns an async generator, consume it to get + # the response response = await request.aio_call() - # 处理异步生成器的情况 + # Handle async generator case if isinstance(response, collections.abc.AsyncGenerator): result = None async for item in response: @@ -236,7 +237,7 @@ async def wait( return rsp else: logger.info("The task %s is %s", task_id, task_status) - await asyncio.sleep(wait_seconds) # 异步等待 + await asyncio.sleep(wait_seconds) # async wait elif rsp.status_code in REPEATABLE_STATUS: logger.warning( "Get task: %s temporary failure, " @@ -246,7 +247,7 @@ async def wait( rsp.code, rsp.message, ) - await asyncio.sleep(wait_seconds) # 异步等待 + await asyncio.sleep(wait_seconds) # async wait else: return rsp diff --git a/dashscope/finetune/reinforcement/common/utils.py b/dashscope/finetune/reinforcement/common/utils.py index d6ee0f7..3757460 100644 --- a/dashscope/finetune/reinforcement/common/utils.py +++ b/dashscope/finetune/reinforcement/common/utils.py @@ -897,7 +897,7 @@ def _parse_decorator_args(decorator) -> Dict[str, Any]: """Extract name and sub_weight from a decorator AST node.""" args_dict = {} if isinstance(decorator, ast.Call): - # 处理 args 和 keywords + # Process args and keywords for i, arg in enumerate(decorator.args): if i == 0: args_dict["name"] = _resolve_str_literal(arg) diff --git a/dashscope/multimodal/dialog_state.py b/dashscope/multimodal/dialog_state.py index 1bead25..a0c76d6 100644 --- a/dashscope/multimodal/dialog_state.py +++ b/dashscope/multimodal/dialog_state.py @@ -6,13 +6,14 @@ class DialogState(Enum): """ - 对话状态枚举类，定义了对话机器人可能处于的不同状态。 + Dialog state enumeration class, defining the possible states + of a dialog bot. Attributes: - IDLE (str): 表示机器人处于空闲状态。 - LISTENING (str): 表示机器人正在监听用户输入。 - THINKING (str): 表示机器人正在思考。 - RESPONDING (str): 表示机器人正在生成或回复中。 + IDLE (str): Bot is in idle state. + LISTENING (str): Bot is listening to user input. + THINKING (str): Bot is thinking. + RESPONDING (str): Bot is generating or responding. """ IDLE = "Idle" @@ -23,36 +24,36 @@ class DialogState(Enum): class StateMachine: """ - 状态机类，用于管理机器人的状态转换。 + State machine class for managing bot state transitions. Attributes: - current_state (DialogState): 当前状态。 + current_state (DialogState): Current state. """ def __init__(self): - # 初始化状态机时设置初始状态为IDLE + # Set initial state to IDLE when initializing the state machine self.current_state = DialogState.IDLE def change_state(self, new_state: str) -> None: """ - 更改当前状态到指定的新状态。 + Change the current state to the specified new state. Args: - new_state (str): 要切换到的新状态。 + new_state (str): The new state to switch to. Raises: - ValueError: 如果尝试切换到一个无效的状态，则抛出此异常。 + ValueError: If attempting to switch to an invalid state. """ if new_state in [state.value for state in DialogState]: self.current_state = DialogState(new_state) else: - raise ValueError("无效的状态类型") + raise ValueError("Invalid state type") def get_current_state(self) -> DialogState: """ - 获取当前状态。 + Get the current state. Returns: - DialogState: 当前状态。 + DialogState: The current state. """ return self.current_state diff --git a/dashscope/multimodal/multimodal_constants.py b/dashscope/multimodal/multimodal_constants.py index 12a2cb0..86aaaeb 100644 --- a/dashscope/multimodal/multimodal_constants.py +++ b/dashscope/multimodal/multimodal_constants.py @@ -1,6 +1,5 @@ # -*- coding: utf-8 -*- # Copyright (c) Alibaba, Inc. and its affiliates. -# -*- coding: utf-8 -*- # multimodal conversation request directive @@ -21,12 +20,17 @@ class RequestToRespondType: RESPONSE_NAME_STATE_CHANGED = "DialogStateChanged" RESPONSE_NAME_REQUEST_ACCEPTED = "RequestAccepted" RESPONSE_NAME_SPEECH_STARTED = "SpeechStarted" -RESPONSE_NAME_SPEECH_ENDED = "SpeechEnded" # 服务端检测到asr语音尾点时下发此事件,可选事件 -RESPONSE_NAME_RESPONDING_STARTED = ( - "RespondingStarted" # AI语音应答开始，sdk要准备接收服务端下发的语音数据 +# Server sends this event when ASR speech endpoint is detected, +# optional event +RESPONSE_NAME_SPEECH_ENDED = "SpeechEnded" +# AI voice response starts, SDK prepares to receive audio +RESPONSE_NAME_RESPONDING_STARTED = "RespondingStarted" +RESPONSE_NAME_RESPONDING_ENDED = "RespondingEnded" # AI voice response ends +RESPONSE_NAME_SPEECH_CONTENT = ( + "SpeechContent" # User speech recognition text, full streaming output +) +RESPONSE_NAME_RESPONDING_CONTENT = ( + "RespondingContent" # System output text, full streaming output ) -RESPONSE_NAME_RESPONDING_ENDED = "RespondingEnded" # AI语音应答结束 -RESPONSE_NAME_SPEECH_CONTENT = "SpeechContent" # 用户语音识别出的文本，流式全量输出 -RESPONSE_NAME_RESPONDING_CONTENT = "RespondingContent" # 统对外输出的文本，流式全量输出 -RESPONSE_NAME_ERROR = "Error" # 服务端对话中报错 -RESPONSE_NAME_HEART_BEAT = "HeartBeat" # 心跳消息 +RESPONSE_NAME_ERROR = "Error" # Server-side error during dialog +RESPONSE_NAME_HEART_BEAT = "HeartBeat" # Heartbeat message diff --git a/dashscope/multimodal/multimodal_dialog.py b/dashscope/multimodal/multimodal_dialog.py index 6f597aa..1517e97 100644 --- a/dashscope/multimodal/multimodal_dialog.py +++ b/dashscope/multimodal/multimodal_dialog.py @@ -39,98 +39,98 @@ class MultiModalCallback: """ - 语音聊天回调类，用于处理语音聊天过程中的各种事件。 + Voice chat callback class for handling various events during voice chat. """ def on_started(self, dialog_id: str) -> None: """ - 通知对话开始 + Notify dialog started. - :param dialog_id: 回调对话ID + :param dialog_id: Callback dialog ID """ def on_stopped(self) -> None: """ - 通知对话停止 + Notify dialog stopped. """ def on_state_changed(self, state: "dialog_state.DialogState") -> None: """ - 对话状态改变 + Dialog state changed. - :param state: 新的对话状态 + :param state: New dialog state """ def on_speech_audio_data(self, data: bytes) -> None: """ - 合成音频数据回调 + Synthesized audio data callback. - :param data: 音频数据 + :param data: Audio data """ def on_error(self, error) -> None: """ - 发生错误时调用此方法。 + Called when an error occurs. - :param error: 错误信息 + :param error: Error message """ def on_connected(self) -> None: """ - 成功连接到服务器后调用此方法。 + Called after successfully connecting to the server. """ def on_responding_started(self): """ - 回复开始回调 + Response started callback. """ def on_responding_ended(self, payload): """ - 回复结束 + Response ended. """ def on_speech_started(self): """ - 检测到语音输入结束 + Speech input started. """ def on_speech_ended(self): """ - 检测到语音输入结束 + Speech input ended. """ def on_speech_content(self, payload): """ - 语音识别文本 + Speech recognition text. :param payload: text """ def on_responding_content(self, payload): """ - 大模型回复文本。 + LLM response text. :param payload: text """ def on_request_accepted(self): """ - 打断请求被接受。 + Interrupt request accepted. """ def on_close(self, close_status_code, close_msg): """ - 连接关闭时调用此方法。 + Called when connection is closed. - :param close_status_code: 关闭状态码 - :param close_msg: 关闭消息 + :param close_status_code: Close status code + :param close_msg: Close message """ class MultiModalDialog: """ - 用于管理WebSocket连接以进行语音聊天的服务类。 + Service class for managing WebSocket connections for voice chat. """ def __init__( @@ -145,17 +145,22 @@ def __init__( model: str = None, ): """ - 创建一个语音对话会话。 - - 此方法用于初始化一个新的voice_chat会话，设置必要的参数以准备开始与模型的交互。 - ：param workspace_id: 客户的workspace_id 主工作空间id,非必填字段 - :param app_id: 客户在管控台创建的应用id，可以根据值规律确定使用哪个对话系统 - :param request_params: 请求参数集合 - :param url: (str) API的URL地址。 - :param multimodal_callback: (MultimodalCallback) 回调对象，用于处理来自服务器的消息。 - :param api_key: (str) 应用程序接入的唯一key - :param dialog_id:对话id，如果传入表示承接上下文继续聊 - :param model: 模型 + Create a voice dialog session. + + This method initializes a new voice_chat session, setting up + the necessary parameters to start interacting with the model. + :param workspace_id: Customer workspace_id, primary workspace ID, + optional field + :param app_id: Application ID created in the console, used to + determine which dialog system to use + :param request_params: Request parameter collection + :param url: (str) API URL address. + :param multimodal_callback: (MultimodalCallback) Callback object + for processing messages from server. + :param api_key: (str) Application unique access key + :param dialog_id: Dialog ID, if provided, continues the + conversation with previous context + :param model: Model """ if request_params is None: raise InputRequired("request_params is required!") @@ -181,7 +186,7 @@ def __init__( self.dialog_state, self._callback, self.close, - ) # 传递 self.close 作为回调 + ) # pass self.close as callback def _on_message( # pylint: disable=unused-argument self, @@ -227,11 +232,15 @@ def _on_open(self, ws): # pylint: disable=unused-argument def start(self, dialog_id, enable_voice_detection=False, task_id=None): """ - 初始化WebSocket连接并发送启动请求 - :param dialog_id: 上下位继承标志位。新对话无需设置。 - 如果继承之前的对话历史，则需要记录之前的dialog_id并传入 - :param enable_voice_detection: 是否开启语音检测,可选参数 默认False - :param task_id: 百炼请求任务 Id，默认会自动生成。您可以指定此 ID 来跟踪请求。 + Initialize WebSocket connection and send start request. + :param dialog_id: Context inheritance flag. Not needed for new + dialogs. + If inheriting previous dialog history, record and pass + the previous dialog_id + :param enable_voice_detection: Whether to enable voice detection, + optional, default False + :param task_id: DashScope request task ID, auto-generated by + default. You can specify this ID to track the request. """ self._voice_detection = enable_voice_detection self._connect(self.api_key) @@ -243,7 +252,7 @@ def start(self, dialog_id, enable_voice_detection=False, task_id=None): ) def start_speech(self): - """开始上传语音数据""" + """Start uploading speech data""" _send_speech_json = self.request.generate_common_direction_request( "SendSpeech", self.dialog_id, @@ -251,11 +260,11 @@ def start_speech(self): self._send_text_frame(_send_speech_json) def send_audio_data(self, speech_data: bytes): - """发送语音数据""" + """Send speech data""" self.__send_binary_frame(speech_data) def stop_speech(self): - """停止上传语音数据""" + """Stop uploading speech data""" _send_speech_json = self.request.generate_common_direction_request( "StopSpeech", self.dialog_id, @@ -263,7 +272,7 @@ def stop_speech(self): self._send_text_frame(_send_speech_json) def interrupt(self): - """请求服务端开始说话""" + """Request server to start speaking""" _send_speech_json = self.request.generate_common_direction_request( "RequestToSpeak", self.dialog_id, @@ -276,7 +285,7 @@ def request_to_respond( text: str, parameters: RequestToRespondParameters = None, ): - """请求服务端直接文本合成语音""" + """Request server to synthesize speech from text directly""" _send_speech_json = self.request.generate_request_to_response_json( direction_name="RequestToRespond", dialog_id=self.dialog_id, @@ -288,11 +297,11 @@ def request_to_respond( @abstractmethod def request_to_respond_prompt(self, text): - """请求服务端通过文本请求回复文本答复""" + """Request server to reply with text response via text request""" return def local_responding_started(self): - """本地tts播放开始""" + """Local TTS playback started""" _send_speech_json = self.request.generate_common_direction_request( "LocalRespondingStarted", self.dialog_id, @@ -300,7 +309,7 @@ def local_responding_started(self): self._send_text_frame(_send_speech_json) def local_responding_ended(self): - """本地tts播放结束""" + """Local TTS playback ended""" _send_speech_json = self.request.generate_common_direction_request( "LocalRespondingEnded", self.dialog_id, @@ -308,7 +317,7 @@ def local_responding_ended(self): self._send_text_frame(_send_speech_json) def send_heart_beat(self): - """发送心跳""" + """Send heartbeat""" _send_speech_json = self.request.generate_common_direction_request( "HeartBeat", self.dialog_id, @@ -316,7 +325,7 @@ def send_heart_beat(self): self._send_text_frame(_send_speech_json) def update_info(self, parameters: RequestToRespondParameters = None): - """更新信息""" + """Update information""" _send_speech_json = self.request.generate_update_info_json( direction_name="UpdateInfo", dialog_id=self.dialog_id, @@ -341,7 +350,7 @@ def get_conversation_mode(self) -> str: """get mode of conversation: support tap2talk/push2talk/duplex""" return self.request_params.upstream.mode - """内部方法""" # pylint: disable=pointless-string-statement + """Internal methods""" # pylint: disable=pointless-string-statement def _send_start_request( self, @@ -349,7 +358,7 @@ def _send_start_request( request_params: RequestParameters, task_id: str = None, ): - """发送'Start'请求""" + """Send 'Start' request""" _start_json = self.request.generate_start_request( workspace_id=self.workspace_id, direction_name="Start", @@ -366,7 +375,7 @@ def _run_forever(self): self.ws.run_forever(ping_interval=None, ping_timeout=None) def _connect(self, api_key: str): - """初始化WebSocket连接并发送启动请求。""" + """Initialize WebSocket connection and send startup request.""" self.ws = websocket.WebSocketApp( self.url, header=self.request.get_websocket_header(api_key), @@ -387,14 +396,14 @@ def close(self): self.ws.close() def _wait_for_connection(self): - """等待WebSocket连接建立""" + """Wait for WebSocket connection to be established""" timeout = 5 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 def _send_text_frame(self, text: str): logger.info(">>>>>> send text frame : %s", text) @@ -408,14 +417,14 @@ def __del__(self): self.cleanup() def cleanup(self): - """清理所有资源""" + """Clean up all resources""" try: if self.ws: self.ws.close() if self.thread and self.thread.is_alive(): - # 设置标志位通知线程退出 + # Set flag to notify thread to exit self.thread.join(timeout=2) - # 清除引用 + # Clear references self.ws = None self.thread = None self._callback = None @@ -459,15 +468,16 @@ def generate_start_request( task_id: str = None, ) -> str: """ - 构建语音聊天服务的启动请求数据. - :param app_id: 管控台应用id - :param request_params: start请求body中的parameters + Build startup request data for voice chat service. + :param app_id: Console application ID + :param request_params: Parameters in start request body :param direction_name: - :param dialog_id: 对话ID. - :param workspace_id: 管控台工作空间id, 非必填字段。 - :param model: 模型 - :param task_id: 百炼请求任务 Id，默认会自动生成。您可以指定此 ID 来跟踪请求。 - :return: 启动请求字典. + :param dialog_id: Dialog ID. + :param workspace_id: Console workspace ID, optional field. + :param model: Model + :param task_id: DashScope request task ID, auto-generated by + default. You can specify this ID to track the request. + :return: Startup request dictionary. """ self.task_id = task_id self._get_dash_request_header(ActionType.START) @@ -492,10 +502,10 @@ def generate_common_direction_request( dialog_id: str, ) -> str: """ - 构建语音聊天服务的命令请求数据. - :param direction_name: 命令. - :param dialog_id: 对话ID. - :return: 命令请求json. + Build command request data for voice chat service. + :param direction_name: Command. + :param dialog_id: Dialog ID. + :return: Command request JSON. """ self._get_dash_request_header(ActionType.CONTINUE) self._get_dash_request_payload(direction_name, dialog_id, self.app_id) @@ -511,10 +521,10 @@ def generate_stop_request( dialog_id: str, ) -> str: """ - 构建语音聊天服务的启动请求数据. - :param direction_name:指令名称 - :param dialog_id: 对话ID. - :return: 启动请求json. + Build stop request data for voice chat service. + :param direction_name: Directive name + :param dialog_id: Dialog ID. + :return: Stop request JSON. """ self._get_dash_request_header(ActionType.FINISHED) self._get_dash_request_payload(direction_name, dialog_id, self.app_id) @@ -534,13 +544,15 @@ def generate_request_to_response_json( parameters: RequestToRespondParameters = None, ) -> str: """ - 构建语音聊天服务的命令请求数据. - :param direction_name: 命令. - :param dialog_id: 对话ID. - :param request_type: 服务应该采取的交互类型，transcript 表示直接把文本转语音，prompt 表示把文本送大模型回答 # noqa: E501 - :param text: 文本. - :param parameters: 命令请求body中的parameters - :return: 命令请求字典. + Build command request data for voice chat service. + :param direction_name: Command. + :param dialog_id: Dialog ID. + :param request_type: Interaction type the service should adopt, + transcript means convert text to speech directly, + prompt means send text to LLM for response + :param text: Text. + :param parameters: Parameters in command request body + :return: Command request dictionary. """ self._get_dash_request_header(ActionType.CONTINUE) @@ -572,10 +584,10 @@ def generate_update_info_json( parameters: RequestToRespondParameters = None, ) -> str: """ - 构建语音聊天服务的命令请求数据. - :param direction_name: 命令. - :param parameters: 命令请求body中的parameters - :return: 命令请求字典. + Build command request data for voice chat service. + :param direction_name: Command. + :param parameters: Parameters in command request body + :return: Command request dictionary. """ self._get_dash_request_header(ActionType.CONTINUE) @@ -600,8 +612,9 @@ def generate_update_info_json( def _get_dash_request_header(self, action: str): """ - 构建多模对话请求的请求协议Header - :param action: ActionType 百炼协议action 支持：run-task, continue-task, finish-task # noqa: E501 + Build request protocol header for multimodal dialog request. + :param action: ActionType DashScope protocol action, supports: + run-task, continue-task, finish-task """ if self.task_id is None: self.task_id = get_random_uuid() @@ -618,13 +631,13 @@ def _get_dash_request_payload( model: str = None, ): """ - 构建多模对话请求的请求协议payload - :param direction_name: 对话协议内部的指令名称 - :param dialog_id: 对话ID. - :param app_id: 管控台应用id - :param request_params: start请求body中的parameters - :param custom_input: 自定义输入 - :param model: 模型 + Build request protocol payload for multimodal dialog request. + :param direction_name: Internal directive name in dialog protocol + :param dialog_id: Dialog ID. + :param app_id: Console application ID + :param request_params: Parameters in start request body + :param custom_input: Custom input + :param model: Model """ if custom_input is not None: input = custom_input # pylint: disable=redefined-builtin @@ -651,20 +664,20 @@ def __init__( close_callback=None, ): super().__init__() - self.dialog_id = None # 对话ID. + self.dialog_id = None # Dialog ID self.dialog_state = state self._callback = callback - self._close_callback = close_callback # 保存关闭回调函数 + self._close_callback = close_callback # Save close callback function # pylint: disable=inconsistent-return-statements def handle_text_response(self, response_json: str): """ - 处理语音聊天服务的响应数据. - :param response_json: 从服务接收到的原始JSON字符串响应。 + Handle response data from voice chat service. + :param response_json: Original JSON string response from server. """ logger.info("<<<<<< server response: %s", response_json) try: - # 尝试将消息解析为JSON + # Attempt to parse message as JSON json_data = json.loads(response_json) if ( "status_code" in json_data["header"] @@ -760,8 +773,8 @@ def _handle_stopped(self): def _handle_state_changed(self, state: str): """ - 处理语音聊天状态流转. - :param state: 状态. + Handle voice chat state transitions. + :param state: State. """ self.dialog_state.change_state(state) self._callback.on_state_changed(self.dialog_state.get_current_state()) diff --git a/dashscope/multimodal/multimodal_request_params.py b/dashscope/multimodal/multimodal_request_params.py index fe88161..1caf609 100644 --- a/dashscope/multimodal/multimodal_request_params.py +++ b/dashscope/multimodal/multimodal_request_params.py @@ -5,7 +5,7 @@ def get_random_uuid() -> str: - """生成并返回32位UUID字符串""" + """Generate and return a 32-character UUID string""" return uuid.uuid4().hex @@ -13,7 +13,7 @@ def get_random_uuid() -> str: class DashHeader: action: str task_id: str = field(default=get_random_uuid()) - streaming: str = field(default="duplex") # 默认为 duplex + streaming: str = field(default="duplex") # default to duplex def to_dict(self): return { @@ -110,12 +110,17 @@ def to_dict(self): class Upstream: """struct for upstream""" - audio_format: str = field(default="pcm") # 上行语音格式，默认pcm.支持pcm/opus + audio_format: str = field( + default="pcm", + ) # upstream audio format, default pcm, supports pcm/opus type: str = field( default="AudioOnly", - ) # 上行类型：AudioOnly 仅语音通话; AudioAndVideo 上传视频 - mode: str = field(default="tap2talk") # 客户端交互模式 push2talk/tap2talk/duplex - sample_rate: int = field(default=16000) # 音频采样率 + ) # upstream type: AudioOnly for voice only; + # AudioAndVideo for video upload + mode: str = field( + default="tap2talk", + ) # client interaction mode: push2talk/tap2talk/duplex + sample_rate: int = field(default=16000) # audio sample rate vocabulary_id: str = field(default=None) asr_post_processing: AsrPostProcessing = field(default=None) pass_through_params: dict = field(default=None) # type: ignore[arg-type] @@ -140,18 +145,24 @@ def to_dict(self): @dataclass class Downstream: - # transcript 返回用户语音识别结果 - # dialog 返回对话系统回答中间结果 - # 可以设置多种，以逗号分割，默认为transcript - voice: str = field(default="") # 语音音色 - sample_rate: int = field(default=0) # 语音音色 # 合成音频采样率 - intermediate_text: str = field(default="transcript") # 控制返回给用户那些中间文本： - debug: bool = field(default=False) # 控制是否返回debug信息 - # type_: str = field(default="Audio", metadata={"alias": "type"}) # 下行类型：Text：不需要下发语音;Audio：输出语音，默认值 # noqa: E501 # pylint: disable=line-too-long - audio_format: str = field(default="pcm") # 下行语音格式，默认pcm 。支持pcm/mp3 - volume: int = field(default=50) # 语音音量 0-100 - pitch_rate: int = field(default=100) # 语音语调 50-200 - speech_rate: int = field(default=100) # 语音语速 50-200 + # transcript returns user speech recognition results + # dialog returns dialog system intermediate results + # Multiple values can be set, comma-separated, default is transcript + voice: str = field(default="") # voice timbre + sample_rate: int = field( + default=0, + ) # voice timbre # synthesis audio sample rate + intermediate_text: str = field( + default="transcript", + ) # Controls which intermediate text is returned to user: + debug: bool = field(default=False) # Controls whether to return debug info + # type_: str = field(default="Audio", metadata={"alias": "type"}) # downstream type: Text: no audio output; Audio: output audio, default # noqa: E501 # pylint: disable=line-too-long + audio_format: str = field( + default="pcm", + ) # downstream audio format, default pcm, supports pcm/mp3 + volume: int = field(default=50) # voice volume 0-100 + pitch_rate: int = field(default=100) # voice pitch 50-200 + speech_rate: int = field(default=100) # voice speed 50-200 pass_through_params: dict = field(default=None) # type: ignore[arg-type] def to_dict(self): diff --git a/dashscope/multimodal/tingwu/tingwu_realtime.py b/dashscope/multimodal/tingwu/tingwu_realtime.py index d553737..71ed4ab 100644 --- a/dashscope/multimodal/tingwu/tingwu_realtime.py +++ b/dashscope/multimodal/tingwu/tingwu_realtime.py @@ -129,7 +129,7 @@ def __init__( self.response = _TingWuResponse( self._callback, self.close, - ) # 传递 self.close 作为回调 + ) # pass self.close as callback def _on_message( # pylint: disable=unused-argument self, @@ -145,13 +145,13 @@ def _on_message( # pylint: disable=unused-argument def _on_error(self, ws, error): # pylint: disable=unused-argument logger.error(f"Error: {error}") if self._callback: - error_code = "" # 默认错误码 + error_code = "" # default error code if "connection" in str(error).lower(): - error_code = "1001" # 连接错误 + error_code = "1001" # connection error elif "timeout" in str(error).lower(): - error_code = "1002" # 超时错误 + error_code = "1002" # timeout error elif "authentication" in str(error).lower(): - error_code = "1003" # 认证错误 + error_code = "1003" # authentication error self._callback.on_error( error_code=error_code, error_msg=str(error), @@ -251,7 +251,7 @@ def _connect(self, api_key: str): on_close=self._on_close, ) self.thread = threading.Thread(target=self._run_forever) - # 统一心跳机制配置 + # Unified heartbeat configuration self.ws.ping_interval = 5 self.ws.ping_timeout = 4 self.thread.daemon = True @@ -272,11 +272,11 @@ def _wait_for_connection(self): 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 def _send_text_frame(self, text: str): - # 避免在日志中记录敏感信息，如API密钥等 - # 只记录非敏感信息 + # Avoid logging sensitive information such as API keys + # Only log non-sensitive information if '"Authorization"' not in text: logger.info(">>>>>> send text frame : %s", text) else: @@ -300,9 +300,9 @@ def cleanup(self): if self.ws: self.ws.close() if self.thread and self.thread.is_alive(): - # 设置标志位通知线程退出 + # Set flag to notify thread to exit self.thread.join(timeout=2) - # 清除引用 + # Clear references self.ws = None self.thread = None self._callback = None @@ -488,9 +488,9 @@ def _get_dash_request_payload( class _TingWuResponse: def __init__(self, callback: TingWuRealtimeCallback, close_callback=None): super().__init__() - self.task_id = None # 对话ID. + self.task_id = None # Task ID self._callback = callback - self._close_callback = close_callback # 保存关闭回调函数 + self._close_callback = close_callback # Save close callback function def handle_text_response(self, response_json: str): """ @@ -558,7 +558,9 @@ def _handle_tingwu_agent_text_response( self._callback.on_recognize_result(response_json) elif action == "ai-result": self._callback.on_ai_result(response_json) - elif action == "speech-end": # ai-result事件永远会先于speech-end事件 + elif ( + action == "speech-end" + ): # ai-result event always arrives before speech-end event self._callback.on_stopped() if self._close_callback is not None: self._close_callback() @@ -598,7 +600,7 @@ def to_dict(self): class DashHeader: action: str task_id: str = field(default=get_random_uuid()) - streaming: str = field(default="duplex") # 默认为 duplex + streaming: str = field(default="duplex") # default to duplex def to_dict(self): return { diff --git a/dashscope/version.py b/dashscope/version.py index fbaaca6..117bf62 100644 --- a/dashscope/version.py +++ b/dashscope/version.py @@ -1,4 +1,4 @@ # -*- coding: utf-8 -*- # Copyright (c) Alibaba, Inc. and its affiliates. -__version__ = "1.25.21" +__version__ = "1.25.22" diff --git a/tests/unit/test_aio_session.py b/tests/unit/test_aio_session.py new file mode 100644 index 0000000..b82061a --- /dev/null +++ b/tests/unit/test_aio_session.py @@ -0,0 +1,166 @@ +# -*- coding: utf-8 -*- +# Copyright (c) Alibaba, Inc. and its affiliates. + +""" +Shared aiohttp session pool unit tests. + +Tests the connection reuse and SSL context caching in aio_session module. +""" + +# pylint: disable=protected-access + +import asyncio +import ssl +from unittest.mock import patch + +import aiohttp +import pytest + +from dashscope.api_entities import aio_session + + +class TestSSLContextCaching: + """Test SSL context is created once and reused.""" + + def setup_method(self): + aio_session._shared_ssl_context = None + + def test_get_ssl_context_returns_ssl_context(self): + ctx = aio_session.get_ssl_context() + assert isinstance(ctx, ssl.SSLContext) + + def test_get_ssl_context_cached(self): + ctx1 = aio_session.get_ssl_context() + ctx2 = aio_session.get_ssl_context() + assert ctx1 is ctx2 + + def test_get_ssl_context_calls_create_default_context_once(self): + with patch( + "ssl.create_default_context", + wraps=ssl.create_default_context, + ) as mock_create: + aio_session._shared_ssl_context = None + aio_session.get_ssl_context() + aio_session.get_ssl_context() + aio_session.get_ssl_context() + assert mock_create.call_count == 1 + + +class TestSharedAioSession: + """Test shared session creation and reuse.""" + + def setup_method(self): + aio_session._shared_ssl_context = None + aio_session._aio_sessions.clear() + + @pytest.mark.asyncio + async def test_get_shared_session_returns_client_session(self): + session = await aio_session.get_shared_aio_session() + try: + assert isinstance(session, aiohttp.ClientSession) + assert not session.closed + finally: + await aio_session.close_shared_aio_session() + + @pytest.mark.asyncio + async def test_get_shared_session_reuses_same_session(self): + s1 = await aio_session.get_shared_aio_session() + s2 = await aio_session.get_shared_aio_session() + try: + assert s1 is s2 + finally: + await aio_session.close_shared_aio_session() + + @pytest.mark.asyncio + async def test_shared_session_has_tcp_connector(self): + session = await aio_session.get_shared_aio_session() + try: + assert isinstance(session.connector, aiohttp.TCPConnector) + finally: + await aio_session.close_shared_aio_session() + + @pytest.mark.asyncio + async def test_shared_session_uses_cached_ssl(self): + session = await aio_session.get_shared_aio_session() + try: + ssl_ctx = aio_session.get_ssl_context() + assert session.connector._ssl is ssl_ctx + finally: + await aio_session.close_shared_aio_session() + + @pytest.mark.asyncio + async def test_close_shared_session(self): + session = await aio_session.get_shared_aio_session() + assert not session.closed + await aio_session.close_shared_aio_session() + assert session.closed + + @pytest.mark.asyncio + async def test_new_session_after_close(self): + s1 = await aio_session.get_shared_aio_session() + await aio_session.close_shared_aio_session() + assert s1.closed + + s2 = await aio_session.get_shared_aio_session() + try: + assert s2 is not s1 + assert not s2.closed + finally: + await aio_session.close_shared_aio_session() + + @pytest.mark.asyncio + async def test_close_idempotent(self): + await aio_session.close_shared_aio_session() + await aio_session.close_shared_aio_session() + + @pytest.mark.asyncio + async def test_stale_sessions_cleaned_up(self): + """Test that closed sessions are replaced in the dict.""" + s1 = await aio_session.get_shared_aio_session() + loop = asyncio.get_running_loop() + + # Manually close without calling close_shared_aio_session + await s1.close() + assert s1.closed + assert loop in aio_session._aio_sessions + + # Getting a new session should replace the stale entry + s2 = await aio_session.get_shared_aio_session() + try: + assert s2 is not s1 + assert not s2.closed + assert aio_session._aio_sessions[loop] is s2 + finally: + await aio_session.close_shared_aio_session() + + +class TestSessionPerLoop: + """Test that different event loops get different sessions.""" + + def setup_method(self): + aio_session._shared_ssl_context = None + aio_session._aio_sessions.clear() + + def test_different_loops_get_different_sessions(self): + sessions = [] + + def run_in_loop(): + loop = asyncio.new_event_loop() + try: + session = loop.run_until_complete( + aio_session.get_shared_aio_session(), + ) + sessions.append(session) + # Keep loop open so session stays valid + loop.run_until_complete( + aio_session.close_shared_aio_session(), + ) + finally: + loop.close() + + run_in_loop() + run_in_loop() + + # Each loop should have gotten its own session + assert len(sessions) == 2 + assert sessions[0] is not sessions[1] diff --git a/tests/unit/test_async_custom_session.py b/tests/unit/test_async_custom_session.py index 180b478..a82f005 100644 --- a/tests/unit/test_async_custom_session.py +++ b/tests/unit/test_async_custom_session.py @@ -134,11 +134,11 @@ async def mock_handle_response(_response): mock_session.close.assert_not_called() @pytest.mark.asyncio - async def test_temporary_aio_session_is_created_when_no_custom_session( + async def test_shared_aio_session_is_used_when_no_custom_session( self, ): - """测试没有自定义 aio_session 时会创建临时 aio_session""" - # 创建 mock session + """测试没有自定义 aio_session 时使用共享 session（不被关闭）""" + # 创建 mock shared session mock_session = AsyncMock() mock_response = AsyncMock() mock_response.status = 200 @@ -172,7 +172,10 @@ async def test_temporary_aio_session_is_created_when_no_custom_session( async def mock_handle_response(_response): yield mock_response - with patch("aiohttp.ClientSession", return_value=mock_session): + with patch( + "dashscope.api_entities.http_request.get_shared_aio_session", + return_value=mock_session, + ): with patch.object( http_request, "_handle_aio_response", @@ -180,8 +183,8 @@ async def mock_handle_response(_response): ): _ = await http_request.aio_call() - # 验证临时 aio_session 被关闭 - mock_session.close.assert_called_once() + # 共享 session 不应被关闭（由 aio_session 模块管理生命周期） + mock_session.close.assert_not_called() class TestAsyncSessionResourceManagement: @@ -237,8 +240,8 @@ async def mock_handle_response(_response): custom_session.close.assert_not_called() @pytest.mark.asyncio - async def test_temporary_aio_session_closed_on_success(self): - """测试临时 aio_session 在成功后被关闭""" + async def test_shared_aio_session_not_closed_on_success(self): + """测试共享 aio_session 在成功后不被关闭（由模块管理生命周期）""" mock_session = AsyncMock() mock_response = AsyncMock() mock_response.status = 200 @@ -269,7 +272,10 @@ async def test_temporary_aio_session_closed_on_success(self): async def mock_handle_response(_response): yield mock_response - with patch("aiohttp.ClientSession", return_value=mock_session): + with patch( + "dashscope.api_entities.http_request.get_shared_aio_session", + return_value=mock_session, + ): with patch.object( http_request, "_handle_aio_response", @@ -277,12 +283,12 @@ async def mock_handle_response(_response): ): _ = await http_request.aio_call() - # 验证临时 aio_session 被关闭 - mock_session.close.assert_called_once() + # 共享 session 不应被关闭 + mock_session.close.assert_not_called() @pytest.mark.asyncio - async def test_temporary_aio_session_closed_on_exception(self): - """测试临时 aio_session 在异常时也被关闭""" + async def test_shared_aio_session_not_closed_on_exception(self): + """测试共享 aio_session 在异常时也不被关闭""" mock_session = AsyncMock() # Make request() raise an exception @@ -311,12 +317,15 @@ async def mock_request(*_args, **_kwargs): http_request.data = request_data # 执行请求应该抛出异常 - with patch("aiohttp.ClientSession", return_value=mock_session): + with patch( + "dashscope.api_entities.http_request.get_shared_aio_session", + return_value=mock_session, + ): with pytest.raises(Exception, match="Network error"): _ = await http_request.aio_call() - # 验证临时 aio_session 仍然被关闭 - mock_session.close.assert_called_once() + # 共享 session 仍然不应被关闭（由模块管理生命周期） + mock_session.close.assert_not_called() class TestAsyncSessionWithCustomConfiguration: @@ -533,8 +542,8 @@ async def test_works_without_aio_session_parameter(self): assert http_request.method == HTTPMethod.POST @pytest.mark.asyncio - async def test_default_behavior_unchanged(self): - """测试默认行为未改变""" + async def test_default_behavior_uses_shared_session(self): + """测试默认行为使用共享 session（不被关闭）""" mock_session = AsyncMock() mock_response = AsyncMock() mock_response.status = 200 @@ -566,7 +575,10 @@ async def test_default_behavior_unchanged(self): async def mock_handle_response(_response): yield mock_response - with patch("aiohttp.ClientSession", return_value=mock_session): + with patch( + "dashscope.api_entities.http_request.get_shared_aio_session", + return_value=mock_session, + ): with patch.object( http_request, "_handle_aio_response", @@ -574,8 +586,8 @@ async def mock_handle_response(_response): ): _ = await http_request.aio_call() - # 验证临时 aio_session 被关闭（原有行为） - mock_session.close.assert_called_once() + # 共享 session 不应被关闭（生命周期由 aio_session 模块管理） + mock_session.close.assert_not_called() class TestAsyncSessionLifecycle: