diff --git a/dashscope/aigc/generation.py b/dashscope/aigc/generation.py index 6ec2a39..054a4f4 100644 --- a/dashscope/aigc/generation.py +++ b/dashscope/aigc/generation.py @@ -179,11 +179,13 @@ def call( # pylint: disable=arguments-renamed,too-many-branches,too-many-statem to_merge_incremental_output = True parameters["incremental_output"] = True - # Pass incremental_to_full flag via headers user-agent - if "headers" not in parameters: - parameters["headers"] = {} + # Pass incremental_to_full flag via user_agent parameter flag = "1" if to_merge_incremental_output else "0" - parameters["headers"]["user-agent"] = f"incremental_to_full/{flag}" + existing_ua = parameters.get("user_agent", "") + new_ua = f"incremental_to_full/{flag}" + parameters["user_agent"] = ( + f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua + ) response = super().call( model=model, @@ -434,11 +436,13 @@ async def call( # type: ignore[override] # pylint: disable=arguments-renamed,to to_merge_incremental_output = True parameters["incremental_output"] = True - # Pass incremental_to_full flag via headers user-agent - if "headers" not in parameters: - parameters["headers"] = {} + # Pass incremental_to_full flag via user_agent parameter flag = "1" if to_merge_incremental_output else "0" - parameters["headers"]["user-agent"] = f"incremental_to_full/{flag}" + existing_ua = parameters.get("user_agent", "") + new_ua = f"incremental_to_full/{flag}" + parameters["user_agent"] = ( + f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua + ) response = await super().call( model=model, diff --git a/dashscope/aigc/image_generation.py b/dashscope/aigc/image_generation.py index 6973108..b830495 100644 --- a/dashscope/aigc/image_generation.py +++ b/dashscope/aigc/image_generation.py @@ -99,14 +99,15 @@ def call( # type: ignore[override] to_merge_incremental_output = True kwargs["incremental_output"] = True - # Pass incremental_to_full flag via headers user-agent - if "headers" not in kwargs: - kwargs["headers"] = {} - + # Pass incremental_to_full flag via user_agent parameter flag = "1" if to_merge_incremental_output else "0" - kwargs["headers"]["user-agent"] = f"incremental_to_full/{flag}" + existing_ua = kwargs.get("user_agent", "") + new_ua = f"incremental_to_full/{flag}" + kwargs["user_agent"] = ( + f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua + ) if kwargs.get("is_async", False): - kwargs["headers"]["X-DashScope-Async"] = "enable" + kwargs.setdefault("headers", {})["X-DashScope-Async"] = "enable" task = cls.async_task else: task = cls.sync_task @@ -416,14 +417,15 @@ async def call( # type: ignore[override] to_merge_incremental_output = True kwargs["incremental_output"] = True - # Pass incremental_to_full flag via headers user-agent - if "headers" not in kwargs: - kwargs["headers"] = {} - + # Pass incremental_to_full flag via user_agent parameter flag = "1" if to_merge_incremental_output else "0" - kwargs["headers"]["user-agent"] = f"incremental_to_full/{flag}" + existing_ua = kwargs.get("user_agent", "") + new_ua = f"incremental_to_full/{flag}" + kwargs["user_agent"] = ( + f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua + ) if kwargs.get("is_async", False): - kwargs["headers"]["X-DashScope-Async"] = "enable" + kwargs.setdefault("headers", {})["X-DashScope-Async"] = "enable" task = cls.async_task else: task = cls.sync_task diff --git a/dashscope/aigc/multimodal_conversation.py b/dashscope/aigc/multimodal_conversation.py index cac2141..8243e77 100644 --- a/dashscope/aigc/multimodal_conversation.py +++ b/dashscope/aigc/multimodal_conversation.py @@ -163,11 +163,13 @@ def call( # pylint: disable=arguments-renamed,too-many-branches,too-many-statem to_merge_incremental_output = True kwargs["incremental_output"] = True - # Pass incremental_to_full flag via headers user-agent - if "headers" not in kwargs: - kwargs["headers"] = {} + # Pass incremental_to_full flag via user_agent parameter flag = "1" if to_merge_incremental_output else "0" - kwargs["headers"]["user-agent"] = f"incremental_to_full/{flag}" + existing_ua = kwargs.get("user_agent", "") + new_ua = f"incremental_to_full/{flag}" + kwargs["user_agent"] = ( + f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua + ) response = super().call( model=model, diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py index d2365db..2757f3c 100644 --- a/dashscope/api_entities/aio_session.py +++ b/dashscope/api_entities/aio_session.py @@ -45,7 +45,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/api_request_factory.py b/dashscope/api_entities/api_request_factory.py index cc58479..c231287 100644 --- a/dashscope/api_entities/api_request_factory.py +++ b/dashscope/api_entities/api_request_factory.py @@ -38,10 +38,16 @@ def _get_protocol_params(kwargs): extra_url_parameters = kwargs.pop("extra_url_parameters", None) session = kwargs.pop("session", None) - # Extract user-agent from headers if present - user_agent = "" + # Extract user_agent from kwargs (preferred) or from headers["user-agent"] + user_agent = kwargs.pop("user_agent", "") if headers and "user-agent" in headers: - user_agent = headers.pop("user-agent") + header_ua = headers.pop("user-agent") + if user_agent: + user_agent = ( + f"{header_ua}; {user_agent}" if header_ua else user_agent + ) + else: + user_agent = header_ua return ( api_protocol, diff --git a/dashscope/api_entities/websocket_request.py b/dashscope/api_entities/websocket_request.py index 0473709..249b8d8 100644 --- a/dashscope/api_entities/websocket_request.py +++ b/dashscope/api_entities/websocket_request.py @@ -119,6 +119,7 @@ async def connection_handler(self): # pylint: disable=too-many-branches timeout=aiohttp.ClientTimeout( total=self.timeout, ), + trust_env=True, ) as session: async with session.ws_connect( self.url, diff --git a/dashscope/app/application.py b/dashscope/app/application.py index 78a77a0..0926cbd 100644 --- a/dashscope/app/application.py +++ b/dashscope/app/application.py @@ -21,6 +21,7 @@ ) from dashscope.common.error import InputRequired, InvalidInput from dashscope.common.logging import logger +from dashscope.utils.message_utils import merge_single_response class Application(BaseApi): @@ -149,6 +150,25 @@ def call( # type: ignore[override] headers["X-DashScope-WorkSpace"] = workspace kwargs["headers"] = headers + # Check if we need to merge incremental output (compute once) + is_stream = kwargs.get("stream", False) + is_incremental_output = kwargs.get("incremental_output", None) + to_merge_incremental_output = ( + is_stream and is_incremental_output is False + ) + + if to_merge_incremental_output: + kwargs["incremental_output"] = True + + # Pass incremental_to_full flag via user_agent parameter to avoid + # overwriting the default SDK user-agent + flag = "1" if to_merge_incremental_output else "0" + existing_ua = kwargs.get("user_agent", "") + new_ua = f"incremental_to_full/{flag}" + kwargs["user_agent"] = ( + f"{existing_ua}; {new_ua}".strip() if existing_ua else new_ua + ) + ( input, # pylint: disable=redefined-builtin parameters, @@ -171,12 +191,15 @@ def call( # type: ignore[override] ) # call request service. response = request.call() - is_stream = kwargs.get("stream", False) if is_stream: - return ( - ApplicationResponse.from_api_response(rsp) for rsp in response - ) + if to_merge_incremental_output: + return cls._merge_application_response(response) + else: + return ( + ApplicationResponse.from_api_response(rsp) + for rsp in response + ) else: return ApplicationResponse.from_api_response(response) @@ -238,3 +261,33 @@ def _build_input_parameters( # pylint: disable=too-many-branches input_param["file_list"] = file_list return input_param, {**parameters, **kwargs} + + @classmethod + def _merge_application_response(cls, response): + """Merge incremental application response chunks. + + Simulate non-incremental output by accumulating text. + """ + accumulated_data = {} + + for rsp in response: + parsed_response = ApplicationResponse.from_api_response(rsp) + if parsed_response.output and not hasattr( + parsed_response.output, + "choices", + ): + parsed_response.output.choices = None + result = merge_single_response( + parsed_response, + accumulated_data, + ) + if result is True: + yield parsed_response + elif isinstance(result, list): + for resp in result: + yield resp + else: + logger.warning( + "Unexpected merge result type: %s, skipping", + type(result).__name__, + ) diff --git a/dashscope/audio/qwen_omni/omni_realtime.py b/dashscope/audio/qwen_omni/omni_realtime.py index 9aeabdd..4f15735 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() + raise RuntimeError(f"Session ended with error: {error}") 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 diff --git a/dashscope/client/base_api.py b/dashscope/client/base_api.py index 68f682a..d67149b 100644 --- a/dashscope/client/base_api.py +++ b/dashscope/client/base_api.py @@ -374,7 +374,7 @@ async def list( **_workspace_header(workspace), **default_headers(api_key), } - async with aiohttp.ClientSession() as session: + async with aiohttp.ClientSession(trust_env=True) as session: response = await session.get( url, params=params, 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() diff --git a/dashscope/tokenizers/qwen_tokenizer.py b/dashscope/tokenizers/qwen_tokenizer.py index bfa1119..9a538f9 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,51 @@ 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] = [] + current_chunk: List[str] = [] + current_len = 0 + for part in parts: + if current_len + len(part) <= chunk_size: + current_chunk.append(part) + current_len += len(part) + else: + if current_chunk: + chunks.append("".join(current_chunk)) + current_chunk = [part] + current_len = len(part) + if current_chunk: + chunks.append("".join(current_chunk)) + return chunks def decode( self, diff --git a/dashscope/version.py b/dashscope/version.py index 117bf62..a4c4f3f 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.22" +__version__ = "1.25.23" diff --git a/tests/unit/test_conversation.py b/tests/unit/test_conversation.py index 52c734d..8dedcc0 100644 --- a/tests/unit/test_conversation.py +++ b/tests/unit/test_conversation.py @@ -234,5 +234,5 @@ def test_not_qwen(self, mock_server: MockServer): assert response.output.finish_reason == "stop" req = mock_server.requests.get(block=True) assert req["model"] == Generation.Models.dolly_12b_v2 - assert req["parameters"] == {} + assert "user_agent" not in req.get("parameters", {}) assert req["input"] == {"prompt": prompt} diff --git a/tests/unit/test_tokenizer.py b/tests/unit/test_tokenizer.py index 903e1f6..682c6a1 100644 --- a/tests/unit/test_tokenizer.py +++ b/tests/unit/test_tokenizer.py @@ -4,6 +4,7 @@ import os from dashscope.tokenizers.tokenizer import get_tokenizer +from dashscope.tokenizers.qwen_tokenizer import _CHUNK_SIZE class TestTokenization: @@ -47,3 +48,33 @@ def test_encode_decode(self): "<|endoftext|>", disallowed_special=set(), ) + + def test_encode_chunk_size_exceed(self): + # Test encoding functionality when text length exceeds _CHUNK_SIZE + tokenizer = get_tokenizer("qwen-7b-chat") + + # Create a long text that exceeds _CHUNK_SIZE + long_text = "Hello world! " * ( + _CHUNK_SIZE // 12 + 10 + ) # Ensure it exceeds the threshold + + # Encode long text, should not raise an exception + tokens = tokenizer.encode(long_text) + + # Decoded text should match the original string + decoded_str = tokenizer.decode(tokens) + assert decoded_str == long_text + + # Ensure the return type is a list + assert isinstance(tokens, list) + + # Test long text with special characters + long_text_with_special = ( + "<|extra_0|> " * (_CHUNK_SIZE // 12 + 5) + ) + "Normal text here." + tokens_with_special = tokenizer.encode( + long_text_with_special, + allowed_special="all", + ) + decoded_with_special = tokenizer.decode(tokens_with_special) + assert decoded_with_special == long_text_with_special