diff --git a/docs/source/features/disagg-serving.md b/docs/source/features/disagg-serving.md index 23752b6d26e8..ba3806c2cd06 100644 --- a/docs/source/features/disagg-serving.md +++ b/docs/source/features/disagg-serving.md @@ -233,9 +233,10 @@ The disaggregated server also serves `/v1/responses`, with both `context_first` - A request with `previous_response_id` is rejected with a 400, because a stored response lives in one worker's memory and a later request may reach another worker. Send the earlier turns in `input` instead. - `store` is ignored: nothing is persisted, and the server logs a warning once when a client asks for it. -One limitation applies to `/v1/responses` on every server, disaggregated or not: +Two limitations apply to `/v1/responses` on every server, disaggregated or not: -- `parallel_tool_calls: false`, and a `tool_choice` other than `auto` or `none` on non-Harmony models, are accepted but not enforced. The server logs a warning once. +- A `tool_choice` other than `auto` or `none` on non-Harmony models is accepted but not enforced. The server logs a warning once. +- `parallel_tool_calls: false` ends generation after the first tool call only when the tool parser declares the text that ends one call, as the DeepSeek DSML parsers (`deepseek_v32`, `deepseek_v4`, `deepseek_v41`) do. With other tool parsers, and on Harmony models, it is accepted but not enforced, and the server logs a warning once. #### Launching disaggregated servers on SLURM clusters diff --git a/tensorrt_llm/serve/anthropic_adapter.py b/tensorrt_llm/serve/anthropic_adapter.py index 85785555dea0..5abc03d674f2 100644 --- a/tensorrt_llm/serve/anthropic_adapter.py +++ b/tensorrt_llm/serve/anthropic_adapter.py @@ -364,8 +364,6 @@ def _convert_tool_choice( if choice is None: # check_tool_choice validator defaults to "auto" when tools are set. return "auto" if tools else None - if choice.disable_parallel_tool_use and choice.type != "none": - raise AnthropicRequestError("tool_choice.disable_parallel_tool_use=true is not supported") if choice.type == "none": return "none" if tools is None: @@ -475,6 +473,11 @@ def convert_anthropic_request(request: AnthropicMessagesRequest) -> ChatCompleti chat_request["tools"] = [t.model_dump() for t in tools] if tool_choice is not None: chat_request["tool_choice"] = tool_choice + choice = request.tool_choice + if choice is not None and choice.disable_parallel_tool_use and choice.type != "none": + # At most one tool_use block: the chat pipeline ends generation after + # the first tool call where the tool-call format allows it. + chat_request["parallel_tool_calls"] = False if request.temperature is not None: chat_request["temperature"] = request.temperature if request.top_p is not None: diff --git a/tensorrt_llm/serve/openai_protocol.py b/tensorrt_llm/serve/openai_protocol.py index 3961aec9e62d..642a6a343b64 100644 --- a/tensorrt_llm/serve/openai_protocol.py +++ b/tensorrt_llm/serve/openai_protocol.py @@ -1106,9 +1106,11 @@ class ChatCompletionRequest(OpenAIBaseModel): tools: Optional[List[ChatCompletionToolsParam]] = None tool_choice: Optional[Union[Literal["none", "auto", "required"], ChatCompletionNamedToolChoiceParam]] = "none" - # Standard OpenAI field, accepted for compatibility. `false` is not - # enforced: the engine does not restrict how many tool calls the model - # emits per turn, so parallel emission remains model behavior either way. + # Standard OpenAI field. `false` ends generation after the first tool + # call when the tool-call format has a text that ends one call + # (BaseToolParser.single_call_stops) and the request has no stops of its + # own that are left out of the output; otherwise it is accepted but not + # enforced. parallel_tool_calls: Optional[bool] = None user: Optional[str] = None reasoning_effort: Optional[ReasoningEffort | Literal[ diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 432e0d8500aa..b26f83643d84 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -297,6 +297,47 @@ def _configure_parser_special_token_decoding( sampling_params.spaces_between_special_tokens = False +def _single_tool_call_stops(tool_parser_name: Optional[str]) -> tuple[str, ...]: + """The texts that end one tool call in this format; empty if it has none. + + See ``BaseToolParser.single_call_stops``. + """ + if not tool_parser_name: + return () + tool_parser_cls = ToolParserFactory.parsers.get(tool_parser_name.lower()) + if tool_parser_cls is None: + return () + return tool_parser_cls().single_call_stops + + +def _stop_after_first_tool_call(sampling_params: SamplingParams, + stops: tuple[str, ...]) -> bool: + """End generation on the first of ``stops`` and keep it in the output. + + This honours a request for at most one tool call per turn without + rewriting what the model wrote: ``stops`` end one call + (``BaseToolParser.single_call_stops``), and the tool parser needs the + matched text to read that call as complete. ``include_stop_str_in_output`` + covers every stop of a request, so nothing changes when the request's + own stop strings or ids are to be left out of the output. + + Returns: + Whether the stops were added. + """ + client_stops = sampling_params.stop + if isinstance(client_stops, str): + client_stops = [client_stops] + client_stops = list(client_stops or []) + if ((client_stops or sampling_params.stop_token_ids) + and not sampling_params.include_stop_str_in_output): + return False + sampling_params.stop = client_stops + [ + stop for stop in stops if stop not in client_stops + ] + sampling_params.include_stop_str_in_output = True + return True + + def _record_generator_termination(generator) -> None: """Classify a fatal generator error without exposing exception details.""" component = "llm" @@ -2298,7 +2339,33 @@ async def chat_stream_generator( "server to be started with --tool_parser."), err_type="BadRequestError", status_code=HTTPStatus.BAD_REQUEST) + # At most one tool call per turn (also Anthropic's + # disable_parallel_tool_use): generation ends after the first call + # where the tool-call format allows it; elsewhere this is accepted + # and said once. + single_call_stops: tuple[str, ...] = () + if (request.parallel_tool_calls is False and all_tools + and request.tool_choice != "none"): + stops = _single_tool_call_stops(self.tool_parser) + client_stops = sampling_params.stop + if isinstance(client_stops, str): + client_stops = [client_stops] + client_stops = set(client_stops or []) + if stops and _stop_after_first_tool_call( + sampling_params, stops): + # A stop the request lists itself keeps its stop reason. + single_call_stops = tuple(stop for stop in stops + if stop not in client_stops) + else: + logger.warning_once( + "'parallel_tool_calls=false' is accepted but not " + "enforced when the tool-call format has no text that " + "ends a single call, or when the request's own stop " + "strings are left out of the output; the model may " + "still emit several tool calls in one turn.", + key="chat_parallel_tool_calls_unenforced") postproc_args = ChatPostprocArgs.from_request(request) + postproc_args.single_call_stops = single_call_stops self._apply_spec_decode_stats_opt_in(postproc_args) # Some models' prompt-token accounting excludes trailing rendered # tokens (kimi_k3: the generation channel opener); the model still @@ -3336,9 +3403,12 @@ async def openai_responses(self, request: ResponsesRequest, raw_request: Request) -> Response: """Serve one /v1/responses request, streaming or not. - ``parallel_tool_calls=false`` and, on the non-Harmony path, - ``tool_choice`` values other than "auto" and "none" are accepted but - not enforced (logged once). ``tool_choice="none"`` skips tool parsing. + ``parallel_tool_calls=false`` ends generation after the first tool + call where the tool-call format has a text that ends one call + (``BaseToolParser.single_call_stops``); elsewhere, and on the Harmony + path, it is accepted but not enforced (logged once). So are, on the + non-Harmony path, ``tool_choice`` values other than "auto" and + "none". ``tool_choice="none"`` skips tool parsing. """ async def create_response( @@ -3425,13 +3495,23 @@ async def create_streaming_generator(promise: RequestOutput, # worker would pull KV cache from. self._validate_internal_disagg_request(request, raw_request) + # parallel_tool_calls=false ends generation after the first tool + # call where the tool-call format allows it (applied below, once + # the sampling params exist); elsewhere it is accepted and said + # once. + single_call_stops: tuple[str, ...] = () if ("parallel_tool_calls" in request.model_fields_set and request.parallel_tool_calls is False): - logger.warning_once( - "Responses API: 'parallel_tool_calls=false' is accepted " - "but not enforced; the model may still emit several tool " - "calls in one turn.", - key="responses_parallel_tool_calls_unenforced") + if not self.use_harmony: + single_call_stops = _single_tool_call_stops( + self.tool_parser) + if not single_call_stops: + logger.warning_once( + "Responses API: 'parallel_tool_calls=false' is " + "accepted but not enforced for this tool-call format; " + "the model may still emit several tool calls in one " + "turn.", + key="responses_parallel_tool_calls_unenforced") if not self.use_harmony and request.tool_choice not in ("auto", "none"): logger.warning_once( @@ -3486,6 +3566,11 @@ async def create_streaming_generator(promise: RequestOutput, reasoning_parser_name=self.generator.args.reasoning_parser, tool_parser_name=self.tool_parser, has_tools=bool(request.tools)) + if (single_call_stops and request.tools + and request.tool_choice != "none"): + # A Responses request sets no stops of its own, so these + # always apply. + _stop_after_first_tool_call(sampling_params, single_call_stops) streaming_processor = None if request.stream: diff --git a/tensorrt_llm/serve/postprocess_handlers.py b/tensorrt_llm/serve/postprocess_handlers.py index b882a2b6e7f3..26e709a00d00 100644 --- a/tensorrt_llm/serve/postprocess_handlers.py +++ b/tensorrt_llm/serve/postprocess_handlers.py @@ -104,6 +104,10 @@ class ChatPostprocArgs(PostprocArgs): reasoning_parser: Optional[str] = None tool_parser: Optional[str] = None tool_parser_grammar_constrained: bool = False + # Stops the server added to end generation after the first tool call + # (parallel_tool_calls=false). The client did not ask for them, so a + # generation they end reports no stop_reason. + single_call_stops: tuple[str, ...] = () reasoning_parser_dict: dict[int, BaseReasoningParser] = field( default_factory=dict) tool_parser_dict: dict[int, BaseToolParser] = field(default_factory=dict) @@ -398,6 +402,14 @@ def forced_tool_arguments_end(text: str) -> Optional[int]: return end +def _client_stop_reason(stop_reason: Optional[Union[int, str]], + args: ChatPostprocArgs) -> Optional[Union[int, str]]: + """The stop reason to report: none for a stop the server added itself.""" + if stop_reason in args.single_call_stops: + return None + return stop_reason + + @nvtx_range_debug("chat_stream_post_processor") def chat_stream_post_processor(rsp: GenerationResultBase, args: ChatPostprocArgs) -> List[str]: @@ -595,7 +607,7 @@ def yield_first_chat(num_tokens: int, speculative_decoding=_build_spec_decode_stats( getattr(output, '_spec_dec_counters', None), args, output.finish_reason), - stop_reason=output.stop_reason, + stop_reason=_client_stop_reason(output.stop_reason, args), ) if args.return_logprobs: logprobs = output.logprobs_diff @@ -608,7 +620,7 @@ def yield_first_chat(num_tokens: int, choice.finish_reason = "tool_calls" else: choice.finish_reason = output.finish_reason - choice.stop_reason = output.stop_reason + choice.stop_reason = _client_stop_reason(output.stop_reason, args) finish_reason_sent[i] = True chunk = ChatCompletionStreamResponse(choices=[choice], model=args.model, @@ -755,7 +767,7 @@ def chat_response_post_processor( choice = ChatCompletionResponseChoice( index=output.index, message=message, - stop_reason=output.stop_reason, + stop_reason=_client_stop_reason(output.stop_reason, args), disaggregated_params=disaggregated_params, avg_decoded_tokens_per_iter=getattr(rsp, 'avg_decoded_tokens_per_iter', diff --git a/tensorrt_llm/serve/tool_parser/base_tool_parser.py b/tensorrt_llm/serve/tool_parser/base_tool_parser.py index daf4f0147cf6..e7bb3a2040f0 100644 --- a/tensorrt_llm/serve/tool_parser/base_tool_parser.py +++ b/tensorrt_llm/serve/tool_parser/base_tool_parser.py @@ -422,6 +422,20 @@ def has_tool_call(self, text: str) -> bool: """ raise NotImplementedError() + @property + def single_call_stops(self) -> tuple[str, ...]: + """Texts that end one complete tool call, to stop generation on. + + A request for at most one tool call per turn + (``parallel_tool_calls=false``) stops generating on the first of these + texts and keeps it in the output, so the parser reads that call as + complete and the output stays a prefix of what the model generated. + Stops match on token ids, so a text is listed in each form the model + tokenizes it in. Empty when the format has no such text; requests + then get every call the model writes. + """ + return () + def supports_structural_tag(self) -> bool: """Return True if this detector supports structural tag format.""" return True diff --git a/tensorrt_llm/serve/tool_parser/deepseekv32_parser.py b/tensorrt_llm/serve/tool_parser/deepseekv32_parser.py index d098f79bb91d..1eb0020260a0 100644 --- a/tensorrt_llm/serve/tool_parser/deepseekv32_parser.py +++ b/tensorrt_llm/serve/tool_parser/deepseekv32_parser.py @@ -95,6 +95,18 @@ def has_tool_call(self, text: str) -> bool: """Check if the text contains a deepseek v32 format tool call.""" return self.bot_token in text + @property + def single_call_stops(self) -> tuple[str, ...]: + # Each call is its own invoke block, so its closing tag ends exactly one + # call. That leaves the calls block open; the whole-text and streaming + # parses both deliver the complete invokes of an open block. + # Stops match on token ids, and the encoders follow the tag with a + # newline that tokenizes together with ">", so the tag is also listed + # with that newline. That form comes first: the decoded text is cut + # after the first listed stop it contains, and the newline belongs to + # the stop's last token. + return (self.invoke_end_token + "\n", self.invoke_end_token) + def _parse_parameters_from_xml(self, invoke_content: str) -> dict: """Parse parameters from either XML-like format or JSON format to dict. @@ -148,16 +160,15 @@ def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult calls = [] try: - # Extract content between function_calls tags - function_calls_match = re.search( - re.escape(self.bot_token) + r"(.*?)" + re.escape(self.eot_token), - text, - re.DOTALL, - ) - if not function_calls_match: - return StreamingParseResult(normal_text=normal_text, calls=[]) - - function_calls_content = function_calls_match.group(1) + # The calls block runs to its closing tag, or to the end of the text + # when generation stopped inside it (single_call_stops, a token + # limit). Only its complete invokes are calls; an unfinished last + # invoke is dropped, as the streaming parse holds it back. + block_start = idx + len(self.bot_token) + block_end = text.find(self.eot_token, block_start) + if block_end == -1: + block_end = len(text) + function_calls_content = text[block_start:block_end] # Find all invoke blocks invoke_pattern = self.invoke_begin_regex + r"(.*?)" + re.escape(self.invoke_end_token) diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml index 517e12501f5d..7212a27b03a3 100644 --- a/tests/integration/test_lists/test-db/l0_cpu.yml +++ b/tests/integration/test_lists/test-db/l0_cpu.yml @@ -268,6 +268,7 @@ l0_cpu: - unittest/llmapi/apps/test_responses_usage.py - unittest/llmapi/apps/test_serve_lazy_import.py - unittest/llmapi/apps/test_serving_extensions.py + - unittest/llmapi/apps/test_single_tool_call_stop.py - unittest/llmapi/apps/test_tool_parsers.py - unittest/llmapi/apps/test_web_search_shared.py - unittest/llmapi/test_bench_async.py diff --git a/tests/unittest/llmapi/apps/test_anthropic_adapter.py b/tests/unittest/llmapi/apps/test_anthropic_adapter.py index 9449c721e1c1..164f5e0ffb57 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_adapter.py +++ b/tests/unittest/llmapi/apps/test_anthropic_adapter.py @@ -630,14 +630,25 @@ def test_tool_choice_any_rejected_instead_of_downgraded(): ) -def test_disable_parallel_tool_use_rejected(): - with pytest.raises(AnthropicRequestError, match="disable_parallel_tool_use"): - convert_anthropic_request( - make_request( - tools=[{"name": "f", "input_schema": {"type": "object"}}], - tool_choice={"type": "auto", "disable_parallel_tool_use": True}, - ) +@pytest.mark.parametrize( + ("tool_choice", "parallel_tool_calls"), + [ + ({"type": "auto", "disable_parallel_tool_use": True}, False), + ({"type": "auto", "disable_parallel_tool_use": False}, None), + ({"type": "auto"}, None), + # "none" allows no call at all, so there is nothing to limit. + ({"type": "none", "disable_parallel_tool_use": True}, None), + ], +) +def test_disable_parallel_tool_use_asks_for_one_call(tool_choice, parallel_tool_calls): + """The chat pipeline then stops generation after the first tool call.""" + chat = convert_anthropic_request( + make_request( + tools=[{"name": "f", "input_schema": {"type": "object"}}], + tool_choice=tool_choice, ) + ) + assert chat.parallel_tool_calls is parallel_tool_calls @pytest.mark.parametrize("choice", [{"type": "any"}, {"type": "tool", "name": "f"}]) diff --git a/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py b/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py index e7e2ceca84cb..dd3f0a1723ef 100644 --- a/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py +++ b/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py @@ -293,6 +293,53 @@ def test_an_unterminated_call_is_text_in_both_views(): assert not mock_logger.warning.called +# What a DeepSeek-V4.1 generation ended by parallel_tool_calls=false leaves: +# the first call, closed by the kept stop, inside a calls block never closed. +_V41_STOPPED_CHUNKS = [ + "Running it.", + "\n\n<｜DSML｜ calls>\n", + '<｜DSML｜ invoke name="exec">\n', + '<｜DSML｜ parameter name="input" string="true">ls\n', + "\n", +] + + +def test_a_dsml_generation_stopped_after_its_first_call_reports_exactly_that_call(): + """The stream and its snapshot both report the call, and nothing after it.""" + processor = _processor(tools=[_tool()], reasoning_parser=None, tool_parser="deepseek_v41") + frames, result = _stream(processor, _V41_STOPPED_CHUNKS) + + streamed = _done_items(frames, "function_call") + assert [(c["name"], json.loads(c["arguments"])) for c in streamed] == [ + ("exec", {"input": "ls"}) + ] + final = _final_output(processor, result) + assert [item for item in final if item["type"] == "function_call"] == streamed + assert _message_texts(final) == _message_texts(_done_items(frames, "message")) + assert "".join(_message_texts(final)).strip() == "Running it." + + +def test_a_non_streamed_dsml_generation_stopped_after_its_first_call_reports_that_call(): + text = "".join(_V41_STOPPED_CHUNKS) + request = ResponsesRequest(model="test-model", input="hi", tools=[_tool()]) + response = create_response_non_store( + generation_result=_generation_chunk(text, text, True), + request=request, + sampling_params=request.to_sampling_params(), + model_name="test-model", + use_harmony=False, + tool_parser="deepseek_v41", + num_prompt_tokens=2, + ).model_dump() + + assert [(item["type"], item.get("name")) for item in response["output"]] == [ + ("message", None), + ("function_call", "exec"), + ] + assert _message_texts(response["output"]) == ["Running it."] + assert response["status"] == "completed" + + def test_tool_choice_none_keeps_the_markup_as_text_in_both_views(): processor = _processor( tools=[_tool()], reasoning_parser="glm47", tool_parser="glm47", tool_choice="none" diff --git a/tests/unittest/llmapi/apps/test_single_tool_call_stop.py b/tests/unittest/llmapi/apps/test_single_tool_call_stop.py new file mode 100644 index 000000000000..6ef1f1585463 --- /dev/null +++ b/tests/unittest/llmapi/apps/test_single_tool_call_stop.py @@ -0,0 +1,531 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""A request for at most one tool call per turn stops generation after the first call. + +``parallel_tool_calls=false`` (Chat Completions, Responses) and Anthropic's +``disable_parallel_tool_use`` are honoured with a stop on the text that ends one +call (``BaseToolParser.single_call_stops``), kept in the output so the tool +parser still reads that call. These tests cover where the stop is configured, +what a client sees of it, and that it matches the tokens the model writes. +""" + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from openai.types.responses.tool import FunctionTool +from utils.llm_data import llm_models_root + +from tensorrt_llm.sampling_params import SamplingParams +from tensorrt_llm.serve import openai_server as openai_server_module +from tensorrt_llm.serve.anthropic_protocol import AnthropicMessagesRequest +from tensorrt_llm.serve.openai_protocol import ( + ChatCompletionRequest, + ChatCompletionResponse, + ChatCompletionResponseChoice, + ChatCompletionToolsParam, + ChatMessage, + FunctionDefinition, + ResponsesRequest, + UsageInfo, +) +from tensorrt_llm.serve.openai_server import ( + OpenAIServer, + _single_tool_call_stops, + _stop_after_first_tool_call, +) +from tensorrt_llm.serve.postprocess_handlers import ( + ChatPostprocArgs, + chat_response_post_processor, + chat_stream_post_processor, +) +from tensorrt_llm.serve.render import chat as render_chat_module +from tensorrt_llm.serve.responses_utils import ConversationHistoryStore + +pytestmark = pytest.mark.cpu_only + +MODEL = "test-model" +V41_STOPS = ("\n", "") +WEATHER_TOOL = { + "type": "function", + "function": { + "name": "get_weather", + "parameters": {"type": "object", "properties": {"location": {"type": "string"}}}, + }, +} +WEATHER_INVOKE = ( + '<｜DSML｜ invoke name="get_weather">\n' + '<｜DSML｜ parameter name="location" string="true">NYC\n' + "" +) + + +# --------------------------------------------------------------------------- +# The stop and the request's own stops +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "tool_parser, stops", + [ + ("deepseek_v41", V41_STOPS), + ("DeepSeek_V41", V41_STOPS), + ("deepseek_v4", ("\n", "")), + ("deepseek_v32", ("\n", "")), + ("qwen3", ()), + ("not_a_parser", ()), + (None, ()), + ], +) +def test_the_stops_come_from_the_tool_parser(tool_parser, stops): + assert _single_tool_call_stops(tool_parser) == stops + + +def test_the_stops_are_added_and_kept_in_the_output(): + sampling_params = SamplingParams() + assert _stop_after_first_tool_call(sampling_params, V41_STOPS) + assert sampling_params.stop == list(V41_STOPS) + assert sampling_params.include_stop_str_in_output is True + + +@pytest.mark.parametrize( + "fields", + [{"stop": "END"}, {"stop": ["END"]}, {"stop_token_ids": [7]}], + ids=["stop_string", "stop_list", "stop_id"], +) +def test_nothing_changes_when_the_request_drops_its_own_stops_from_the_output(fields): + """Keeping stop text applies to every stop, so the client's would show.""" + sampling_params = SamplingParams(**fields) + assert not _stop_after_first_tool_call(sampling_params, V41_STOPS) + assert sampling_params.stop == fields.get("stop") + assert sampling_params.include_stop_str_in_output is False + + +def test_the_stops_follow_the_request_stops_it_keeps_in_the_output(): + sampling_params = SamplingParams(stop=["END", V41_STOPS[0]], include_stop_str_in_output=True) + assert _stop_after_first_tool_call(sampling_params, V41_STOPS) + assert sampling_params.stop == ["END", *V41_STOPS] + + +@pytest.mark.parametrize( + "model_name, tool_parser", + [ + ("DeepSeek-V3.2-hf", "deepseek_v32"), + ("DeepSeek-V4-Flash", "deepseek_v4"), + ("DeepSeek-V4-Pro", "deepseek_v4"), + ("DeepSeek-V4.1-Flash", "deepseek_v41"), + ], +) +def test_the_stop_matches_the_tokens_the_model_writes(model_name, tool_parser): + """Stops match token ids, and '>' plus the newline after it are one token. + + So in each encoder's layout of DeepSeek tool calls the stop with the + newline matches once per call, and the bare closing tag never does. + """ + root = llm_models_root() + tokenizer_file = None if root is None else root / model_name / "tokenizer.json" + if tokenizer_file is None or not tokenizer_file.is_file(): + pytest.skip(f"{model_name} is not under LLM_MODELS_ROOT") + from transformers import PreTrainedTokenizerFast + + from tensorrt_llm.tokenizer.deepseek_v4 import DeepseekV4Tokenizer + from tensorrt_llm.tokenizer.deepseek_v32 import DeepseekV32Tokenizer + from tensorrt_llm.tokenizer.deepseek_v41 import DeepseekV41Tokenizer + + tokenizer_class = { + "deepseek_v32": DeepseekV32Tokenizer, + "deepseek_v4": DeepseekV4Tokenizer, + "deepseek_v41": DeepseekV41Tokenizer, + }[tool_parser] + + tokenizer = PreTrainedTokenizerFast(tokenizer_file=str(tokenizer_file)) + # The ids the engine matches: SamplingParams tokenizes its stop strings. + stops = _single_tool_call_stops(tool_parser) + sampling_params = SamplingParams(stop=list(stops)) + sampling_params._setup(tokenizer, None, None) + with_newline, bare = sampling_params._get_stop_words() + + calls = [ + { + "id": f"call_{index}", + "type": "function", + "function": {"name": name, "arguments": json.dumps(arguments)}, + } + for index, (name, arguments) in enumerate([("exec", {"cmd": "ls"}), ("wait", {"s": 2})]) + ] + prompt = tokenizer_class(tokenizer).apply_chat_template( + [ + {"role": "user", "content": "List the files, then wait."}, + {"role": "assistant", "content": "", "tool_calls": calls}, + ], + thinking=False, + tokenize=False, + ) + ids = tokenizer.encode(prompt, add_special_tokens=False) + + def matches(word): + return sum(ids[i : i + len(word)] == word for i in range(len(ids))) + + assert prompt.count(stops[0]) == 2 + assert matches(with_newline) == 2 + assert matches(bare) == 0 + + +# --------------------------------------------------------------------------- +# Chat Completions and Anthropic Messages +# --------------------------------------------------------------------------- + + +class _ModelConfig: + model_type = "llama" + vocab_size = 1024 + + +def _chat_server(monkeypatch, tool_parser): + """A server whose real ``openai_chat`` records what reaches the engine.""" + + async def render(**kwargs): + return "rendered prompt" + + monkeypatch.setattr(render_chat_module, "async_apply_chat_template", render) + seen = {} + + def generate_async(*, inputs, **kwargs): + seen["sampling_params"] = kwargs["sampling_params"] + return SimpleNamespace(prompt_token_ids=[1, 2, 3], finished=True) + + async def create_chat_response(promise, postproc_params, raw_request, disaggregated_params): + seen["postproc_args"] = postproc_params.postproc_args + return ChatCompletionResponse( + model=MODEL, + choices=[ + ChatCompletionResponseChoice( + index=0, + message=ChatMessage(role="assistant", content="ok"), + finish_reason="stop", + ) + ], + usage=UsageInfo(prompt_tokens=3, completion_tokens=1, total_tokens=4), + ) + + server = object.__new__(OpenAIServer) + server.model = MODEL + server.allow_request_chat_template = False + server.model_config = _ModelConfig() + server.processor = None + server.tokenizer = SimpleNamespace(tokenizer=SimpleNamespace(vocab_size=1024)) + server.chat_template = None + server.tool_parser = tool_parser + server.tool_call_id_type = "random" + server.multimodal_server_config = None + server._input_proc_executor = None + server._is_visual_gen = False + server.generator = SimpleNamespace( + args=SimpleNamespace( + gather_generation_logits=False, + reasoning_parser=None, + backend="pytorch", + guided_decoding_backend=None, + num_postprocess_workers=0, + ), + generate_async=generate_async, + ) + server.await_disconnected = AsyncMock() + server._create_chat_response = create_chat_response + return server, seen + + +def _raw_request(): + return SimpleNamespace(state=SimpleNamespace(), headers={}) + + +@pytest.fixture +def server_logger(monkeypatch): + log = Mock() + monkeypatch.setattr(openai_server_module, "logger", log) + return log + + +def _warned(log, key): + return key in [call.kwargs.get("key") for call in log.warning_once.call_args_list] + + +@pytest.mark.parametrize( + "fields, tool_parser, stop, enforced", + [ + ({"parallel_tool_calls": False}, "deepseek_v41", list(V41_STOPS), True), + ({}, "deepseek_v41", [], False), + ({"parallel_tool_calls": True}, "deepseek_v41", [], False), + ({"parallel_tool_calls": False, "tool_choice": "none"}, "deepseek_v41", [], False), + ({"parallel_tool_calls": False, "tools": None}, "deepseek_v41", [], False), + ({"parallel_tool_calls": False}, "qwen3", [], False), + ({"parallel_tool_calls": False, "stop": ["END"]}, "deepseek_v41", ["END"], False), + ( + {"parallel_tool_calls": False, "stop": ["END"], "include_stop_str_in_output": True}, + "deepseek_v41", + ["END", *V41_STOPS], + True, + ), + ], + ids=[ + "asked", + "omitted", + "parallel_allowed", + "tool_choice_none", + "no_tools", + "format_without_stop", + "own_stops_dropped", + "own_stops_kept", + ], +) +def test_chat_configures_the_stop_only_when_asked_and_possible( + monkeypatch, server_logger, fields, tool_parser, stop, enforced +): + server, seen = _chat_server(monkeypatch, tool_parser) + request = ChatCompletionRequest( + **{ + "model": MODEL, + "messages": [{"role": "user", "content": "weather?"}], + "tools": [WEATHER_TOOL], + **fields, + } + ) + response = asyncio.run(server.openai_chat(request, _raw_request())) + + assert response.status_code == 200, response.body + sampling_params = seen["sampling_params"] + assert sampling_params.stop == stop + assert sampling_params.include_stop_str_in_output is ( + "include_stop_str_in_output" in fields or enforced + ) + assert seen["postproc_args"].single_call_stops == (V41_STOPS if enforced else ()) + asked = fields.get("parallel_tool_calls") is False and fields.get("tools", [WEATHER_TOOL]) + asked = asked and fields.get("tool_choice") != "none" + assert _warned(server_logger, "chat_parallel_tool_calls_unenforced") is bool( + asked and not enforced + ) + + +def test_a_stop_the_request_lists_itself_keeps_its_stop_reason(monkeypatch, server_logger): + """Only the stops the server adds are hidden from the reply's stop reason.""" + server, seen = _chat_server(monkeypatch, "deepseek_v41") + request = ChatCompletionRequest( + model=MODEL, + messages=[{"role": "user", "content": "weather?"}], + tools=[WEATHER_TOOL], + parallel_tool_calls=False, + stop=[V41_STOPS[0]], + include_stop_str_in_output=True, + ) + response = asyncio.run(server.openai_chat(request, _raw_request())) + + assert response.status_code == 200, response.body + assert seen["sampling_params"].stop == list(V41_STOPS) + assert seen["postproc_args"].single_call_stops == V41_STOPS[1:] + + +@pytest.mark.parametrize( + "tool_parser, stop_sequences, enforced", + [ + ("deepseek_v41", [], True), + ("deepseek_v41", ["END"], False), + ("qwen3", [], False), + (None, [], False), + ], + ids=["supported", "own_stops_dropped", "format_without_stop", "no_tool_parser"], +) +def test_anthropic_disable_parallel_tool_use_configures_the_stop_when_possible( + monkeypatch, server_logger, tool_parser, stop_sequences, enforced +): + server, seen = _chat_server(monkeypatch, tool_parser) + request = AnthropicMessagesRequest( + model=MODEL, + max_tokens=64, + messages=[{"role": "user", "content": "weather?"}], + tools=[{"name": "get_weather", "input_schema": {"type": "object"}}], + tool_choice={"type": "auto", "disable_parallel_tool_use": True}, + stop_sequences=stop_sequences, + ) + response = asyncio.run(server.anthropic_messages(request, _raw_request())) + + assert response.status_code == 200, response.body + assert seen["sampling_params"].stop == (list(V41_STOPS) if enforced else stop_sequences) + assert seen["sampling_params"].include_stop_str_in_output is enforced + assert seen["postproc_args"].single_call_stops == (V41_STOPS if enforced else ()) + assert _warned(server_logger, "chat_parallel_tool_calls_unenforced") is not enforced + + +# --------------------------------------------------------------------------- +# What a chat client receives +# --------------------------------------------------------------------------- + + +def _chat_args(single_call_stops=V41_STOPS): + args = ChatPostprocArgs( + role="assistant", + model=MODEL, + tools=[ChatCompletionToolsParam(function=FunctionDefinition(**WEATHER_TOOL["function"]))], + tool_choice="auto", + single_call_stops=single_call_stops, + ) + args.tool_parser = "deepseek_v41" + args.num_prompt_tokens = 3 + return args + + +def _output(text, finish_reason, stop_reason): + return SimpleNamespace( + index=0, + text=text, + text_diff=text, + token_ids=[1, 2], + token_ids_diff=[1, 2], + length=2, + finish_reason=finish_reason, + stop_reason=stop_reason, + logprobs=None, + logprobs_diff=None, + disaggregated_params=None, + ) + + +# The output a generation stopped after its first call leaves: the calls block +# is open and the kept stop ends it. +STOPPED_REPLY = "Checking.\n\n<｜DSML｜ calls>\n" + WEATHER_INVOKE + "\n" + + +@pytest.mark.parametrize( + "stop_reason, reported", [(V41_STOPS[0], None), (V41_STOPS[1], None), ("END", "END")] +) +def test_a_chat_reply_reports_its_call_and_no_stop_the_server_added(stop_reason, reported): + rsp = SimpleNamespace( + outputs=[_output(STOPPED_REPLY, "stop", stop_reason)], prompt_token_ids=[1], cached_tokens=0 + ) + choice = chat_response_post_processor(rsp, _chat_args()).choices[0] + + assert choice.finish_reason == "tool_calls" + assert choice.stop_reason == reported + assert choice.message.content == "Checking." + calls = [(c.function.name, json.loads(c.function.arguments)) for c in choice.message.tool_calls] + assert calls == [("get_weather", {"location": "NYC"})] + + +def test_a_streamed_chat_reply_reports_its_call_and_no_stop_the_server_added(): + args = _chat_args() + head, tail = STOPPED_REPLY.split("<｜DSML｜ invoke", 1) + frames = [] + for diff, done in [(head, False), ("<｜DSML｜ invoke" + tail, True)]: + output = _output(diff, "stop" if done else None, V41_STOPS[0] if done else None) + rsp = SimpleNamespace(id=1, outputs=[output], cached_tokens=0, _done=done) + frames += chat_stream_post_processor(rsp, args) + + choices = [json.loads(frame[len("data: ") :])["choices"][0] for frame in frames] + deltas = [call for choice in choices for call in choice["delta"].get("tool_calls") or []] + assert [d["function"]["name"] for d in deltas if d["function"].get("name")] == ["get_weather"] + assert json.loads("".join(d["function"].get("arguments") or "" for d in deltas)) == { + "location": "NYC" + } + assert choices[-1]["finish_reason"] == "tool_calls" + assert all(choice.get("stop_reason") is None for choice in choices) + assert "｜DSML｜" not in "".join(choice["delta"].get("content") or "" for choice in choices) + + +# --------------------------------------------------------------------------- +# Responses +# --------------------------------------------------------------------------- + + +class _EngineReached(Exception): + pass + + +def _responses_server(monkeypatch, tool_parser): + """A server whose real ``openai_responses`` records what reaches the engine.""" + seen = {} + + async def preprocess(*, request, **kwargs): + return [1, 2, 3], request.to_sampling_params(default_sampling_params={"stop_token_ids": []}) + + def generate_async(**kwargs): + seen["sampling_params"] = kwargs["sampling_params"] + raise _EngineReached() + + monkeypatch.setattr(openai_server_module, "responses_api_request_preprocess", preprocess) + server = object.__new__(OpenAIServer) + server.model = MODEL + server.enable_store = False + server._is_visual_gen = False + server.use_harmony = False + server.tokenizer = None + server.model_config = None + server.processor = None + server.multimodal_server_config = None + server.chat_template = None + server.tool_parser = tool_parser + server.metrics_collector = None + server.conversation_store = ConversationHistoryStore() + server.generator = SimpleNamespace( + args=SimpleNamespace(reasoning_parser=None, num_postprocess_workers=0), + generate_async=generate_async, + ) + return server, seen + + +@pytest.mark.parametrize( + "fields, tool_parser, enforced, warned", + [ + ({"parallel_tool_calls": False}, "deepseek_v41", True, False), + ({}, "deepseek_v41", False, False), + ({"parallel_tool_calls": True}, "deepseek_v41", False, False), + ({"parallel_tool_calls": False, "tool_choice": "none"}, "deepseek_v41", False, False), + ({"parallel_tool_calls": False, "tools": []}, "deepseek_v41", False, False), + ({"parallel_tool_calls": False}, "qwen3", False, True), + ({"parallel_tool_calls": False}, None, False, True), + ], + ids=[ + "asked", + "omitted", + "parallel_allowed", + "tool_choice_none", + "no_tools", + "format_without_stop", + "no_tool_parser", + ], +) +def test_responses_configures_the_stop_only_when_asked_and_possible( + monkeypatch, server_logger, fields, tool_parser, enforced, warned +): + server, seen = _responses_server(monkeypatch, tool_parser) + tool = FunctionTool( + name="get_weather", + type="function", + strict=False, + parameters=WEATHER_TOOL["function"]["parameters"], + ) + request = ResponsesRequest(**{"model": MODEL, "input": "weather?", "tools": [tool], **fields}) + raw_request = SimpleNamespace( + state=SimpleNamespace(), + headers={}, + url=SimpleNamespace(path="/v1/responses"), + json=AsyncMock(return_value={}), + ) + asyncio.run(server.openai_responses(request, raw_request)) + + sampling_params = seen["sampling_params"] + assert sampling_params.stop == (list(V41_STOPS) if enforced else None) + assert sampling_params.include_stop_str_in_output is enforced + assert _warned(server_logger, "responses_parallel_tool_calls_unenforced") is warned diff --git a/tests/unittest/llmapi/apps/test_tool_parsers.py b/tests/unittest/llmapi/apps/test_tool_parsers.py index 691820959447..3a32dcbdb1b9 100644 --- a/tests/unittest/llmapi/apps/test_tool_parsers.py +++ b/tests/unittest/llmapi/apps/test_tool_parsers.py @@ -47,6 +47,7 @@ from tensorrt_llm.serve.tool_parser.qwen3_coder_parser import \ Qwen3CoderToolParser from tensorrt_llm.serve.tool_parser.qwen3_tool_parser import Qwen3ToolParser +from tensorrt_llm.serve.tool_parser.tool_parser_factory import ToolParserFactory from tensorrt_llm.tokenizer.deepseek_v32.encoding import encode_messages from tensorrt_llm.serve.tool_parser.gemma4_parser import ( # isort: skip @@ -3303,6 +3304,207 @@ def test_deepseek_streaming_prefix_is_delta_independent( assert names == ["get_weather"], f"{deltas!r} called {names!r}" +# ============================================================================ +# DeepSeek DSML: generation stopped after one call, calls block left open +# ============================================================================ + +# Each DSML generation: parser, calls tag, invoke tag, parameter tag. +_SINGLE_CALL_DSML_FORMATS = [ + (DeepSeekV32Parser, "function_calls", "invoke", "parameter"), + (DeepSeekV4Parser, "tool_calls", "invoke", "parameter"), + (DeepSeekV41Parser, " calls", " invoke", " parameter"), +] +_SINGLE_CALL_DSML_IDS = ["v32", "v4", "v41"] + +_WEATHER_CALL = ("get_weather", {"location": "NYC"}) +_SEARCH_CALL = ("search_web", {"query": "AI news"}) + + +def _encoder_invoke(invoke_tag: str, parameter_tag: str, name: str, + arguments: dict[str, str]) -> str: + """One call as the checkpoints' encoders write it.""" + parameters = "\n".join( + f'<｜DSML｜{parameter_tag} name="{key}" string="true">{value}' + f"" for key, value in arguments.items()) + return (f'<｜DSML｜{invoke_tag} name="{name}">\n{parameters}\n' + f"") + + +def _encoder_reply(calls_tag: str, invoke_tag: str, parameter_tag: str, + content: str, calls: list) -> str: + """A reply with tool calls as the encoders write it, block closed.""" + invokes = "\n".join( + _encoder_invoke(invoke_tag, parameter_tag, name, arguments) + for name, arguments in calls) + return (f"{content}\n\n<｜DSML｜{calls_tag}>\n{invokes}\n" + f"") + + +def _delta_splits(text: str) -> Iterator[list[str]]: + """Ways the detokenizer may cut ``text`` into deltas, and more. + + The DSML marker is one token, so it is never cut; other text is cut at + any character, which covers every token boundary. Yields the whole text, + one unit per delta, every cut into two deltas, and fixed-width deltas of + 2 to 7 units. + """ + marker = "｜DSML｜" + units, pos = [], 0 + while pos < len(text): + step = len(marker) if text.startswith(marker, pos) else 1 + units.append(text[pos:pos + step]) + pos += step + yield [text] + yield units + for cut in range(1, len(units)): + yield ["".join(units[:cut]), "".join(units[cut:])] + for width in range(2, 8): + yield [ + "".join(units[start:start + width]) + for start in range(0, len(units), width) + ] + + +def _whole_text_view(parser_cls, text, tools): + result = parser_cls().detect_and_parse(text, tools) + return result.normal_text.strip(), [(call.name, json.loads(call.parameters)) + for call in result.calls] + + +def _streamed_view(parser_cls, deltas, tools): + """What a streaming client receives, finish() included.""" + parser = parser_cls() + normal, names, arguments = [], {}, {} + results = [parser.parse_streaming_increment(d, tools) for d in deltas] + for result in results + [parser.finish(tools)]: + normal.append(result.normal_text) + for call in result.calls: + if call.name: + names[call.tool_index] = call.name + arguments[call.tool_index] = (arguments.get(call.tool_index, "") + + call.parameters) + return "".join(normal).strip(), [ + (names[index], json.loads(arguments[index])) for index in sorted(names) + ] + + +def _assert_both_views(parser_cls, text, tools, expected): + """The whole-text parse and every streaming split read ``expected``.""" + assert _whole_text_view(parser_cls, text, tools) == expected + for deltas in _delta_splits(text): + assert _streamed_view(parser_cls, deltas, tools) == expected, deltas + + +@pytest.mark.parametrize("parser_cls, calls_tag, invoke_tag, parameter_tag", + _SINGLE_CALL_DSML_FORMATS, + ids=_SINGLE_CALL_DSML_IDS) +def test_dsml_single_call_stops_are_the_invoke_close_tag( + parser_cls, calls_tag, invoke_tag, parameter_tag): + """The close tag with the newline the encoders write after it comes first. + + '>' and that newline are one token, and stops match token ids, so the bare + tag alone would never match the encoders' layout. + """ + close_tag = f"" + assert parser_cls().single_call_stops == (close_tag + "\n", close_tag) + + +@pytest.mark.parametrize("stop_index", [0, 1], ids=["with_newline", "bare"]) +@pytest.mark.parametrize("parser_cls, calls_tag, invoke_tag, parameter_tag", + _SINGLE_CALL_DSML_FORMATS, + ids=_SINGLE_CALL_DSML_IDS) +def test_dsml_generation_stopped_after_its_first_call_delivers_that_call( + sample_tools, parser_cls, calls_tag, invoke_tag, parameter_tag, + stop_index): + """The output ends on the kept stop, inside a calls block never closed.""" + reply = _encoder_reply(calls_tag, invoke_tag, parameter_tag, + "Checking both.", [_WEATHER_CALL, _SEARCH_CALL]) + stop = parser_cls().single_call_stops[stop_index] + stopped = reply[:reply.index(stop) + len(stop)] + + _assert_both_views(parser_cls, stopped, sample_tools, + ("Checking both.", [_WEATHER_CALL])) + + +@pytest.mark.parametrize( + "tail", + [ + "", + "\n", + # Cut inside the next invoke: header, parameter, close tags. + "\n<", + "\n<｜DSML｜{invoke}", + '\n<｜DSML｜{invoke} name="search', + '\n<｜DSML｜{invoke} name="search_web">\n<｜DSML｜{parameter} name="query" ' + 'string="true">AI', + '\n<｜DSML｜{invoke} name="search_web">\n<｜DSML｜{parameter} name="query" ' + 'string="true">AI news\n\n" + + _encoder_invoke(invoke_tag, parameter_tag, *_WEATHER_CALL) + + tail.format(invoke=invoke_tag, + parameter=parameter_tag, + calls_prefix=calls_tag[:-2])) + + _assert_both_views(parser_cls, text, sample_tools, + ("Checking.", [_WEATHER_CALL])) + + +@pytest.mark.parametrize("parser_cls, calls_tag, invoke_tag, parameter_tag", + _SINGLE_CALL_DSML_FORMATS, + ids=_SINGLE_CALL_DSML_IDS) +def test_dsml_open_block_delivers_every_complete_invoke(sample_tools, + parser_cls, calls_tag, + invoke_tag, + parameter_tag): + """The parser does not limit the calls; only the stop does.""" + reply = _encoder_reply(calls_tag, invoke_tag, parameter_tag, "Checking.", + [_WEATHER_CALL, _SEARCH_CALL]) + open_block = reply[:reply.rindex("\n\n" + invoke[:-1] + + _assert_both_views(parser_cls, text, sample_tools, ("Checking.", [])) + + +def test_only_the_dsml_parsers_declare_single_call_stops(): + """Other formats have no stop, so their requests keep every call.""" + declaring = { + name + for name, parser_cls in ToolParserFactory.parsers.items() + if parser_cls().single_call_stops + } + assert declaring == {"deepseek_v32", "deepseek_v4", "deepseek_v41"} + + # ============================================================================ # Glm4ToolParser Tests # ============================================================================