fix: raise ValueError instead of assert in DeepSeek DSML completion parsing

lmdeploy/deepseek_v32_encoding.py

@@ -318,34 +318,40 @@ def parse_tool_calls(index: int, text: str):
 
     while index < len(text):
         index, _, stop_token = _read_until_stop(index, text, [f'<{dsml_token}invoke', tool_calls_end_token])
-        assert _ == '>\n', 'Tool call format error'
+        if _ != '>\n':
+            raise ValueError('Tool call format error')
 
         if stop_token == tool_calls_end_token:
             break
 
-        assert stop_token is not None, 'Missing special token'
+        if stop_token is None:
+            raise ValueError('Missing special token')
 
         index, tool_name_content, stop_token = _read_until_stop(
             index, text, [f'<{dsml_token}parameter', f'</{dsml_token}invoke'])
 
         p_tool_name = re.findall(r'^\s*name="(.*?)">\n$', tool_name_content, flags=re.DOTALL)
-        assert len(p_tool_name) == 1, 'Tool name format error'
+        if len(p_tool_name) != 1:
+            raise ValueError('Tool name format error')
         tool_name = p_tool_name[0]
 
         tool_args: dict[str, tuple[str, str]] = {}
         while stop_token == f'<{dsml_token}parameter':
             index, param_content, stop_token = _read_until_stop(index, text, [f'/{dsml_token}parameter'])
 
             param_kv = re.findall(r'^ name="(.*?)" string="(true|false)">(.*?)<$', param_content, flags=re.DOTALL)
-            assert len(param_kv) == 1, 'Parameter format error'
+            if len(param_kv) != 1:
+                raise ValueError('Parameter format error')
             param_name, string, param_value = param_kv[0]
 
-            assert param_name not in tool_args, 'Duplicate parameter name'
+            if param_name in tool_args:
+                raise ValueError('Duplicate parameter name')
             tool_args[param_name] = (param_value, string)
 
             index, content, stop_token = _read_until_stop(
                 index, text, [f'<{dsml_token}parameter', f'</{dsml_token}invoke'])
-            assert content == '>\n', 'Parameter format error'
+            if content != '>\n':
+                raise ValueError('Parameter format error')
 
         tool_call = decode_dsml_to_arguments(tool_name=tool_name, tool_args=tool_args)
         tool_calls.append(tool_call)
@@ -364,27 +370,30 @@ def parse_message_from_completion_text(text: str, thinking_mode: str):
     if is_thinking:
         index, content_delta, stop_token = _read_until_stop(index, text, [thinking_end_token, tool_calls_start_token])
         reasoning_content = content_delta
-        assert stop_token == thinking_end_token, 'Invalid thinking format'
+        if stop_token != thinking_end_token:
+            raise ValueError('Invalid thinking format')
 
     index, content_delta, stop_token = _read_until_stop(index, text, [eos_token, tool_calls_start_token])
     summary_content = content_delta
     if stop_token == tool_calls_start_token:
         is_tool_calling = True
     else:
-        assert stop_token == eos_token, 'Invalid summary format'
+        if stop_token != eos_token:
+            raise ValueError('Invalid summary format')
 
     if is_tool_calling:
         index, stop_token, tool_calls = parse_tool_calls(index, text)
 
         index, tool_ends_text, stop_token = _read_until_stop(index, text, [eos_token])
-        assert not tool_ends_text, 'Unexpected content after tool calls'
+        if tool_ends_text:
+            raise ValueError('Unexpected content after tool calls')
 
-    assert len(text) == index and stop_token in [eos_token, None], 'Unexpected content at end'
+    if len(text) != index or stop_token not in [eos_token, None]:
+        raise ValueError('Unexpected content at end')
 
     for sp_token in [bos_token, eos_token, thinking_start_token, thinking_end_token, dsml_token]:
-        assert (
-            sp_token not in summary_content and sp_token not in reasoning_content
-        ), 'Unexpected special token in content'
+        if sp_token in summary_content or sp_token in reasoning_content:
+            raise ValueError('Unexpected special token in content')
 
     return {
         'role': 'assistant',

lmdeploy/deepseek_v4_encoding.py

@@ -721,26 +721,30 @@ def parse_message_from_completion_text(text: str, thinking_mode: str) -> dict[st
     if is_thinking:
         index, content_delta, stop_token = _read_until_stop(index, text, [thinking_end_token, tool_calls_start_token])
         reasoning_content = content_delta
-        assert stop_token == thinking_end_token, 'Invalid thinking format: missing </think>'
+        if stop_token != thinking_end_token:
+            raise ValueError('Invalid thinking format: missing </think>')
 
     index, content_delta, stop_token = _read_until_stop(index, text, [eos_token, tool_calls_start_token])
     summary_content = content_delta
     if stop_token == tool_calls_start_token:
         is_tool_calling = True
     else:
-        assert stop_token == eos_token, 'Invalid format: missing EOS token'
+        if stop_token != eos_token:
+            raise ValueError('Invalid format: missing EOS token')
 
     if is_tool_calling:
         index, stop_token, tool_calls = parse_tool_calls(index, text)
 
         index, tool_ends_text, stop_token = _read_until_stop(index, text, [eos_token])
-        assert not tool_ends_text, 'Unexpected content after tool calls'
+        if tool_ends_text:
+            raise ValueError('Unexpected content after tool calls')
 
-    assert len(text) == index and stop_token in [eos_token, None], 'Unexpected content at end'
+    if len(text) != index or stop_token not in [eos_token, None]:
+        raise ValueError('Unexpected content at end')
 
     for sp_token in [bos_token, eos_token, thinking_start_token, thinking_end_token, dsml_token]:
-        assert sp_token not in summary_content and sp_token not in reasoning_content, \
-            f"Unexpected special token '{sp_token}' in content"
+        if sp_token in summary_content or sp_token in reasoning_content:
+            raise ValueError(f"Unexpected special token '{sp_token}' in content")
 
     return {
         'role': 'assistant',

tests/test_lmdeploy/test_deepseek_v32_encoding.py

@@ -1,6 +1,8 @@
 # Copyright (c) OpenMMLab. All rights reserved.
 import json
 
+import pytest
+
 from lmdeploy.deepseek_v32_encoding import (
     bos_token,
     encode_messages,
@@ -253,3 +255,10 @@ def test_deepseek_v32_response_parser_streaming_dsml_function_calls():
     assert tool_deltas[0].function.name == 'search'
     arguments = ''.join(tool_call.function.arguments or '' for tool_call in tool_deltas)
     assert json.loads(arguments) == {'query': 'DeepSeek V3.2'}
+
+
+def test_deepseek_v32_parse_completion_text_malformed_raises_value_error():
+    with pytest.raises(ValueError, match='Invalid summary format'):
+        parse_message_from_completion_text('partial answer, no terminal token', thinking_mode='chat')
+    with pytest.raises(ValueError, match='Invalid thinking format'):
+        parse_message_from_completion_text(f'no end of thinking{eos_token}', thinking_mode='thinking')

tests/test_lmdeploy/test_deepseek_v4_encoding.py

@@ -1,6 +1,8 @@
 # Copyright (c) OpenMMLab. All rights reserved.
 import json
 
+import pytest
+
 from lmdeploy.deepseek_v4_encoding import (
     REASONING_EFFORT_MAX,
     bos_token,
@@ -321,3 +323,10 @@ def test_deepseek_v4_response_parser_reasoning_effort_does_not_enable_thinking()
     assert tool_emitted is False
     assert delta.content == 'hello'
     assert delta.reasoning_content is None
+
+
+def test_deepseek_v4_parse_completion_text_malformed_raises_value_error():
+    with pytest.raises(ValueError, match='missing EOS token'):
+        parse_message_from_completion_text('partial answer, no terminal token', thinking_mode='chat')
+    with pytest.raises(ValueError, match='missing </think>'):
+        parse_message_from_completion_text(f'no end of thinking{eos_token}', thinking_mode='thinking')