Dev/issue 1

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,

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

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,

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
 

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,

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,

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__,
+                )

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

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,

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()
 

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,

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"

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}