# SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project """Tests for DeepSeek V4-specific parser engine semantics.""" import json import pytest from tests.parser.engine.conftest import make_mock_tokenizer from tests.parser.engine.replay_harness import ( DUMMY_TOOLS, MockTokenizer, _test_request, collect_output, replay_streaming, ) from tests.parser.engine.streaming_helpers import ( collect_content, collect_function_name, collect_tool_arguments, simulate_reasoning_streaming, simulate_tool_streaming, ) from vllm.parser.abstract_parser import DelegatingParser from vllm.parser.deepseek_v4 import ( DSML_INVOKE_END, DSML_INVOKE_NAME_END, DSML_INVOKE_PREFIX, DSML_THINK_END, DSML_THINK_START, DSML_TOOL_END, DSML_TOOL_START, DeepSeekV4Parser, _dsml_arg_converter, _unwrap_wrapper_args, deepseek_v4_config, ) from vllm.parser.engine.registered_adapters import ( DeepSeekV4ParserReasoningAdapter, DeepSeekV4ParserToolAdapter, ) _THINK_START_ID = 50 _THINK_END_ID = 51 _PARAM_OPEN = '｜DSML｜parameter name="{name}" string="{is_str}">' _PARAM_CLOSE = "" def _param(name: str, is_str: str, value: str) -> str: return f"<{_PARAM_OPEN.format(name=name, is_str=is_str)}{value}{_PARAM_CLOSE}" @pytest.fixture def mock_tokenizer(): return make_mock_tokenizer( { DSML_THINK_START: _THINK_START_ID, DSML_THINK_END: _THINK_END_ID, } ) # ── Arg converter unit tests ───────────────────────────────────────── class TestArgConverter: def _raw(self, *params: tuple[str, str, str]) -> str: lines = [_param(n, s, v) for n, s, v in params] return "\n" + "\n".join(lines) + "\n" def test_string_param(self): raw = self._raw(("city", "true", "杭州")) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result == {"city": "杭州"} def test_string_with_spaces_and_quotes(self): raw = self._raw(("msg", "true", 'He said "hello world"')) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result["msg"] == 'He said "hello world"' def test_integer_param(self): raw = self._raw(("count", "false", "42")) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result["count"] == 42 assert isinstance(result["count"], int) def test_float_param(self): raw = self._raw(("ratio", "false", "3.14")) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert abs(result["ratio"] - 3.14) < 1e-9 def test_bool_param(self): raw = self._raw(("flag", "false", "true")) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result["flag"] is True def test_array_param(self): raw = self._raw(("items", "false", '["a", "b", "c"]')) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result["items"] == ["a", "b", "c"] def test_object_param(self): raw = self._raw(("opts", "false", '{"key": "val"}')) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result["opts"] == {"key": "val"} def test_mixed_types(self): raw = self._raw( ("location", "true", "Tokyo"), ("limit", "false", "10"), ("active", "false", "false"), ) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result == {"location": "Tokyo", "limit": 10, "active": False} def test_empty_args(self): result = json.loads(_dsml_arg_converter("", partial=False)) assert result == {} def test_invalid_json_fallback(self): raw = self._raw(("data", "false", "[broken")) result = json.loads(_dsml_arg_converter(raw, partial=False)) assert result["data"] == "[broken" def test_chinese_chars_preserved_in_json(self): raw = self._raw(("query", "true", "你好世界")) raw_json = _dsml_arg_converter(raw, partial=False) assert "你好世界" in raw_json result = json.loads(raw_json) assert result["query"] == "你好世界" def test_partial_complete_plus_in_progress(self): raw = self._raw(("city", "true", "Tokyo")) raw += f"<{_PARAM_OPEN.format(name='unit', is_str='true')}celsi" result = json.loads(_dsml_arg_converter(raw, partial=True)) assert result["city"] == "Tokyo" assert result["unit"] == "celsi" def test_partial_no_in_progress(self): raw = self._raw(("city", "true", "Tokyo")) result = json.loads(_dsml_arg_converter(raw, partial=True)) assert result == {"city": "Tokyo"} def test_partial_value_with_angle_bracket(self): raw = f"<{_PARAM_OPEN.format(name='code', is_str='true')}a absorption and duplicate absorption ───────── class TestThinkTagAbsorption: def test_bare_think_end_not_leaked(self, mock_tokenizer): parser = DeepSeekV4Parser(mock_tokenizer) chunks = ["", "Here is the direct answer."] reasoning, content = simulate_reasoning_streaming(parser, chunks) assert reasoning == "" assert "" not in content assert "Here is the direct answer" in content def test_duplicate_think_start_absorbed(self, mock_tokenizer): parser = DeepSeekV4Parser( mock_tokenizer, chat_template_kwargs={"thinking": True} ) chunks = [ "\n", "Some reasoning.\n", "\n", "Answer.", ] reasoning, content = simulate_reasoning_streaming(parser, chunks) assert "Some reasoning" in reasoning assert "Answer" in content # ── Missing before ──────────── class TestMissingInvokeEnd: def test_non_streaming(self, mock_tokenizer, mock_request): parser = DeepSeekV4Parser(mock_tokenizer) text = ( f"{DSML_TOOL_START}" f"{DSML_INVOKE_PREFIX}get_weather{DSML_INVOKE_NAME_END}\n" f"{_param('location', 'true', 'NYC')}\n" f"{DSML_TOOL_END}" ) result = parser.extract_tool_calls(text, mock_request) assert result.tools_called is True assert len(result.tool_calls) == 1 assert result.tool_calls[0].function.name == "get_weather" args = json.loads(result.tool_calls[0].function.arguments) assert args == {"location": "NYC"} def test_streaming_with_trailing_content(self, mock_tokenizer, mock_request): parser = DeepSeekV4Parser(mock_tokenizer) chunks = [ DSML_TOOL_START, f"{DSML_INVOKE_PREFIX}get_weather{DSML_INVOKE_NAME_END}\n" f"{_param('location', 'true', 'NYC')}\n", DSML_TOOL_END, "Done.", ] results = simulate_tool_streaming(parser, mock_request, chunks) assert collect_function_name(results) == "get_weather" args = json.loads(collect_tool_arguments(results)) assert args == {"location": "NYC"} assert "Done." in collect_content(results) # ── Thinking mode initial state ────────────────────────────────────── class TestThinkingModeConfig: def test_thinking_true_starts_in_reasoning(self): cfg = deepseek_v4_config(thinking=True) assert cfg.initial_state.name == "REASONING" def test_thinking_false_starts_in_content(self): cfg = deepseek_v4_config(thinking=False) assert cfg.initial_state.name == "CONTENT" def test_enable_thinking_kwarg(self, mock_tokenizer): p = DeepSeekV4Parser( mock_tokenizer, chat_template_kwargs={"enable_thinking": True} ) assert p.parser_engine_config.initial_state.name == "REASONING" def test_no_thinking_kwarg_defaults_to_content(self, mock_tokenizer): p = DeepSeekV4Parser(mock_tokenizer) assert p.parser_engine_config.initial_state.name == "CONTENT" def test_thinking_mode_reasoning_without_tags(self, mock_tokenizer): parser = DeepSeekV4Parser( mock_tokenizer, chat_template_kwargs={"thinking": True} ) chunks = [ "\n\nLet me consider ", "this carefully.\n", "\n", "Here is the result.", ] reasoning, content = simulate_reasoning_streaming(parser, chunks) assert "Let me consider" in reasoning assert "Here is the result" in content def test_thinking_mode_all_reasoning_no_end_tag(self, mock_tokenizer): parser = DeepSeekV4Parser( mock_tokenizer, chat_template_kwargs={"thinking": True} ) chunks = ["I'll review ", "the PR."] reasoning, content = simulate_reasoning_streaming(parser, chunks) assert "review" in reasoning assert "the PR" in reasoning assert content == "" def test_reasoning_effort_none_overrides_enable_thinking(self, mock_tokenizer): p = DeepSeekV4Parser( mock_tokenizer, chat_template_kwargs={ "enable_thinking": True, "reasoning_effort": "none", }, ) assert p.parser_engine_config.initial_state.name == "CONTENT" # ── Implicit reasoning end (missing before tool calls) ───── class TestImplicitReasoningEnd: """Tool call markers end reasoning implicitly when is missing. DeepSeek V4 models occasionally omit before emitting tool calls. The (REASONING, TOOL_START) transition handles this gracefully. """ @pytest.fixture def thinking_parser(self, mock_tokenizer): return DeepSeekV4Parser(mock_tokenizer, chat_template_kwargs={"thinking": True}) def _reasoning_then_tool(self, reasoning_text: str) -> str: return reasoning_text + _tool_calls( _invoke("get_weather", ("location", "true", "NYC")), ) def test_non_streaming_extract_reasoning_implicit_end(self, thinking_parser): text = self._reasoning_then_tool("Let me look up the weather.\n\n") reasoning, content = thinking_parser.extract_reasoning(text, None) assert reasoning == "Let me look up the weather." assert DSML_TOOL_START not in reasoning assert DSML_INVOKE_PREFIX not in reasoning assert content is None def test_non_streaming_extract_tool_calls_implicit_end( self, thinking_parser, mock_request ): text = self._reasoning_then_tool("Let me look up the weather.\n\n") result = thinking_parser.extract_tool_calls(text, mock_request) assert result.tools_called is True assert len(result.tool_calls) == 1 assert result.tool_calls[0].function.name == "get_weather" args = json.loads(result.tool_calls[0].function.arguments) assert args == {"location": "NYC"} def test_non_streaming_parse_implicit_end(self, thinking_parser, mock_request): text = self._reasoning_then_tool("Let me look up the weather.\n\n") reasoning, content, tool_calls = thinking_parser.parse(text, mock_request) assert reasoning == "Let me look up the weather." assert content is None assert tool_calls is not None assert len(tool_calls) == 1 assert tool_calls[0].name == "get_weather" args = json.loads(tool_calls[0].arguments) assert args == {"location": "NYC"} def test_streaming_reasoning_implicit_end(self, thinking_parser): chunks = [ "Let me look up the weather.\n\n", DSML_TOOL_START, DSML_INVOKE_PREFIX + "get_weather" + DSML_INVOKE_NAME_END, ] reasoning, content = simulate_reasoning_streaming(thinking_parser, chunks) assert reasoning == "Let me look up the weather." assert DSML_TOOL_START not in reasoning assert DSML_INVOKE_PREFIX not in reasoning def test_streaming_tool_extraction_implicit_end( self, thinking_parser, mock_request ): chunks = [ "Let me check.\n\n", DSML_TOOL_START, DSML_INVOKE_PREFIX + "get_weather" + DSML_INVOKE_NAME_END + "\n" + _param("location", "true", "NYC") + "\n" + DSML_INVOKE_END, DSML_TOOL_END, ] results = simulate_tool_streaming(thinking_parser, mock_request, chunks) assert collect_function_name(results) == "get_weather" args = json.loads(collect_tool_arguments(results)) assert args == {"location": "NYC"} def test_thinking_false_explicit_think_then_tool_call(self, mock_tokenizer): parser = DeepSeekV4Parser(mock_tokenizer) chunks = [ DSML_THINK_START, "Let me check the weather.", DSML_TOOL_START, DSML_INVOKE_PREFIX + "get_weather" + DSML_INVOKE_NAME_END, ] reasoning, content = simulate_reasoning_streaming(parser, chunks) assert "Let me check the weather" in reasoning assert DSML_TOOL_START not in reasoning assert DSML_THINK_START not in reasoning def test_non_streaming_parallel_tools_after_implicit_end( self, thinking_parser, mock_request ): text = "I need both.\n\n" + _tool_calls( _invoke("get_weather", ("location", "true", "NYC")), _invoke("get_time", ("timezone", "true", "EST")), ) result = thinking_parser.extract_tool_calls(text, mock_request) assert result.tools_called is True assert len(result.tool_calls) == 2 assert result.tool_calls[0].function.name == "get_weather" assert result.tool_calls[1].function.name == "get_time" def test_streaming_implicit_end_trailing_whitespace_stripped(self, thinking_parser): chunks = [ "Reasoning.\n\n\n", DSML_TOOL_START, DSML_INVOKE_PREFIX + "func" + DSML_INVOKE_NAME_END, ] reasoning, content = simulate_reasoning_streaming(thinking_parser, chunks) assert reasoning == "Reasoning." # ── Wrapper argument unwrapping ────────────────────────────────────── class TestWrapperUnwrapping: def test_unwrap_arguments_wrapper(self): from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionToolsParam, ) tool = ChatCompletionToolsParam( type="function", function={ "name": "get_weather", "parameters": { "type": "object", "properties": {"location": {"type": "string"}}, }, }, ) result = _unwrap_wrapper_args( '{"arguments": {"location": "Beijing"}}', [tool], "get_weather", ) assert json.loads(result) == {"location": "Beijing"} def test_unwrap_input_wrapper(self): from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionToolsParam, ) tool = ChatCompletionToolsParam( type="function", function={ "name": "get_weather", "parameters": { "type": "object", "properties": {"location": {"type": "string"}}, }, }, ) result = _unwrap_wrapper_args( '{"input": {"location": "Beijing"}}', [tool], "get_weather", ) assert json.loads(result) == {"location": "Beijing"} def test_no_unwrap_when_key_in_schema(self): from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionToolsParam, ) tool = ChatCompletionToolsParam( type="function", function={ "name": "func", "parameters": { "type": "object", "properties": {"arguments": {"type": "string"}}, }, }, ) result = _unwrap_wrapper_args( '{"arguments": "some value"}', [tool], "func", ) assert json.loads(result) == {"arguments": "some value"} def test_no_unwrap_when_no_tools(self): result = _unwrap_wrapper_args( '{"arguments": {"location": "Beijing"}}', None, "get_weather", ) assert json.loads(result) == {"arguments": {"location": "Beijing"}} def test_unwrap_json_string_inner(self): from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionToolsParam, ) tool = ChatCompletionToolsParam( type="function", function={ "name": "get_weather", "parameters": { "type": "object", "properties": {"location": {"type": "string"}}, }, }, ) result = _unwrap_wrapper_args( '{"arguments": "{\\"location\\": \\"Beijing\\"}"}', [tool], "get_weather", ) assert json.loads(result) == {"location": "Beijing"} # ── Parallel tool call wrapper unwrapping ─────────────────────────── def _make_tool(name, properties): from vllm.entrypoints.openai.chat_completion.protocol import ( # noqa: E501 ChatCompletionToolsParam, ) return ChatCompletionToolsParam( type="function", function={ "name": name, "parameters": { "type": "object", "properties": properties, }, }, ) def _invoke(name, *params): body = "\n".join(_param(n, s, v) for n, s, v in params) return ( f"{DSML_INVOKE_PREFIX}{name}{DSML_INVOKE_NAME_END}\n{body}\n{DSML_INVOKE_END}" ) def _tool_calls(*invokes): return DSML_TOOL_START + "\n".join(invokes) + DSML_TOOL_END class TestParallelUnwrapping: @pytest.fixture def weather_tool(self): return _make_tool( "get_weather", { "location": {"type": "string"}, "unit": {"type": "string"}, }, ) @pytest.fixture def time_tool(self): return _make_tool( "get_time", {"timezone": {"type": "string"}}, ) @pytest.mark.parametrize( "weather_args, expected", [ ( '{"location": "NYC", "unit": "celsius"}', {"location": "NYC", "unit": "celsius"}, ), ('{"location": "NYC"}', {"location": "NYC"}), ], ids=["all_props", "subset_props"], ) def test_unwrap_parallel_uses_correct_schema( self, mock_tokenizer, mock_request, weather_tool, time_tool, weather_args, expected, ): tools = [weather_tool, time_tool] parser = DeepSeekV4Parser(mock_tokenizer, tools=tools) mock_request.tools = tools text = _tool_calls( _invoke("get_weather", ("arguments", "false", weather_args)), _invoke("get_time", ("timezone", "true", "EST")), ) result = parser.extract_tool_calls(text, mock_request) assert result.tools_called is True assert len(result.tool_calls) == 2 assert result.tool_calls[0].function.name == "get_weather" args0 = json.loads(result.tool_calls[0].function.arguments) assert args0 == expected assert result.tool_calls[1].function.name == "get_time" args1 = json.loads(result.tool_calls[1].function.arguments) assert args1 == {"timezone": "EST"} def test_unwrap_parallel_streaming( self, mock_tokenizer, mock_request, weather_tool, time_tool ): tools = [weather_tool, time_tool] parser = DeepSeekV4Parser(mock_tokenizer, tools=tools) mock_request.tools = tools chunks = [ DSML_TOOL_START, _invoke( "get_weather", ("arguments", "false", '{"location": "NYC"}'), ), _invoke("get_time", ("timezone", "true", "EST")), DSML_TOOL_END, ] results = simulate_tool_streaming(parser, mock_request, chunks) final_delta, _ = results[-1] finish_delta = parser.finish_streaming() extracted = parser._build_extracted_result(final_delta, finish_delta) assert extracted.tools_called is True assert len(extracted.tool_calls) == 2 args0 = json.loads(extracted.tool_calls[0].function.arguments) assert args0 == {"location": "NYC"} args1 = json.loads(extracted.tool_calls[1].function.arguments) assert args1 == {"timezone": "EST"} def test_no_unwrap_parallel_when_no_match( self, mock_tokenizer, mock_request, weather_tool, time_tool ): tools = [weather_tool, time_tool] parser = DeepSeekV4Parser(mock_tokenizer, tools=tools) mock_request.tools = tools text = _tool_calls( _invoke( "get_weather", ("arguments", "false", '{"unknown_key": "val"}'), ), _invoke("get_time", ("timezone", "true", "EST")), ) result = parser.extract_tool_calls(text, mock_request) assert len(result.tool_calls) == 2 args0 = json.loads(result.tool_calls[0].function.arguments) assert args0 == {"arguments": {"unknown_key": "val"}} args1 = json.loads(result.tool_calls[1].function.arguments) assert args1 == {"timezone": "EST"} def test_unwrap_single_tool_still_works(self, mock_tokenizer, mock_request): tool = _make_tool("get_weather", {"location": {"type": "string"}}) tools = [tool] parser = DeepSeekV4Parser(mock_tokenizer, tools=tools) mock_request.tools = tools text = _tool_calls( _invoke( "get_weather", ("arguments", "false", '{"location": "Beijing"}'), ), ) result = parser.extract_tool_calls(text, mock_request) assert result.tools_called is True assert len(result.tool_calls) == 1 args = json.loads(result.tool_calls[0].function.arguments) assert args == {"location": "Beijing"} # ── Streaming wrapper consistency ───────────────────────────────────── class TestStreamingWrapperConsistency: """Streamed arg deltas must stay consistent with final extraction when wrapper params like 'arguments' are unwrapped.""" def test_streaming_wrapper_unwrap_consistency(self, mock_tokenizer, mock_request): tool = _make_tool("get_weather", {"location": {"type": "string"}}) tools = [tool] parser = DeepSeekV4Parser(mock_tokenizer, tools=tools) mock_request.tools = tools chunks = [ DSML_TOOL_START, _invoke( "get_weather", ("arguments", "false", '{"location": "NYC"}'), ), DSML_TOOL_END, ] results = simulate_tool_streaming(parser, mock_request, chunks) streamed_args = collect_tool_arguments(results) final_delta, _ = results[-1] finish_delta = parser.finish_streaming() extracted = parser._build_extracted_result(final_delta, finish_delta) assert extracted.tools_called is True assert len(extracted.tool_calls) == 1 final_args = extracted.tool_calls[0].function.arguments assert json.loads(final_args) == {"location": "NYC"} assert '"arguments"' not in streamed_args, ( f"Streamed args should not contain wrapper key, got: {streamed_args!r}" ) assert final_args.startswith(streamed_args), ( f"Extracted args {final_args!r} " f"should start with streamed args {streamed_args!r}" ) # ── DelegatingParser: large delta with + tool calls ───────── _DSV4_FULL_VOCAB = { DSML_THINK_START: 128821, DSML_THINK_END: 128822, DSML_TOOL_START: 128823, DSML_TOOL_END: 128824, } class _DeepSeekV4Delegating(DelegatingParser): reasoning_parser_cls = DeepSeekV4ParserReasoningAdapter tool_parser_cls = DeepSeekV4ParserToolAdapter def _dsv4_tokens( reasoning: str, tool_name: str, params: list[tuple[str, str, str]], ) -> list[tuple[int, str]]: """Build a token sequence: reasoning + + DSML tool block.""" tokens: list[tuple[int, str]] = [] tid = 100 for word in reasoning.split(" "): prefix = " " if tokens else "" tokens.append((tid, prefix + word)) tid += 1 tokens.append((_DSV4_FULL_VOCAB[DSML_THINK_END], DSML_THINK_END)) tokens.append((tid, "\n\n")) tid += 1 tokens.append((_DSV4_FULL_VOCAB[DSML_TOOL_START], DSML_TOOL_START)) tokens.append((tid, "\n")) tid += 1 invoke_prefix_text = f"{DSML_INVOKE_PREFIX}{tool_name}{DSML_INVOKE_NAME_END}" tokens.append((tid, invoke_prefix_text)) tid += 1 tokens.append((tid, "\n")) tid += 1 for name, is_str, value in params: param_text = _param(name, is_str, value) tokens.append((tid, param_text)) tid += 1 tokens.append((tid, "\n")) tid += 1 tokens.append((tid, DSML_INVOKE_END)) tid += 1 tokens.append((tid, "\n")) tid += 1 tokens.append((_DSV4_FULL_VOCAB[DSML_TOOL_END], DSML_TOOL_END)) return tokens class TestDelegatingParserLargeDelta: """Regression: tool calls lost when + DSML arrive in same delta. The DelegatingParser used by the serving layer splits reasoning and tool parsing across two separate engine instances. When and the entire DSML tool block arrive in a single large streaming delta, the content transfer from reasoning adapter to tool adapter must preserve the tool call text. """ @pytest.fixture def dsv4_tokens(self): return _dsv4_tokens( reasoning="The user wants the current weather in Berlin.", tool_name="get_weather", params=[ ("location", "true", "Berlin"), ("units", "true", "celsius"), ], ) @pytest.fixture def dsv4_tokenizer(self, dsv4_tokens): return MockTokenizer( vocab=dict(_DSV4_FULL_VOCAB), tokens=dsv4_tokens, ) @pytest.mark.parametrize( "chunk_size", [1, 2, 3, 5, None], ids=lambda c: f"chunk={c}", ) def test_tool_calls_extracted_at_all_chunk_sizes( self, dsv4_tokenizer, dsv4_tokens, chunk_size ): parser = _DeepSeekV4Delegating( dsv4_tokenizer, chat_template_kwargs={"thinking": True}, ) deltas = replay_streaming( parser, dsv4_tokens, chunk_size=chunk_size, finished_on_last=True, tools=DUMMY_TOOLS, ) output = collect_output(deltas) assert "The user wants" in output.reasoning assert len(output.tool_calls) == 1, ( f"Expected 1 tool call but got {len(output.tool_calls)}; " f"reasoning={output.reasoning!r}, content={output.content!r}" ) assert output.tool_calls[0]["name"] == "get_weather" args = json.loads(output.tool_calls[0]["arguments"]) assert args == {"location": "Berlin", "units": "celsius"} def test_eos_drop_token_does_not_swallow_tool_calls(self): """Tool calls must survive when an EOS DROP token's ID is in delta_token_ids but its text is absent from delta_text. At large stream_interval the EOS token ID arrives in the same delta as + tool calls but the detokenizer strips the EOS text. The scanner's _rebuild_from_anchors defers all text after when it can't find the EOS anchor text. The reasoning adapter's finish_streaming must flush deferred text as content (with skip_tool_parsing), not as tool calls. """ eos_text = "<｜end▁of▁sentence｜>" eos_id = 128801 vocab = { DSML_THINK_START: 128821, DSML_THINK_END: 128822, eos_text: eos_id, } reasoning = "The user wants weather." tool_block = ( "\n\n" + DSML_TOOL_START + "\n" + DSML_INVOKE_PREFIX + "get_weather" + DSML_INVOKE_NAME_END + "\n" + _param("location", "true", "Berlin") + "\n" + DSML_INVOKE_END + "\n" + DSML_TOOL_END ) # delta_text does NOT include EOS text (detokenizer strips it) full_text = reasoning + DSML_THINK_END + tool_block # Build token list: word-split reasoning, then special tokens, # then word-split tool block content, then EOS. # EOS ID is present but its text is NOT in delta_text. tokens: list[tuple[int, str]] = [] tid = 100 for word in reasoning.split(" "): pfx = " " if tokens else "" tokens.append((tid, pfx + word)) tid += 1 tokens.append((128822, DSML_THINK_END)) for ch in tool_block: tokens.append((tid, ch)) tid += 1 tokens.append((eos_id, eos_text)) all_ids = [t[0] for t in tokens] tokenizer = MockTokenizer(vocab=vocab, tokens=tokens) request = _test_request(tools=DUMMY_TOOLS) # All-in-one delta: EOS ID in token_ids but text NOT in # delta_text (detokenizer strips EOS). This is the scenario # at large stream_interval. parser = _DeepSeekV4Delegating( tokenizer, chat_template_kwargs={"thinking": True}, ) deltas = [ parser.parse_delta( full_text, all_ids, request, prompt_token_ids=[], finished=True, ) ] output = collect_output(deltas) assert "The user wants" in output.reasoning assert len(output.tool_calls) == 1, ( f"Expected 1 tool call but got {len(output.tool_calls)}; " f"reasoning={output.reasoning!r}, content={output.content!r}" ) assert output.tool_calls[0]["name"] == "get_weather" args = json.loads(output.tool_calls[0]["arguments"]) assert args == {"location": "Berlin"}