From 677c3a71767d8d5d1d9c97eb025f837eed36bfb4 Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Fri, 12 Jun 2026 14:12:34 +0800 Subject: [PATCH 1/9] Introduce shared aiohttp session pool for connection reuse Replace per-request aiohttp.ClientSession creation with a shared session pool (one per event loop) to enable HTTP connection reuse across async API calls. SSL context is cached and shared across all sessions. Co-Authored-By: Claude Opus 4.7 --- dashscope/api_entities/aio_session.py | 71 +++++++++++ dashscope/api_entities/aiohttp_request.py | 65 +++++----- dashscope/api_entities/http_request.py | 97 +++++++------- tests/unit/test_aio_session.py | 146 ++++++++++++++++++++++ tests/unit/test_async_custom_session.py | 54 ++++---- 5 files changed, 326 insertions(+), 107 deletions(-) create mode 100644 dashscope/api_entities/aio_session.py create mode 100644 tests/unit/test_aio_session.py diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py new file mode 100644 index 0000000..6b2b3b2 --- /dev/null +++ b/dashscope/api_entities/aio_session.py @@ -0,0 +1,71 @@ +# -*- 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 atexit +import ssl +from typing import Dict, Optional + +import aiohttp +import certifi + +_shared_ssl_context: Optional[ssl.SSLContext] = None +_aio_sessions: Dict[int, aiohttp.ClientSession] = {} + + +def get_ssl_context() -> ssl.SSLContext: + global _shared_ssl_context + 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() + loop_id = id(loop) + session = _aio_sessions.get(loop_id) + 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_id] = session + return session + + +async def close_shared_aio_session() -> None: + """Close the shared session for the current event loop.""" + loop = asyncio.get_running_loop() + loop_id = id(loop) + session = _aio_sessions.pop(loop_id, None) + if session is not None and not session.closed: + await session.close() + + +def _atexit_close_sessions() -> None: + sessions = list(_aio_sessions.values()) + _aio_sessions.clear() + for session in sessions: + if not session.closed: + try: + loop = asyncio.new_event_loop() + loop.run_until_complete(session.close()) + loop.close() + except Exception: + pass + + +atexit.register(_atexit_close_sessions) diff --git a/dashscope/api_entities/aiohttp_request.py b/dashscope/api_entities/aiohttp_request.py index 75b3965..01e0aa8 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 ( @@ -246,41 +247,43 @@ async def _handle_response( # 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: - logger.debug("Starting request: %s", self.url) - if self.method == HTTPMethod.POST: - 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: - response = await session.get( + session = await get_shared_aio_session() + request_timeout = aiohttp.ClientTimeout(total=self.timeout) + + logger.debug("Starting request: %s", self.url) + if self.method == HTTPMethod.POST: + is_form, obj = self.data.get_aiohttp_payload() + if is_form: + headers = {**self.headers, **obj.headers} + response = await session.post( url=self.url, - params=self.data.parameters, - 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: + response = await session.get( + url=self.url, + params=self.data.parameters, + 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 except aiohttp.ClientConnectorError as e: logger.error(e) raise e diff --git a/dashscope/api_entities/http_request.py b/dashscope/api_entities/http_request.py index 85ee959..ba6fcc5 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 ( @@ -162,67 +161,55 @@ 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() - 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( + 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_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 except aiohttp.ClientConnectorError as e: logger.error(e) raise e diff --git a/tests/unit/test_aio_session.py b/tests/unit/test_aio_session.py new file mode 100644 index 0000000..ebe14bb --- /dev/null +++ b/tests/unit/test_aio_session.py @@ -0,0 +1,146 @@ +# -*- 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, MagicMock + +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() + + +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: From ac32bc5aad1459acaf13640350bd1c89e7baad81 Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Fri, 12 Jun 2026 14:20:56 +0800 Subject: [PATCH 2/9] Fix async streaming timeout: use sock_read instead of total Streaming requests were using aiohttp.ClientTimeout(total=...) which kills the connection after the deadline even if the model is still actively producing tokens. Switch to ClientTimeout(sock_read=...) for streaming requests so the timeout only triggers on idle connections (no data between chunks). Non-streaming requests still use total timeout. Co-Authored-By: Claude Opus 4.7 --- dashscope/api_entities/aiohttp_request.py | 9 +++++++-- dashscope/api_entities/http_request.py | 9 +++++++-- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/dashscope/api_entities/aiohttp_request.py b/dashscope/api_entities/aiohttp_request.py index 01e0aa8..8f3cfb1 100644 --- a/dashscope/api_entities/aiohttp_request.py +++ b/dashscope/api_entities/aiohttp_request.py @@ -39,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 ''. @@ -248,7 +250,10 @@ async def _handle_response( # pylint: disable=too-many-branches async def _handle_request(self): try: session = await get_shared_aio_session() - request_timeout = aiohttp.ClientTimeout(total=self.timeout) + if self.stream: + request_timeout = aiohttp.ClientTimeout(sock_read=self.timeout) + else: + request_timeout = aiohttp.ClientTimeout(total=self.timeout) logger.debug("Starting request: %s", self.url) if self.method == HTTPMethod.POST: diff --git a/dashscope/api_entities/http_request.py b/dashscope/api_entities/http_request.py index ba6fcc5..1ab8e96 100644 --- a/dashscope/api_entities/http_request.py +++ b/dashscope/api_entities/http_request.py @@ -52,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 ''. @@ -167,7 +169,10 @@ async def _handle_aio_request(self): # pylint: disable=too-many-branches else: session = await get_shared_aio_session() - request_timeout = aiohttp.ClientTimeout(total=self.timeout) + if self.stream: + request_timeout = aiohttp.ClientTimeout(sock_read=self.timeout) + else: + request_timeout = aiohttp.ClientTimeout(total=self.timeout) logger.debug("Starting request: %s", self.url) if self.method == HTTPMethod.POST: From 3a8e5fa18da35f158fb9041d6fd0598ea536f926 Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Fri, 12 Jun 2026 17:13:56 +0800 Subject: [PATCH 3/9] Fix session pool: remove broken atexit handler and clean stale entries The atexit handler created a new event loop to close sessions bound to the original loop, which silently failed. OS reclaims resources on exit, so explicit close is unnecessary. Also adds stale-entry cleanup in get_shared_aio_session() to prevent unbounded dict growth when event loops are repeatedly created and closed. Co-Authored-By: Claude Opus 4.7 --- dashscope/api_entities/aio_session.py | 31 +++++------------------ dashscope/api_entities/aiohttp_request.py | 5 +++- dashscope/api_entities/http_request.py | 5 +++- tests/unit/test_aio_session.py | 22 +++++++++++++++- 4 files changed, 36 insertions(+), 27 deletions(-) diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py index 6b2b3b2..ee64eec 100644 --- a/dashscope/api_entities/aio_session.py +++ b/dashscope/api_entities/aio_session.py @@ -7,15 +7,15 @@ is created once and shared across all sessions. """ import asyncio -import atexit import ssl -from typing import Dict, Optional +import weakref +from typing import Optional import aiohttp import certifi _shared_ssl_context: Optional[ssl.SSLContext] = None -_aio_sessions: Dict[int, aiohttp.ClientSession] = {} +_aio_sessions: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary() def get_ssl_context() -> ssl.SSLContext: @@ -35,37 +35,20 @@ async def get_shared_aio_session() -> aiohttp.ClientSession: is handled by the underlying TCPConnector. """ loop = asyncio.get_running_loop() - loop_id = id(loop) - session = _aio_sessions.get(loop_id) + + 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_id] = session + _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() - loop_id = id(loop) - session = _aio_sessions.pop(loop_id, None) + session = _aio_sessions.pop(loop, None) if session is not None and not session.closed: await session.close() - - -def _atexit_close_sessions() -> None: - sessions = list(_aio_sessions.values()) - _aio_sessions.clear() - for session in sessions: - if not session.closed: - try: - loop = asyncio.new_event_loop() - loop.run_until_complete(session.close()) - loop.close() - except Exception: - pass - - -atexit.register(_atexit_close_sessions) diff --git a/dashscope/api_entities/aiohttp_request.py b/dashscope/api_entities/aiohttp_request.py index 8f3cfb1..95aeeef 100644 --- a/dashscope/api_entities/aiohttp_request.py +++ b/dashscope/api_entities/aiohttp_request.py @@ -251,7 +251,10 @@ async def _handle_request(self): try: session = await get_shared_aio_session() if self.stream: - request_timeout = aiohttp.ClientTimeout(sock_read=self.timeout) + request_timeout = aiohttp.ClientTimeout( + total=None, + sock_read=self.timeout, + ) else: request_timeout = aiohttp.ClientTimeout(total=self.timeout) diff --git a/dashscope/api_entities/http_request.py b/dashscope/api_entities/http_request.py index 1ab8e96..b175586 100644 --- a/dashscope/api_entities/http_request.py +++ b/dashscope/api_entities/http_request.py @@ -170,7 +170,10 @@ async def _handle_aio_request(self): # pylint: disable=too-many-branches session = await get_shared_aio_session() if self.stream: - request_timeout = aiohttp.ClientTimeout(sock_read=self.timeout) + request_timeout = aiohttp.ClientTimeout( + total=None, + sock_read=self.timeout, + ) else: request_timeout = aiohttp.ClientTimeout(total=self.timeout) diff --git a/tests/unit/test_aio_session.py b/tests/unit/test_aio_session.py index ebe14bb..b82061a 100644 --- a/tests/unit/test_aio_session.py +++ b/tests/unit/test_aio_session.py @@ -11,7 +11,7 @@ import asyncio import ssl -from unittest.mock import patch, MagicMock +from unittest.mock import patch import aiohttp import pytest @@ -113,6 +113,26 @@ 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.""" From 1574ccb98bfac52f0d31f5e6cb2339a15536467b Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Mon, 15 Jun 2026 11:12:22 +0800 Subject: [PATCH 4/9] Bump version to 1.25.22 and translate all Chinese comments and docstrings to English Co-Authored-By: Claude Opus 4.7 --- dashscope/aigc/image_synthesis.py | 6 +- dashscope/api_entities/encryption.py | 36 ++-- dashscope/audio/asr/recognition.py | 2 +- dashscope/audio/asr/translation_recognizer.py | 4 +- dashscope/audio/asr/vocabulary.py | 36 ++-- dashscope/audio/qwen_omni/omni_realtime.py | 18 +- .../qwen_tts_realtime/qwen_tts_realtime.py | 18 +- dashscope/audio/tts_v2/enrollment.py | 38 ++-- dashscope/audio/tts_v2/speech_synthesizer.py | 38 ++-- dashscope/client/base_api.py | 8 +- .../finetune/reinforcement/common/utils.py | 2 +- dashscope/multimodal/dialog_state.py | 28 +-- dashscope/multimodal/multimodal_constants.py | 14 +- dashscope/multimodal/multimodal_dialog.py | 199 +++++++++--------- .../multimodal/multimodal_request_params.py | 36 ++-- .../multimodal/tingwu/tingwu_realtime.py | 30 +-- dashscope/version.py | 2 +- 17 files changed, 258 insertions(+), 257 deletions(-) diff --git a/dashscope/aigc/image_synthesis.py b/dashscope/aigc/image_synthesis.py index 41bd060..4b251d3 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/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/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..103f08a 100644 --- a/dashscope/audio/asr/vocabulary.py +++ b/dashscope/audio/asr/vocabulary.py @@ -86,11 +86,11 @@ 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,11 @@ 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 +157,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 +185,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 +210,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..7b8e37b 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,14 @@ 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 +332,7 @@ def close(self) -> None: """ self.ws.close() - # 监听消息的回调函数 + # Callback for listening to messages def on_message( # pylint: disable=unused-argument self, ws, @@ -344,7 +344,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 +372,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 +394,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..073fcaf 100644 --- a/dashscope/audio/tts_v2/enrollment.py +++ b/dashscope/audio/tts_v2/enrollment.py @@ -92,13 +92,13 @@ 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 +133,10 @@ 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 +170,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 +194,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 +219,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..45bb329 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 +918,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 +995,7 @@ 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 +1009,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..9f17f01 100644 --- a/dashscope/client/base_api.py +++ b/dashscope/client/base_api.py @@ -81,9 +81,9 @@ 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 +236,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 +246,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..3b54904 100644 --- a/dashscope/multimodal/dialog_state.py +++ b/dashscope/multimodal/dialog_state.py @@ -6,13 +6,13 @@ 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 +23,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..406ad84 100644 --- a/dashscope/multimodal/multimodal_constants.py +++ b/dashscope/multimodal/multimodal_constants.py @@ -21,12 +21,12 @@ 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_SPEECH_ENDED = "SpeechEnded" # Server sends this event when ASR speech endpoint is detected, optional event RESPONSE_NAME_RESPONDING_STARTED = ( - "RespondingStarted" # AI语音应答开始，sdk要准备接收服务端下发的语音数据 + "RespondingStarted" # AI voice response starts, SDK should prepare to receive audio data from server ) -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_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_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..ec4bbf5 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,18 @@ def __init__( model: str = None, ): """ - 创建一个语音对话会话。 + Create a voice dialog session. - 此方法用于初始化一个新的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: 模型 + 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 +182,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 +228,11 @@ 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 +244,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 +252,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 +264,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 +277,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 +289,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 +301,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 +309,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 +317,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 +342,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 +350,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 +367,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 +388,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 +409,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 +460,15 @@ 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 +493,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 +512,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 +535,13 @@ 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 # noqa: E501 + :param text: Text. + :param parameters: Parameters in command request body + :return: Command request dictionary. """ self._get_dash_request_header(ActionType.CONTINUE) @@ -572,10 +573,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 +601,8 @@ 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 # noqa: E501 """ if self.task_id is None: self.task_id = get_random_uuid() @@ -618,13 +619,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 +652,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 +761,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..00fc3e9 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,12 @@ 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 +140,18 @@ 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..f3cbeaf 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,7 @@ 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 +598,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" From 5fdd2e393a30ca795f67945a61920c1c24060b0f Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Mon, 15 Jun 2026 11:13:15 +0800 Subject: [PATCH 5/9] Remove redundant HTML header block from README Co-Authored-By: Claude Opus 4.7 --- README.md | 9 --------- 1 file changed, 9 deletions(-) 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. From 93a0d28599da4d99c0f70b7de9f5465f36043895 Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Mon, 15 Jun 2026 11:27:30 +0800 Subject: [PATCH 6/9] Fix pre-commit lint issues Co-Authored-By: Claude Opus 4.7 --- dashscope/audio/asr/vocabulary.py | 6 ++- .../qwen_tts_realtime/qwen_tts_realtime.py | 4 +- dashscope/audio/tts_v2/enrollment.py | 9 +++-- dashscope/audio/tts_v2/speech_synthesizer.py | 6 ++- dashscope/client/base_api.py | 3 +- dashscope/multimodal/dialog_state.py | 3 +- dashscope/multimodal/multimodal_constants.py | 18 +++++---- dashscope/multimodal/multimodal_dialog.py | 38 ++++++++++++------- .../multimodal/multimodal_request_params.py | 23 ++++++++--- .../multimodal/tingwu/tingwu_realtime.py | 4 +- 10 files changed, 77 insertions(+), 37 deletions(-) diff --git a/dashscope/audio/asr/vocabulary.py b/dashscope/audio/asr/vocabulary.py index 103f08a..8fd2ce8 100644 --- a/dashscope/audio/asr/vocabulary.py +++ b/dashscope/audio/asr/vocabulary.py @@ -88,7 +88,8 @@ def create_vocabulary( """ 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: 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 """ @@ -120,7 +121,8 @@ def list_vocabularies( ) -> List[dict]: """ List all created hot word tables. - param: prefix Custom prefix, if set only returns hot word table identifiers with the specified prefix. + 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 diff --git a/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py b/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py index 7b8e37b..415c098 100644 --- a/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py +++ b/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py @@ -222,7 +222,9 @@ def update_session( if speech_rate is not None: self.config["speech_rate"] = speech_rate if audio_format is not None: - self.config["response_format"] = audio_format # update if configured + 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: diff --git a/dashscope/audio/tts_v2/enrollment.py b/dashscope/audio/tts_v2/enrollment.py index 073fcaf..9a728c3 100644 --- a/dashscope/audio/tts_v2/enrollment.py +++ b/dashscope/audio/tts_v2/enrollment.py @@ -94,10 +94,12 @@ def create_voice( """ 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: 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: 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 """ @@ -136,7 +138,8 @@ def list_voices( 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. + return: List[dict] Voice list, including id, creation time, + modification time, and status for each voice. """ if prefix: # pylint: disable=no-value-for-parameter diff --git a/dashscope/audio/tts_v2/speech_synthesizer.py b/dashscope/audio/tts_v2/speech_synthesizer.py index 45bb329..4420456 100644 --- a/dashscope/audio/tts_v2/speech_synthesizer.py +++ b/dashscope/audio/tts_v2/speech_synthesizer.py @@ -770,7 +770,8 @@ def call(self, text: str, timeout_millis=None): it will wait for the corresponding number of milliseconds; otherwise, it will wait indefinitely. """ - # print('non-streaming TTS not yet supported for LLM calls, using streaming simulation') + # print('non-streaming TTS not yet supported for LLM calls,' + # ' using streaming simulation') if self.additional_params is None: self.additional_params = {"enable_ssml": True} else: @@ -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 + # Iterate over object pool, return pre-connected object + # if available for idx, poolObject in enumerate(self._pool): if ( self._avaliable[idx] diff --git a/dashscope/client/base_api.py b/dashscope/client/base_api.py index 9f17f01..a609584 100644 --- a/dashscope/client/base_api.py +++ b/dashscope/client/base_api.py @@ -81,7 +81,8 @@ def _handle_kwargs( @classmethod async def _handle_request(cls, request): - # If aio_call returns an async generator, consume it to get the response + # 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): diff --git a/dashscope/multimodal/dialog_state.py b/dashscope/multimodal/dialog_state.py index 3b54904..a0c76d6 100644 --- a/dashscope/multimodal/dialog_state.py +++ b/dashscope/multimodal/dialog_state.py @@ -6,7 +6,8 @@ class DialogState(Enum): """ - Dialog state enumeration class, defining the possible states of a dialog bot. + Dialog state enumeration class, defining the possible states + of a dialog bot. Attributes: IDLE (str): Bot is in idle state. diff --git a/dashscope/multimodal/multimodal_constants.py b/dashscope/multimodal/multimodal_constants.py index 406ad84..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" # Server sends this event when ASR speech endpoint is detected, optional event -RESPONSE_NAME_RESPONDING_STARTED = ( - "RespondingStarted" # AI voice response starts, SDK should prepare to receive audio data from server -) +# 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_SPEECH_CONTENT = ( + "SpeechContent" # User speech recognition text, full streaming output +) +RESPONSE_NAME_RESPONDING_CONTENT = ( + "RespondingContent" # System output text, full streaming output +) 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 ec4bbf5..1517e97 100644 --- a/dashscope/multimodal/multimodal_dialog.py +++ b/dashscope/multimodal/multimodal_dialog.py @@ -147,15 +147,19 @@ def __init__( """ 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 + 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 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 dialog_id: Dialog ID, if provided, continues the + conversation with previous context :param model: Model """ if request_params is None: @@ -229,10 +233,14 @@ def _on_open(self, ws): # pylint: disable=unused-argument def start(self, dialog_id, enable_voice_detection=False, task_id=None): """ 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. + :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) @@ -467,7 +475,8 @@ def generate_start_request( :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. + :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 @@ -538,7 +547,9 @@ def generate_request_to_response_json( 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 # noqa: E501 + :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. @@ -602,7 +613,8 @@ def generate_update_info_json( def _get_dash_request_header(self, action: str): """ Build request protocol header for multimodal dialog request. - :param action: ActionType DashScope protocol action, supports: run-task, continue-task, finish-task # noqa: E501 + :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() diff --git a/dashscope/multimodal/multimodal_request_params.py b/dashscope/multimodal/multimodal_request_params.py index 00fc3e9..1caf609 100644 --- a/dashscope/multimodal/multimodal_request_params.py +++ b/dashscope/multimodal/multimodal_request_params.py @@ -110,11 +110,16 @@ def to_dict(self): class Upstream: """struct for upstream""" - audio_format: str = field(default="pcm") # upstream audio format, default pcm, supports pcm/opus + audio_format: str = field( + default="pcm", + ) # upstream audio format, default pcm, supports pcm/opus type: str = field( default="AudioOnly", - ) # upstream type: AudioOnly for voice only; AudioAndVideo for video upload - mode: str = field(default="tap2talk") # client interaction mode: push2talk/tap2talk/duplex + ) # 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) @@ -144,11 +149,17 @@ class Downstream: # 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: + 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 + 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 diff --git a/dashscope/multimodal/tingwu/tingwu_realtime.py b/dashscope/multimodal/tingwu/tingwu_realtime.py index f3cbeaf..71ed4ab 100644 --- a/dashscope/multimodal/tingwu/tingwu_realtime.py +++ b/dashscope/multimodal/tingwu/tingwu_realtime.py @@ -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 event always arrives before speech-end event + 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() From eadf165012e6bb4af8a41489e6bd59dd78b845ac Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Wed, 17 Jun 2026 15:01:50 +0800 Subject: [PATCH 7/9] Fix end_session() to handle server errors immediately The server currently does not support session.finish event for qwen3.5-omni-flash-realtime, returning an error instead of session.finished. Previously end_session() would wait the full timeout duration (20s) before falling back to ws.close(). Now the method detects server error events immediately and closes the connection without waiting, reducing wait time from 20s to <0.1s when the server rejects session.finish. Related to issue #140 --- dashscope/audio/qwen_omni/omni_realtime.py | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/dashscope/audio/qwen_omni/omni_realtime.py b/dashscope/audio/qwen_omni/omni_realtime.py index 9aeabdd..ec4a1ab 100644 --- a/dashscope/audio/qwen_omni/omni_realtime.py +++ b/dashscope/audio/qwen_omni/omni_realtime.py @@ -153,6 +153,7 @@ def __init__( self.metrics = [] # Add event for synchronously waiting on connection close self.disconnect_event = None + self._disconnect_error = None def _generate_event_id(self): """ @@ -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() + return if not finish_success: self.close() raise TimeoutError( @@ -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 From ea82cff4917824d33d9995b418620f5fc8ef5079 Mon Sep 17 00:00:00 2001 From: "zhansheng.lzs" Date: Wed, 17 Jun 2026 15:28:42 +0800 Subject: [PATCH 8/9] Fix tiktoken StackOverflow on long text inputs tiktoken's recursive BPE merging overflows the call stack on very long inputs, causing pyo3_runtime.PanicException: StackOverflow. Split text into chunks (<= 100k chars) at line boundaries before encoding. Short text takes the original fast path unchanged. Fixes #116 --- dashscope/tokenizers/qwen_tokenizer.py | 47 +++++++++++++++++++++++--- 1 file changed, 42 insertions(+), 5 deletions(-) diff --git a/dashscope/tokenizers/qwen_tokenizer.py b/dashscope/tokenizers/qwen_tokenizer.py index bfa1119..d06d755 100644 --- a/dashscope/tokenizers/qwen_tokenizer.py +++ b/dashscope/tokenizers/qwen_tokenizer.py @@ -33,6 +33,11 @@ ) SPECIAL_TOKENS_SET = set(t for i, t in SPECIAL_TOKENS) +# tiktoken's BPE merges tokens recursively in Rust, which can overflow the +# call stack on very long inputs (pyo3_runtime.PanicException: StackOverflow). +# Split text into chunks below this threshold before encoding. +_CHUNK_SIZE = 100_000 + class QwenTokenizer(Tokenizer): @staticmethod @@ -102,11 +107,43 @@ def encode( # type: ignore[override] disallowed_special: Union[Collection, str] = (), ) -> Union[List[List], List]: text = unicodedata.normalize("NFC", text) - return self._tokenizer.encode( - text, - allowed_special=allowed_special, - disallowed_special=disallowed_special, - ) + if len(text) <= _CHUNK_SIZE: + return self._tokenizer.encode( + text, + allowed_special=allowed_special, + disallowed_special=disallowed_special, + ) + + result = [] + for chunk in self._split_text(text): + result.extend( + self._tokenizer.encode( + chunk, + allowed_special=allowed_special, + disallowed_special=disallowed_special, + ), + ) + return result + + @staticmethod + def _split_text(text: str, chunk_size: int = _CHUNK_SIZE) -> List[str]: + """Split text into chunks at safe tokenization boundaries.""" + parts: List[str] = [] + for i, line in enumerate(text.split("\n")): + piece = line if i == 0 else "\n" + line + if len(piece) <= chunk_size: + parts.append(piece) + else: + for j in range(0, len(piece), chunk_size): + parts.append(piece[j : j + chunk_size]) + + chunks: List[str] = [] + for part in parts: + if chunks and len(chunks[-1]) + len(part) <= chunk_size: + chunks[-1] += part + else: + chunks.append(part) + return chunks def decode( self, From 4dcbba598bb18964ea9d29b6710203a6177890e8 Mon Sep 17 00:00:00 2001 From: kevin Date: Wed, 17 Jun 2026 16:35:28 +0800 Subject: [PATCH 9/9] feat: support env proxy and merge incremental stream output for Application API --- dashscope/api_entities/aio_session.py | 2 +- dashscope/api_entities/websocket_request.py | 1 + dashscope/app/application.py | 57 +++++++++++++++++-- .../finetune/reinforcement/common/utils.py | 1 + 4 files changed, 56 insertions(+), 5 deletions(-) diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py index ee64eec..f243fca 100644 --- a/dashscope/api_entities/aio_session.py +++ b/dashscope/api_entities/aio_session.py @@ -41,7 +41,7 @@ async def get_shared_aio_session() -> aiohttp.ClientSession: return session connector = aiohttp.TCPConnector(ssl=get_ssl_context()) - session = aiohttp.ClientSession(connector=connector) + session = aiohttp.ClientSession(connector=connector, trust_env=True) _aio_sessions[loop] = session return session 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..08b612e 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,26 @@ 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 (append) + if "headers" not in kwargs: + kwargs["headers"] = {} + flag = "1" if to_merge_incremental_output else "0" + existing_ua = kwargs["headers"].get("user-agent", "") + new_ua = f"incremental_to_full/{flag}" + kwargs["headers"]["user-agent"] = ( + f"{existing_ua} {new_ua}".strip() if existing_ua else new_ua + ) + ( input, # pylint: disable=redefined-builtin parameters, @@ -171,12 +192,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 +262,28 @@ 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) + 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/finetune/reinforcement/common/utils.py b/dashscope/finetune/reinforcement/common/utils.py index 3757460..27da3d9 100644 --- a/dashscope/finetune/reinforcement/common/utils.py +++ b/dashscope/finetune/reinforcement/common/utils.py @@ -74,6 +74,7 @@ async def _make_request() -> Dict[str, Any]: async with aiohttp.ClientSession( headers=headers, timeout=aiohttp.ClientTimeout(total=timeout), + trust_env=True, ) as session: method_upper = method.upper()